diff --git a/ddev/changelog.d/25331.added b/ddev/changelog.d/25331.added new file mode 100644 index 0000000000000..5c5c8437aa79c --- /dev/null +++ b/ddev/changelog.d/25331.added @@ -0,0 +1 @@ +Add Dispatcher metrics for GitHub requests, retries, throttled requests, rate-limit budgets, throttle waits, and polling intervals. diff --git a/ddev/src/ddev/cli/ci/dispatch_run.py b/ddev/src/ddev/cli/ci/dispatch_run.py index ab0e96f5aea90..45da73ef694c0 100644 --- a/ddev/src/ddev/cli/ci/dispatch_run.py +++ b/ddev/src/ddev/cli/ci/dispatch_run.py @@ -296,14 +296,29 @@ def resolve_pull_request_run( from ddev.utils.github_errors import GitHubAuthenticationError client_logger = None + observer = None + rate_limiter = None if monitor is not None: + from ddev.cli.ci.tests.github_monitor import GitHubMonitor from ddev.monitoring.adapter import ComponentLogAdapter + from ddev.utils.github_async.defaults import default_github_rate_limiter, log_rate_limit_events + from ddev.utils.rate_limiting import RateLimitEvent client_logger = ComponentLogAdapter(monitor) + github_monitor = observer = GitHubMonitor(monitor) + log_event = log_rate_limit_events(client_logger) + + def on_rate_limit_event(event: RateLimitEvent) -> None: + log_event(event) + github_monitor.rate_limit_event(event) + + rate_limiter = default_github_rate_limiter(on_event=on_rate_limit_event) monitor.logger.info('Resolving pull request', pr_number=resolver.number, all_targets=all_targets) async def resolve() -> ResolvedRun | None: - async with async_github_client(token=token, logger=client_logger) as client: + async with async_github_client( + token=token, logger=client_logger, observer=observer, rate_limiter=rate_limiter + ) as client: pull = await resolver.resolve(client) if pull is None: if monitor is not None: diff --git a/ddev/src/ddev/cli/ci/dispatch_tests.py b/ddev/src/ddev/cli/ci/dispatch_tests.py index 05d804190f291..4ef666aab8f4d 100644 --- a/ddev/src/ddev/cli/ci/dispatch_tests.py +++ b/ddev/src/ddev/cli/ci/dispatch_tests.py @@ -197,6 +197,7 @@ def dispatch_tests( tag_fields, ) from ddev.cli.ci.tests.dispatcher_config import DispatcherConfig + from ddev.cli.ci.tests.dispatcher_logging import ci_pipeline_id from ddev.monitoring import MonitoringRuntime, console_formatter tested_repository = f'{owner}/{repo}' @@ -221,6 +222,8 @@ def dispatch_tests( **BASE_FIELDS, **repository_fields(owner, repo), **UNRESOLVED_RUN_FIELDS, + # Bind even when absent, so caller tags and scopes cannot forge the workflow's identity. + 'ci_pipeline_id': ci_pipeline_id(), }, ) datadog_handler = attach_datadog_log_handler(app, monitoring, level=output_level) diff --git a/ddev/src/ddev/cli/ci/tests/dispatcher.py b/ddev/src/ddev/cli/ci/tests/dispatcher.py index bea612078a7cd..a33dd48f658e1 100644 --- a/ddev/src/ddev/cli/ci/tests/dispatcher.py +++ b/ddev/src/ddev/cli/ci/tests/dispatcher.py @@ -240,12 +240,22 @@ def build_dispatcher( all buckets share the provider's budget and pauses. The caller owns `monitoring`; each processor reports its own metrics through the monitor it is given. """ + from ddev.cli.ci.tests.github_monitor import GitHubMonitor from ddev.utils.github_async import AsyncGitHubClient - client_logger = ComponentLogAdapter(monitoring.component('github-async')) + client_monitor = monitoring.component('github-async') + client_logger = ComponentLogAdapter(client_monitor) + github_monitor = GitHubMonitor(client_monitor) integrations = frozenset(integration for batch in batches for integration in batch.integrations) - rate_limiters = RateLimiterFactory(config.github_rate_limits, client_logger) - client = AsyncGitHubClient(token, rate_limiter=rate_limiters.get_limiter(integrations), logger=client_logger) + rate_limiters = RateLimiterFactory( + config.github_rate_limits, client_logger, on_event=github_monitor.rate_limit_event + ) + client = AsyncGitHubClient( + token, + rate_limiter=rate_limiters.get_limiter(integrations), + logger=client_logger, + observer=github_monitor, + ) canonical_run_fields = run_fields(run, tags=tags) runner = TaskTestRunner( diff --git a/ddev/src/ddev/cli/ci/tests/dispatcher_attributes.py b/ddev/src/ddev/cli/ci/tests/dispatcher_attributes.py index 79097656e7bf3..1340af9005a77 100644 --- a/ddev/src/ddev/cli/ci/tests/dispatcher_attributes.py +++ b/ddev/src/ddev/cli/ci/tests/dispatcher_attributes.py @@ -15,6 +15,7 @@ if TYPE_CHECKING: from ddev.cli.ci.dispatch_run import ResolvedRun + from ddev.monitoring import Metrics DEFAULT_TEAM = 'agent-integrations' @@ -46,6 +47,12 @@ class AttributeSpec: # Compact operational context helps read a line; payloads and run-wide identity do not, so they # stay console-hidden and remain available in the structured event. ATTRIBUTE_SPECS: Mapping[str, AttributeSpec] = { + # Logs take the pipeline ID from `ci_attributes`, with the rest of the workflow's identity. + 'ci_pipeline_id': AttributeSpec( + 'ci.pipeline.id', + log_tag=False, + metric_tag=True, + ), 'repo': AttributeSpec( 'git.repository.id_v2', metric_tag=True, @@ -308,6 +315,12 @@ class AttributeSpec: console_tag=True, metric_tag=True, ), + 'status_code': AttributeSpec( + 'http.status_code', + ), + 'rate_limit_resource': AttributeSpec( + 'github.rate_limit.resource', + ), 'error': AttributeSpec( 'error.message', console_tag=True, @@ -320,6 +333,7 @@ class AttributeSpec: PROTECTED_RUN_FIELDS = frozenset( { 'team', + 'ci_pipeline_id', 'repo', 'repository_url', 'head_branch', @@ -411,6 +425,27 @@ def metric_tag_mapping(fields: Mapping[str, Any]) -> dict[str, str]: return _policy_mapping(fields, lambda spec: spec.metric_tag, _stringify_value) +# The GitHub request metrics are sampled per attempt, so every run, batch or job dimension the +# context holds would multiply their series. They carry only these, whatever else is bound. +GITHUB_METRIC_FIELDS = frozenset({'ci_pipeline_id', 'status_code', 'reason', 'rate_limit_resource'}) + + +def github_metric_tag_mapping(fields: Mapping[str, Any]) -> dict[str, str]: + """Render the few dimensions the GitHub request and throttle metrics carry.""" + return attribute_mapping({name: value for name, value in fields.items() if name in GITHUB_METRIC_FIELDS}) + + +def github_metrics(metrics: Metrics) -> Metrics: + """A view of *metrics* for the GitHub request and throttle family. + + Only the pipeline ID may come from the context. Every other allowlisted dimension describes one + emission, so a caller's `--tags` or an enclosing scope must not supply it. + """ + return metrics.with_tag_projector(github_metric_tag_mapping).bind( + **dict.fromkeys(GITHUB_METRIC_FIELDS - PROTECTED_RUN_FIELDS) + ) + + def tag_fields(tags: Sequence[str]) -> dict[str, str]: fields: dict[str, str] = {} for tag in tags: diff --git a/ddev/src/ddev/cli/ci/tests/dispatcher_logging.py b/ddev/src/ddev/cli/ci/tests/dispatcher_logging.py index 5f0b62c80869a..308f4c8f48bd9 100644 --- a/ddev/src/ddev/cli/ci/tests/dispatcher_logging.py +++ b/ddev/src/ddev/cli/ci/tests/dispatcher_logging.py @@ -47,9 +47,14 @@ def _stringify(value: Any) -> str: return stringify(redact_value(value)) +def ci_pipeline_id() -> str | None: + """The ID of the GitHub Actions workflow running the Dispatcher, which a rerun keeps.""" + return os.getenv('GITHUB_RUN_ID') or None + + def ci_attributes() -> dict[str, str]: """Describe the GitHub Actions workflow running the Dispatcher.""" - run_id = os.getenv('GITHUB_RUN_ID') + run_id = ci_pipeline_id() if not run_id: return {} @@ -73,7 +78,7 @@ def ci_attributes() -> dict[str, str]: def get_dispatcher_logs_url(*, terminal: bool = False, now: datetime | None = None) -> str | None: """Link to this run's logs, freezing the time window for terminal reports.""" - run_id = os.getenv('GITHUB_RUN_ID') + run_id = ci_pipeline_id() if not run_id: return None from_ts: int | str diff --git a/ddev/src/ddev/cli/ci/tests/github_monitor.py b/ddev/src/ddev/cli/ci/tests/github_monitor.py new file mode 100644 index 0000000000000..b183c2bbcd601 --- /dev/null +++ b/ddev/src/ddev/cli/ci/tests/github_monitor.py @@ -0,0 +1,119 @@ +# (C) Datadog, Inc. 2026-present +# All rights reserved +# Licensed under a 3-clause BSD style license (see LICENSE) +"""Dispatcher metrics and logs for the GitHub requests it sends and the waits its limiters impose.""" + +from __future__ import annotations + +import math +import time +from collections.abc import Callable +from contextlib import suppress +from typing import Any + +import httpx + +from ddev.cli.ci.tests.dispatcher_attributes import github_metrics +from ddev.monitoring import ComponentMonitor +from ddev.utils.github_async.client import failure_reason, parse_header, with_query_masked +from ddev.utils.github_async.observer import RequestAttempt, RequestFailure, RequestFault +from ddev.utils.rate_limiting import RateLimitEvent, WaitEvent + +# GitHub names the budget a response drew from. One without the header is kept apart rather than +# assumed to have drawn from `core`. +UNKNOWN_RESOURCE = 'unknown' + +RATE_LIMIT_FAULTS = frozenset( + {RequestFault.PRIMARY_RATE_LIMIT, RequestFault.SECONDARY_RATE_LIMIT, RequestFault.UNKNOWN_RATE_LIMIT} +) + + +def failure_detail(error: Exception) -> str: + """The failure's type and reason, with no query string that could carry a signature.""" + reason = failure_reason(error) if isinstance(error, httpx.HTTPError) else with_query_masked(str(error)) + return f'{type(error).__name__}: {reason}' + + +class GitHubMonitor: + """The GitHub client's request observer and the rate limiters' event handler for one component.""" + + def __init__(self, monitor: ComponentMonitor, *, now: Callable[[], float] = time.time) -> None: + self._logger = monitor.logger + self._metrics = github_metrics(monitor.metrics) + self._now = now + + def attempt_finished(self, attempt: RequestAttempt) -> None: + fault = attempt.fault + status = {} if attempt.response is None else {'status_code': attempt.response.status_code} + self._metrics.count('requests.count', 1, **status) + is_fault = fault not in (RequestFault.NONE, RequestFault.CANCELLED) + self._metrics.count('requests.fault', int(is_fault), reason=fault, **status) + self._metrics.count('requests.throttled', int(fault in RATE_LIMIT_FAULTS), **status) + self._metrics.count('requests.retried', int(attempt.number > 1)) + self._metrics.distribution('requests.duration', attempt.duration_seconds, **status) + if attempt.response is not None: + self._emit_budget(attempt.response.headers) + + fields: dict[str, Any] = { + 'method': attempt.method, + 'endpoint': attempt.endpoint, + 'attempt': attempt.number, + 'duration_seconds': attempt.duration_seconds, + **status, + } + if attempt.cancelled: + self._logger.debug('GitHub request cancelled', reason=fault, **fields) + elif attempt.error is None: + self._logger.debug('GitHub request completed', **fields) + else: + self._logger.error( + 'GitHub request attempt failed', + reason=fault, + terminal=False, + error=failure_detail(attempt.error), + **fields, + ) + + def request_failed(self, failure: RequestFailure) -> None: + last = failure.last_attempt + fields: dict[str, Any] = {} + if last is not None and last.error is not None: + fields['reason'] = last.fault + if last.response is not None: + fields['status_code'] = last.response.status_code + self._logger.error( + 'GitHub request failed', + method=failure.method, + endpoint=failure.endpoint, + attempts=0 if last is None else last.number, + terminal=True, + error=failure_detail(failure.error), + **fields, + ) + + def rate_limit_event(self, event: RateLimitEvent) -> None: + # Called from inside the limiter, which must not fail because monitoring did. + with suppress(Exception): + if not isinstance(event, WaitEvent): + return + self._metrics.distribution('throttle.wait.duration', event.elapsed_seconds, reason=event.reason) + # DEBUG even when abandoned: the limiter's own events already warn about that. + self._logger.debug( + 'GitHub rate limit wait ended', + reason=event.reason, + outcome=event.outcome, + elapsed_seconds=event.elapsed_seconds, + requested_seconds=event.requested_seconds, + limiter=event.name or None, + ) + + def _emit_budget(self, headers: httpx.Headers) -> None: + """Gauge the budget a response reports. Absent or unparseable values are not reported as zero.""" + resource = headers.get('x-ratelimit-resource') or UNKNOWN_RESOURCE + for name in ('remaining', 'limit', 'used'): + if (value := parse_header(headers, f'x-ratelimit-{name}', int)) is not None: + self._metrics.gauge(f'requests.rate_limit.{name}', value, rate_limit_resource=resource) + if (reset_at := parse_header(headers, 'x-ratelimit-reset', float)) is not None and math.isfinite(reset_at): + self._metrics.gauge( + 'requests.rate_limit.reset_in', max(0.0, reset_at - self._now()), rate_limit_resource=resource + ) diff --git a/ddev/src/ddev/cli/ci/tests/rate_limiting.py b/ddev/src/ddev/cli/ci/tests/rate_limiting.py index 20a169f67dbdc..8f82cbf53f5df 100644 --- a/ddev/src/ddev/cli/ci/tests/rate_limiting.py +++ b/ddev/src/ddev/cli/ci/tests/rate_limiting.py @@ -100,10 +100,17 @@ def __init__( self, config: RateLimiterFactoryConfig | None = None, logger: logging.Logger | None = None, + on_event: Callable[[RateLimitEvent], None] | None = None, ) -> None: cfg = config or RateLimiterFactoryConfig() self.slow_integrations = cfg.slow_integrations - on_event = event_logger(logger) if logger else None + handlers = [handler for handler in (event_logger(logger) if logger else None, on_event) if handler is not None] + + def handle(event: RateLimitEvent) -> None: + for handler in handlers: + handler(event) + + on_event = handle if handlers else None budget_governor = BudgetGovernor( reserve_fraction=cfg.reserve_fraction, buffer_seconds=cfg.budget_buffer_seconds, diff --git a/ddev/src/ddev/cli/ci/tests/task_test_runner.py b/ddev/src/ddev/cli/ci/tests/task_test_runner.py index a62ba9a8f89a5..b6b4feb8e8fda 100644 --- a/ddev/src/ddev/cli/ci/tests/task_test_runner.py +++ b/ddev/src/ddev/cli/ci/tests/task_test_runner.py @@ -13,6 +13,7 @@ from datetime import datetime from itertools import count from pathlib import Path +from time import monotonic from typing import Any from pydantic import ValidationError @@ -280,7 +281,12 @@ async def _poll_until_complete( sequences = count(1) known_jobs: dict[str, WorkflowJob] = {} previous_state: ExecutionState | None = None + previous_poll: float | None = None while True: + poll_started = monotonic() + if previous_poll is not None: + self.monitor.metrics.distribution('requests.polling_interval', poll_started - previous_poll) + previous_poll = poll_started try: run = await self._client.get_workflow_run(self._options.owner, self._options.repo, run_id) except ValidationError as error: diff --git a/ddev/src/ddev/monitoring/metrics.py b/ddev/src/ddev/monitoring/metrics.py index 7f0a14e48b364..c7865ad98fed7 100644 --- a/ddev/src/ddev/monitoring/metrics.py +++ b/ddev/src/ddev/monitoring/metrics.py @@ -87,6 +87,16 @@ def bind(self, **fields: Any) -> Metrics: is_closed=self._is_closed, ) + def with_tag_projector(self, tag_projector: TagProjector) -> Metrics: + """The same view, rendering its tags with *tag_projector* instead of the runtime's.""" + return Metrics( + self._context, + sink=self._sink, + tag_projector=tag_projector, + fields=self._fields, + is_closed=self._is_closed, + ) + def count( self, name: str, diff --git a/ddev/src/ddev/utils/github_async/client.py b/ddev/src/ddev/utils/github_async/client.py index 6258c97c1d3cc..2236a3d7e9d01 100644 --- a/ddev/src/ddev/utils/github_async/client.py +++ b/ddev/src/ddev/utils/github_async/client.py @@ -5,6 +5,7 @@ from __future__ import annotations +import asyncio import io import logging import re @@ -14,6 +15,7 @@ from copy import copy from dataclasses import dataclass, replace from pathlib import Path +from time import monotonic from typing import Any, Literal, Self, overload import httpx @@ -55,6 +57,7 @@ WorkflowJobsList, WorkflowRun, ) +from .observer import RequestAttempt, RequestFailure, RequestObserver from .retry import ( DEFAULT_RETRY_POLICIES, NO_RETRY, @@ -200,6 +203,17 @@ def __call__(self, exc: Exception) -> bool: return True +@dataclass +class RequestAttempts: + """The sends of one logical request so far, shared by every retry layer that replays it.""" + + last: RequestAttempt | None = None + + @property + def count(self) -> int: + return 0 if self.last is None else self.last.number + + def github_rate_limit_snapshot(headers: httpx.Headers) -> BudgetSnapshot | None: """Parse GitHub's `x-ratelimit-*` / `retry-after` response headers into a BudgetSnapshot.""" snapshot = BudgetSnapshot( @@ -252,6 +266,9 @@ class AsyncGitHubClient: default rate limiter, while a caller-supplied `rate_limiter` keeps whatever logging it was built with, since that choice belongs to whoever built it. transport: Optional custom HTTPX transport (useful for testing with MockTransport). + observer: Receives every request sent to the GitHub API and every request that finally + fails. A retry of a failure it was already shown is then not logged again through + `logger`. Views made with `with_rate_limit` report to the same observer. """ def __init__( @@ -264,11 +281,13 @@ def __init__( retry_policies: RetryPolicies | None = None, logger: logging.Logger | None = None, transport: httpx.AsyncBaseTransport | None = None, + observer: RequestObserver | None = None, ) -> None: if not token: raise ValueError("GitHub token must not be empty.") self._logger = logger + self._observer = observer # A None limiter means "use the default protection," not "no protection." The local bucket # is deliberately permissive because the governor is the protection; with a healthy budget # and no secondary limits the governor adds zero wait, so this default is invisible to @@ -347,10 +366,17 @@ def _refuses_retry(self, exc: Exception) -> bool: return self._is_rate_limit_response(exc.response) or exc.response.has_redirect_location return False - def _log_retry(self, description: str, cause: RetryCause, attempt: stamina.Attempt) -> None: - """Report a retry that is about to run. Silent on the first attempt, and with no logger.""" + def _log_retry( + self, description: str, cause: RetryCause, attempt: stamina.Attempt, attempts: RequestAttempts + ) -> None: + """Report a retry that is about to run. Silent on the first attempt, and with no logger. + + Also silent when the failure being retried is the attempt the observer was just shown. + """ if attempt.num == 1 or self._logger is None: return + if self._observer is not None and attempts.last is not None and cause.error is attempts.last.error: + return self._logger.warning( "Retrying %s after %r (attempt %s)", description, @@ -374,22 +400,63 @@ def _is_rate_limit_response(response: httpx.Response) -> bool: or response.headers.get("x-ratelimit-remaining") == "0" ) + def _report_attempt( + self, + attempts: RequestAttempts, + method: str, + endpoint: str, + duration_seconds: float, + response: httpx.Response | None, + error: Exception | None, + *, + cancelled: bool = False, + ) -> None: + attempts.last = RequestAttempt( + method=method, + endpoint=with_query_masked(endpoint), + number=attempts.count + 1, + duration_seconds=duration_seconds, + response=response, + error=error, + cancelled=cancelled, + ) + if self._observer is not None: + with suppress(Exception): + self._observer.attempt_finished(attempts.last) + + def _report_failure(self, attempts: RequestAttempts, method: str, endpoint: str, error: Exception) -> None: + if self._observer is not None: + with suppress(Exception): + self._observer.request_failed( + RequestFailure( + method=method, endpoint=with_query_masked(endpoint), error=error, last_attempt=attempts.last + ) + ) + async def _execute_request( self, method: str, endpoint: str, timeout: float, + attempts: RequestAttempts, *, expect_redirect: bool = False, **kwargs: Any, ) -> httpx.Response: + started = monotonic() try: response = await self._client.request(method, endpoint, timeout=timeout, **kwargs) except httpx.TransportError as exc: # Rewritten in place rather than replaced by a copy, which would drop the request httpx # attached and leave `exc.request` raising RuntimeError for the caller. exc.args = (f"{method} {endpoint}: {exc}",) + self._report_attempt(attempts, method, endpoint, monotonic() - started, None, exc) raise + except asyncio.CancelledError: + # The request may already have reached GitHub, so it still counts as sent. + self._report_attempt(attempts, method, endpoint, monotonic() - started, None, None, cancelled=True) + raise + duration_seconds = monotonic() - started # Observe before raise_for_status, never after: learning must not be gated on success. A # failed response's rate-limit headers arm the shared pause even if the caller swallows the # exception, so one request's 403 protects every other in-flight and future request in this @@ -400,19 +467,31 @@ async def _execute_request( snapshot = replace(snapshot or NULL_SNAPSHOT, retry_after=secondary_rate_limit_wait) if snapshot is not None: self._rate_limiter.observe(snapshot) - # The artifact endpoint checks the redirect itself, and reports a bad one more precisely. - if expect_redirect and response.is_redirect: - return response - # Not `is_redirect`, which spans the whole 3xx range: a 304 carries no Location to refuse. - if response.has_redirect_location: - raise GitHubUnexpectedRedirectError.from_response(method, endpoint, response) - response.raise_for_status() + error: httpx.HTTPError | None = None + if not (expect_redirect and response.is_redirect): + # Not `is_redirect`, which spans the whole 3xx range: a 304 carries no Location to refuse. + if response.has_redirect_location: + error = GitHubUnexpectedRedirectError.from_response(method, endpoint, response) + else: + try: + response.raise_for_status() + except httpx.HTTPStatusError as exc: + error = exc + if expect_redirect and error is None: + if response.status_code != 302: + error = httpx.HTTPError(f"Expected 302 redirect from {endpoint}, got {response.status_code}") + elif not response.headers.get("location"): + error = httpx.HTTPError(f"Missing Location header on redirect from {endpoint}") + self._report_attempt(attempts, method, endpoint, duration_seconds, response, error) + if error is not None: + raise error return response async def _rate_limited_request( self, method: str, endpoint: str, + attempts: RequestAttempts, timeout: float | None = None, *, expect_redirect: bool = False, @@ -427,7 +506,7 @@ async def _rate_limited_request( async with self._rate_limiter: try: return await self._execute_request( - method, endpoint, effective_timeout, expect_redirect=expect_redirect, **kwargs + method, endpoint, effective_timeout, attempts, expect_redirect=expect_redirect, **kwargs ) except httpx.HTTPStatusError as exc: # Safe to replay even for non-idempotent endpoints: GitHub rejected the request @@ -450,21 +529,32 @@ async def _request( *, retry: RetryPolicy | None = None, expect_redirect: bool = False, + attempts: RequestAttempts | None = None, **kwargs: Any, ) -> httpx.Response: """Send one request, retrying the failures its policy accepts. Wraps the rate-limit layer rather than living inside it, so every attempt re-acquires the limiter and waits out any pause the governor holds. + + A caller that retries this request itself passes `attempts`, so its replays keep counting + from here and it alone reports the final failure. """ policy = self._effective_retry(retry if retry is not None else self._retry_policies.for_method(method)) cause = self._retry_cause(policy) - async for attempt in retry_attempts(policy, cause): - with attempt: - self._log_retry(f"{method} {endpoint}", cause, attempt) - return await self._rate_limited_request( - method, endpoint, timeout, expect_redirect=expect_redirect, **kwargs - ) + owns_attempts = attempts is None + attempts = RequestAttempts() if attempts is None else attempts + try: + async for attempt in retry_attempts(policy, cause): + with attempt: + self._log_retry(f"{method} {endpoint}", cause, attempt, attempts) + return await self._rate_limited_request( + method, endpoint, attempts, timeout, expect_redirect=expect_redirect, **kwargs + ) + except Exception as error: + if owns_attempts: + self._report_failure(attempts, method, endpoint, error) + raise raise RuntimeError("unreachable: the retry loop always returns or raises") # pragma: no cover async def _paginated_request( @@ -1443,6 +1533,7 @@ async def _resolve_artifact_redirect( timeout: float | None = None, *, retry: RetryPolicy | None = None, + attempts: RequestAttempts | None = None, ) -> str: """Authenticated GET; return the unauthenticated signed URL from the 302 Location header. @@ -1455,16 +1546,10 @@ async def _resolve_artifact_redirect( timeout=timeout, retry=retry, expect_redirect=True, + attempts=attempts, follow_redirects=False, ) - if redirect_response.status_code != 302: - raise httpx.HTTPError( - f"Expected 302 redirect from {archive_download_url}, got {redirect_response.status_code}" - ) - location = redirect_response.headers.get("location") - if not location: - raise httpx.HTTPError(f"Missing Location header on redirect from {archive_download_url}") - return location + return redirect_response.headers["location"] async def _download_and_extract_zip( self, @@ -1531,13 +1616,24 @@ async def download_artifact( """ policy = retry if retry is not None else self._artifact_retry cause = self._retry_cause(policy) - async for attempt in retry_attempts(policy, cause): - with attempt: - self._log_retry(f"artifact download {with_query_masked(archive_download_url)}", cause, attempt) - # NO_RETRY on the inner call: this loop is the only ladder, or the two would multiply. - location = await self._resolve_artifact_redirect(archive_download_url, timeout, retry=NO_RETRY) - await self._download_and_extract_zip(location, dest_path, timeout) - return + # Shared by every pass so a refreshed redirect counts as a retry. The storage fetch is not a + # GitHub API request and is never counted. + attempts = RequestAttempts() + try: + async for attempt in retry_attempts(policy, cause): + with attempt: + self._log_retry( + f"artifact download {with_query_masked(archive_download_url)}", cause, attempt, attempts + ) + # NO_RETRY on the inner call: this loop is the only ladder, or the two would multiply. + location = await self._resolve_artifact_redirect( + archive_download_url, timeout, retry=NO_RETRY, attempts=attempts + ) + await self._download_and_extract_zip(location, dest_path, timeout) + return + except Exception as error: + self._report_failure(attempts, "GET", archive_download_url, error) + raise # --------------------------------------------------------------------------- @@ -1555,6 +1651,7 @@ async def async_github_client( retry_policies: RetryPolicies | None = None, logger: logging.Logger | None = None, transport: httpx.AsyncBaseTransport | None = None, + observer: RequestObserver | None = None, ) -> AsyncIterator[AsyncGitHubClient]: """ Async context manager that creates an AsyncGitHubClient and ensures it is closed on exit. @@ -1575,6 +1672,7 @@ async def async_github_client( retry_policies: Overrides the per-endpoint defaults for failures that are not rate limiting. logger: Where retries are reported; None keeps the client silent. transport: Optional custom HTTPX transport (useful for testing with MockTransport). + observer: Receives every request sent to the GitHub API and every request that finally fails. Yields: AsyncGitHubClient: A ready-to-use async GitHub client. @@ -1587,6 +1685,7 @@ async def async_github_client( retry_policies=retry_policies, logger=logger, transport=transport, + observer=observer, ) try: yield client diff --git a/ddev/src/ddev/utils/github_async/defaults.py b/ddev/src/ddev/utils/github_async/defaults.py index 1dc4da452e29d..be94fe67c4a11 100644 --- a/ddev/src/ddev/utils/github_async/defaults.py +++ b/ddev/src/ddev/utils/github_async/defaults.py @@ -25,12 +25,11 @@ PacingReason, RateLimitEvent, SecondaryLimitEvent, + WaitEvent, ) -logger = logging.getLogger(__name__) - -def log_rate_limit_events(logger: logging.Logger = logger) -> Callable[[RateLimitEvent], None]: +def log_rate_limit_events(logger: logging.Logger) -> Callable[[RateLimitEvent], None]: """Return an on_event callback that logs rate-limit events for operators. Stateless: a low budget logs at DEBUG on every observe, so it can be chatty. An edge-detecting @@ -72,6 +71,8 @@ def handle(event: RateLimitEvent) -> None: logger.debug("%s rate limiter throttling request", event.name) case BucketEvent(): pass # Bucket had capacity: nothing worth an operator's attention. + case WaitEvent(): + pass # The PacingEvent and BucketEvent above already announced this wait. case _: # New event types must not break logging. logger.debug("unhandled rate limit event: %r", event) @@ -91,14 +92,15 @@ def default_github_rate_limiter( shared budget and secondary limits reported in response headers. With a healthy budget and no secondary limits the governor adds zero wait, so this default is invisible to well-behaved callers and engages only once GitHub has already signaled backpressure. + + Without `on_event` the limiter reports nothing. """ - handler = on_event or log_rate_limit_events() - governor = budget_governor or BudgetGovernor(on_event=handler) + governor = budget_governor or BudgetGovernor(on_event=on_event) # The governor and the InstrumentedAsyncLimiter each carry their own on_event slot; wiring only # one silently drops half the events (bucket-throttle vs budget/pacing), so wire both. return InstrumentedAsyncLimiter( AsyncLimiter(max_rate=5000, time_period=RATE_LIMIT_TIME_PERIOD), - on_event=handler, + on_event=on_event, budget_governor=governor, name=name, ) diff --git a/ddev/src/ddev/utils/github_async/observer.py b/ddev/src/ddev/utils/github_async/observer.py new file mode 100644 index 0000000000000..a905ab9010fe1 --- /dev/null +++ b/ddev/src/ddev/utils/github_async/observer.py @@ -0,0 +1,98 @@ +# (C) Datadog, Inc. 2026-present +# All rights reserved +# Licensed under a 3-clause BSD style license (see LICENSE) +"""What the client reports about the requests it sends, for a caller that monitors them.""" + +from __future__ import annotations + +from dataclasses import dataclass +from enum import StrEnum, auto +from typing import Protocol + +import httpx + +from ddev.utils.github_errors import github_secondary_rate_limit_wait + + +class RequestFault(StrEnum): + """Why an attempt did not succeed, in a small fixed set of categories. + + `CANCELLED` is the caller's own cancellation of a request in flight, not a failure of GitHub's. + """ + + NONE = auto() + CANCELLED = auto() + AUTHENTICATION = auto() + PERMISSION = auto() + PRIMARY_RATE_LIMIT = auto() + SECONDARY_RATE_LIMIT = auto() + UNKNOWN_RATE_LIMIT = auto() + TIMEOUT = auto() + TRANSPORT = auto() + CLIENT_ERROR = auto() + SERVER_ERROR = auto() + UNEXPECTED_RESPONSE = auto() + + +@dataclass(frozen=True) +class RequestAttempt: + """One request actually sent to the GitHub API, once it has a response, a transport failure or + was cancelled in flight. + + `endpoint` has its query masked. `number` counts the sends of one logical request across every + retry layer, starting at 1. `response` is None for a transport failure or a cancellation. + """ + + method: str + endpoint: str + number: int + duration_seconds: float + response: httpx.Response | None + error: Exception | None + cancelled: bool = False + + @property + def fault(self) -> RequestFault: + if self.cancelled: + return RequestFault.CANCELLED + if self.error is None: + return RequestFault.NONE + if self.response is None: + return RequestFault.TIMEOUT if isinstance(self.error, httpx.TimeoutException) else RequestFault.TRANSPORT + status = self.response.status_code + # The client's own rule for a retryable rate limit: a 403 alone is a permission denial. + if status in (403, 429): + if self.response.headers.get("x-ratelimit-remaining") == "0": + return RequestFault.PRIMARY_RATE_LIMIT + if github_secondary_rate_limit_wait(self.response) is not None: + return RequestFault.SECONDARY_RATE_LIMIT + if status == 429: + return RequestFault.UNKNOWN_RATE_LIMIT + return RequestFault.PERMISSION + if status == 401: + return RequestFault.AUTHENTICATION + if status < 400: + return RequestFault.UNEXPECTED_RESPONSE + return RequestFault.CLIENT_ERROR if status < 500 else RequestFault.SERVER_ERROR + + +@dataclass(frozen=True) +class RequestFailure: + """A logical request that failed for good, after whatever retries it was allowed. + + `last_attempt` is None when the request failed before anything was sent, for example when a + rate-limit wait was abandoned. + """ + + method: str + endpoint: str + error: Exception + last_attempt: RequestAttempt | None + + +class RequestObserver(Protocol): + """Receives the client's observations. Exceptions it raises are ignored by the client.""" + + def attempt_finished(self, attempt: RequestAttempt) -> None: ... + + def request_failed(self, failure: RequestFailure) -> None: ... diff --git a/ddev/src/ddev/utils/rate_limiting.py b/ddev/src/ddev/utils/rate_limiting.py index c9bc191f1afb0..61b16c9362ad8 100644 --- a/ddev/src/ddev/utils/rate_limiting.py +++ b/ddev/src/ddev/utils/rate_limiting.py @@ -9,7 +9,8 @@ import dataclasses import time from collections.abc import Callable -from enum import StrEnum +from enum import StrEnum, auto +from time import monotonic from typing import Any, Literal from aiolimiter import AsyncLimiter @@ -74,6 +75,7 @@ class RateLimitEventType(StrEnum): BUDGET = "budget" SECONDARY_LIMIT = "secondary_limit" PACING = "pacing" + WAIT = "wait" class PacingReason(StrEnum): @@ -86,6 +88,23 @@ class PacingReason(StrEnum): ABANDONED = "abandoned" +class WaitReason(StrEnum): + """What held a request back for a WaitEvent.""" + + BUCKET = auto() + RATIONING = auto() + EXHAUSTED = auto() + SECONDARY_LIMIT = auto() + + +class WaitOutcome(StrEnum): + """How a WaitEvent's wait ended.""" + + COMPLETED = auto() + ABANDONED = auto() + CANCELLED = auto() + + @dataclasses.dataclass(frozen=True) class BucketEvent: """Fired once per InstrumentedAsyncLimiter acquire, carrying whether it had to throttle.""" @@ -126,7 +145,23 @@ class PacingEvent: type: Literal[RateLimitEventType.PACING] = RateLimitEventType.PACING -RateLimitEvent = BucketEvent | BudgetEvent | SecondaryLimitEvent | PacingEvent +@dataclasses.dataclass(frozen=True) +class WaitEvent: + """Fired when a wait that actually blocked ends, however it ended. + + `elapsed_seconds` is measured on the monotonic clock. `requested_seconds` is the delay the wait + set out to cover, which the local bucket does not expose, so it is None for a bucket wait. + """ + + reason: WaitReason + elapsed_seconds: float + outcome: WaitOutcome + requested_seconds: float | None = None + name: str = "" + type: Literal[RateLimitEventType.WAIT] = RateLimitEventType.WAIT + + +RateLimitEvent = BucketEvent | BudgetEvent | SecondaryLimitEvent | PacingEvent | WaitEvent class RateLimitWaitAbandoned(TimeoutError): @@ -278,27 +313,48 @@ async def wait(self) -> None: self.on_event(PacingEvent(wait_seconds=wait_seconds, reason=reason)) give_up_at = None if self.max_wait_seconds is None else start + self.max_wait_seconds floor = max(deadline, self.pause_until) - for _ in range(MAX_WAIT_ITERATIONS): - current_floor = max(deadline, self.pause_until) - now = self.now() - delay = current_floor - now - if delay <= 0: - return - if give_up_at is not None and current_floor > give_up_at: - # The floor is past the budget: the request is doomed, so fail now rather than - # sleep into a wait we already know we will abandon. - self.on_event(PacingEvent(wait_seconds=delay, reason=PacingReason.ABANDONED)) - raise RateLimitWaitAbandoned(waited_seconds=now - start, remaining_seconds=delay) - if current_floor > floor: - # The floor only grows when pause_until does, i.e. a secondary-limit observe - # extended the wait mid-sleep; emit so the extension is observable. - self.on_event(PacingEvent(wait_seconds=delay, reason=PacingReason.SECONDARY_LIMIT)) - floor = current_floor - await asyncio.sleep(delay) - raise RuntimeError( - f"BudgetGovernor.wait failed to converge after {MAX_WAIT_ITERATIONS} iterations " - f"(remaining delay: {max(deadline, self.pause_until) - self.now()}s)" - ) + started = monotonic() + slept = False + outcome = WaitOutcome.CANCELLED + try: + for _ in range(MAX_WAIT_ITERATIONS): + current_floor = max(deadline, self.pause_until) + now = self.now() + delay = current_floor - now + if delay <= 0: + outcome = WaitOutcome.COMPLETED + return + if give_up_at is not None and current_floor > give_up_at: + # The floor is past the budget: the request is doomed, so fail now rather than + # sleep into a wait we already know we will abandon. + self.on_event(PacingEvent(wait_seconds=delay, reason=PacingReason.ABANDONED)) + outcome = WaitOutcome.ABANDONED + raise RateLimitWaitAbandoned(waited_seconds=now - start, remaining_seconds=delay) + if current_floor > floor: + # The floor only grows when pause_until does, i.e. a secondary-limit observe + # extended the wait mid-sleep; emit so the extension is observable. + self.on_event(PacingEvent(wait_seconds=delay, reason=PacingReason.SECONDARY_LIMIT)) + floor = current_floor + reason = PacingReason.SECONDARY_LIMIT + slept = True + await asyncio.sleep(delay) + outcome = WaitOutcome.ABANDONED + raise RuntimeError( + f"BudgetGovernor.wait failed to converge after {MAX_WAIT_ITERATIONS} iterations " + f"(remaining delay: {max(deadline, self.pause_until) - self.now()}s)" + ) + finally: + # One event per wait, however many extensions it absorbed. A wait abandoned before its + # first sleep never blocked, and its PacingEvent already reports the abandonment. + if slept and reason is not PacingReason.NONE: + self.on_event( + WaitEvent( + reason=WaitReason(reason), + elapsed_seconds=monotonic() - started, + outcome=outcome, + requested_seconds=wait_seconds, + ) + ) @dataclasses.dataclass(frozen=True) @@ -357,7 +413,21 @@ async def __aenter__(self) -> InstrumentedAsyncLimiter: if self.budget_governor is not None: await self.budget_governor.wait() throttled = not self.limiter.has_capacity() - await self.limiter.__aenter__() + started = monotonic() + outcome = WaitOutcome.CANCELLED + try: + await self.limiter.__aenter__() + outcome = WaitOutcome.COMPLETED + finally: + if throttled: + self.on_event( + WaitEvent( + reason=WaitReason.BUCKET, + elapsed_seconds=monotonic() - started, + outcome=outcome, + name=self.name, + ) + ) self.on_event(BucketEvent(throttled=throttled, name=self.name)) return self diff --git a/ddev/tests/cli/ci/test_dispatch_tests.py b/ddev/tests/cli/ci/test_dispatch_tests.py index a925f091eeaf3..08195a5ac6a88 100644 --- a/ddev/tests/cli/ci/test_dispatch_tests.py +++ b/ddev/tests/cli/ci/test_dispatch_tests.py @@ -13,6 +13,7 @@ from typing import TYPE_CHECKING, Any from unittest.mock import ANY +import httpx import pytest from ddev.cli.application import Application @@ -26,9 +27,12 @@ from ddev.monitoring.datadog import DatadogLogHandler from ddev.monitoring.datadog_metrics import DatadogMetricsSink from ddev.utils.git import ChangedFile, ChangeType, GitCommit +from ddev.utils.github_async import async_github_client from ddev.utils.github_async.models import PullRequest, WorkflowRun +from ddev.utils.rate_limiting import BudgetGovernor from tests.cli.ci.helpers import HEAD_SHA, PR_NUMBER, decode_job_list, listed_pull_request, mock_job_result, pulls_page from tests.cli.ci.tests.helpers import make_batch, make_job +from tests.helpers.clock import FakeClock, advance_clock_on_sleep from tests.helpers.datadog import FakeLogSubmitter, FakeMetricsSubmitter from tests.helpers.monitoring import RecordingJsonHandler, RecordingSink, projector_for @@ -615,6 +619,77 @@ def observe_plan(app: Application, *, monitor: ComponentMonitor, **kwargs: Any) } +@pytest.mark.parametrize( + ('environ', 'pipeline_id'), + [ + ({'GITHUB_RUN_ID': '111', 'GITHUB_RUN_ATTEMPT': '1'}, '111'), + ({'GITHUB_RUN_ID': '111', 'GITHUB_RUN_ATTEMPT': '2'}, '111'), + ({'GITHUB_RUN_ID': '222', 'GITHUB_RUN_ATTEMPT': '1'}, '222'), + ({}, None), + ], + ids=['first-attempt', 'rerun', 'another-workflow', 'outside-github-actions'], +) +def test_metrics_are_tagged_with_the_workflow_running_the_dispatcher( + ddev, fake_async_github, resolved_changes, mocker, monkeypatch, environ: dict[str, str], pipeline_id: str | None +): + """A rerun reports as the same emitter, since it cannot run concurrently with the attempt it replaces.""" + for variable in ('GITHUB_RUN_ID', 'GITHUB_RUN_ATTEMPT'): + monkeypatch.delenv(variable, raising=False) + for variable, value in environ.items(): + monkeypatch.setenv(variable, value) + sink = recording_runtime(mocker) + + def observe_plan(app: Application, *, monitor: ComponentMonitor, **kwargs: Any) -> list[TestBatch]: + monitor.metrics.count('plan') + return [] + + mocker.patch('ddev.cli.ci.dispatch_tests.build_plan', observe_plan) + + result = ddev('ci', 'dispatch-tests', '--commit', 'a-sha', '--tags', 'ci_pipeline_id:forged') + + assert result.exit_code == 0, result.output + assert sink.records_named('plan') + assert {record.tags.get('ci.pipeline.id') for record in sink.records} == {pipeline_id} + + +def test_pull_request_resolution_meters_the_rate_limit_wait_it_sits_out(ddev, resolved_changes, mocker, monkeypatch): + """Resolution builds its own client before the Dispatcher's exists, so its waits need their own wiring.""" + clock = FakeClock() + advance_clock_on_sleep(clock, monkeypatch) + monkeypatch.setattr('ddev.utils.rate_limiting.monotonic', clock) + monkeypatch.setattr('ddev.utils.github_async.defaults.BudgetGovernor', partial(BudgetGovernor, now=clock)) + responses = [ + httpx.Response(429, headers={'retry-after': '30'}), + httpx.Response(200, json=pull_request().model_dump(mode='json')), + ] + transport = httpx.MockTransport(lambda request: responses.pop(0)) + mocker.patch('ddev.utils.github_async.async_github_client', partial(async_github_client, transport=transport)) + mocker.patch.dict('os.environ', {'DD_GITHUB_TOKEN': 'ghp_test'}) + mocker.patch('ddev.cli.ci.dispatch_tests.build_plan', return_value=[]) + monkeypatch.setenv('GITHUB_RUN_ID', '12345') + handler = RecordingJsonHandler() + sink = recording_runtime(mocker, handler) + + result = ddev('ci', 'dispatch-tests', '--pr', str(PR_NUMBER)) + + assert result.exit_code == 0, result.output + # The 30 seconds GitHub asked for, plus the governor's one-second buffer. + assert [(record.value, dict(record.tags)) for record in sink.records_named('throttle.wait.duration')] == [ + (31, {'ci.pipeline.id': '12345', 'dispatcher.reason': 'secondary_limit'}) + ] + # Resolution's own requests are observed too, under the same pipeline ID. + assert [ + (record.value, record.tags.get('ci.pipeline.id')) for record in sink.records_named('requests.throttled') + ] == [ + (1, '12345'), + (0, '12345'), + ] + assert [event['event'] for event in handler.events if event['level'] == 'warning'] == [ + 'GitHub secondary rate limit hit: asked to retry after 30s, pausing all requests for 31s', + 'rate limit secondary pause: waiting 31.0s before the next request', + ] + + @pytest.mark.usefixtures('resolved_changes') @pytest.mark.parametrize( ('level_options', 'visible', 'hidden'), @@ -1082,9 +1157,11 @@ def test_a_resolution_failure_keeps_unresolved_metric_dimensions( ddev: CliRunner, fake_async_github: FakeAsyncGitHubClient, mocker: MockerFixture, + monkeypatch: pytest.MonkeyPatch, tmp_path: Path, from_manifest: bool, ): + monkeypatch.setenv('GITHUB_RUN_ID', '12345') mocker.patch('ddev.cli.ci.dispatch_run.MERGE_COMMIT_REFRESH_SECONDS', 0.0) fake_async_github.mock_response('get_pull_request', pull_request(merge_commit_sha=None)) handler = RecordingJsonHandler() @@ -1115,6 +1192,7 @@ def test_a_resolution_failure_keeps_unresolved_metric_dimensions( 'git.branch': 'unresolved', 'dispatcher.base_branch': 'unresolved', 'dispatcher.run.is_fork': 'unresolved', + 'ci.pipeline.id': '12345', 'dispatcher.component': 'dispatcher', } [started] = [event for event in handler.events if event['event'] == 'Dispatcher invocation started'] diff --git a/ddev/tests/cli/ci/tests/test_dispatcher.py b/ddev/tests/cli/ci/tests/test_dispatcher.py index d7096f182abe0..e642b7f7674ab 100644 --- a/ddev/tests/cli/ci/tests/test_dispatcher.py +++ b/ddev/tests/cli/ci/tests/test_dispatcher.py @@ -11,6 +11,7 @@ import signal import sys import time +from collections import Counter from collections.abc import AsyncIterator from io import StringIO from pathlib import Path @@ -67,6 +68,7 @@ WorkflowJobStatus, WorkflowRun, ) +from ddev.utils.github_async.observer import RequestObserver from ddev.utils.rate_limiting import BucketEvent, InstrumentedAsyncLimiter, RateLimitEvent from tests.cli.ci.helpers import mock_job_result from tests.cli.ci.tests.helpers import ( @@ -208,12 +210,14 @@ def test_dispatcher_assembly_routes_artifact_requests_to_the_artifact_tier( ): events: list[RateLimitEvent] = [] requests: dict[str, str] = {} + requests_sent: Counter[str] = Counter() job = make_job() run_url = "https://github.com/DataDog/integrations-core/actions/runs/123" def handle(request: httpx.Request) -> httpx.Response: bucket = next(event for event in reversed(events) if isinstance(event, BucketEvent)) requests[request.url.path] = bucket.name + requests_sent[request.url.path] += 1 if request.url.path.endswith("/dispatches"): return httpx.Response(200, json={"workflow_run_id": 123, "run_url": str(request.url), "html_url": run_url}) if request.url.path.endswith("/artifacts"): @@ -234,13 +238,20 @@ def handle(request: httpx.Request) -> httpx.Response: ) def make_client( - token: str, *, rate_limiter: InstrumentedAsyncLimiter, logger: logging.Logger | None = None + token: str, + *, + rate_limiter: InstrumentedAsyncLimiter, + logger: logging.Logger | None = None, + observer: RequestObserver | None = None, ) -> AsyncGitHubClient: - return AsyncGitHubClient(token, rate_limiter=rate_limiter, logger=logger, transport=httpx.MockTransport(handle)) + return AsyncGitHubClient( + token, rate_limiter=rate_limiter, logger=logger, observer=observer, transport=httpx.MockTransport(handle) + ) monkeypatch.setattr("ddev.utils.github_async.AsyncGitHubClient", make_client) monkeypatch.setattr(rate_limiting, "event_logger", lambda _: events.append) - monitoring = MonitoringRuntime() + sink = RecordingSink() + monitoring = MonitoringRuntime(metrics_sink=sink) dispatcher = build_dispatcher( batches=[make_batch(job)], run=CONTEXT.model_copy(update={"pr_number": None}), @@ -258,6 +269,8 @@ def make_client( assert requests[f"{prefix}/artifacts"] == "artifacts" assert requests[f"{prefix}/jobs"] == "default" assert requests[prefix] == "default" + # Both tiers report through the Dispatcher: one attempt per request the transport answered. + assert len(sink.records_named("requests.count")) == sum(requests_sent.values()) def test_progress_reaches_the_comment_before_artifact_collection_finishes( diff --git a/ddev/tests/cli/ci/tests/test_github_monitor.py b/ddev/tests/cli/ci/tests/test_github_monitor.py new file mode 100644 index 0000000000000..135b99bd1d3c6 --- /dev/null +++ b/ddev/tests/cli/ci/tests/test_github_monitor.py @@ -0,0 +1,262 @@ +# (C) Datadog, Inc. 2026-present +# All rights reserved +# Licensed under a 3-clause BSD style license (see LICENSE) +"""Tests for the Dispatcher's GitHub request and throttle metrics and logs.""" + +from __future__ import annotations + +import asyncio +import json +from contextlib import suppress + +import httpx +import pytest +from aiolimiter import AsyncLimiter + +from ddev.cli.ci.tests.github_monitor import GitHubMonitor +from ddev.monitoring.metrics import MetricRecord +from ddev.utils.github_async import AsyncGitHubClient +from ddev.utils.github_async.retry import NO_RETRY +from ddev.utils.rate_limiting import ( + BudgetGovernor, + BudgetSnapshot, + InstrumentedAsyncLimiter, + RateLimitWaitAbandoned, + WaitEvent, + WaitOutcome, + WaitReason, +) +from tests.cli.ci.tests.helpers import recording_runtime +from tests.helpers.clock import FakeClock, advance_clock_on_sleep +from tests.helpers.monitoring import RecordingJsonHandler, RecordingSink + +NOW = 1_700_000_000.0 +RUN_PATH = '/repos/o/r/actions/runs/42' +SIGNATURE = 'sig-secret' +PIPELINE = {'ci.pipeline.id': '12345'} +RUN_PAYLOAD = {'id': 42, 'status': 'completed', 'conclusion': 'success', 'html_url': 'https://github.com/o/r/runs/42'} + + +def monitored( + transport: httpx.MockTransport, rate_limiter: InstrumentedAsyncLimiter | None = None +) -> tuple[AsyncGitHubClient, GitHubMonitor, RecordingSink, RecordingJsonHandler]: + """A client reporting through a runtime whose context holds dimensions the family must not carry. + + That includes a caller's `reason`, `status_code` and `rate_limit_resource` tags, which must never + stand in for an emission's own. + """ + monitoring, sink = recording_runtime() + handler = RecordingJsonHandler() + monitoring.add_log_handler(handler) + monitoring.set_run_fields( + ci_pipeline_id='12345', + repo='o/r', + context='pr', + team='agent-integrations', + reason='forged', + status_code='999', + rate_limit_resource='forged', + ) + monitor = GitHubMonitor(monitoring.component('github-async', integration='ntp'), now=lambda: NOW) + client = AsyncGitHubClient('token', rate_limiter=rate_limiter, transport=transport, observer=monitor) + return client, monitor, sink, handler + + +def samples(sink: RecordingSink, name: str) -> list[tuple[float, dict[str, str]]]: + return [(record.value, dict(record.tags)) for record in sink.records_named(name)] + + +async def test_every_attempt_is_counted_under_the_family_tags_only(monkeypatch: pytest.MonkeyPatch): + advance_clock_on_sleep(FakeClock(), monkeypatch) + budget = { + 'x-ratelimit-remaining': '4990', + 'x-ratelimit-limit': '5000', + 'x-ratelimit-reset': str(int(NOW) + 90), + 'x-ratelimit-resource': 'core', + } + responses = [httpx.Response(503), httpx.Response(200, json=RUN_PAYLOAD, headers=budget)] + client, _, sink, handler = monitored(httpx.MockTransport(lambda request: responses.pop(0))) + + await client.get_workflow_run('o', 'r', 42) + + failed = {**PIPELINE, 'http.status_code': '503'} + succeeded = {**PIPELINE, 'http.status_code': '200'} + assert samples(sink, 'requests.count') == [(1, failed), (1, succeeded)] + assert samples(sink, 'requests.fault') == [ + (1, {**failed, 'dispatcher.reason': 'server_error'}), + (0, {**succeeded, 'dispatcher.reason': 'none'}), + ] + assert samples(sink, 'requests.throttled') == [(0, failed), (0, succeeded)] + assert samples(sink, 'requests.retried') == [(0, PIPELINE), (1, PIPELINE)] + assert [tags for _, tags in samples(sink, 'requests.duration')] == [failed, succeeded] + # Only the headers the response sent: the 503 had none, and the 200 sent no `used`. + core = {**PIPELINE, 'github.rate_limit.resource': 'core'} + assert [(record.name, record.value, dict(record.tags)) for record in budget_gauges(sink)] == [ + ('requests.rate_limit.remaining', 4990, core), + ('requests.rate_limit.limit', 5000, core), + ('requests.rate_limit.reset_in', 90, core), + ] + [failure] = [event for event in handler.events if event['level'] == 'error'] + assert failure['event'] == 'GitHub request attempt failed' + assert (failure['endpoint'], failure['attempt'], failure['status_code']) == (RUN_PATH, 1, 503) + assert (failure['reason'], failure['terminal'], failure['integration']) == ('server_error', False, 'ntp') + [completed] = [event for event in handler.events if event['event'] == 'GitHub request completed'] + assert (completed['level'], completed['attempt'], completed['status_code']) == ('debug', 2, 200) + + +def budget_gauges(sink: RecordingSink) -> list[MetricRecord]: + return [record for record in sink.records if record.name.startswith('requests.rate_limit.')] + + +@pytest.mark.parametrize( + ('resource', 'tagged'), + [(None, 'unknown'), ('search', 'search')], + ids=['unnamed', 'another-resource'], +) +async def test_a_failed_response_gauges_its_own_budget_and_logs_no_header(resource: str | None, tagged: str): + """Budgets GitHub keeps apart stay apart, and an unparseable value is not reported as a number.""" + headers = { + 'x-ratelimit-remaining': '29', + 'x-ratelimit-used': '1', + 'x-ratelimit-limit': 'not-a-number', + 'x-ratelimit-reset': 'nan', + 'location': f'https://storage.example/a?signature={SIGNATURE}', + } + if resource is not None: + headers['x-ratelimit-resource'] = resource + client, _, sink, handler = monitored(httpx.MockTransport(lambda request: httpx.Response(404, headers=headers))) + + with pytest.raises(httpx.HTTPStatusError): + await client.get_workflow_run('o', 'r', 42, retry=NO_RETRY) + + budget = {**PIPELINE, 'github.rate_limit.resource': tagged} + assert [(record.name, record.value, dict(record.tags)) for record in budget_gauges(sink)] == [ + ('requests.rate_limit.remaining', 29, budget), + ('requests.rate_limit.used', 1, budget), + ] + [attempt, terminal] = handler.events + assert (attempt['method'], attempt['endpoint'], attempt['attempt']) == ('GET', RUN_PATH, 1) + assert (attempt['status_code'], attempt['reason'], terminal['status_code']) == (404, 'client_error', 404) + assert SIGNATURE not in json.dumps(handler.events, default=str) + + +def governed_limiter(monkeypatch: pytest.MonkeyPatch) -> InstrumentedAsyncLimiter: + clock = FakeClock() + advance_clock_on_sleep(clock, monkeypatch) + return InstrumentedAsyncLimiter(AsyncLimiter(5000, 3600), budget_governor=BudgetGovernor(now=clock)) + + +@pytest.mark.parametrize( + ('rejection', 'throttled', 'replayed'), + [ + pytest.param(httpx.Response(403, headers={'x-ratelimit-remaining': '0'}), 1, True, id='primary-limit'), + pytest.param(httpx.Response(403, headers={'retry-after': '5'}), 1, True, id='secondary-limit'), + # Rate limiting by its status alone, which the client does not wait out without headers. + pytest.param(httpx.Response(429), 1, False, id='unexplained-429'), + pytest.param(httpx.Response(403), 0, False, id='permission-denied'), + pytest.param(httpx.Response(500), 0, False, id='server-error'), + ], +) +async def test_only_an_attempt_github_rejects_for_rate_limiting_counts_as_throttled( + rejection: httpx.Response, throttled: int, replayed: bool, monkeypatch: pytest.MonkeyPatch +): + responses = [rejection, httpx.Response(200, json=RUN_PAYLOAD)] + client, _, sink, _ = monitored(httpx.MockTransport(lambda request: responses.pop(0)), governed_limiter(monkeypatch)) + + with suppress(httpx.HTTPStatusError): + await client.get_workflow_run('o', 'r', 42, retry=NO_RETRY) + + rejected = [(throttled, {**PIPELINE, 'http.status_code': str(rejection.status_code)})] + succeeded = [(0, {**PIPELINE, 'http.status_code': '200'})] if replayed else [] + assert samples(sink, 'requests.throttled') == rejected + succeeded + # Throttled or not, a rejection is still GitHub's fault. + assert samples(sink, 'requests.fault')[0][0] == 1 + + +def abandoning_limiter() -> InstrumentedAsyncLimiter: + governor = BudgetGovernor(max_wait_seconds=1) + governor.observe(BudgetSnapshot(retry_after=3600)) + return InstrumentedAsyncLimiter(AsyncLimiter(100, 1), budget_governor=governor) + + +@pytest.mark.parametrize( + ('transport', 'rate_limiter', 'error', 'sent'), + [ + pytest.param( + httpx.MockTransport(lambda request: httpx.Response(404)), None, httpx.HTTPStatusError, 1, id="not-found" + ), + pytest.param( + httpx.MockTransport(lambda request: httpx.Response(200)), + abandoning_limiter(), + RateLimitWaitAbandoned, + 0, + id="abandoned-before-send", + ), + ], +) +async def test_a_request_that_fails_for_good_is_logged_once_as_terminal( + transport: httpx.MockTransport, + rate_limiter: InstrumentedAsyncLimiter | None, + error: type[Exception], + sent: int, +): + client, _, sink, handler = monitored(transport, rate_limiter) + + with pytest.raises(error): + await client.get_workflow_run('o', 'r', 42) + + # One sample per send, and none for the terminal failure or for a wait that never sent. + assert len(sink.records_named('requests.count')) == len(sink.records_named('requests.throttled')) == sent + [terminal] = [event for event in handler.events if event.get('terminal')] + assert (terminal['level'], terminal['endpoint'], terminal['attempts']) == ('error', RUN_PATH, sent) + assert terminal['error'].startswith(error.__name__) + # Nothing was sent to fail, so an abandoned wait adds no HTTP reason and the log keeps the context's. + expected = ('client_error', 404) if sent else ('forged', '999') + assert (terminal['reason'], terminal['status_code']) == expected + + +@pytest.mark.parametrize( + ('error', 'fault', 'reason', 'logged'), + [ + pytest.param( + httpx.ConnectError('refused'), + 1, + 'transport', + [('error', 'GitHub request attempt failed'), ('error', 'GitHub request failed')], + id='transport-failure', + ), + # The caller's own cancellation was sent, but it is neither GitHub's fault nor a failure. + pytest.param(asyncio.CancelledError(), 0, 'cancelled', [('debug', 'GitHub request cancelled')], id='cancelled'), + ], +) +async def test_a_send_without_a_response_has_elapsed_time_but_no_status( + error: BaseException, fault: int, reason: str, logged: list[tuple[str, str]] +): + def unanswered(request: httpx.Request) -> httpx.Response: + raise error + + client, _, sink, handler = monitored(httpx.MockTransport(unanswered)) + + with pytest.raises(type(error)): + await client.get_workflow_run('o', 'r', 42, retry=NO_RETRY) + + assert samples(sink, 'requests.count') == [(1, PIPELINE)] + assert samples(sink, 'requests.fault') == [(fault, {**PIPELINE, 'dispatcher.reason': reason})] + assert samples(sink, 'requests.throttled') == [(0, PIPELINE)] + assert samples(sink, 'requests.retried') == [(0, PIPELINE)] + assert [tags for _, tags in samples(sink, 'requests.duration')] == [PIPELINE] + assert [(event['level'], event['event']) for event in handler.events] == logged + assert handler.events[0]['reason'] == reason + + +@pytest.mark.parametrize('outcome', [WaitOutcome.COMPLETED, WaitOutcome.ABANDONED, WaitOutcome.CANCELLED]) +def test_a_throttle_wait_reports_its_actual_time_and_keeps_the_requested_one_in_the_log(outcome: WaitOutcome): + _, monitor, sink, handler = monitored(httpx.MockTransport(lambda request: httpx.Response(200))) + + monitor.rate_limit_event( + WaitEvent(reason=WaitReason.EXHAUSTED, elapsed_seconds=12.5, outcome=outcome, requested_seconds=51.0) + ) + + assert samples(sink, 'throttle.wait.duration') == [(12.5, {**PIPELINE, 'dispatcher.reason': 'exhausted'})] + [event] = handler.events + assert (event['outcome'], event['elapsed_seconds'], event['requested_seconds']) == (outcome, 12.5, 51.0) diff --git a/ddev/tests/cli/ci/tests/test_task_test_runner.py b/ddev/tests/cli/ci/tests/test_task_test_runner.py index e4d278167316b..7096defe704e6 100644 --- a/ddev/tests/cli/ci/tests/test_task_test_runner.py +++ b/ddev/tests/cli/ci/tests/test_task_test_runner.py @@ -16,7 +16,7 @@ from pydantic import ValidationError from ddev.cli.ci.dispatch_run import ResolvedRun -from ddev.cli.ci.tests import messages +from ddev.cli.ci.tests import messages, task_test_runner from ddev.cli.ci.tests.dispatcher_attributes import run_fields from ddev.cli.ci.tests.messages import BatchFinished, BatchJob, TestBatch from ddev.cli.ci.tests.progress import ExecutionState @@ -234,6 +234,40 @@ async def test_healthy_attempts_report_zero_operation_failures(tmp_path: Path): assert [record.kind.value for record in sink.records_named('artifacts.download.duration')] == ['distribution'] +async def test_polling_intervals_are_measured_between_polls_of_the_same_batch( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +): + """A batch's first poll has no predecessor, even when an earlier batch was polled by the same runner. + + The samples are the runner's own, so they carry the run's metric dimensions and its component. + """ + fake = FakeAsyncGitHubClient() + mock_artifacts(fake, []) + fake.mock_response("get_workflow_run", make_workflow_run("completed", "success")) + ticks = iter(range(0, 1000, 10)) + monkeypatch.setattr(task_test_runner, "monotonic", lambda: float(next(ticks))) + monitoring, sink = recording_runtime() + monitoring.set_run_fields(ci_pipeline_id="12345", context="pr", team="agent-integrations") + runner = make_runner(fake, tmp_path, monitor=monitoring.component("test-runner")) + + for batch_id, polls in (("batch-1", 3), ("batch-2", 2)): + for _ in range(polls - 1): + fake.mock_response("get_workflow_run", make_workflow_run("in_progress"), once=True) + await runner.process_message(make_batch(batch_id)) + + intervals = sink.records_named("requests.polling_interval") + assert [record.value for record in intervals] == [10.0, 10.0, 10.0] + assert {tuple(sorted(record.tags.items())) for record in intervals} == { + ( + ("ci.pipeline.id", "12345"), + ("dispatcher.component", "test-runner"), + ("dispatcher.context", "pr"), + ("team", "agent-integrations"), + ) + } + assert {record.kind for record in intervals} == {MetricKind.DISTRIBUTION} + + @pytest.mark.asyncio async def test_an_accepted_dispatch_counts_the_batch_and_its_jobs(tmp_path: Path): fake = FakeAsyncGitHubClient() diff --git a/ddev/tests/conftest.py b/ddev/tests/conftest.py index 0f297aeae4a03..53b0e2176397d 100644 --- a/ddev/tests/conftest.py +++ b/ddev/tests/conftest.py @@ -92,6 +92,8 @@ async def fake_context( default_timeout: float = 30.0, transport: Any = None, logger: Any = None, + observer: Any = None, + rate_limiter: Any = None, ) -> AsyncIterator[FakeAsyncGitHubClient]: yield fake diff --git a/ddev/tests/utils/github_async/helpers.py b/ddev/tests/utils/github_async/helpers.py index 8f7885cedc59a..bf1aad384fcf6 100644 --- a/ddev/tests/utils/github_async/helpers.py +++ b/ddev/tests/utils/github_async/helpers.py @@ -16,6 +16,7 @@ from aiolimiter import AsyncLimiter from ddev.utils.github_async import AsyncGitHubClient, GitHubResponse +from ddev.utils.github_async.observer import RequestAttempt, RequestFailure from ddev.utils.rate_limiting import RATE_LIMIT_TIME_PERIOD, BudgetGovernor, InstrumentedAsyncLimiter from tests.helpers.clock import FakeClock from tests.utils.github_async.payloads import ( @@ -87,6 +88,7 @@ def governed_client( transport: httpx.MockTransport, on_event: Any = None, max_rate_limit_retries: int = 2, + observer: RecordingObserver | None = None, ) -> AsyncGitHubClient: """Client whose governor runs on *clock*, so retry waits are deterministic under a fake sleep.""" governor = BudgetGovernor(now=clock, on_event=on_event) @@ -97,10 +99,26 @@ def governed_client( name="github", ) return AsyncGitHubClient( - token=TOKEN, rate_limiter=limiter, transport=transport, max_rate_limit_retries=max_rate_limit_retries + token=TOKEN, + rate_limiter=limiter, + transport=transport, + max_rate_limit_retries=max_rate_limit_retries, + observer=observer, ) +@dataclasses.dataclass +class RecordingObserver: + attempts: list[RequestAttempt] = dataclasses.field(default_factory=list) + failures: list[RequestFailure] = dataclasses.field(default_factory=list) + + def attempt_finished(self, attempt: RequestAttempt) -> None: + self.attempts.append(attempt) + + def request_failed(self, failure: RequestFailure) -> None: + self.failures.append(failure) + + async def first_page(pages: AsyncIterator[GitHubResponse[Any]]) -> GitHubResponse[Any]: """Return the first page of a paginated endpoint (enough to exercise errors and headers).""" async for page in pages: diff --git a/ddev/tests/utils/github_async/test_download_artifact.py b/ddev/tests/utils/github_async/test_download_artifact.py index 9e8a1fb08e26b..dfc54bece9311 100644 --- a/ddev/tests/utils/github_async/test_download_artifact.py +++ b/ddev/tests/utils/github_async/test_download_artifact.py @@ -4,13 +4,15 @@ import logging import traceback +from pathlib import Path import httpx import pytest from ddev.utils.github_async import AsyncGitHubClient +from ddev.utils.github_async.observer import RequestFault from ddev.utils.github_errors import GitHubAuthenticationError -from tests.utils.github_async.helpers import TOKEN, make_client, make_zip, patch_signed_download +from tests.utils.github_async.helpers import TOKEN, RecordingObserver, make_client, make_zip, patch_signed_download pytestmark = pytest.mark.usefixtures("instant_backoff") @@ -35,14 +37,32 @@ def signed_handler(request: httpx.Request) -> httpx.Response: assert (tmp_path / "out" / "hello.txt").read_bytes() == b"hi" -async def test_download_artifact_non_302_raises(tmp_path) -> None: - def handler(request: httpx.Request) -> httpx.Response: - return httpx.Response(200, content=b"not a redirect") - - client = make_client(httpx.MockTransport(handler)) - with pytest.raises(httpx.HTTPError, match="Expected 302"): +@pytest.mark.parametrize( + ('status_code', 'headers', 'message'), + [ + pytest.param(200, {}, 'Expected 302', id='non-redirect'), + pytest.param(301, {'location': 'https://signed.example/zip'}, 'Expected 302', id='wrong-redirect'), + pytest.param(302, {}, 'Missing Location', id='missing-location'), + pytest.param(302, {'location': ''}, 'Missing Location', id='empty-location'), + ], +) +async def test_download_artifact_reports_invalid_redirect( + tmp_path: Path, status_code: int, headers: dict[str, str], message: str +): + observer = RecordingObserver() + client = AsyncGitHubClient( + token=TOKEN, + transport=httpx.MockTransport(lambda request: httpx.Response(status_code, headers=headers)), + observer=observer, + ) + with pytest.raises(httpx.HTTPError, match=message) as exc_info: await client.download_artifact("/repos/o/r/actions/artifacts/1/zip", tmp_path / "out") + assert [(attempt.number, attempt.fault) for attempt in observer.attempts] == [ + (1, RequestFault.UNEXPECTED_RESPONSE), + ] + assert observer.attempts[0].error is exc_info.value + async def test_download_artifact_authentication_error_remains_actionable(tmp_path) -> None: client = make_client(httpx.MockTransport(lambda request: httpx.Response(403))) @@ -69,15 +89,6 @@ def signed_handler(request: httpx.Request) -> httpx.Response: await client.download_artifact("/repos/o/r/actions/artifacts/1/zip", tmp_path / "out") -async def test_download_artifact_missing_location_header_raises(tmp_path) -> None: - def handler(request: httpx.Request) -> httpx.Response: - return httpx.Response(302) - - client = make_client(httpx.MockTransport(handler)) - with pytest.raises(httpx.HTTPError, match="Missing Location"): - await client.download_artifact("/repos/o/r/actions/artifacts/1/zip", tmp_path / "out") - - @pytest.mark.parametrize( "malicious_member", [ @@ -133,6 +144,36 @@ def signed_handler(request: httpx.Request) -> httpx.Response: assert (tmp_path / "out" / "hello.txt").read_bytes() == b"hi" +async def test_only_github_api_requests_are_reported_and_a_storage_retry_is_still_logged( + monkeypatch, tmp_path, caplog +) -> None: + """The storage fetch is not a GitHub API request, so it must not count as one. + + Its failure is not an attempt the observer was shown, so the retry it causes stays in the log. + """ + github_handler, signed_handler, _ = _signed_download( + httpx.Response(403, content=b"AccessDenied"), fail_every_attempt=False + ) + patch_signed_download(monkeypatch, signed_handler) + observer = RecordingObserver() + client = AsyncGitHubClient( + token=TOKEN, + transport=httpx.MockTransport(github_handler), + logger=logging.getLogger("test-client"), + observer=observer, + ) + + with caplog.at_level(logging.WARNING, logger="test-client"): + await client.download_artifact("/repos/o/r/actions/artifacts/1/zip", tmp_path / "out") + + assert [(attempt.number, attempt.fault) for attempt in observer.attempts] == [ + (1, RequestFault.NONE), + (2, RequestFault.NONE), + ] + assert observer.failures == [] + assert len([record for record in caplog.records if record.name == "test-client"]) == 1 + + async def test_a_denial_from_github_itself_is_not_retried_as_an_expired_url(tmp_path) -> None: """The 403 the artifact policy retries is the storage host's, not GitHub's. diff --git a/ddev/tests/utils/github_async/test_rate_limiting.py b/ddev/tests/utils/github_async/test_rate_limiting.py index cde4d9f2b4510..e8870a3e9d0de 100644 --- a/ddev/tests/utils/github_async/test_rate_limiting.py +++ b/ddev/tests/utils/github_async/test_rate_limiting.py @@ -2,6 +2,7 @@ from __future__ import annotations +import asyncio import dataclasses import logging @@ -10,9 +11,10 @@ from aiolimiter import AsyncLimiter from ddev.utils.github_async import AsyncGitHubClient, async_github_client -from ddev.utils.github_async.client import github_rate_limit_snapshot +from ddev.utils.github_async.client import RequestAttempts, github_rate_limit_snapshot from ddev.utils.github_async.defaults import default_github_rate_limiter, log_rate_limit_events from ddev.utils.github_async.models import WorkflowRun +from ddev.utils.github_async.observer import RequestFault from ddev.utils.github_errors import GitHubAuthenticationError from ddev.utils.rate_limiting import ( BucketEvent, @@ -28,6 +30,7 @@ from tests.helpers.clock import FakeClock, advance_clock_on_sleep from tests.utils.github_async.helpers import ( TOKEN, + RecordingObserver, governed_client, json_response, make_zip, @@ -36,7 +39,7 @@ ) from tests.utils.github_async.payloads import workflow_run_payload -LOGGER_NAME = "ddev.utils.github_async.defaults" +LOGGER_NAME = "test-rate-limit-events" async def test_client_request_with_rate_limiter_consumes_token() -> None: @@ -114,7 +117,7 @@ async def test_default_rate_limiter_is_constructed_and_observes_403() -> None: assert governor is not None with pytest.raises(httpx.HTTPStatusError): - await client._rate_limited_request("GET", "/x") + await client._rate_limited_request("GET", "/x", RequestAttempts()) # The 403's retry-after was observed (before raise_for_status), arming the shared pause; # exact pause arithmetic is covered by the clocked governor tests. @@ -130,7 +133,7 @@ async def test_retry_on_secondary_limit_returns_success(monkeypatch: pytest.Monk transport, calls = recording_transport([httpx.Response(403, headers={"retry-after": "5"}), httpx.Response(200)]) client = governed_client(clock, transport, on_event=events.append) - response = await client._rate_limited_request("GET", "/x") + response = await client._rate_limited_request("GET", "/x", RequestAttempts()) assert response.status_code == 200 assert len(calls) == 2 @@ -161,7 +164,7 @@ async def test_retry_on_secondary_limit_without_valid_wait_returns_success( transport, calls = recording_transport([rate_limited_response, httpx.Response(200)]) client = governed_client(clock, transport, on_event=events.append) - response = await client._rate_limited_request("GET", "/x") + response = await client._rate_limited_request("GET", "/x", RequestAttempts()) assert response.status_code == 200 assert len(calls) == 2 @@ -185,7 +188,7 @@ async def test_retry_on_primary_exhaustion_waits_until_reset(monkeypatch: pytest ) client = governed_client(clock, transport, on_event=events.append) - response = await client._rate_limited_request("GET", "/x") + response = await client._rate_limited_request("GET", "/x", RequestAttempts()) assert response.status_code == 200 assert len(calls) == 2 @@ -194,6 +197,107 @@ async def test_retry_on_primary_exhaustion_waits_until_reset(monkeypatch: pytest assert any(isinstance(e, PacingEvent) and e.reason is PacingReason.EXHAUSTED for e in events) +async def test_attempts_are_counted_across_both_retry_layers_and_timed_without_their_waits( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A rate-limit replay is as much a retry as a 5xx replay, and waiting for either is not latency.""" + clock = FakeClock() + advance_clock_on_sleep(clock, monkeypatch) + monkeypatch.setattr("ddev.utils.github_async.client.monotonic", clock) + responses = [ + httpx.Response(403, headers={"retry-after": "30"}), + httpx.Response(503), + json_response(workflow_run_payload()), + ] + + def handler(request: httpx.Request) -> httpx.Response: + clock.advance(2) + return responses.pop(0) + + observer = RecordingObserver() + client = governed_client(clock, httpx.MockTransport(handler), observer=observer) + + await client.get_workflow_run("o", "r", 42) + + assert [(attempt.number, attempt.fault, attempt.duration_seconds) for attempt in observer.attempts] == [ + (1, RequestFault.SECONDARY_RATE_LIMIT, 2), + (2, RequestFault.SERVER_ERROR, 2), + (3, RequestFault.NONE, 2), + ] + + +BLOCK = object() + + +@pytest.mark.parametrize( + ("responses", "expected"), + [ + pytest.param([BLOCK], [(1, RequestFault.CANCELLED)], id="in-flight"), + pytest.param( + [httpx.Response(503), BLOCK], + [(1, RequestFault.SERVER_ERROR), (2, RequestFault.CANCELLED)], + id="in-flight-replay", + ), + pytest.param( + [httpx.Response(403, headers={"retry-after": "30"}), BLOCK], + [(1, RequestFault.SECONDARY_RATE_LIMIT)], + id="during-rate-limit-wait", + ), + ], +) +async def test_a_cancelled_request_reports_only_the_sends_it_made_and_stays_cancelled( + monkeypatch: pytest.MonkeyPatch, responses: list[httpx.Response | object], expected: list[tuple[int, RequestFault]] +) -> None: + """A send cut off by the caller still reached the wire, but a wait it never finished sent nothing. + + The cancellation is the caller's, so it must propagate as-is rather than be retried or reported + as a failed request. + """ + clock = FakeClock() + real_sleep = asyncio.sleep + + async def sleep(delay: float) -> None: + clock.advance(delay) + # Unlike `advance_clock_on_sleep`, suspend, so a cancellation requested before the wait lands in it. + await real_sleep(0) + + monkeypatch.setattr(asyncio, "sleep", sleep) + monkeypatch.setattr("ddev.utils.github_async.client.monotonic", clock) + calls: list[httpx.Request] = [] + + def cancel_request() -> None: + task = asyncio.current_task() + assert task is not None + task.cancel() + + async def handler(request: httpx.Request) -> httpx.Response: + item = responses[len(calls)] + calls.append(request) + clock.advance(2) + if item is BLOCK: + cancel_request() + await asyncio.Event().wait() + assert isinstance(item, httpx.Response) + return item + + def on_event(event: RateLimitEvent) -> None: + if isinstance(event, PacingEvent) and event.reason is PacingReason.SECONDARY_LIMIT: + cancel_request() + + observer = RecordingObserver() + client = governed_client(clock, httpx.MockTransport(handler), on_event=on_event, observer=observer) + request = asyncio.create_task(client.get_workflow_run("o", "r", 42)) + + with pytest.raises(asyncio.CancelledError): + await request + + assert request.cancelled() + assert [(attempt.number, attempt.fault) for attempt in observer.attempts] == expected + assert {attempt.duration_seconds for attempt in observer.attempts} == {2} + assert len(calls) == len(expected) + assert observer.failures == [] + + @pytest.mark.parametrize("status_code", [401, 403]) async def test_authentication_error_is_actionable_and_not_retried(status_code: int) -> None: """Authentication failures are actionable and raise on the first attempt.""" @@ -201,7 +305,7 @@ async def test_authentication_error_is_actionable_and_not_retried(status_code: i client = AsyncGitHubClient(token=TOKEN, transport=transport) with pytest.raises(GitHubAuthenticationError) as exc_info: - await client._rate_limited_request("GET", "/x") + await client._rate_limited_request("GET", "/x", RequestAttempts()) assert len(calls) == 1 assert exc_info.value.response.status_code == status_code @@ -218,7 +322,7 @@ async def test_the_rate_limit_layer_does_not_retry_a_transport_error() -> None: client = AsyncGitHubClient(token=TOKEN, transport=transport) with pytest.raises(httpx.ConnectError): - await client._rate_limited_request("GET", "/x") + await client._rate_limited_request("GET", "/x", RequestAttempts()) assert len(calls) == 1 @@ -246,7 +350,7 @@ async def test_the_rate_limit_layer_stops_re_acquiring_when_it_is_out_of_attempt client.enter_shutdown_mode() with pytest.raises(httpx.HTTPStatusError): - await client._rate_limited_request("GET", "/x") + await client._rate_limited_request("GET", "/x", RequestAttempts()) assert len(calls) == expected_calls @@ -354,7 +458,7 @@ def test_log_rate_limit_events_level_mapping( caplog: pytest.LogCaptureFixture, event: RateLimitEvent, expected_level: int ) -> None: with caplog.at_level(logging.DEBUG, logger=LOGGER_NAME): - log_rate_limit_events()(event) + log_rate_limit_events(logging.getLogger(LOGGER_NAME))(event) assert caplog.records assert caplog.records[-1].levelno == expected_level @@ -368,6 +472,6 @@ class FutureEvent: type: str = "future" with caplog.at_level(logging.DEBUG, logger=LOGGER_NAME): - log_rate_limit_events()(FutureEvent()) # type: ignore[arg-type] + log_rate_limit_events(logging.getLogger(LOGGER_NAME))(FutureEvent()) # type: ignore[arg-type] assert caplog.records[-1].levelno == logging.DEBUG diff --git a/ddev/tests/utils/github_async/test_retry.py b/ddev/tests/utils/github_async/test_retry.py index 561bfedd80aac..b8b8a2e183d6c 100644 --- a/ddev/tests/utils/github_async/test_retry.py +++ b/ddev/tests/utils/github_async/test_retry.py @@ -15,6 +15,7 @@ import pytest from ddev.utils.github_async import AsyncGitHubClient +from ddev.utils.github_async.observer import RequestAttempt, RequestFailure, RequestFault from ddev.utils.github_async.retry import ( DEFAULT_ATTEMPTS, MUTATION_RETRY, @@ -26,7 +27,13 @@ on_transport_error, ) from ddev.utils.github_errors import GitHubUnexpectedRedirectError -from tests.utils.github_async.helpers import ENDPOINT_CALLS, TOKEN, json_response, recording_transport +from tests.utils.github_async.helpers import ( + ENDPOINT_CALLS, + TOKEN, + RecordingObserver, + json_response, + recording_transport, +) from tests.utils.github_async.payloads import ( full_pull_request_payload, issue_comment_payload, @@ -302,6 +309,69 @@ async def test_a_retry_is_reported_with_its_cause(caplog: pytest.LogCaptureFixtu assert records[0].attempt == 2 +async def test_an_observer_is_shown_every_send_and_the_log_does_not_repeat_it( + caplog: pytest.LogCaptureFixture, +) -> None: + """A monitor counts sends from these reports, so each one has to arrive once with its outcome.""" + transport, _ = recording_transport( + [httpx.Response(503), httpx.ConnectTimeout("slow"), json_response(workflow_run_payload())] + ) + observer = RecordingObserver() + logger = logging.getLogger("test-github-client") + client = AsyncGitHubClient(token=TOKEN, transport=transport, logger=logger, observer=observer) + + with caplog.at_level(logging.WARNING, logger="test-github-client"): + await client.get_workflow_run("o", "r", 42) + + assert [(attempt.number, attempt.fault) for attempt in observer.attempts] == [ + (1, RequestFault.SERVER_ERROR), + (2, RequestFault.TIMEOUT), + (3, RequestFault.NONE), + ] + assert {attempt.endpoint for attempt in observer.attempts} == {"/repos/o/r/actions/runs/42"} + assert observer.failures == [] + assert not [record for record in caplog.records if record.name == "test-github-client"] + + +@pytest.mark.parametrize( + ("responses", "expected_attempts"), + [ + pytest.param([httpx.Response(404)], 1, id="not-retryable"), + pytest.param([httpx.Response(503)], DEFAULT_ATTEMPTS, id="retries-exhausted"), + ], +) +async def test_a_request_that_fails_for_good_is_reported_once( + responses: list[httpx.Response], expected_attempts: int +) -> None: + transport, _ = recording_transport(responses) + observer = RecordingObserver() + client = AsyncGitHubClient(token=TOKEN, transport=transport, observer=observer) + + with pytest.raises(httpx.HTTPStatusError) as exc_info: + await client.get_workflow_run("o", "r", 42) + + assert len(observer.attempts) == expected_attempts + [failure] = observer.failures + assert failure.error is exc_info.value + assert failure.last_attempt is observer.attempts[-1] + + +async def test_a_failing_observer_does_not_fail_the_request() -> None: + """Monitoring is secondary to the request it watches.""" + + class BrokenObserver: + def attempt_finished(self, attempt: RequestAttempt) -> None: + raise RuntimeError("monitor down") + + def request_failed(self, failure: RequestFailure) -> None: + raise RuntimeError("monitor down") + + transport, _ = recording_transport([json_response(workflow_run_payload())]) + client = AsyncGitHubClient(token=TOKEN, transport=transport, observer=BrokenObserver()) + + assert (await client.get_workflow_run("o", "r", 42)).data.id == 42 + + # --------------------------------------------------------------------------- # Composing policies # --------------------------------------------------------------------------- diff --git a/ddev/tests/utils/test_rate_limiting.py b/ddev/tests/utils/test_rate_limiting.py index b6e4eef0896bf..29a3ed61a2109 100644 --- a/ddev/tests/utils/test_rate_limiting.py +++ b/ddev/tests/utils/test_rate_limiting.py @@ -3,6 +3,7 @@ from __future__ import annotations import asyncio +import contextlib import inspect import itertools from collections.abc import Callable @@ -25,6 +26,9 @@ RateLimitWaitAbandoned, RelaxedRateLimits, SecondaryLimitEvent, + WaitEvent, + WaitOutcome, + WaitReason, ) from tests.helpers.assertions import assert_blocks from tests.helpers.clock import FakeClock @@ -76,6 +80,17 @@ def governor(clock: FakeClock) -> BudgetGovernor: return BudgetGovernor(now=clock, reserve_fraction=0.15) +@pytest.fixture +def monotonic_clock(clock: FakeClock, monkeypatch: pytest.MonkeyPatch) -> FakeClock: + """Measure elapsed waits on the fake clock too.""" + monkeypatch.setattr("ddev.utils.rate_limiting.monotonic", clock) + return clock + + +def wait_events(events: list[RateLimitEvent]) -> list[WaitEvent]: + return [event for event in events if isinstance(event, WaitEvent)] + + @pytest.fixture def slept(clock: FakeClock, monkeypatch: pytest.MonkeyPatch) -> list[float]: """Monkeypatch asyncio.sleep to advance the fake clock and return the recorded sleep durations.""" @@ -488,7 +503,7 @@ async def test_two_sequential_paced_waits_are_spaced_by_one_interval( async def test_wait_extends_when_retry_after_observed_mid_wait( - clock: FakeClock, monkeypatch: pytest.MonkeyPatch + clock: FakeClock, monotonic_clock: FakeClock, monkeypatch: pytest.MonkeyPatch ) -> None: """A retry_after pause raised while a request is mid-wait must extend the wait (pause_until floor).""" events: list[RateLimitEvent] = [] @@ -516,6 +531,15 @@ async def fake_sleep(delay: float) -> None: assert clock.current - start == pytest.approx(interval + 50.0) pacing_events = [event for event in events if isinstance(event, PacingEvent)] assert pacing_events[-1] == PacingEvent(wait_seconds=pytest.approx(50.0), reason=PacingReason.SECONDARY_LIMIT) + # One wait, reported once for all of it, under the cause that held it last. + assert wait_events(events) == [ + WaitEvent( + reason=WaitReason.SECONDARY_LIMIT, + elapsed_seconds=pytest.approx(interval + 50.0), + outcome=WaitOutcome.COMPLETED, + requested_seconds=pytest.approx(interval), + ) + ] async def test_wait_raises_when_max_iterations_exceeded_without_hanging( @@ -549,7 +573,7 @@ async def test_wait_abandons_immediately_when_exhausted_window_exceeds_budget( async def test_wait_abandons_when_flood_pushes_floor_past_budget( - clock: FakeClock, monkeypatch: pytest.MonkeyPatch + clock: FakeClock, monotonic_clock: FakeClock, monkeypatch: pytest.MonkeyPatch ) -> None: """A secondary-limit extension that crosses the budget mid-wait must raise, not be slept out.""" events: list[RateLimitEvent] = [] @@ -577,6 +601,14 @@ async def fake_sleep(delay: float) -> None: assert exc_info.value.remaining_seconds == pytest.approx(100.0) abandoned = [e for e in events if isinstance(e, PacingEvent) and e.reason is PacingReason.ABANDONED] assert abandoned == [PacingEvent(wait_seconds=pytest.approx(100.0), reason=PacingReason.ABANDONED)] + assert wait_events(events) == [ + WaitEvent( + reason=WaitReason.RATIONING, + elapsed_seconds=pytest.approx(interval), + outcome=WaitOutcome.ABANDONED, + requested_seconds=pytest.approx(interval), + ) + ] async def test_wait_with_ample_budget_sleeps_identically_to_no_budget(monkeypatch: pytest.MonkeyPatch) -> None: @@ -662,6 +694,108 @@ async def test_wait_fires_pacing_event_with_reason( assert pacing_events == [PacingEvent(wait_seconds=pytest.approx(expected_wait_seconds), reason=expected_reason)] +@pytest.mark.parametrize( + ("arm", "expected_reason"), + [ + pytest.param(None, None, id="healthy"), + pytest.param(arm_rationing, WaitReason.RATIONING, id="rationing"), + pytest.param( + lambda clock, governor: governor.observe(make_snapshot(clock=clock, limit=100, remaining=0, reset_in=50)), + WaitReason.EXHAUSTED, + id="exhausted", + ), + ], +) +async def test_only_a_wait_that_blocked_is_reported_with_its_measured_time( + clock: FakeClock, + monotonic_clock: FakeClock, + slept: list[float], + arm: Callable[[FakeClock, BudgetGovernor], None] | None, + expected_reason: WaitReason | None, +) -> None: + events: list[RateLimitEvent] = [] + governor = BudgetGovernor(now=clock, buffer_seconds=1.0, on_event=events.append) + if arm is not None: + arm(clock, governor) + + await governor.wait() + + expected = ( + [] + if expected_reason is None + else [ + WaitEvent( + reason=expected_reason, + elapsed_seconds=pytest.approx(sum(slept)), + outcome=WaitOutcome.COMPLETED, + requested_seconds=pytest.approx(sum(slept)), + ) + ] + ) + assert wait_events(events) == expected + + +class FakeBucket: + """An empty bucket whose acquire takes *wait_seconds* on *clock*, then optionally is cancelled.""" + + def __init__(self, clock: FakeClock, wait_seconds: float, *, cancelled: bool) -> None: + self.clock = clock + self.wait_seconds = wait_seconds + self.cancelled = cancelled + + def has_capacity(self) -> bool: + return False + + async def __aenter__(self) -> None: + self.clock.advance(self.wait_seconds) + if self.cancelled: + raise asyncio.CancelledError + + +@pytest.mark.parametrize("cancel_during", [None, "governor", "bucket"]) +async def test_a_cancelled_wait_is_still_reported_for_the_time_it_blocked( + clock: FakeClock, monotonic_clock: FakeClock, monkeypatch: pytest.MonkeyPatch, cancel_during: str | None +) -> None: + """A wait cut short by cancellation was still time a request spent blocked.""" + events: list[RateLimitEvent] = [] + governor = BudgetGovernor(now=clock, buffer_seconds=1.0, on_event=events.append) + governor.observe(make_snapshot(retry_after=30.0)) + + async def fake_sleep(delay: float) -> None: + clock.advance(5.0) + if cancel_during == "governor": + raise asyncio.CancelledError + clock.advance(delay - 5.0) + + monkeypatch.setattr(asyncio, "sleep", fake_sleep) + limiter = InstrumentedAsyncLimiter( + FakeBucket(clock, 2.0, cancelled=cancel_during == "bucket"), # type: ignore[arg-type] + on_event=events.append, + budget_governor=governor, + name="github", + ) + + with pytest.raises(asyncio.CancelledError) if cancel_during else contextlib.nullcontext(): + await limiter.__aenter__() + + def outcome(stage: str) -> WaitOutcome: + return WaitOutcome.CANCELLED if cancel_during == stage else WaitOutcome.COMPLETED + + expected = [ + WaitEvent( + reason=WaitReason.SECONDARY_LIMIT, + elapsed_seconds=pytest.approx(5.0 if cancel_during == "governor" else 31.0), + outcome=outcome("governor"), + requested_seconds=pytest.approx(31.0), + ) + ] + if cancel_during != "governor": + expected.append( + WaitEvent(reason=WaitReason.BUCKET, elapsed_seconds=2.0, outcome=outcome("bucket"), name="github") + ) + assert wait_events(events) == expected + + # --------------------------------------------------------------------------- # InstrumentedAsyncLimiter + BudgetGovernor wiring # ---------------------------------------------------------------------------