metrics.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oct/mcp/metrics.py
 
"""
Purpose
-------
Optional Prometheus instrumentation for the ``oct-mcp`` server (Phase 5C).
 
Exposes five metrics:
 
* ``oct_mcp_tool_calls_total`` — Counter; labels ``tool``, ``decision``.
* ``oct_mcp_tool_duration_seconds`` — Histogram; label ``tool``.
* ``oct_mcp_rate_limit_hits_total`` — Counter; label ``tool``.
* ``oct_mcp_redactions_total`` — Counter; label ``tool``.
* ``oct_mcp_validation_failures_total`` — Counter; label ``tool``.
 
All public methods are **no-ops** when ``prometheus_client`` is not
installed, so the rest of the pipeline never needs to guard with
``if metrics is not None``.
 
Enabling metrics
----------------
Install the optional extra::
 
    pip install 'oct[metrics]'
 
Then start the server with ``--metrics-port``::
 
    python -m oct.mcp --project-root /path/to/project --metrics-port 9090
 
Diagnostics
-----------
Domain: MCP
Levels:
    L1 — errors: metrics server startup failures
    L2 — lifecycle: metrics server started
 
Contracts
---------
- This module never raises for the caller — all methods swallow errors
  and degrade gracefully.
- When ``prometheus_client`` is absent, ``PROMETHEUS_AVAILABLE`` is
  ``False`` and :class:`McpMetrics` is a lightweight no-op stub.
- The metrics HTTP server is started in a **daemon** background thread
  so it does not prevent process exit.
- :class:`McpMetrics` accepts an optional ``registry`` parameter
  (``CollectorRegistry``). The default is the module-level
  ``prometheus_client.REGISTRY`` so production ``/metrics`` works
  without any caller change. Tests and multi-tenant callers pass a
  private ``CollectorRegistry()`` to avoid ``Duplicated timeseries``
  collisions across instances.
"""
 
from __future__ import annotations
 
import threading
 
try:
    from prometheus_client import (
        Counter,
        Histogram,
        REGISTRY,
        start_http_server as _start_http,
    )
    from prometheus_client.registry import CollectorRegistry
 
    PROMETHEUS_AVAILABLE: bool = True
except ImportError:  # pragma: no cover — only absent when oct[metrics] not installed
    PROMETHEUS_AVAILABLE = False
    REGISTRY = None  # type: ignore[assignment]
    CollectorRegistry = None  # type: ignore[assignment,misc]
 
try:
    from oc_diagnostics import _dbg as _real_dbg
 
    def _dbg(*args, **kwargs) -> None:
        _real_dbg(*args, **kwargs)
except ImportError:  # pragma: no cover
    def _dbg(*args, **kwargs) -> None:
        return None
 
 
class McpMetrics:
    """Prometheus instrumentation for the oct-mcp pipeline.
 
    All record methods are no-ops when ``prometheus_client`` is absent.
    Instantiate once per server session and pass into :class:`ServerState`.
    """
 
    def __init__(self, registry: "CollectorRegistry | None" = None) -> None:
        """Build the metric vectors.
 
        Parameters
        ----------
        registry
            Prometheus :class:`CollectorRegistry` to register metrics on.
            ``None`` (default) uses the module-level global ``REGISTRY``,
            which is what ``prometheus_client.start_http_server()`` exposes
            on ``/metrics`` by default. Pass a private
            ``CollectorRegistry()`` for tests or multi-tenant scenarios
            that need isolation from other ``McpMetrics`` instances.
        """
        if not PROMETHEUS_AVAILABLE:
            return
 
        # Default to the global REGISTRY so production /metrics keeps
        # working without any caller change. Tests pass their own
        # CollectorRegistry to avoid `Duplicated timeseries` collisions.
        self._registry = registry if registry is not None else REGISTRY
 
        self._tool_calls: Counter = Counter(
            "oct_mcp_tool_calls_total",
            "Total MCP tool calls processed by oct-mcp",
            ["tool", "decision"],
            registry=self._registry,
        )
        self._tool_duration: Histogram = Histogram(
            "oct_mcp_tool_duration_seconds",
            "Wall-clock time from tool call receipt to response (seconds)",
            ["tool"],
            registry=self._registry,
        )
        self._rate_limit_hits: Counter = Counter(
            "oct_mcp_rate_limit_hits_total",
            "Tool calls rejected by the rate limiter",
            ["tool"],
            registry=self._registry,
        )
        self._redactions: Counter = Counter(
            "oct_mcp_redactions_total",
            "Redaction substitutions applied to tool output",
            ["tool"],
            registry=self._registry,
        )
        self._validation_failures: Counter = Counter(
            "oct_mcp_validation_failures_total",
            "Tool calls rejected by input validation",
            ["tool"],
            registry=self._registry,
        )
 
    # ------------------------------------------------------------------
    # Public API
    # ------------------------------------------------------------------
 
    def start_server(self, port: int) -> None:
        """Start the Prometheus ``/metrics`` HTTP server on *port*.
 
        Runs in a **daemon** background thread so it does not block the
        MCP event loop. Silently no-ops when ``prometheus_client`` is
        absent.
 
        Parameters
        ----------
        port
            TCP port to bind. Must be in the range 1–65535.
        """
        if not PROMETHEUS_AVAILABLE:
            return
        func = "McpMetrics.start_server"
        try:
            # Pass the same registry the metrics were built on; otherwise
            # an isolated-registry instance would expose nothing on /metrics
            # because start_http_server defaults to prometheus_client.REGISTRY.
            t = threading.Thread(
                target=_start_http,
                args=(port,),
                kwargs={"registry": self._registry},
                daemon=True,
            )
            t.start()
            _dbg("MCP", func, f"Prometheus metrics server started on port {port}", 2)
        except Exception as exc:  # pragma: no cover
            _dbg("MCP", func, f"SYSTEM_ERROR: failed to start metrics server: {exc}", 1)
 
    def record_call(
        self,
        tool: str,
        decision: str,
        duration_s: float,
        redactions: int,
    ) -> None:
        """Record a completed (or denied) tool call.
 
        Parameters
        ----------
        tool
            MCP tool name, e.g. ``"oct_lint"``.
        decision
            Pipeline decision: ``"allowed"``, ``"denied"``, or
            ``"requires_confirmation"``.
        duration_s
            Wall-clock duration in seconds.
        redactions
            Number of redaction substitutions applied to the output.
        """
        if not PROMETHEUS_AVAILABLE:
            return
        try:
            self._tool_calls.labels(tool=tool, decision=decision).inc()
            self._tool_duration.labels(tool=tool).observe(duration_s)
            if redactions > 0:
                self._redactions.labels(tool=tool).inc(redactions)
        except Exception:  # pragma: no cover
            pass
 
    def record_rate_limit_hit(self, tool: str) -> None:
        """Increment the rate-limit rejection counter for *tool*."""
        if not PROMETHEUS_AVAILABLE:
            return
        try:
            self._rate_limit_hits.labels(tool=tool).inc()
        except Exception:  # pragma: no cover
            pass
 
    def record_validation_failure(self, tool: str) -> None:
        """Increment the validation-failure counter for *tool*."""
        if not PROMETHEUS_AVAILABLE:
            return
        try:
            self._validation_failures.labels(tool=tool).inc()
        except Exception:  # pragma: no cover
            pass