orchestrata_ultimate.gateway.metrics

Reference › orchestrata_ultimate › gateway › metrics

metrics defines the Prometheus instruments of the gateway (SPEC-014 US-009) behind a set of small recording functions. session calls connection_opened and connection_closed, protocol calls message_received and message_sent, rpc calls rpc_latency with the method name and elapsed seconds, pubsub calls subscription_added and subscription_removed with the topic, and auth_handler calls auth_success and auth_failure. The /metrics route in http_surface exposes the result.

_get_or_create registers each instrument once and reuses it afterwards, so importing the module repeatedly does not duplicate metrics. Every helper runs through _safe, which logs a warning and never raises, so a fault in instrumentation cannot break the gateway operation it measures. See gateway.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/gateway/metrics.py

"""Prometheus instruments for the WebSocket gateway (SPEC-014 US-009).

Registers the eight ``orchestrata_gateway_*`` metrics named in
REQ-014-142..149 on the ``prometheus_client`` DEFAULT registry -- the same
registry ``http_surface.py``'s ``/metrics`` endpoint serves via
``generate_latest()`` (REQ-014-141), so anything registered here is
scrape-visible with no extra wiring.

Registration is idempotent (get-or-create by name): re-importing this
module, or constructing the gateway more than once in the same process
(parallel test workers, repeated fixtures), must never raise
prometheus_client's "Duplicated timeseries in CollectorRegistry".

Every ``record_*``/``observe_*`` helper below swallows its own failures --
a metrics-recording bug must never break the gateway operation it is
instrumenting (session design ruling, US-009).

Naming note (REQ-014-143): ``orchestrata_gateway_total_connections`` has
"total" in the middle of the name rather than as an OpenMetrics-style
trailing ``_total`` suffix. ``prometheus_client``'s ``Counter`` type
*unconditionally* appends ``_total`` to both the rendered sample name and
the ``# TYPE``/``# HELP`` comment lines (see
``prometheus_client.exposition.generate_latest``), which would render this
metric as ``orchestrata_gateway_total_connections_total`` -- violating the
exact-name contract. A ``Gauge`` renders the given name verbatim, so this
one instrument is implemented as a ``Gauge`` that is only ever
incremented (never decremented), i.e. a monotonic running total. Every
other instrument here uses the prometheus_client type its REQ names.
"""

from __future__ import annotations

import logging
from collections.abc import Callable
from typing import Any

from prometheus_client import REGISTRY, Counter, Gauge, Histogram

from orchestrata_ultimate.core.base.traces import traces

logger = logging.getLogger(__name__)


def _safe(op: Callable[[], None], warning: str) -> None:
    """Run *op*, logging (never raising) on failure.

    Every ``record_*``/``observe_*`` helper below goes through here -- a
    bug in the metrics layer must never break the gateway operation it is
    instrumenting (US-009 design ruling).
    """
    try:
        op()
    except Exception:
        logger.warning(warning, exc_info=True)


def _get_or_create(
    metric_cls: type[Any],
    name: str,
    documentation: str,
    labelnames: tuple[str, ...] = (),
) -> Any:
    """Return the named collector, creating it once on the default registry.

    ``prometheus_client`` raises ``ValueError`` on a second registration of
    the same metric name. This module is imported by both ``http_surface``
    and the modules that record against these instruments, and tests may
    reimport/reload it -- so look the collector up by name first, and fall
    back to catching the race if two callers create it concurrently.
    """
    existing = REGISTRY._names_to_collectors.get(name)
    if existing is not None:
        return existing
    try:
        return metric_cls(name, documentation, labelnames)
    except ValueError:
        return REGISTRY._names_to_collectors[name]


CONNECTED_CLIENTS = _get_or_create(
    Gauge,
    "orchestrata_gateway_connected_clients",
    "Current number of active WebSocket connections to the gateway.",
)

# See module docstring: implemented as a Gauge (inc-only) so the exact
# name REQ-014-143 pins is rendered verbatim -- a real Counter would
# render this as "..._total_total"-adjacent (the auto-appended "_total"
# stacking on the word "total" already in the name).
TOTAL_CONNECTIONS = _get_or_create(
    Gauge,
    "orchestrata_gateway_total_connections",
    "Total WebSocket connections accepted by the gateway since start "
    "(monotonic; never decremented).",
)

MESSAGES_RECEIVED = _get_or_create(
    Counter,
    "orchestrata_gateway_messages_received_total",
    "Total inbound WebSocket messages processed by the gateway.",
)

MESSAGES_SENT = _get_or_create(
    Counter,
    "orchestrata_gateway_messages_sent_total",
    "Total outbound WebSocket messages sent by the gateway.",
)

RPC_LATENCY_SECONDS = _get_or_create(
    Histogram,
    "orchestrata_gateway_rpc_latency_seconds",
    "RPC dispatch latency in seconds, labeled by method.",
    ("method",),
)

SUBSCRIPTION_COUNT = _get_or_create(
    Gauge,
    "orchestrata_gateway_subscription_count",
    "Current number of active subscriptions, labeled by topic.",
    ("topic",),
)

AUTH_SUCCESS = _get_or_create(
    Counter,
    "orchestrata_gateway_auth_success_total",
    "Total successful gateway authentication attempts.",
)

AUTH_FAILURE = _get_or_create(
    Counter,
    "orchestrata_gateway_auth_failure_total",
    "Total failed gateway authentication attempts.",
)


@traces(spec="014", story="US-009", reqs=["REQ-014-142", "REQ-014-143"])
def connection_opened() -> None:
    """Record a newly registered WebSocket connection."""

    def _do() -> None:
        CONNECTED_CLIENTS.inc()
        TOTAL_CONNECTIONS.inc()

    _safe(_do, "Failed to record connection-opened metric")


@traces(spec="014", story="US-009", reqs=["REQ-014-142"])
def connection_closed() -> None:
    """Record a WebSocket connection going away."""
    _safe(CONNECTED_CLIENTS.dec, "Failed to record connection-closed metric")


@traces(spec="014", story="US-009", reqs=["REQ-014-144"])
def message_received() -> None:
    """Record one inbound WebSocket frame."""
    _safe(MESSAGES_RECEIVED.inc, "Failed to record message-received metric")


@traces(spec="014", story="US-009", reqs=["REQ-014-145"])
def message_sent() -> None:
    """Record one outbound WebSocket frame."""
    _safe(MESSAGES_SENT.inc, "Failed to record message-sent metric")


@traces(spec="014", story="US-009", reqs=["REQ-014-146"])
def rpc_latency(method: str, seconds: float) -> None:
    """Record one RPC dispatch's latency, labeled by method.

    Args:
        method: RPC method name (e.g. ``"gateway.ping"``).
        seconds: Wall-clock dispatch duration in seconds.
    """
    _safe(
        lambda: RPC_LATENCY_SECONDS.labels(method=method).observe(seconds),
        "Failed to record rpc-latency metric",
    )


@traces(spec="014", story="US-009", reqs=["REQ-014-147"])
def subscription_added(topic: str) -> None:
    """Record one new subscription on *topic*.

    Args:
        topic: The topic pattern the subscription was created for.
    """
    _safe(
        lambda: SUBSCRIPTION_COUNT.labels(topic=topic).inc(),
        "Failed to record subscription-added metric",
    )


@traces(spec="014", story="US-009", reqs=["REQ-014-147"])
def subscription_removed(topic: str) -> None:
    """Record one subscription removed from *topic*.

    Args:
        topic: The topic pattern the subscription was removed from.
    """
    _safe(
        lambda: SUBSCRIPTION_COUNT.labels(topic=topic).dec(),
        "Failed to record subscription-removed metric",
    )


@traces(spec="014", story="US-009", reqs=["REQ-014-148"])
def auth_success() -> None:
    """Record a successful authentication verdict."""
    _safe(AUTH_SUCCESS.inc, "Failed to record auth-success metric")


@traces(spec="014", story="US-009", reqs=["REQ-014-149"])
def auth_failure() -> None:
    """Record a failed authentication verdict."""
    _safe(AUTH_FAILURE.inc, "Failed to record auth-failure metric")