feat(metrics): emit conserver.ingress_list.length observable gauge

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.
 

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)."
+        ),
+    )

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()

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)