"""Datadog DogStatsD metrics for the hardening kit, via the public ``datadog`` package. The client targets the agent through the standard ``DD_AGENT_HOST``/``DD_DOGSTATSD_PORT`` env vars (default ``localhost:8125``), so it reaches the agent whether it runs as a sidecar or a node-level daemon. Tags are key-TYPE only, never the key value, to avoid PII leakage and tag-cardinality blowups. Emission is best-effort over UDP (non-blocking); an unwired/unreachable agent just drops the datagrams. """ import logging import os from datadog.dogstatsd import DogStatsd logger = logging.getLogger(__name__) try: _statsd: DogStatsd | None = DogStatsd( host=os.environ.get('DD_AGENT_HOST', 'localhost'), port=int(os.environ.get('DD_DOGSTATSD_PORT', '8125')), ) except Exception: # pragma: no cover -- never let metrics setup break module import # A bad DD_DOGSTATSD_PORT (or any construct failure) disables ALL hardening # metrics service-wide, so leave a breadcrumb rather than failing silently. logger.warning('DogStatsD init failed; hardening metrics disabled.', exc_info=True) _statsd = None def _fmt(tags: dict[str, str]) -> list[str]: return [f'{k}:{v}' for k, v in tags.items()] def _count(metric: str, tags: dict[str, str]) -> None: if _statsd is None: return # Metrics are best-effort UDP: never break a request on an emit failure. try: _statsd.increment(metric, tags=_fmt(tags)) except Exception: logger.debug('statsd emit failed for %s', metric, exc_info=True) def _count_by(metric: str, value: int, tags: dict[str, str]) -> None: if _statsd is None: return # Metrics are best-effort UDP: never break a request on an emit failure. try: _statsd.increment(metric, value=value, tags=_fmt(tags)) except Exception: logger.debug('statsd emit failed for %s', metric, exc_info=True) def _gauge(metric: str, value: float, tags: dict[str, str]) -> None: if _statsd is None: return # Metrics are best-effort UDP: never break a request on an emit failure. try: _statsd.gauge(metric, value, tags=_fmt(tags)) except Exception: logger.debug('statsd emit failed for %s', metric, exc_info=True) def _distribution(metric: str, value: float, tags: dict[str, str]) -> None: if _statsd is None: return # Metrics are best-effort UDP: never break a request on an emit failure. try: _statsd.distribution(metric, value, tags=_fmt(tags)) except Exception: logger.debug('statsd emit failed for %s', metric, exc_info=True) def rate_limit_rejected(category: str, key_type: str) -> None: """Count a rate-limit rejection, tagged by limit category and the key TYPE (not value).""" _count('hardening.ratelimit.rejected', {'category': category, 'key_type': key_type}) def ratelimit_fallback(reason: str) -> None: """Count a rate-limit fallback to the in-memory store, tagged by reason. reason in {'breaker_open', 'redis_error', 'other_error', 'boot_error', 'fallback_error'} -- makes degradation (including the non-Redis-bug case that never trips the breaker) graphable without log-scraping. """ _count('hardening.ratelimit.fallback', {'reason': reason}) def body_too_large() -> None: """Count a request rejected for exceeding the body-size limit.""" _count('hardening.body_limit.rejected', {}) def request_body_bytes(size: int, endpoint: str) -> None: """Record the observed request body size as a distribution, tagged by the matched endpoint. A distribution (not a gauge) so percentiles are computed across all pods; the endpoint name is a bounded route identifier, never a principal key value, so it is cardinality- and PII-safe. """ _distribution('hardening.request.body_bytes', size, {'endpoint': endpoint}) def downstream_failure(resource: str) -> None: """Count a downstream failure, tagged by the resource name.""" _count('hardening.downstream.failure', {'resource': resource}) def dataloader_batch( entity_name: str, route: str, requested: int, unresolved: int ) -> None: """Record a batch /dataloader request's size and unresolved-id count, tagged by entity+route. Two instruments, each chosen for what it must answer: - ``requested_ids`` is a distribution so batch-size percentiles vs the OWS_BATCH_LIMIT cap are a real capacity signal, and its sum is the ratio denominator. - ``unresolved_ids`` is a counter incremented by N. A distribution would be wrong here: it is 0 on almost every request, so its percentiles carry no signal, and two distributions cannot be combined per batch. As a counter its sum gives the numerator, so the unresolved ratio -- sum(unresolved) / sum(requested) -- is graphable and alertable per entity without log-scraping. A ratio spike toward 1.0 is the abnormal 'resolver returned None for a healthy batch' case; a steady low ratio is the normal partial-resolution state. The ``route`` tag disambiguates endpoints that share an entity name (several contract dataloaders all pass ``entity='Contract'``) so per-endpoint signal is not blended. """ tags = {'entity': entity_name, 'route': route} _distribution('dataloader.requested_ids', requested, tags) _count_by('dataloader.unresolved_ids', unresolved, tags) def dataloader_key_field_mismatch(entity_name: str, route: str) -> None: """Count a /dataloader fetch whose records carried no usable key_field value. A durable signal for the copy-paste / field-rename bug that would otherwise ship a silent all-null 200 with only an error log; alert on any nonzero rate per entity+route. """ _count('dataloader.key_field_mismatch', {'entity': entity_name, 'route': route}) def on_breaker_state(name: str, old: str, new: str) -> None: """Emit per-state breaker gauges on a transition: clear the old state, set the new.""" # Per-state 0/1 gauge so "how many pods are open" is answerable; UDP gauge is non-blocking. _gauge('hardening.breaker.state', 0, {'resource': name, 'state': old}) _gauge('hardening.breaker.state', 1, {'resource': name, 'state': new})