diff --git a/common/lib/metrics.py b/common/lib/metrics.py index 3f10a16..6311384 100644 --- a/common/lib/metrics.py +++ b/common/lib/metrics.py @@ -15,6 +15,7 @@ meter = None counter_metrics = {} histogram_metrics = {} +observable_gauges = {} _otel_initialized = False logger = logging.getLogger(__name__) @@ -121,6 +122,43 @@ def increment_counter(metric_name, value=1, attributes=None): logger.warning(f"Failed to publish counter metric to OpenTelemetry: {e}") +def register_observable_gauge(metric_name, callback, description=None): + """Register an OpenTelemetry observable gauge. + + The callback is invoked by the OTel SDK on each metric-export tick + (interval governed by ``OTEL_METRIC_EXPORT_INTERVAL``). It receives a + single ``options`` argument (unused by callers here) and must return + an iterable of ``opentelemetry.metrics.Observation`` instances — one + per dimension combination. + + No-op when ``OTEL_EXPORTER_OTLP_ENDPOINT`` is unset (tests, dev) or when + a gauge of the same name has already been registered in this process. + + Args: + metric_name: Name of the metric (e.g. ``"conserver.ingress_list.length"``) + callback: Zero-arg callable returning an iterable of ``Observation`` + description: Optional description; defaults to a generic string + """ + # Lazy initialization on first use + _init_otel_metrics() + + if not OTEL_EXPORTER_OTLP_ENDPOINT or not meter: + return + + # Idempotent: a gauge is process-lifetime, so subsequent calls are no-ops. + if metric_name in observable_gauges: + return + + try: + observable_gauges[metric_name] = meter.create_observable_gauge( + name=metric_name, + callbacks=[lambda _options: callback()], + description=description or f"Observable gauge for {metric_name}", + ) + except Exception as e: + logger.warning(f"Failed to register OpenTelemetry observable gauge {metric_name!r}: {e}") + + def record_histogram(metric_name, value, attributes=None): """Record a value in a histogram metric in OpenTelemetry. diff --git a/common/lib/queue_metrics.py b/common/lib/queue_metrics.py new file mode 100644 index 0000000..c8df85b --- /dev/null +++ b/common/lib/queue_metrics.py @@ -0,0 +1,107 @@ +"""Observable gauge emitting current ingress-list and DLQ depths. + +The gauge ``conserver.ingress_list.length`` is sampled on every metric-export +tick. Each observation carries: + + - ``ingress_list`` — the configured ingress list name as it appears in the + chain config (e.g. ``"transcribe"``) + - ``kind`` — ``"ingress"`` for the live queue, ``"dlq"`` for the derived + dead-letter queue + +Splitting ``kind`` into its own attribute (rather than baking ``DLQ:`` into +the value) lets monitoring queries select live vs DLQ depth without regex +parsing the metric label. + +The ingress-list set is re-read from configuration on every tick, so chain +config changes propagate without a conserver restart. +""" + +from opentelemetry.metrics import Observation + +from dlq_utils import get_ingress_list_dlq_name +from lib.logging_utils import init_logger +from lib.metrics import register_observable_gauge + +logger = init_logger(__name__) + + +def _get_configured_ingress_lists(): + """Return the set of ingress list names from the live chain config. + + Imported lazily so test fixtures that monkey-patch ``Configuration`` + see the patched value. Returns an empty list on any config-read + failure so the gauge degrades to "no series" rather than crashing + the export tick. + """ + try: + from config import Configuration + + chains = Configuration.get_config().get("chains", {}) or {} + names = set() + for chain_config in chains.values(): + for name in chain_config.get("ingress_lists", []) or []: + if name: + names.add(name) + return sorted(names) + except Exception as e: + logger.warning("Failed to read ingress list config for gauge: %s", e) + return [] + + +def _build_callback(client): + """Return a zero-arg callback that yields one Observation per + ``(ingress_list, kind)`` combination. + + Each call performs one LLEN per series — typically a handful of cheap + Redis round-trips per export interval. Errors on individual LLENs are + logged and skipped; the export tick still publishes the series that + succeeded. + """ + + def _callback(): + observations = [] + for ingress_list in _get_configured_ingress_lists(): + for kind, key in ( + ("ingress", ingress_list), + ("dlq", get_ingress_list_dlq_name(ingress_list)), + ): + try: + length = client.llen(key) + except Exception as e: + logger.warning( + "LLEN failed for %s (ingress_list=%s, kind=%s): %s", + key, ingress_list, kind, e, + ) + continue + observations.append( + Observation( + value=length, + attributes={"ingress_list": ingress_list, "kind": kind}, + ) + ) + return observations + + return _callback + + +def register_ingress_list_length_gauge(client): + """Register the ``conserver.ingress_list.length`` observable gauge. + + Idempotent — safe to call multiple times per process. The callback + captures ``client`` by reference, so the same gauge instance follows + any Redis client swap done via the same ``client`` object. + + Args: + client: A Redis client with an ``llen(key)`` method. In production + this is the worker's ``redis_mgr`` client; in tests, a + ``MagicMock``. + """ + register_observable_gauge( + metric_name="conserver.ingress_list.length", + callback=_build_callback(client), + description=( + "Current Redis LLEN for each configured ingress list and its " + "derived DLQ. Attributes: ingress_list (configured name), " + "kind (ingress|dlq)." + ), + ) diff --git a/common/tests/test_queue_metrics.py b/common/tests/test_queue_metrics.py new file mode 100644 index 0000000..f3f581f --- /dev/null +++ b/common/tests/test_queue_metrics.py @@ -0,0 +1,152 @@ +"""Unit tests for the ``conserver.ingress_list.length`` observable gauge. + +The callback construction can be tested without any OpenTelemetry SDK +involvement — the helpers in ``lib.queue_metrics`` return a plain +zero-arg callable whose output is a list of ``Observation``. This file +exercises that callable directly. +""" + +from unittest.mock import MagicMock, patch + +from lib.queue_metrics import _build_callback, _get_configured_ingress_lists + + +class TestConfiguredIngressLists: + def test_empty_config_returns_empty_list(self): + with patch("config.Configuration.get_config", return_value={}): + assert _get_configured_ingress_lists() == [] + + def test_dedupes_and_sorts_ingress_lists_across_chains(self): + """Two chains can share an ingress list; the gauge should only see + each name once, and the order is deterministic so tests don't flap.""" + with patch( + "config.Configuration.get_config", + return_value={ + "chains": { + "main_chain": {"ingress_lists": ["default", "transcribe"]}, + "transcription_chain": {"ingress_lists": ["transcribe"]}, + "scitt_chain": {"ingress_lists": ["scitt_backfill"]}, + }, + }, + ): + assert _get_configured_ingress_lists() == [ + "default", + "scitt_backfill", + "transcribe", + ] + + def test_returns_empty_list_on_config_failure(self): + """A broken config read must degrade to 'no series' rather than + propagate an exception into the export tick.""" + with patch( + "config.Configuration.get_config", + side_effect=RuntimeError("config server down"), + ): + assert _get_configured_ingress_lists() == [] + + def test_skips_empty_ingress_list_names(self): + """Config edge case: tolerate empty strings or missing list.""" + with patch( + "config.Configuration.get_config", + return_value={ + "chains": { + "weird_chain": {"ingress_lists": ["", "transcribe", None]}, + }, + }, + ): + assert _get_configured_ingress_lists() == ["transcribe"] + + +class TestCallbackEmitsObservations: + def _make_client(self, llen_values): + """Build a Redis-shaped MagicMock that returns the given LLEN + result for each configured key.""" + client = MagicMock() + client.llen = MagicMock(side_effect=lambda key: llen_values[key]) + return client + + def test_yields_one_observation_per_ingress_and_dlq(self): + client = self._make_client( + { + "transcribe": 1234, + "DLQ:transcribe": 5, + "default": 0, + "DLQ:default": 0, + } + ) + cb = _build_callback(client) + + with patch( + "config.Configuration.get_config", + return_value={ + "chains": { + "main_chain": {"ingress_lists": ["default"]}, + "transcription_chain": {"ingress_lists": ["transcribe"]}, + }, + }, + ): + obs = cb() + + emitted = { + (o.attributes["ingress_list"], o.attributes["kind"]): o.value for o in obs + } + assert emitted == { + ("default", "ingress"): 0, + ("default", "dlq"): 0, + ("transcribe", "ingress"): 1234, + ("transcribe", "dlq"): 5, + } + + def test_attribute_value_is_bare_ingress_name_not_dlq_prefixed(self): + """Critical: the ``ingress_list`` attribute must match the + configured ingress name (``transcribe``) for BOTH the live and + DLQ observations. The ``kind`` attribute is what discriminates, + so alert specs read ``{kind="dlq"}`` clean instead of regex- + matching ``^DLQ:`` in the label value.""" + client = self._make_client({"transcribe": 0, "DLQ:transcribe": 0}) + cb = _build_callback(client) + + with patch( + "config.Configuration.get_config", + return_value={ + "chains": {"c": {"ingress_lists": ["transcribe"]}}, + }, + ): + obs = cb() + + for o in obs: + assert o.attributes["ingress_list"] == "transcribe" + assert {o.attributes["kind"] for o in obs} == {"ingress", "dlq"} + + def test_individual_llen_failure_skips_that_series_only(self): + """If LLEN errors on one key (e.g. transient Redis hiccup), the + callback must still emit the series that succeeded — never raise + from inside the export tick.""" + client = MagicMock() + def _llen(key): + if key == "DLQ:transcribe": + raise RuntimeError("transient redis error") + return 42 + client.llen.side_effect = _llen + + cb = _build_callback(client) + with patch( + "config.Configuration.get_config", + return_value={ + "chains": {"c": {"ingress_lists": ["transcribe"]}}, + }, + ): + obs = cb() + + emitted = {(o.attributes["ingress_list"], o.attributes["kind"]) for o in obs} + assert emitted == {("transcribe", "ingress")} + + def test_empty_config_emits_zero_observations(self): + """No ingress lists configured → callback returns an empty list, + SDK publishes nothing this tick. Never raises.""" + client = MagicMock() + cb = _build_callback(client) + with patch("config.Configuration.get_config", return_value={}): + obs = cb() + assert obs == [] + client.llen.assert_not_called() diff --git a/conserver/main.py b/conserver/main.py index d60acc0..b417202 100644 --- a/conserver/main.py +++ b/conserver/main.py @@ -781,7 +781,13 @@ def worker_loop(worker_id: int) -> None: # Re-initialize Redis client + queue in worker process r = redis_mgr.get_client() queue = VconQueue(r) - + + # Register the ingress-list / DLQ depth gauge. The callback inside reads + # the live chain config on every tick, so new ingress lists begin + # emitting series automatically without a worker restart. + from lib.queue_metrics import register_ingress_list_length_gauge + register_ingress_list_length_gauge(r) + # Re-register signal handler in worker process signal.signal(signal.SIGTERM, signal_handler) signal.signal(signal.SIGINT, signal_handler)