# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/beneath-what-cardinality-and-tail-sampling-bounds-must-an-instance-constrain-telemetry-emission-to-prevent-metric-explosion-and-observability-induced-outages.md
# Remnant Warrant:    https://corpusfetch.com/c/beneath-what-cardinality-and-tail-sampling-bounds-must-an-instance-constrain-telemetry-emission-to-prevent-metric-explosion-and-observability-induced-outages.warrant.json
# Markdown SHA-256:   bf81c1c1b9a571f4b54578cc42c3885787e716ac759a599e0c83a13335e0dfdb
# Code AST SHA-256:   e733edbab418e7ebfd96e0cf0192af5bfbabd0909f8b2067873820154525b2d1
# Verified Classes:   CardinalitySafeCounter, RateLimitedTelemetryFilter
# Defensive Guards:   2 fail-closed invariants (zero external deps)

import re
import time
from collections import Counter
from dataclasses import dataclass, field
from typing import Mapping

# Replaces UUIDs, hex hashes, and numeric path segments with ":id"
_DYNAMIC_SEGMENT_RE = re.compile(
    r"/(?:[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}|[0-9a-fA-F]{16,}|\d+)(?=/|$)"
)

_FORBIDDEN_LABEL_KEYS = frozenset({
    "user_id", "userid", "request_id", "trace_id", "span_id",
    "email", "ip", "client_ip", "session_id", "url", "raw_path", "error_message",
})


def normalize_route_for_metric(raw_path: str) -> str:
    '''
    Collapses dynamic numeric IDs, UUIDs, and hex tokens in URL paths into
    a bounded template (`/users/:id/orders/:id`) safe for metric labels.
    '''
    path_only = raw_path.split("?", 1)[0].strip() or "/"
    normalized = _DYNAMIC_SEGMENT_RE.sub("/:id", path_only)
    return normalized[:128]


@dataclass
class CardinalitySafeCounter:
    '''
    A metric counter that enforces:
    1. A strict allowlist of label keys (rejecting user_id, request_id, etc.).
    2. A hard ceiling on total unique label-set combinations (`max_series`).
    3. An automatic overflow bucket (`__overflow__`) so cardinality spikes never OOM.
    '''
    name: str
    allowed_label_keys: tuple[str, ...]
    max_series: int = 500
    _series: Counter[tuple[tuple[str, str], ...]] = field(default_factory=Counter)
    overflow_drops: int = 0

    def __post_init__(self) -> None:
        forbidden = set(self.allowed_label_keys) & _FORBIDDEN_LABEL_KEYS
        if forbidden:
            raise ValueError(
                f"Metric {self.name!r} configured with forbidden high-cardinality label keys: {sorted(forbidden)}"
            )

    def inc(self, labels: Mapping[str, str], amount: int = 1) -> None:
        if set(labels.keys()) != set(self.allowed_label_keys):
            raise ValueError(
                f"Metric {self.name!r} expects exact label keys {self.allowed_label_keys}, "
                f"got {tuple(labels.keys())}"
            )

        canonical_key = tuple(
            (k, str(labels[k])[:64]) for k in self.allowed_label_keys
        )

        # Gate: Cap total unique time series in memory; route excess to a single overflow series
        if canonical_key not in self._series and len(self._series) >= self.max_series:
            self.overflow_drops += amount
            overflow_key = tuple((k, "__overflow__") for k in self.allowed_label_keys)
            self._series[overflow_key] += amount
            return

        self._series[canonical_key] += amount


class RateLimitedTelemetryFilter:
    '''
    Prevents error-storm log floods and tail-samples trace spans:
    - Always keeps rare errors up to `burst_per_window` per signature, then suppresses.
    - Samples routine fast 2xx operations at `routine_sample_rate`.
    '''

    def __init__(
        self,
        *,
        window_sec: float = 10.0,
        burst_per_window: int = 5,
        routine_sample_every_n: int = 100,
    ) -> None:
        self.window_sec = window_sec
        self.burst_per_window = burst_per_window
        self.routine_sample_every_n = max(1, routine_sample_every_n)
        self._window_start = time.monotonic()
        self._counts: Counter[str] = Counter()
        self._suppressed: Counter[str] = Counter()
        self._routine_counter = 0

    def should_emit_error_log(self, error_signature: str) -> tuple[bool, int]:
        '''
        Returns `(should_log, suppressed_count_in_previous_window)`.
        Bounds identical error logs to `burst_per_window` every `window_sec`.
        '''
        now = time.monotonic()
        if now - self._window_start >= self.window_sec:
            self._window_start = now
            self._counts.clear()
            self._suppressed.clear()

        self._counts[error_signature] += 1
        if self._counts[error_signature] <= self.burst_per_window:
            flushed_suppressed = self._suppressed.pop(error_signature, 0)
            return True, flushed_suppressed

        self._suppressed[error_signature] += 1
        return False, 0

    def should_record_trace_span(self, *, is_error: bool, latency_ms: float, slow_threshold_ms: float = 500.0) -> bool:
        '''
        Tail-sampling rule: retain 100% of error or high-latency spans (up to rate limit),
        and sample 1-in-N routine fast spans.
        '''
        if is_error or latency_ms >= slow_threshold_ms:
            allowed, _ = self.should_emit_error_log(f"span:{is_error}:{int(latency_ms // 100)}")
            return allowed

        self._routine_counter += 1
        return (self._routine_counter % self.routine_sample_every_n) == 0