AI-7163 Monitor Dispatcher GitHub requests and throttling

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.

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:

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)

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(

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:

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

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
+            )

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,

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:

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,