"""Prometheus metrics for MCP Hangar. Production-grade metrics following Prometheus/OpenMetrics best practices: - Consistent naming: mcp_hangar___ - Proper label cardinality control - Thread-safe implementations - Standard histogram buckets for different use cases """ from collections import defaultdict from dataclasses import dataclass, field from typing import Any import threading import time # ============================================================================= # Core Metric Types # ============================================================================= @dataclass class MetricSample: """Single metric sample with labels.""" value: float labels: dict[str, str] = field(default_factory=dict) class Counter: """ Prometheus counter - monotonically increasing value. Use for: requests, errors, completions, bytes transferred. """ def __init__(self, name: str, description: str, labels: list[str] | None = None): self.name = name self.description = description self.label_names = labels or [] self._values: dict[tuple, float] = defaultdict(float) self._created: dict[tuple, float] = {} self._lock = threading.Lock() def inc(self, value: float = 1.0, **labels) -> None: """Increment counter by value (must be >= 0).""" if value < 0: raise ValueError("Counter can only increase") key = self._make_key(labels) with self._lock: if key not in self._created: self._created[key] = time.time() self._values[key] += value def _make_key(self, labels: dict) -> tuple: return tuple(labels.get(label_name, "") for label_name in self.label_names) def labels(self, **label_values) -> "_LabeledCounter": """Return counter with preset labels for reuse.""" return _LabeledCounter(self, label_values) def collect(self) -> list[MetricSample]: """Collect all samples.""" with self._lock: return [ MetricSample(value=v, labels=dict(zip(self.label_names, k, strict=False))) for k, v in self._values.items() ] class Gauge: """ Prometheus gauge - value that can go up and down. Use for: in-progress operations, current state, temperature, queue size. """ def __init__(self, name: str, description: str, labels: list[str] | None = None): self.name = name self.description = description self.label_names = labels or [] self._values: dict[tuple, float] = {} self._lock = threading.Lock() def set(self, value: float, **labels) -> None: """Set gauge to value.""" key = self._make_key(labels) with self._lock: self._values[key] = value def inc(self, value: float = 1.0, **labels) -> None: """Increment gauge.""" key = self._make_key(labels) with self._lock: self._values[key] = self._values.get(key, 0) + value def dec(self, value: float = 1.0, **labels) -> None: """Decrement gauge.""" key = self._make_key(labels) with self._lock: self._values[key] = self._values.get(key, 0) - value def set_to_current_time(self, **labels) -> None: """Set gauge to current Unix timestamp.""" self.set(time.time(), **labels) def set_max(self, value: float, **labels) -> None: """Set gauge to value unless it already holds a larger one. For a timestamp written from events: two can be handled out of the order they happened in, and the older one must not move the gauge back. """ key = self._make_key(labels) with self._lock: self._values[key] = max(value, self._values.get(key, value)) def remove(self, **labels) -> None: """Drop every series whose labels include ``labels``. For a subject that no longer exists. A gauge left behind reads as a live value forever: a deleted server stuck at `state == 4`. """ with self._lock: for key in list(self._values): current = dict(zip(self.label_names, key, strict=False)) if all(current.get(name) == value for name, value in labels.items()): del self._values[key] def _make_key(self, labels: dict) -> tuple: return tuple(labels.get(label_name, "") for label_name in self.label_names) def labels(self, **label_values) -> "_LabeledGauge": """Return gauge with preset labels.""" return _LabeledGauge(self, label_values) def collect(self) -> list[MetricSample]: """Collect all samples.""" with self._lock: return [ MetricSample(value=v, labels=dict(zip(self.label_names, k, strict=False))) for k, v in self._values.items() ] class Histogram: """ Prometheus histogram - distribution of values in buckets. Use for: request latencies, response sizes. """ # Standard bucket presets DEFAULT_BUCKETS = (0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0) LATENCY_BUCKETS = ( 0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0, ) SIZE_BUCKETS = (100, 1000, 10000, 100000, 1000000, 10000000) def __init__( self, name: str, description: str, labels: list[str] | None = None, buckets: tuple[Any, ...] | None = None, ): self.name = name self.description = description self.label_names = labels or [] self.buckets = tuple(sorted(buckets or self.DEFAULT_BUCKETS)) + (float("inf"),) self._lock = threading.Lock() self._buckets: dict[tuple, dict[float, int]] = defaultdict(lambda: dict.fromkeys(self.buckets, 0)) self._sums: dict[tuple, float] = defaultdict(float) self._counts: dict[tuple, int] = defaultdict(int) def observe(self, value: float, **labels) -> None: """Record an observation.""" key = self._make_key(labels) with self._lock: self._sums[key] += value self._counts[key] += 1 # Add to the first bucket that fits (buckets are sorted) for bucket in self.buckets: if value <= bucket: self._buckets[key][bucket] += 1 break # Only add to the first matching bucket def _make_key(self, labels: dict) -> tuple: return tuple(labels.get(label_name, "") for label_name in self.label_names) def labels(self, **label_values) -> "_LabeledHistogram": """Return histogram with preset labels.""" return _LabeledHistogram(self, label_values) def time(self) -> "_Timer": """Context manager for timing code blocks.""" return _Timer(self, {}) def collect(self) -> tuple: """Collect buckets, sum, and count samples.""" buckets = [] sums = [] counts = [] with self._lock: for key, bucket_values in self._buckets.items(): base_labels = dict(zip(self.label_names, key, strict=False)) cumulative = 0 for bucket in self.buckets: cumulative += bucket_values.get(bucket, 0) le = "+Inf" if bucket == float("inf") else str(bucket) buckets.append(MetricSample(value=cumulative, labels={**base_labels, "le": le})) sums.append(MetricSample(value=self._sums[key], labels=base_labels)) counts.append(MetricSample(value=self._counts[key], labels=base_labels)) return buckets, sums, counts class Summary: """ Prometheus summary - streaming quantiles. Simpler implementation using min/max/avg for now. Use for: streaming data where quantiles aren't critical. """ def __init__(self, name: str, description: str, labels: list[str] | None = None): self.name = name self.description = description self.label_names = labels or [] self._lock = threading.Lock() self._sums: dict[tuple, float] = defaultdict(float) self._counts: dict[tuple, int] = defaultdict(int) def observe(self, value: float, **labels) -> None: """Record an observation.""" key = self._make_key(labels) with self._lock: self._sums[key] += value self._counts[key] += 1 def _make_key(self, labels: dict) -> tuple: return tuple(labels.get(label_name, "") for label_name in self.label_names) def collect(self) -> tuple: """Collect sum and count samples.""" sums = [] counts = [] with self._lock: for key in self._sums: base_labels = dict(zip(self.label_names, key, strict=False)) sums.append(MetricSample(value=self._sums[key], labels=base_labels)) counts.append(MetricSample(value=self._counts[key], labels=base_labels)) return sums, counts class Info: """ Prometheus info metric - static key-value pairs. Use for: version info, build metadata, configuration. """ def __init__(self, name: str, description: str): self.name = name self.description = description self._labels: dict[str, str] = {} self._lock = threading.Lock() def info(self, **labels) -> None: """Set info labels.""" with self._lock: self._labels = {k: str(v) for k, v in labels.items()} def collect(self) -> list[MetricSample]: """Collect info sample.""" with self._lock: if self._labels: return [MetricSample(value=1.0, labels=self._labels)] return [] # ============================================================================= # Labeled Metric Helpers # ============================================================================= class _LabeledCounter: """Counter with preset labels.""" def __init__(self, counter: Counter, labels: dict): self._counter = counter self._labels = labels def inc(self, value: float = 1.0) -> None: self._counter.inc(value, **self._labels) class _LabeledGauge: """Gauge with preset labels.""" def __init__(self, gauge: Gauge, labels: dict): self._gauge = gauge self._labels = labels def set(self, value: float) -> None: self._gauge.set(value, **self._labels) def inc(self, value: float = 1.0) -> None: self._gauge.inc(value, **self._labels) def dec(self, value: float = 1.0) -> None: self._gauge.dec(value, **self._labels) class _LabeledHistogram: """Histogram with preset labels.""" def __init__(self, histogram: Histogram, labels: dict): self._histogram = histogram self._labels = labels def observe(self, value: float) -> None: self._histogram.observe(value, **self._labels) def time(self) -> "_Timer": return _Timer(self._histogram, self._labels) class _Timer: """Context manager for timing operations.""" def __init__(self, histogram: Histogram, labels: dict): self._histogram = histogram self._labels = labels self._start: float | None = None def __enter__(self) -> "_Timer": self._start = time.perf_counter() return self def __exit__(self, *args) -> None: assert self._start is not None duration = time.perf_counter() - self._start self._histogram.observe(duration, **self._labels) # ============================================================================= # Metrics Registry # ============================================================================= class CollectorRegistry: """Central registry for all metrics with Prometheus exposition format output.""" def __init__(self): self._collectors: dict[str, Any] = {} self._lock = threading.Lock() def register(self, collector) -> None: """Register a metric collector.""" with self._lock: if collector.name in self._collectors: raise ValueError(f"Metric {collector.name} already registered") self._collectors[collector.name] = collector def unregister(self, name: str) -> None: """Unregister a metric.""" with self._lock: self._collectors.pop(name, None) def get(self, name: str): """Get collector by name.""" return self._collectors.get(name) def get_metrics_output(self) -> str: return self.collect() def collect(self) -> str: """Generate Prometheus exposition format output.""" lines = [] with self._lock: collectors = list(self._collectors.items()) for name, collector in collectors: lines.extend(self._format_metric(name, collector)) lines.append("") return "\n".join(lines) @staticmethod def _family_name(name: str, collector) -> str: """The name a collector's samples are exposed under. `# HELP` and `# TYPE` must name this family, not the declared name: a counter declared as `x` is sampled as `x_total` and an info metric as `x_info`, which is also how `prometheus_client` writes their headers. Headers carrying the bare name made Prometheus file every counter's type and help text under a family with no samples, leaving the series anyone queries untyped (#1260). """ if isinstance(collector, Counter): return f"{name}_total" if isinstance(collector, Info): return f"{name}_info" return name def _format_metric(self, name: str, collector) -> list[str]: """Format a single metric in Prometheus text format 0.0.4.""" family = self._family_name(name, collector) lines = [] lines.append(f"# HELP {family} {collector.description}") if isinstance(collector, Counter): lines.append(f"# TYPE {family} counter") for sample in collector.collect(): labels = self._format_labels(sample.labels) lines.append(f"{family}{labels} {sample.value}") elif isinstance(collector, Gauge): lines.append(f"# TYPE {name} gauge") for sample in collector.collect(): labels = self._format_labels(sample.labels) lines.append(f"{name}{labels} {sample.value}") elif isinstance(collector, Histogram): lines.append(f"# TYPE {name} histogram") buckets, sums, counts = collector.collect() for sample in buckets: labels = self._format_labels(sample.labels) lines.append(f"{name}_bucket{labels} {int(sample.value)}") for sample in sums: labels = self._format_labels(sample.labels) lines.append(f"{name}_sum{labels} {sample.value}") for sample in counts: labels = self._format_labels(sample.labels) lines.append(f"{name}_count{labels} {int(sample.value)}") elif isinstance(collector, Summary): lines.append(f"# TYPE {name} summary") sums, counts = collector.collect() for sample in sums: labels = self._format_labels(sample.labels) lines.append(f"{name}_sum{labels} {sample.value}") for sample in counts: labels = self._format_labels(sample.labels) lines.append(f"{name}_count{labels} {int(sample.value)}") elif isinstance(collector, Info): lines.append(f"# TYPE {family} gauge") for sample in collector.collect(): labels = self._format_labels(sample.labels) lines.append(f"{family}{labels} 1") return lines def _format_labels(self, labels: dict[str, str]) -> str: """Format labels in Prometheus format.""" if not labels: return "" # Escape label values properly escaped = [] for k, v in sorted(labels.items()): if v is None: v = "" v = str(v).replace("\\", "\\\\").replace('"', '\\"').replace("\n", "\\n") escaped.append(f'{k}="{v}"') return "{" + ",".join(escaped) + "}" # ============================================================================= # Global Registry # ============================================================================= REGISTRY = CollectorRegistry() # ============================================================================= # MCP Hangar Metrics - Following Best Practices # ============================================================================= # ----------------------------------------------------------------------------- # Build/Version Info # ----------------------------------------------------------------------------- BUILD_INFO = Info( name="mcp_hangar_build", description="Build and version information for MCP Hangar", ) # ----------------------------------------------------------------------------- # Process Metrics # ----------------------------------------------------------------------------- PROCESS_START_TIME = Gauge( name="mcp_hangar_process_start_time_seconds", description="Unix timestamp of process start time", ) # ----------------------------------------------------------------------------- # McpServer Lifecycle Metrics # ----------------------------------------------------------------------------- PROVIDER_INFO = Gauge( name="mcp_hangar_mcp_server_info", description="McpServer configuration info (always 1, labels contain metadata)", labels=["mcp_server", "mode"], ) # 0 is COLD: not running, and not failing -- never started, stopped, or reaped # for being idle. A server Hangar gave up on is 4 and stays 4 until an explicit # start or a call; health checks do not probe it (#1361). A server the recovery # saga gave up on used to read 0. PROVIDER_STATE_CURRENT = Gauge( name="mcp_hangar_mcp_server_state", description="Current mcp_server state (0=cold, 1=initializing, 2=ready, 3=degraded, 4=dead)", labels=["mcp_server"], ) PROVIDER_UP = Gauge( name="mcp_hangar_mcp_server_up", description="Whether mcp_server is up and ready (1=up, 0=down)", labels=["mcp_server"], ) PROVIDER_INITIALIZED = Gauge( name="mcp_hangar_mcp_server_initialized", description="Whether mcp_server has been initialized at least once (1=yes, 0=no/cold)", labels=["mcp_server"], ) PROVIDER_LAST_STATE_CHANGE_SECONDS = Gauge( name="mcp_hangar_mcp_server_last_state_change_timestamp_seconds", description="Unix timestamp of last mcp_server state change", labels=["mcp_server"], ) # When Hangar last saw the server working (#1359). Written from three events: # a passing health check, a completed start, a successful tool call. Nothing # clears it -- not going cold, not being given up on -- and it never moves back. # Absent until the first of the three. # # A cold server is not probed, so the value ages while it is cold, and the bare # `time() - ... > N` also fires for a server reaped for being idle more than N # ago. Leave cold servers out: # # time() - mcp_hangar_mcp_server_last_healthy_timestamp_seconds > 900 # unless mcp_hangar_mcp_server_state == 0 # # Default matching, on every label. `on(mcp_server)` drops `instance`, so with # more than one replica a server cold on one replica would hide it dead on # another. # # That catches a server given up on (4), one still being retried (3) and one # stuck starting (1). A server that was never healthy has no series: catch the # give-up itself with `mcp_hangar_mcp_server_state == 4`. PROVIDER_LAST_HEALTHY_SECONDS = Gauge( name="mcp_hangar_mcp_server_last_healthy_timestamp_seconds", description=( "Unix timestamp of the last passing health check, completed start or successful tool call; " "kept when the mcp_server goes cold or dead" ), labels=["mcp_server"], ) PROVIDER_STARTS_TOTAL = Counter( name="mcp_hangar_mcp_server_starts", description="Total number of mcp_server start attempts", labels=["mcp_server", "result"], # result: success, failure ) #: Every value of the `reason` label on `mcp_hangar_mcp_server_stops_total` #: (#1360). A closed set, so a rule such as `reason!="idle"` is written against #: a known list, and the metric's HELP line carries it: #: #: - `idle`: the GC stopped a server unused past its idle TTL. #: - `shutdown`: Hangar stopped the server itself: a reload, unload or delete, a #: group's stop_all, process exit. #: - `user_request`: hangar_stop, or the REST stop without a reason. #: - `manual`: the REST stop with an empty reason, or one not in this list. #: - `failback`, `compensation`: the failover saga stopped a backup. #: - `detection_enforcement:block`: a detection rule, or the REST block, stopped it. #: - `max_retries_exceeded`: the recovery saga gave up on the server, which is #: now `dead` (`STOPPED_BY_GIVING_UP` in `domain.events`). #: #: A stop is counted once, from its `McpServerStopped`, whose reason is the one #: the stop was made for (#1466). MCP_SERVER_STOP_REASONS = ( "idle", "shutdown", "user_request", "manual", "failback", "compensation", "detection_enforcement:block", "max_retries_exceeded", ) #: What a reason outside `MCP_SERVER_STOP_REASONS` is counted as. MCP_SERVER_STOP_REASON_OTHER = "manual" PROVIDER_STOPS_TOTAL = Counter( name="mcp_hangar_mcp_server_stops", description="Total number of mcp_server stops, by reason: " + ", ".join(MCP_SERVER_STOP_REASONS), labels=["mcp_server", "reason"], ) PROVIDER_COLD_START_SECONDS = Histogram( name="mcp_hangar_mcp_server_cold_start_seconds", description="Time from cold start to ready state (critical UX metric)", labels=["mcp_server", "mode"], buckets=(0.1, 0.25, 0.5, 1.0, 2.0, 3.0, 5.0, 10.0, 15.0, 30.0, 60.0), ) PROVIDER_COLD_START_IN_PROGRESS = Gauge( name="mcp_hangar_mcp_server_cold_start_in_progress", description="Number of mcp_servers currently in cold start", labels=["mcp_server"], ) # ----------------------------------------------------------------------------- # Tool Invocation Metrics (RED method: Rate, Errors, Duration) # ----------------------------------------------------------------------------- TOOL_CALLS_TOTAL = Counter( name="mcp_hangar_tool_calls", description="Total number of tool calls", labels=["mcp_server", "tool", "status"], # status: success, error ) TOOL_CALL_DURATION_SECONDS = Histogram( name="mcp_hangar_tool_call_duration_seconds", description="Duration of tool calls in seconds", labels=["mcp_server", "tool"], buckets=Histogram.LATENCY_BUCKETS, ) TOOL_CALL_ERRORS_TOTAL = Counter( name="mcp_hangar_tool_call_errors", description="Total number of tool call errors by error type", labels=["mcp_server", "tool", "error_type"], ) # ----------------------------------------------------------------------------- # Health Check Metrics # ----------------------------------------------------------------------------- HEALTH_CHECK_TOTAL = Counter( name="mcp_hangar_health_checks", description="Total number of health check executions", labels=["mcp_server", "result"], # result: cold, healthy, unhealthy ) HEALTH_CHECK_DURATION_SECONDS = Histogram( name="mcp_hangar_health_check_duration_seconds", description="Duration of health checks in seconds", labels=["mcp_server"], buckets=(0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0), ) HEALTH_CHECK_CONSECUTIVE_FAILURES = Gauge( name="mcp_hangar_health_check_consecutive_failures", description="Number of consecutive health check failures", labels=["mcp_server"], ) # ----------------------------------------------------------------------------- # Connection Pool Metrics # ----------------------------------------------------------------------------- CONNECTIONS_ACTIVE = Gauge( name="mcp_hangar_connections_active", description="Whether a client connection to the mcp_server is currently open (1/0)", labels=["mcp_server"], ) # CONNECTIONS_TOTAL and CONNECTION_DURATION_SECONDS were removed by the # observability audit: never emitted, no dashboard/alert referenced them, and # they duplicated the server-lifecycle signals (starts_total, cold_start_seconds, # server lifetime). connections_active is kept and now wired (a provider-details # panel uses it). # ----------------------------------------------------------------------------- # Message Metrics # ----------------------------------------------------------------------------- MESSAGES_SENT_TOTAL = Counter( name="mcp_hangar_messages_sent", description="Total number of JSON-RPC messages sent", labels=["mcp_server", "method"], ) MESSAGES_RECEIVED_TOTAL = Counter( name="mcp_hangar_messages_received", description="Total number of JSON-RPC messages received", labels=["mcp_server", "type"], # values: response, notification, error ) MESSAGE_SIZE_BYTES = Histogram( name="mcp_hangar_message_size_bytes", description="Size of JSON-RPC messages in bytes", labels=["mcp_server", "direction"], # direction: sent, received buckets=Histogram.SIZE_BUCKETS, ) # ----------------------------------------------------------------------------- # GC (Garbage Collection) Metrics # ----------------------------------------------------------------------------- GC_CYCLES_TOTAL = Counter( name="mcp_hangar_gc_cycles", description="Total number of garbage collection cycles", ) GC_CYCLE_DURATION_SECONDS = Histogram( name="mcp_hangar_gc_cycle_duration_seconds", description="Duration of garbage collection cycles in seconds", buckets=(0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5), ) GC_PROVIDERS_COLLECTED_TOTAL = Counter( name="mcp_hangar_gc_mcp_servers_collected", description="Total number of mcp_servers collected by GC", labels=["reason"], # reason: idle, dead, error ) # ----------------------------------------------------------------------------- # Error Metrics # ----------------------------------------------------------------------------- ERRORS_TOTAL = Counter( name="mcp_hangar_errors", description="Total number of errors by type and component", labels=["component", "error_type"], # component: mcp_server, tool, health, gc, server ) # ----------------------------------------------------------------------------- # Rate Limiter Metrics # ----------------------------------------------------------------------------- RATE_LIMIT_HITS_TOTAL = Counter( name="mcp_hangar_rate_limit_hits", description="Total number of rate limit decisions by result (allowed or rejected)", labels=["result"], # result: allowed, rejected ) RATE_LIMIT_ACTIVE_BUCKETS = Gauge( name="mcp_hangar_rate_limit_active_buckets", description="Number of active rate limit token buckets", labels=[], ) # ----------------------------------------------------------------------------- # Discovery Metrics # ----------------------------------------------------------------------------- DISCOVERY_SOURCES_TOTAL = Gauge( name="mcp_hangar_discovery_sources", description="Number of configured discovery sources", labels=["source_type", "mode"], ) DISCOVERY_SOURCES_HEALTHY = Gauge( name="mcp_hangar_discovery_sources_healthy", description="Whether discovery source is healthy (1=healthy, 0=unhealthy)", labels=["source_type"], ) DISCOVERY_PROVIDERS_TOTAL = Gauge( name="mcp_hangar_discovery_mcp_servers", description="Number of discovered mcp_servers", labels=["source_type", "status"], # status: discovered, registered, quarantined ) DISCOVERY_CYCLES_TOTAL = Counter( name="mcp_hangar_discovery_cycles", description="Total number of discovery cycles executed", labels=["source_type"], ) DISCOVERY_CYCLE_DURATION_SECONDS = Histogram( name="mcp_hangar_discovery_cycle_duration_seconds", description="Duration of discovery cycles in seconds", labels=["source_type"], buckets=(0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0), ) DISCOVERY_REGISTRATIONS_TOTAL = Counter( name="mcp_hangar_discovery_registrations", description="Total mcp_server registrations from discovery", labels=["source_type"], ) DISCOVERY_DEREGISTRATIONS_TOTAL = Counter( name="mcp_hangar_discovery_deregistrations", description="Total mcp_server deregistrations from discovery", labels=["source_type", "reason"], # reason: ttl_expired, source_removed, manual ) DISCOVERY_QUARANTINE_TOTAL = Counter( name="mcp_hangar_discovery_quarantine", description="Total mcp_servers quarantined", labels=["reason"], # reason: health_check_failed, validation_failed, rate_limited ) DISCOVERY_ERRORS_TOTAL = Counter( name="mcp_hangar_discovery_errors", description="Total discovery errors", labels=["source_type", "error_type"], ) DISCOVERY_LAST_CYCLE_TIMESTAMP = Gauge( name="mcp_hangar_discovery_last_cycle_timestamp_seconds", description="Unix timestamp of last discovery cycle", labels=["source_type"], ) DISCOVERY_VALIDATION_FAILURES_TOTAL = Counter( name="mcp_hangar_discovery_validation_failures", description="Total discovery validation failures", labels=["source_type", "validation_type"], ) DISCOVERY_VALIDATION_DURATION_SECONDS = Histogram( name="mcp_hangar_discovery_validation_duration_seconds", description="Duration of discovery validation in seconds", labels=["source_type"], buckets=(0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0), ) # ----------------------------------------------------------------------------- # HTTP Transport Metrics (for remote mcp_servers) # ----------------------------------------------------------------------------- HTTP_REQUESTS_TOTAL = Counter( name="mcp_hangar_http_requests", description="Total number of HTTP requests to remote mcp_servers", labels=["mcp_server", "method", "status_code"], ) HTTP_REQUEST_DURATION_SECONDS = Histogram( name="mcp_hangar_http_request_duration_seconds", description="Duration of HTTP requests to remote mcp_servers in seconds", labels=["mcp_server", "method"], buckets=Histogram.LATENCY_BUCKETS, ) HTTP_ERRORS_TOTAL = Counter( name="mcp_hangar_http_errors", description="Total number of HTTP errors by type", labels=["mcp_server", "error_type"], # error_type: connection_refused, timeout, auth_failed, ssl_error ) HTTP_RETRIES_TOTAL = Counter( name="mcp_hangar_http_retries", description="Total number of HTTP request retries", labels=["mcp_server", "retry_reason"], # retry_reason: 502, 503, 504, connection_error ) # ----------------------------------------------------------------------------- # Batch Invocation Metrics # ----------------------------------------------------------------------------- BATCH_CALLS_TOTAL = Counter( name="mcp_hangar_batch_calls", description="Total number of batch invocations", labels=["result"], # result: success, partial, failure, validation_error ) BATCH_VALIDATION_FAILURES_TOTAL = Counter( name="mcp_hangar_batch_validation_failures", description="Total number of batch validation failures", ) BATCH_SIZE_HISTOGRAM = Histogram( name="mcp_hangar_batch_size", description="Distribution of batch sizes (number of calls per batch)", buckets=(1, 2, 5, 10, 20, 50, 100), ) BATCH_DURATION_SECONDS = Histogram( name="mcp_hangar_batch_duration_seconds", description="Duration of batch invocations in seconds", buckets=(0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0, 60.0, 120.0, 300.0), ) BATCH_CONCURRENCY_GAUGE = Gauge( name="mcp_hangar_batch_concurrency", description="Current number of parallel batch executions", ) BATCH_TRUNCATIONS_TOTAL = Counter( name="mcp_hangar_batch_truncations", description="Total number of response truncations in batches", labels=["reason"], # reason: per_call, total_size ) BATCH_CIRCUIT_BREAKER_REJECTIONS_TOTAL = Counter( name="mcp_hangar_batch_circuit_breaker_rejections", description="Total calls rejected due to circuit breaker in batches", labels=["mcp_server"], ) BATCH_CANCELLATIONS_TOTAL = Counter( name="mcp_hangar_batch_cancellations", description="Total number of batch cancellations", labels=["reason"], # reason: timeout, fail_fast ) # ----------------------------------------------------------------------------- # Batch Concurrency Metrics # ----------------------------------------------------------------------------- BATCH_INFLIGHT_CALLS = Gauge( name="mcp_hangar_batch_inflight_calls", description="Number of MCP tool calls currently in flight (global)", ) BATCH_INFLIGHT_CALLS_PER_PROVIDER = Gauge( name="mcp_hangar_batch_inflight_calls_per_mcp_server", description="Number of MCP tool calls currently in flight per mcp_server", labels=["mcp_server"], ) BATCH_CONCURRENCY_WAIT_SECONDS = Histogram( name="mcp_hangar_batch_concurrency_wait_seconds", description="Time spent waiting for a concurrency slot", labels=["mcp_server"], buckets=(0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0), ) BATCH_CONCURRENCY_QUEUED_TOTAL = Counter( name="mcp_hangar_batch_concurrency_queued", description="Total calls that had to wait for a concurrency slot", labels=["mcp_server"], ) # ----------------------------------------------------------------------------- # Circuit Breaker Metrics # ----------------------------------------------------------------------------- CIRCUIT_BREAKER_STATE = Gauge( name="mcp_hangar_circuit_breaker_state", description="Current circuit breaker state per mcp_server and state (1=active, 0=inactive)", labels=["mcp_server", "state"], ) # A group's circuit, as the replica that exposes it sees it (#1357). Each # replica keeps its own breaker (#1358), so two can disagree about one group; # the scrape's `instance` label is what tells them apart. The fleet view is a # query, not a metric (#1380): # # max by (group) (mcp_hangar_group_circuit_open) # - min by (group) (mcp_hangar_group_circuit_open) > 0 # # is 1 for every group the replicas disagree about. # # 0/1, not a closed/half-open/open enum. A group never half-opens its circuit: # the breaker does that only in `allow_request()`, which the group does not # call. A half-open breaker is one the group reports closed and routes through, # so it would read 0 here, as `circuit_open` in `hangar_group_list` does. # # Only `group`: one series per group this replica has loaded, seeded when it is # loaded, dropped when it is deleted. GROUP_CIRCUIT_OPEN = Gauge( name="mcp_hangar_group_circuit_open", description="Whether this replica has the group's circuit breaker open (1=open, 0=closed)", labels=["group"], ) # ----------------------------------------------------------------------------- # Event Store Compaction Metrics # ----------------------------------------------------------------------------- EVENTS_COMPACTED_TOTAL = Counter( name="mcp_hangar_events_compacted", description="Total number of events removed by stream compaction", # No per-stream label: stream IDs are unbounded identifiers and would be a # cardinality bomb. Compaction is a fleet-wide signal; aggregate is enough. ) # ----------------------------------------------------------------------------- # Tool Access Policy Metrics # ----------------------------------------------------------------------------- TOOL_ACCESS_DENIED_TOTAL = Counter( name="mcp_hangar_tool_access_denied", description="Total tool invocations denied by access policy", labels=["mcp_server", "tool", "reason"], # reason: tool_not_in_access_policy ) # A call refused by a per-tenant execution budget (#1445), after every policy # gate passed it. `budget` is the configured entry that refused it -- a tenant # id from `execution.tenant_limits`, or "*" -- or "none" when no entry applied, # so its values are bounded by the configuration, not by who calls. TENANT_QUOTA_REFUSALS_TOTAL = Counter( name="mcp_hangar_tenant_quota_refusals", description="Total tool calls refused by a per-tenant execution budget", labels=["budget", "reason"], # reason: no_budget, concurrency, rate ) TOOLS_FILTERED_TOTAL = Gauge( name="mcp_hangar_tools_filtered", description="Number of tools filtered by access policy per mcp_server", labels=["mcp_server"], ) TOOL_ACCESS_POLICY_ACTIVE = Gauge( name="mcp_hangar_tool_access_policy_active", description="Whether tool access policy is active (1) or unrestricted (0) per mcp_server", labels=["mcp_server"], ) # ----------------------------------------------------------------------------- # Capability Enforcement Metrics # ----------------------------------------------------------------------------- CAPABILITY_VIOLATIONS_TOTAL = Counter( name="mcp_hangar_capability_violations", description="Total number of capability violations detected", labels=["mcp_server", "violation_type"], ) EGRESS_POLICY_VIOLATIONS_OBSERVED_TOTAL = Counter( name="mcp_hangar_egress_policy_violations_observed", description="Total L7 egress-policy violations observed in Audit mode (recorded, not blocked)", labels=["mcp_server", "would_be_action"], # would_be_action: deny | require_approval ) # The enforcing half of the pair above (#1128). Audit mode -- which changes # nothing -- was counted; Enforce mode, refusing the call for real, was not, so # "which calls did this policy refuse" had no answer in any metric. # `mcp_hangar_tool_call_errors_total` cannot cover it: it is fed from # `ToolInvocationFailed`, whose three emitters are all past the gate. EGRESS_POLICY_ENFORCED_TOTAL = Counter( name="mcp_hangar_egress_policy_enforced", description="Total tool calls refused by an Enforce-mode L7 egress policy", # action: deny | require_approval -- rule_kind: header | tool | arguments labels=["mcp_server", "action", "rule_kind"], ) # ----------------------------------------------------------------------------- # Cost Attribution Metrics # ----------------------------------------------------------------------------- COST_CENTS_TOTAL = Counter( # The registry appends "_total" to counter names on exposition; the base # name must NOT include it, or the series renders as "..._total_total". name="mcp_hangar_cost_cents", description="Total attributed cost in hundredths of a cent", labels=["mcp_server", "tool", "cost_model"], ) COST_ATTRIBUTIONS_TOTAL = Counter( name="mcp_hangar_cost_attributions", description="Total number of cost attribution computations", labels=["mcp_server", "tool"], ) # ----------------------------------------------------------------------------- # OTLP Trace Export Metrics # ----------------------------------------------------------------------------- OTLP_EXPORT_FAILURES_TOTAL = Counter( # BatchSpanProcessor exports on a background thread and swallows failures, # so a down/unreachable collector is otherwise silent. This counter makes # that signal observable; each failed batch also drops its buffered spans. name="mcp_hangar_otlp_export_failures", description="Total number of failed OTLP span-export batches (collector unreachable or export error)", ) OTLP_AUDIT_EXPORT_FAILURES_TOTAL = Counter( # The audit log pipeline's twin of the counter above (#1289): its # BatchLogRecordProcessor also exports on a background thread and swallows # failures, and each failed batch drops the audit records in it. name="mcp_hangar_otlp_audit_export_failures", description="Total number of failed OTLP audit log-record export batches (collector unreachable or export error)", ) # ----------------------------------------------------------------------------- # Task Relay Metrics (ADR-014 Phase 3) # ----------------------------------------------------------------------------- # The relay seam emits Task* lifecycle events; these counters make the async # task lifecycle observable per tenant. The fail-closed paths (TaskFailed, # DigestMismatchInTask) are the ones most in need of alerting, so each is a # distinct series. tenant_id is normalized to "unknown" when absent so the # label never carries a None (which would break exposition). TASK_RELAYED_TOTAL = Counter( name="mcp_hangar_task_relayed", description="Total async tasks relayed/created through the task relay seam", labels=["tenant_id"], ) # ----------------------------------------------------------------------------- # Front-door Projection Metrics # ----------------------------------------------------------------------------- # A front door that serves nothing looks identical from outside whether the # tenant genuinely has no tools, the replica has discovered nothing yet, or the # caller arrived without an identity (#862, #887). Same 200, same empty list, # nothing in the log. This splits them by cause so "this gateway is answering # nobody" is visible on a dashboard rather than inferred from a support ticket. # # Deliberately NOT labelled by tenant: a public front door has unbounded tenant # cardinality, and the question a dashboard asks here is "which cause", not # "which tenant" -- the log line carries the tenant for the follow-up. EMPTY_PROJECTION_TOTAL = Counter( name="mcp_hangar_empty_projection", description="Total front-door tool projections that resolved to zero tools, by cause", # reason: no_identity (fail-closed, no tenant), nothing_discovered (cold # replica), filtered (policy or withdrawal removed everything) labels=["reason"], ) # One sample per start the front door's catalogue retry makes for a required # server its boot warm-up could not project (#1446). `mcp_server` is bounded by # `tool_access.required_catalogue.servers`, which is checked against the config. CATALOGUE_RETRIES_TOTAL = Counter( name="mcp_hangar_catalogue_retries", description="Total starts the front door's required-catalogue retry made, by outcome", # outcome: projected, started (not projected yet), failed, refused (the # server's own backoff or capability block refused the start) labels=["mcp_server", "outcome"], ) # How big the answer to `tools/list` is, which nothing on the server side could # see. The surface sits in an agent's prompt prefix and is paid for on every # turn, so a client with a small context window can be pushed over the limit # before it calls anything -- and the first report of that arrived from a client # hitting its own limit, not from here (#904). # # Split by kind so "the upstream grew" and "the control plane is being projected" # are separable, which is the whole question the surface split exists to answer. # Unlabelled by tenant for the same reason as EMPTY_PROJECTION_TOTAL: a public # front door has unbounded tenant cardinality. PROJECTED_TOOLS = Histogram( name="mcp_hangar_projected_tools", description="Tools returned by a front-door tools/list, by kind", labels=["kind"], # governed (upstream) | management (hangar_*) buckets=(0, 1, 2, 5, 10, 25, 50, 100, 250, 500), ) # What the projected surface weighs and whether it moved (#1369). The diagnosis # that opened #1365 was "46 tools and 174 KB" held by a gateway whose client saw # none, and nothing on the server could say either number. All three are # written in one place, `projection_metrics.observe_served_listing`, on a # listing the client received and never on the SDK's pre-dispatch listing (#1049). # # No tenant label on any of them, for #895's reason: a public front door has # unbounded tenant cardinality. `mcp_server` is the upstream (a group reads as # its group id): a set the operator configures, as on every per-server metric. _SURFACE_BYTE_BUCKETS = (0, 1024, 4096, 16384, 65536, 131072, 262144, 524288, 1048576, 4194304) PROJECTED_SURFACE_BYTES = Histogram( name="mcp_hangar_projected_surface_bytes", description="Bytes of tool definitions returned by a front-door tools/list, by kind", labels=["kind"], # governed (upstream) | management (hangar_*) buckets=_SURFACE_BYTE_BUCKETS, ) PROJECTED_UPSTREAM_BYTES = Histogram( name="mcp_hangar_projected_upstream_bytes", description="Bytes of tool definitions one upstream contributes to a front-door tools/list that includes it", labels=["mcp_server"], buckets=_SURFACE_BYTE_BUCKETS, ) # A listing that served an identity a projection different from the one this # replica last served it: the event a `tools/list_changed` exists to announce. # A front door seeds it at zero when it installs its handlers, because a counter # that first appears at 1 reads as no increase to `increase()`. PROJECTION_CHANGES_TOTAL = Counter( name="mcp_hangar_projection_changes", description="Front-door tools/list responses whose projection differed from the one last served to the same caller", ) # The SDK's Mcp-Param-* check is fail-open: a tools/list that cannot produce a # schema skips validation and the call still runs (#1053). Some of those # branches log; none of them were a metric. Reasons match the SDK skip arms # Hangar can see without wrapping the transport (pagination is absent here: # the front door returns one unpaged list). PARAM_HEADER_VALIDATION_SKIPPED_TOTAL = Counter( name="mcp_hangar_param_header_validation_skipped", description="Times Mcp-Param-* header/body validation did not run, by cause", # listing_failed | tool_not_listed | invalid_annotation | legacy_protocol labels=["reason"], ) # A tool Hangar declines to project, though nothing withdrew it and policy # allows it: the definition itself is unusable to a conforming client (#1056). # Counted once per tool per schema version, not per listing. PROJECTION_WITHDRAWALS_TOTAL = Counter( name="mcp_hangar_projection_withdrawals", description="Tools withheld from the front-door projection by their own definition, by cause", # invalid_x_mcp_header labels=["reason"], ) # A handed-out resource_link the front door forgot: the per-tenant map hit its # cap, or the tenant-map LRU dropped a whole tenant (#1139). Either way the # victim's `resources/list` just gets shorter, with nothing in the log -- the # shape #1128 argued against on the egress path. Unlabelled by tenant for the # same reason as EMPTY_PROJECTION_TOTAL (#895): unbounded cardinality. RESOURCE_LINKS_EVICTED_TOTAL = Counter( name="mcp_hangar_resource_links_evicted", description="Handed-out resource_links forgotten by the front door, by cause", # tenant_cap (oldest link at the per-tenant cap) | tenant_map_cap (every # link of a tenant dropped by the tenant-map LRU) labels=["reason"], ) # ----------------------------------------------------------------------------- # Approval Gate Metrics # ----------------------------------------------------------------------------- # A gate that holds calls and notifies nobody is indistinguishable from a broken # gateway: every gated call waits out `approval_timeout_seconds` and then denies # (#914). Nothing leaks -- the failure is closed -- but the remediation an # operator reaches for under that pressure is emptying `approval_list`, which is # fail-closed in code and fail-open in the organisation. # # The pair below is what makes "armed and unmanned" visible before someone # reaches for that. Requests counts what the gate held; deliveries counts what # actually left, by outcome. Requests climbing while deliveries stay at zero -- # or `outcome="not_notified"` tracking requests one-for-one -- is the shape. # # Labelled by channel, not by tool or tenant: the question is which notification # path is dead, and channels are a small closed set an operator configures. APPROVAL_REQUESTS_TOTAL = Counter( name="mcp_hangar_approval_requests", description="Total tool invocations held by the approval gate, by channel label", labels=["channel"], ) APPROVAL_DELIVERIES_TOTAL = Counter( name="mcp_hangar_approval_deliveries", description="Total approval notifications handed to a delivery channel, by outcome", # outcome: sent (the adapter accepted it), failed (it raised -- the gate # swallows this so the hold survives), not_notified (the resolved channel # does not reach out of the process at all, so nobody was told) labels=["channel", "outcome"], ) APPROVAL_DECISIONS_TOTAL = Counter( name="mcp_hangar_approval_decisions", description="Total approval holds by how they ended", # decision: granted, denied, expired. `expired` climbing alongside a flat # `sent` is the same story from the other end. labels=["channel", "decision"], ) TASK_COMPLETED_TOTAL = Counter( name="mcp_hangar_task_completed", description="Total relayed tasks that finished successfully", labels=["tenant_id"], ) TASK_FAILED_TOTAL = Counter( name="mcp_hangar_task_failed", description="Total relayed tasks that terminated with an error, by failure reason", labels=["tenant_id", "reason"], ) TASK_CANCELLED_TOTAL = Counter( name="mcp_hangar_task_cancelled", description="Total relayed tasks cancelled before completion", labels=["tenant_id"], ) TASK_INPUT_REQUIRED_TOTAL = Counter( name="mcp_hangar_task_input_required", description="Total relayed tasks that paused awaiting caller input", labels=["tenant_id"], ) TASK_DIGEST_DRIFT_TOTAL = Counter( name="mcp_hangar_task_digest_drift", description="Total relayed tasks failed fail-closed on pinned-tool digest drift at result time", labels=["tenant_id"], ) TASK_CONSENT_DECIDED_TOTAL = Counter( name="mcp_hangar_task_consent_decided", description="Total mid-flight task input-required consent decisions (labeled by grant outcome)", labels=["tenant_id", "granted"], ) # ============================================================================= # Register All Metrics # ============================================================================= def _register_all_metrics(): """Register all predefined metrics. This hand-maintained list is the failure mode, not any one metric: a collector defined above and forgotten here accumulates in process memory and never reaches a scrape, which looks from outside exactly like a feature that was never built. Four had (#1059). If you add a metric, add it here -- `test_every_metric_is_registered.py` walks this module and fails if you do not.""" metrics = [ BUILD_INFO, PROCESS_START_TIME, PROVIDER_INFO, PROVIDER_STATE_CURRENT, PROVIDER_UP, PROVIDER_INITIALIZED, PROVIDER_LAST_STATE_CHANGE_SECONDS, PROVIDER_LAST_HEALTHY_SECONDS, PROVIDER_STARTS_TOTAL, PROVIDER_STOPS_TOTAL, PROVIDER_COLD_START_SECONDS, PROVIDER_COLD_START_IN_PROGRESS, TOOL_CALLS_TOTAL, TOOL_CALL_DURATION_SECONDS, TOOL_CALL_ERRORS_TOTAL, HEALTH_CHECK_TOTAL, HEALTH_CHECK_DURATION_SECONDS, HEALTH_CHECK_CONSECUTIVE_FAILURES, CONNECTIONS_ACTIVE, MESSAGES_SENT_TOTAL, MESSAGES_RECEIVED_TOTAL, MESSAGE_SIZE_BYTES, GC_CYCLES_TOTAL, GC_CYCLE_DURATION_SECONDS, GC_PROVIDERS_COLLECTED_TOTAL, ERRORS_TOTAL, RATE_LIMIT_HITS_TOTAL, RATE_LIMIT_ACTIVE_BUCKETS, # Discovery metrics DISCOVERY_SOURCES_TOTAL, DISCOVERY_SOURCES_HEALTHY, DISCOVERY_PROVIDERS_TOTAL, DISCOVERY_CYCLES_TOTAL, DISCOVERY_CYCLE_DURATION_SECONDS, DISCOVERY_REGISTRATIONS_TOTAL, DISCOVERY_DEREGISTRATIONS_TOTAL, DISCOVERY_QUARANTINE_TOTAL, DISCOVERY_ERRORS_TOTAL, DISCOVERY_LAST_CYCLE_TIMESTAMP, DISCOVERY_VALIDATION_FAILURES_TOTAL, DISCOVERY_VALIDATION_DURATION_SECONDS, # HTTP transport metrics HTTP_REQUESTS_TOTAL, HTTP_REQUEST_DURATION_SECONDS, HTTP_ERRORS_TOTAL, HTTP_RETRIES_TOTAL, # Batch invocation metrics BATCH_CALLS_TOTAL, BATCH_VALIDATION_FAILURES_TOTAL, BATCH_SIZE_HISTOGRAM, BATCH_DURATION_SECONDS, BATCH_CONCURRENCY_GAUGE, BATCH_TRUNCATIONS_TOTAL, BATCH_CIRCUIT_BREAKER_REJECTIONS_TOTAL, BATCH_CANCELLATIONS_TOTAL, # Cost attribution metrics COST_CENTS_TOTAL, COST_ATTRIBUTIONS_TOTAL, # OTLP trace and audit export metrics OTLP_EXPORT_FAILURES_TOTAL, OTLP_AUDIT_EXPORT_FAILURES_TOTAL, ] # Concurrency metrics (defined above alongside other batch metrics) metrics.extend( [ BATCH_INFLIGHT_CALLS, BATCH_INFLIGHT_CALLS_PER_PROVIDER, BATCH_CONCURRENCY_WAIT_SECONDS, BATCH_CONCURRENCY_QUEUED_TOTAL, ] ) # Tool access policy metrics metrics.extend( [ TOOL_ACCESS_DENIED_TOTAL, TENANT_QUOTA_REFUSALS_TOTAL, TOOLS_FILTERED_TOTAL, TOOL_ACCESS_POLICY_ACTIVE, ] ) # Circuit breaker and event store compaction metrics metrics.extend( [ CIRCUIT_BREAKER_STATE, GROUP_CIRCUIT_OPEN, EVENTS_COMPACTED_TOTAL, ] ) # Capability enforcement metrics metrics.append(CAPABILITY_VIOLATIONS_TOTAL) # Task relay metrics (ADR-014 Phase 3) metrics.extend( [ TASK_RELAYED_TOTAL, TASK_COMPLETED_TOTAL, TASK_FAILED_TOTAL, TASK_CANCELLED_TOTAL, TASK_INPUT_REQUIRED_TOTAL, TASK_DIGEST_DRIFT_TOTAL, TASK_CONSENT_DECIDED_TOTAL, ] ) # Front-door projection / SEP-2243 header validation / link map (#887, #904, #1053, #1139). metrics.extend( [ EMPTY_PROJECTION_TOTAL, CATALOGUE_RETRIES_TOTAL, PROJECTED_TOOLS, # Surface size, composition and churn (#1369). PROJECTED_SURFACE_BYTES, PROJECTED_UPSTREAM_BYTES, PROJECTION_CHANGES_TOTAL, PARAM_HEADER_VALIDATION_SKIPPED_TOTAL, PROJECTION_WITHDRAWALS_TOTAL, RESOURCE_LINKS_EVICTED_TOTAL, ] ) # Approval gate (#920) and Audit-mode egress observations (ADR-013). Both # were defined, incremented on the live path, and documented with PromQL -- # and absent from every /metrics scrape, because nothing appended them here # (#1059). metrics.extend( [ APPROVAL_REQUESTS_TOTAL, APPROVAL_DELIVERIES_TOTAL, APPROVAL_DECISIONS_TOTAL, EGRESS_POLICY_VIOLATIONS_OBSERVED_TOTAL, EGRESS_POLICY_ENFORCED_TOTAL, ] ) for metric in metrics: REGISTRY.register(metric) _register_all_metrics() # ============================================================================= # Convenience Functions # ============================================================================= def get_metrics() -> str: """Get all metrics in Prometheus exposition format.""" return REGISTRY.collect() def observe_tool_call(mcp_server: str, tool: str, duration: float, success: bool, error_type: str | None = None): """Record a tool call observation. The duration histogram observes only successful calls: failures carry no meaningful duration (the failure path has none to report), and recording a 0-second observation for every failure poisons the latency percentiles. Failures are counted separately via TOOL_CALL_ERRORS_TOTAL. """ status = "success" if success else "error" TOOL_CALLS_TOTAL.inc(mcp_server=mcp_server, tool=tool, status=status) if success: TOOL_CALL_DURATION_SECONDS.observe(duration, mcp_server=mcp_server, tool=tool) elif error_type: TOOL_CALL_ERRORS_TOTAL.inc(mcp_server=mcp_server, tool=tool, error_type=error_type) def observe_health_check( mcp_server: str, duration: float, healthy: bool, is_cold: bool = False, consecutive_failures: int = 0, ): """Record a health check observation. Args: mcp_server: McpServer ID duration: Health check duration in seconds healthy: Whether the check passed (only meaningful if not cold) is_cold: Whether mcp_server is in cold state (not started yet) consecutive_failures: Number of consecutive failures """ if is_cold: result = "cold" elif healthy: result = "healthy" else: result = "unhealthy" HEALTH_CHECK_TOTAL.inc(mcp_server=mcp_server, result=result) HEALTH_CHECK_DURATION_SECONDS.observe(duration, mcp_server=mcp_server) HEALTH_CHECK_CONSECUTIVE_FAILURES.set(consecutive_failures, mcp_server=mcp_server) def update_mcp_server_state(mcp_server: str, state: str, mode: str = "subprocess", *, record_change: bool = True): """Update mcp_server state metrics. ``record_change=False`` sets the state without claiming it changed now: a state read back from the event log at boot changed before this process. """ state_map = {"cold": 0, "initializing": 1, "ready": 2, "degraded": 3, "dead": 4} PROVIDER_STATE_CURRENT.set(state_map.get(state, 0), mcp_server=mcp_server) PROVIDER_UP.set(1 if state == "ready" else 0, mcp_server=mcp_server) PROVIDER_INITIALIZED.set(0 if state == "cold" else 1, mcp_server=mcp_server) PROVIDER_INFO.set(1, mcp_server=mcp_server, mode=mode) if record_change: PROVIDER_LAST_STATE_CHANGE_SECONDS.set(time.time(), mcp_server=mcp_server) def remove_mcp_server_series(mcp_server: str) -> None: """Drop the lifecycle gauges of an mcp_server that was deleted, unloaded or reloaded away. Left behind, they read as live forever: a removed dead server at `state == 4` and an ever-older last-healthy time. Counters stay; `rate()` and `increase()` already read a series that stops moving correctly. An event about the server handled after this writes its series again. That is accepted: every removal path publishes the server's own events before the one that triggers this, so it takes an event published late by another thread, and the series it leaves is the state that event reported. """ for gauge in ( PROVIDER_INFO, PROVIDER_STATE_CURRENT, PROVIDER_UP, PROVIDER_INITIALIZED, PROVIDER_LAST_STATE_CHANGE_SECONDS, PROVIDER_LAST_HEALTHY_SECONDS, PROVIDER_COLD_START_IN_PROGRESS, HEALTH_CHECK_CONSECUTIVE_FAILURES, CONNECTIONS_ACTIVE, ): gauge.remove(mcp_server=mcp_server) def record_mcp_server_healthy(mcp_server: str, at: float) -> None: """Record that the mcp_server was seen working at ``at`` (Unix seconds). ``at`` is when it happened -- the event's ``occurred_at`` -- not when the event was handled, and an older one never replaces a newer one. """ PROVIDER_LAST_HEALTHY_SECONDS.set_max(at, mcp_server=mcp_server) def record_mcp_server_start(mcp_server: str, success: bool): """Record a mcp_server start attempt.""" result = "success" if success else "failure" PROVIDER_STARTS_TOTAL.inc(mcp_server=mcp_server, result=result) if success: PROVIDER_INITIALIZED.set(1, mcp_server=mcp_server) def mcp_server_stop_reason(reason: object) -> str: """The one of `MCP_SERVER_STOP_REASONS` a stop for `reason` is recorded under. Any other reason is `manual`: the REST stop takes its reason from the request body, which can hold any JSON value, and the label stays the closed set its HELP line lists. The stop command records its stop under this (#1466), so the event carries a reason every consumer of it knows. """ if isinstance(reason, str) and reason in MCP_SERVER_STOP_REASONS: return reason return MCP_SERVER_STOP_REASON_OTHER def record_mcp_server_stop(mcp_server: str, reason: str): """Record a mcp_server stop, under `mcp_server_stop_reason(reason)`. Called for each `McpServerStopped` only, so a stop is counted once (#1466). """ PROVIDER_STOPS_TOTAL.inc(mcp_server=mcp_server, reason=mcp_server_stop_reason(reason)) def record_catalogue_retry(mcp_server: str, outcome: str) -> None: """Record one start by the front door's required-catalogue retry (#1446).""" CATALOGUE_RETRIES_TOTAL.inc(mcp_server=mcp_server, outcome=outcome) def record_cold_start(mcp_server: str, duration: float, mode: str = "subprocess"): """Record cold start duration - the critical UX metric. This measures time from user request to mcp_server ready state. High values here directly impact user experience. Args: mcp_server: McpServer ID duration: Time in seconds from start to ready mode: McpServer mode (subprocess, docker, etc.) """ PROVIDER_COLD_START_SECONDS.observe(duration, mcp_server=mcp_server, mode=mode) def cold_start_begin(mcp_server: str): """Mark beginning of cold start (for in-progress tracking).""" PROVIDER_COLD_START_IN_PROGRESS.set(1, mcp_server=mcp_server) def cold_start_end(mcp_server: str): """Mark end of cold start.""" PROVIDER_COLD_START_IN_PROGRESS.set(0, mcp_server=mcp_server) def record_gc_cycle(duration: float, collected: dict[str, int] | None = None): """Record a GC cycle.""" GC_CYCLES_TOTAL.inc() GC_CYCLE_DURATION_SECONDS.observe(duration) if collected: for reason, count in collected.items(): for _ in range(count): GC_PROVIDERS_COLLECTED_TOTAL.inc(reason=reason) def record_error(component: str, error_type: str): """Record an error.""" ERRORS_TOTAL.inc(component=component, error_type=error_type) def update_circuit_breaker_state(mcp_server: str, new_state: str) -> None: """Update circuit breaker state gauge for a mcp_server. Sets the active state label to 1 and all others to 0 so that PromQL can filter by state label. Args: mcp_server: McpServer ID. new_state: New circuit breaker state value (closed, open, half_open). """ for state in ("closed", "open", "half_open"): CIRCUIT_BREAKER_STATE.set(1.0 if state == new_state else 0.0, mcp_server=mcp_server, state=state) def set_group_circuit_open(group: str, is_open: bool) -> None: """Record whether this replica has the group's circuit open.""" GROUP_CIRCUIT_OPEN.set(1.0 if is_open else 0.0, group=group) def remove_group_series(group: str) -> None: """Drop the gauges of a group that was deleted. Left behind, a deleted group whose circuit was open would read as open for good, and every disagreement query would keep finding it. """ GROUP_CIRCUIT_OPEN.remove(group=group) def record_events_compacted(stream_id: str, count: int) -> None: """Record events removed by compaction. Args: stream_id: The stream that was compacted. count: Number of events removed. """ if count > 0: # stream_id kept in the signature for callers/logging but intentionally # not used as a metric label (unbounded cardinality). EVENTS_COMPACTED_TOTAL.inc(count) def set_connection_active(mcp_server: str, active: bool) -> None: """Set whether a client connection to a server is currently open. Args: mcp_server: The server ID. active: True when a client is connected/ready, False on close. """ CONNECTIONS_ACTIVE.set(1 if active else 0, mcp_server=mcp_server) def classify_jsonrpc_message(message: dict) -> str: """Classify a received JSON-RPC message as response / notification / error. - ``error``: carries a top-level ``error`` member. - ``notification``: a request with no ``id`` (server-initiated). - ``response``: everything else (a normal result envelope). """ if "error" in message: return "error" if "method" in message and "id" not in message: return "notification" return "response" def record_message_sent(mcp_server: str, method: str, size_bytes: int) -> None: """Record an outgoing JSON-RPC message to an upstream server. Args: mcp_server: The upstream server ID (``"unknown"`` if unlabeled). method: JSON-RPC method (e.g. ``tools/call``). size_bytes: Serialized request size in bytes. """ MESSAGES_SENT_TOTAL.inc(mcp_server=mcp_server, method=method) MESSAGE_SIZE_BYTES.observe(size_bytes, mcp_server=mcp_server, direction="sent") def record_message_received(mcp_server: str, message_type: str, size_bytes: int) -> None: """Record an incoming JSON-RPC message from an upstream server. Args: mcp_server: The upstream server ID (``"unknown"`` if unlabeled). message_type: ``response`` / ``notification`` / ``error`` (see :func:`classify_jsonrpc_message`). size_bytes: Raw message size in bytes. """ MESSAGES_RECEIVED_TOTAL.inc(mcp_server=mcp_server, type=message_type) MESSAGE_SIZE_BYTES.observe(size_bytes, mcp_server=mcp_server, direction="received") def record_cost(mcp_server: str, tool: str, cost_cents: int, cost_model: str) -> None: """Record attributed cost for a tool invocation. Args: mcp_server: The server the tool ran on. tool: The tool name. cost_cents: Attributed cost (hundredths of a cent) to add. cost_model: The pricing model used for attribution. """ COST_CENTS_TOTAL.inc(cost_cents, mcp_server=mcp_server, tool=tool, cost_model=cost_model) COST_ATTRIBUTIONS_TOTAL.inc(mcp_server=mcp_server, tool=tool) def record_capability_violation(mcp_server: str, violation_type: str) -> None: """Record a capability violation detection. Args: mcp_server: McpServer ID that triggered the violation. violation_type: Type of violation (egress_denied, capability_drift, etc.). """ CAPABILITY_VIOLATIONS_TOTAL.inc(mcp_server=mcp_server, violation_type=violation_type) def record_egress_policy_violation_observed(mcp_server: str, would_be_action: str) -> None: """Record an Audit-mode L7 egress-policy violation (observed, not blocked). Args: mcp_server: McpServer ID whose tool call tripped the policy. would_be_action: Verdict Enforce mode would have applied (``deny`` or ``require_approval``). """ EGRESS_POLICY_VIOLATIONS_OBSERVED_TOTAL.inc(mcp_server=mcp_server, would_be_action=would_be_action) def record_egress_policy_enforced(mcp_server: str, action: str, rule_kind: str) -> None: """Record an Enforce-mode L7 egress-policy refusal (the call was blocked). Args: mcp_server: McpServer ID whose tool call the policy refused. action: Verdict applied -- ``deny`` or ``require_approval``. rule_kind: Which part of the policy decided -- ``header``, ``tool`` or ``arguments``. """ EGRESS_POLICY_ENFORCED_TOTAL.inc(mcp_server=mcp_server, action=action, rule_kind=rule_kind) def record_resource_links_evicted(reason: str, count: int) -> None: """Record handed-out resource_links the front door forgot (#1139). Args: reason: ``tenant_cap`` (a tenant's oldest link at its own cap) or ``tenant_map_cap`` (a whole tenant dropped by the tenant-map LRU). count: Links that went away -- for ``tenant_map_cap`` that is every link the evicted tenant held, not one. """ if count > 0: RESOURCE_LINKS_EVICTED_TOTAL.inc(count, reason=reason) def record_otlp_export_failure() -> None: """Record a failed OTLP span-export batch. Called when the OTLP exporter's ``export()`` returns a failure result or raises. The export runs on the BatchSpanProcessor's background thread and would otherwise fail silently, so this counter is the observable signal of a down or unreachable collector (and of the spans dropped with that batch). """ OTLP_EXPORT_FAILURES_TOTAL.inc() def record_otlp_audit_export_failure() -> None: """Record a failed OTLP audit log-record export batch. The audit counterpart of `record_otlp_export_failure`, called from the metered exporter the audit log pipeline wraps its OTLP exporter in. """ OTLP_AUDIT_EXPORT_FAILURES_TOTAL.inc() # ============================================================================= # Task Relay Metrics Functions (ADR-014 Phase 3) # ============================================================================= def record_task_relayed(tenant_id: str | None) -> None: """Record an async task relayed/created (TaskCreated).""" TASK_RELAYED_TOTAL.inc(tenant_id=tenant_id or "unknown") def record_task_completed(tenant_id: str | None) -> None: """Record a relayed task that finished successfully (TaskCompleted).""" TASK_COMPLETED_TOTAL.inc(tenant_id=tenant_id or "unknown") def record_task_failed(tenant_id: str | None, reason: str) -> None: """Record a relayed task that terminated with an error (TaskFailed). Args: tenant_id: Owning tenant, normalized to "unknown" when absent. reason: The failure reason (the event's ``error_type``). """ TASK_FAILED_TOTAL.inc(tenant_id=tenant_id or "unknown", reason=reason or "unknown") def record_task_cancelled(tenant_id: str | None) -> None: """Record a relayed task cancelled before completion (TaskCancelled).""" TASK_CANCELLED_TOTAL.inc(tenant_id=tenant_id or "unknown") def record_task_input_required(tenant_id: str | None) -> None: """Record a relayed task that paused awaiting caller input (TaskInputRequired).""" TASK_INPUT_REQUIRED_TOTAL.inc(tenant_id=tenant_id or "unknown") def record_task_digest_drift(tenant_id: str | None) -> None: """Record a relayed task failed fail-closed on digest drift (DigestMismatchInTask).""" TASK_DIGEST_DRIFT_TOTAL.inc(tenant_id=tenant_id or "unknown") def record_task_consent_decided(tenant_id: str | None, granted: bool) -> None: """Record a mid-flight task input-required consent decision (TaskConsentDecided).""" TASK_CONSENT_DECIDED_TOTAL.inc(tenant_id=tenant_id or "unknown", granted="true" if granted else "false") # ============================================================================= # Discovery Metrics Functions # ============================================================================= def update_discovery_source(source_type: str, mode: str, is_healthy: bool, mcp_servers_count: int): """Update discovery source metrics. Args: source_type: Type of source (filesystem, docker, kubernetes, entrypoint) mode: Discovery mode (additive, authoritative) is_healthy: Whether the source is healthy mcp_servers_count: Number of mcp_servers discovered by this source """ DISCOVERY_SOURCES_TOTAL.set(1, source_type=source_type, mode=mode) DISCOVERY_SOURCES_HEALTHY.set(1 if is_healthy else 0, source_type=source_type) DISCOVERY_PROVIDERS_TOTAL.set(mcp_servers_count, source_type=source_type, status="discovered") def record_discovery_cycle( source_type: str, duration: float, discovered: int = 0, registered: int = 0, quarantined: int = 0, ): """Record a discovery cycle execution. Args: source_type: Type of source duration: Duration of the cycle in seconds discovered: Number of mcp_servers discovered registered: Number of mcp_servers registered quarantined: Number of mcp_servers quarantined """ DISCOVERY_CYCLES_TOTAL.inc(source_type=source_type) DISCOVERY_CYCLE_DURATION_SECONDS.observe(duration, source_type=source_type) DISCOVERY_LAST_CYCLE_TIMESTAMP.set(time.time(), source_type=source_type) # Update mcp_server counts DISCOVERY_PROVIDERS_TOTAL.set(discovered, source_type=source_type, status="discovered") DISCOVERY_PROVIDERS_TOTAL.set(registered, source_type=source_type, status="registered") DISCOVERY_PROVIDERS_TOTAL.set(quarantined, source_type=source_type, status="quarantined") def record_discovery_registration(source_type: str): """Record a mcp_server registration from discovery.""" DISCOVERY_REGISTRATIONS_TOTAL.inc(source_type=source_type) def record_discovery_deregistration(source_type: str, reason: str): """Record a mcp_server deregistration from discovery. Args: source_type: Type of source reason: Reason for deregistration (ttl_expired, source_removed, manual) """ DISCOVERY_DEREGISTRATIONS_TOTAL.inc(source_type=source_type, reason=reason) def record_discovery_quarantine(reason: str): """Record a mcp_server quarantine. Args: reason: Reason for quarantine (health_check_failed, validation_failed, rate_limited) """ DISCOVERY_QUARANTINE_TOTAL.inc(reason=reason) def record_discovery_error(source_type: str, error_type: str): """Record a discovery error. Args: source_type: Type of source error_type: Type of error """ DISCOVERY_ERRORS_TOTAL.inc(source_type=source_type, error_type=error_type) def record_discovery_validation_failure(source_type: str, validation_type: str): """Record a discovery validation failure. Args: source_type: Type of source validation_type: The validation result/type that failed """ DISCOVERY_VALIDATION_FAILURES_TOTAL.inc(source_type=source_type, validation_type=validation_type) def record_discovery_validation_duration(source_type: str, duration: float): """Record the duration of a discovery validation. Args: source_type: Type of source duration: Validation duration in seconds """ DISCOVERY_VALIDATION_DURATION_SECONDS.observe(duration, source_type=source_type) # ============================================================================= # Timing Decorator # =============================================================================