diff --git a/src/deriver/__main__.py b/src/deriver/__main__.py index c56ed6a0..9103f6f7 100644 --- a/src/deriver/__main__.py +++ b/src/deriver/__main__.py @@ -10,6 +10,7 @@ from src.db import engine, register_db_query_instrumentation from src.startup import validate_embedding_schema from src.telemetry import ( initialize_telemetry_async, + prometheus_metrics, register_db_pool_collector, shutdown_telemetry, ) @@ -25,6 +26,10 @@ def start_metrics_server() -> None: # Expose DB connection-pool stats for this deriver instance. register_db_pool_collector("deriver") register_db_query_instrumentation("deriver") + + # Pre-materialize bounded-label counter children at 0 so metrics are visible + # in Prometheus before the first event (no-op if metrics off). + prometheus_metrics.initialize_bounded_metrics(instance_type="deriver") logger.info("Prometheus metrics server started on port 9090") diff --git a/src/main.py b/src/main.py index 3c8bf3e8..95d316e5 100644 --- a/src/main.py +++ b/src/main.py @@ -109,6 +109,10 @@ async def lifespan(_: FastAPI): register_db_pool_collector("api") register_db_query_instrumentation("api") + # Pre-materialize bounded-label counter children at 0 so metrics are visible + # in Prometheus before the first event (no-op if metrics off). + prometheus_metrics.initialize_bounded_metrics(instance_type="api") + # Validate embedding schema before serving any traffic. Fails closed: if # the configured EMBEDDING_VECTOR_DIMENSIONS does not match the physical # pgvector columns, the process refuses to start rather than silently diff --git a/src/reconciler/sync_vectors.py b/src/reconciler/sync_vectors.py index 68e9ac59..2166c9ab 100644 --- a/src/reconciler/sync_vectors.py +++ b/src/reconciler/sync_vectors.py @@ -23,6 +23,7 @@ from src.config import settings from src.dependencies import tracked_db from src.embedding_client import embedding_client from src.exceptions import VectorStoreError +from src.telemetry import prometheus_metrics from src.telemetry.events import EmbeddingCallPurpose from src.utils.types import embedding_call_purpose from src.vector_store import VectorRecord, VectorStore, get_external_vector_store @@ -700,6 +701,30 @@ async def _cleanup_pgvector_batch( return True +async def _record_pending_embeddings_backlog() -> None: + """Set the pending-embeddings backlog gauge to the current count of + MessageEmbedding rows awaiting a vector (sync_state='pending'). + + Called at the end of each reconciliation cycle so the gauge reflects the + residual backlog after the sweep. Best-effort: a metrics/DB hiccup here must + never fail the reconciliation cycle. + """ + if not settings.METRICS.ENABLED: + return + try: + async with tracked_db("reconciler_pending_count", read_only=True) as db: + count = await db.scalar( + select(func.count()) + .select_from(models.MessageEmbedding) + .where(models.MessageEmbedding.sync_state == "pending") + ) + prometheus_metrics.set_message_embeddings_pending(count=count or 0) + except Exception: + logger.warning( + "Failed to record pending-embeddings backlog gauge", exc_info=True + ) + + async def run_vector_reconciliation_cycle() -> ReconciliationMetrics: """ Run a complete reconciliation cycle. @@ -729,6 +754,7 @@ async def run_vector_reconciliation_cycle() -> ReconciliationMetrics: if not (embs_work or cleanup_work): break logger.debug("Vector reconciliation cycle completed (pgvector mode)") + await _record_pending_embeddings_backlog() return metrics # External vector store mode - reconcile documents, embeddings, and cleanup @@ -756,4 +782,5 @@ async def run_vector_reconciliation_cycle() -> ReconciliationMetrics: break logger.debug("Vector reconciliation cycle completed") + await _record_pending_embeddings_backlog() return metrics diff --git a/src/telemetry/events/__init__.py b/src/telemetry/events/__init__.py index 000e70b4..dfba4084 100644 --- a/src/telemetry/events/__init__.py +++ b/src/telemetry/events/__init__.py @@ -138,9 +138,64 @@ __all__ = [ # Lifecycle "initialize_telemetry_events", "shutdown_telemetry_events", + # Zero-init registry + "ALL_EVENT_TYPES", + "HIGH_VOLUME_EVENT_TYPES", ] +# Explicit registry of CloudEvents `type` values, used to pre-materialize the +# `telemetry_events_emitted` / `telemetry_events_sampled_out` counter children at +# 0 so those metrics are visible in Prometheus before any event fires (see +# src/telemetry/prometheus/metrics.py:initialize_bounded_metrics and +# .meta design telemetry-counter-zero-init). +# +# ⚠️ When you add a new BaseEvent subclass, add its `_event_type` here (and to +# HIGH_VOLUME_EVENT_TYPES if `_volume_class == "high_volume"`). The drift-guard +# test tests/telemetry/test_metric_zero_init.py fails until you do — it asserts +# this registry equals the set discovered by walking BaseEvent subclasses. +ALL_EVENT_TYPES: tuple[str, ...] = ( + # api + "message.created", + "file.uploaded", + "context.retrieved", + # agent + "agent.iteration", + "agent.tool.conclusions.created", + "agent.tool.conclusions.deleted", + "agent.tool.peer_card.updated", + "agent.tool.summary.created", + "agent.tool.call.completed", + # deletion / dialectic / dream / representation + "deletion.completed", + "dialectic.completed", + "dream.run", + "dream.specialist", + "representation.completed", + # llm / embedding + "llm.call.completed", + "embedding.call.completed", + # reconciliation + "reconciliation.sync_vectors.completed", + "reconciliation.cleanup_stale_items.completed", + # trace stream (only emitted when TELEMETRY.TRACE_PAYLOADS_ENABLED, but they + # flow through the same emit() path and increment the same counters) + "llm.call.traced", + "embedding.call.traced", + "trace.content", +) + +# Subset of ALL_EVENT_TYPES whose `_volume_class == "high_volume"` — only these +# can ever be counted by `telemetry_events_sampled_out` (ground_truth events skip +# the sampler entirely). +HIGH_VOLUME_EVENT_TYPES: tuple[str, ...] = ( + "agent.iteration", + "agent.tool.call.completed", + "llm.call.completed", + "embedding.call.completed", +) + + def emit(event: BaseEvent) -> None: """Queue an event for emission to the telemetry backend. diff --git a/src/telemetry/prometheus/metrics.py b/src/telemetry/prometheus/metrics.py index 76a879ef..319a6d18 100644 --- a/src/telemetry/prometheus/metrics.py +++ b/src/telemetry/prometheus/metrics.py @@ -5,7 +5,7 @@ from __future__ import annotations import logging from collections.abc import Iterator from enum import Enum -from typing import cast, final +from typing import cast, final, get_args from prometheus_client import ( CONTENT_TYPE_LATEST, @@ -20,7 +20,7 @@ from prometheus_client.core import GaugeMetricFamily from starlette.requests import Request from starlette.responses import Response -from src.config import settings +from src.config import ReasoningLevel, settings disable_created_metrics() @@ -66,6 +66,24 @@ class DialecticComponents(Enum): TOTAL = "total" +# Bounded label domains used to zero-initialize counter children at startup (see +# initialize_bounded_metrics). REASONING_LEVELS is derived from the config +# Literal so it never drifts. +REASONING_LEVELS: tuple[str, ...] = get_args(ReasoningLevel) + +# Valid (token_type, component) pairs for deriver_tokens_processed. NOT the full +# cartesian product: input tokens only have input components, output tokens only +# OUTPUT_TOTAL. Enumerating the cartesian product would fabricate impossible +# always-0 series (e.g. output/prompt). Kept as an explicit literal, drift-guarded +# by tests/telemetry/test_metric_zero_init.py. +_DERIVER_TOKEN_COMBOS: tuple[tuple[str, str], ...] = ( + (TokenTypes.INPUT.value, DeriverComponents.PROMPT.value), + (TokenTypes.INPUT.value, DeriverComponents.MESSAGES.value), + (TokenTypes.INPUT.value, DeriverComponents.PREVIOUS_SUMMARY.value), + (TokenTypes.OUTPUT.value, DeriverComponents.OUTPUT_TOTAL.value), +) + + api_requests_counter = NamespacedCounter( "api_requests", "Total API requests", @@ -155,6 +173,16 @@ telemetry_buffer_size_gauge = NamespacedGauge( ["namespace"], ) +# Embedding backlog: MessageEmbedding rows still awaiting a vector +# (sync_state='pending'). Distinct from embed_now_tasks_in_flight (which counts +# in-flight fast-path work in the API process) — this is the durable, DB-wide +# backlog the reconciler drains. Set once per reconciliation cycle in the deriver. +message_embeddings_pending_gauge = NamespacedGauge( + "message_embeddings_pending", + "MessageEmbedding rows awaiting embedding (sync_state='pending')", + ["namespace"], +) + # DB connection-pool health. The in-flight gauge counts statements actually # executing on the wire, so checked_out minus in_flight reveals connections held # but parked (the "idle in transaction during an external call" antipattern). @@ -325,26 +353,109 @@ class PrometheusMetrics: except Exception as e: self._handle_metric_error("record_telemetry_event_dropped", e) + def _touch(self, counter: NamespacedCounter, **labels: str) -> None: + """Pre-create a counter child series at 0 without incrementing it. + + A labeled Prometheus counter exports no time series until its first + ``labels(...)`` call, so pre-touching a child keeps it present at 0 — + a missing series then signals a broken scrape rather than "no events". + Fail-soft (like the recorders): a bad init must never crash startup. + """ + try: + counter.labels(**labels) + except Exception as e: + self._handle_metric_error("_touch", e) + def initialize_telemetry_dropped_metrics(self, *, reasons: list[str]) -> None: """Pre-create telemetry_events_dropped child series at 0. - A labeled Prometheus counter exports no time series until its first - ``labels(...)`` call, so ``telemetry_events_dropped`` stays invisible in - Prometheus/Grafana until an event is actually dropped — you cannot alert - on or graph a metric that does not exist yet. Materializing the - ``(namespace, reason)`` children at startup keeps the metric present at 0, - so a missing series signals a broken scrape rather than "no drops". + ``telemetry_events_dropped`` stays invisible in Prometheus/Grafana until + an event is actually dropped — you cannot alert on or graph a metric that + does not exist yet. Materializing the ``(namespace, reason)`` children at + startup keeps the metric present at 0, so a missing series signals a broken + scrape rather than "no drops". + + Called per-emitter from ``TelemetryEmitter.start()`` because the reason + values are prefix-dependent (the trace emitter uses a ``trace_`` prefix), + which the process-level ``initialize_bounded_metrics`` does not know. Args: reasons: The reason label values the calling emitter can produce. """ for reason in reasons: - try: - telemetry_events_dropped_counter.labels(reason=reason) - except Exception as e: - self._handle_metric_error( - "initialize_telemetry_dropped_metrics", e - ) + self._touch(telemetry_events_dropped_counter, reason=reason) + + def initialize_bounded_metrics(self, *, instance_type: str) -> None: + """Pre-create bounded-label counter children at 0 for this process. + + See .meta design telemetry-counter-zero-init. Only counters whose full + label domain is bounded, enumerable at startup, and actually emitted by + THIS process are materialized; high-cardinality labels (endpoint, + workspace_name) and impossible label tuples are deliberately left absent. + + ``telemetry_events_dropped`` is handled separately, per-emitter, in + ``TelemetryEmitter.start()`` (prefix-dependent — see above). + + Args: + instance_type: "api" or "deriver" — selects the process-specific + counters. Event-type and buffer metrics are initialized in both. + """ + if not settings.METRICS.ENABLED: + return + + # Lazy import to avoid any import-time cycle (metrics is imported widely). + from src.telemetry.events import ALL_EVENT_TYPES, HIGH_VOLUME_EVENT_TYPES + + # --- Common: both processes run a TelemetryEmitter, so both emit their + # own subset of event types. The domain is bounded/low-cardinality (~21 + # types), so init the full set in each process rather than maintain a + # fragile per-event-type -> process map. + for event_type in ALL_EVENT_TYPES: + self._touch(telemetry_events_emitted_counter, type=event_type) + for event_type in HIGH_VOLUME_EVENT_TYPES: + self._touch(telemetry_events_sampled_out_counter, type=event_type) + telemetry_buffer_size_gauge.labels().set(0) + + if instance_type == "api": + # dialectic tokens: token_type x component(total) x reasoning_level + for token_type in TokenTypes: + for level in REASONING_LEVELS: + self._touch( + dialectic_tokens_processed_counter, + token_type=token_type.value, + component=DialecticComponents.TOTAL.value, + reasoning_level=level, + ) + # embed_now fast path runs as an API-process background task + self._touch(embed_now_tasks_shed_counter) + embed_now_tasks_in_flight_gauge.labels().set(0) + + elif instance_type == "deriver": + # deriver tokens: only the VALID (task_type, token_type, component) + # tuples (see _DERIVER_TOKEN_COMBOS). + for task_type in DeriverTaskTypes: + for token_type_value, component_value in _DERIVER_TOKEN_COMBOS: + self._touch( + deriver_tokens_processed_counter, + task_type=task_type.value, + token_type=token_type_value, + component=component_value, + ) + # dreamer tokens: specialist_name x token_type. Specialist names are + # derived from the concrete BaseSpecialist subclasses so a new + # specialist can't silently miss init. + from src.dreamer.specialists import BaseSpecialist + + for specialist in BaseSpecialist.__subclasses__(): + for token_type in TokenTypes: + self._touch( + dreamer_tokens_processed_counter, + specialist_name=specialist.name, + token_type=token_type.value, + ) + # embedding backlog gauge — set live each reconciliation cycle; init + # at 0 so it's visible before the first cycle. + message_embeddings_pending_gauge.labels().set(0) def set_telemetry_buffer_size(self, *, size: int) -> None: try: @@ -352,6 +463,12 @@ class PrometheusMetrics: except Exception as e: self._handle_metric_error("set_telemetry_buffer_size", e) + def set_message_embeddings_pending(self, *, count: int) -> None: + try: + message_embeddings_pending_gauge.labels().set(count) + except Exception as e: + self._handle_metric_error("set_message_embeddings_pending", e) + prometheus_metrics = PrometheusMetrics() diff --git a/tests/telemetry/test_metric_zero_init.py b/tests/telemetry/test_metric_zero_init.py new file mode 100644 index 00000000..0b68a6da --- /dev/null +++ b/tests/telemetry/test_metric_zero_init.py @@ -0,0 +1,269 @@ +"""Tests for startup zero-initialization of bounded-label metrics. + +Backfills the coverage PR #927 shipped without, and covers the generalization: +- bounded-label counter children are materialized at 0 before any event, +- high-cardinality / impossible label combinations are deliberately NOT, +- per-process init doesn't materialize the other process's counters, +- the explicit registries stay in sync with the source of truth (drift guards). + +Reads use ``REGISTRY.get_sample_value`` (returns the value if a series exists, +``None`` if it does not) rather than ``counter.labels(...)``, because ``.labels`` +would itself materialize the child and destroy the presence/absence signal. +""" + +from collections.abc import Iterator + +import pytest +from prometheus_client import REGISTRY + +from src.telemetry.events import ALL_EVENT_TYPES, HIGH_VOLUME_EVENT_TYPES +from src.telemetry.events.base import BaseEvent +from src.telemetry.prometheus.metrics import ( + _DERIVER_TOKEN_COMBOS, # pyright: ignore[reportPrivateUsage] + REASONING_LEVELS, + DeriverComponents, + DeriverTaskTypes, + DialecticComponents, + TokenTypes, + prometheus_metrics, +) + +NS = "test" + + +@pytest.fixture +def metrics_enabled(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]: + monkeypatch.setattr("src.config.settings.METRICS.ENABLED", True) + monkeypatch.setattr("src.config.settings.METRICS.NAMESPACE", NS) + yield + + +def sample(name: str, **labels: str) -> float | None: + """Value of a series if it exists, else None. Never materializes it.""" + return REGISTRY.get_sample_value(name, {"namespace": NS, **labels}) + + +# --------------------------------------------------------------------------- +# Drift guards (pure logic — no registry). These are the repo-visible collaborator +# notes: adding an event type / token component without updating the registry +# fails here with a pointer to what to fix. +# --------------------------------------------------------------------------- + + +def _walk_event_subclasses(cls: type[BaseEvent]) -> Iterator[type[BaseEvent]]: + for sub in cls.__subclasses__(): + yield sub + yield from _walk_event_subclasses(sub) + + +def test_all_event_types_registry_matches_subclasses(): + """ALL_EVENT_TYPES must equal every BaseEvent subclass's _event_type. + + If this fails you added/removed a BaseEvent subclass without updating + ALL_EVENT_TYPES in src/telemetry/events/__init__.py — its Prometheus counter + would not be zero-initialized. Update the registry. + """ + discovered = { + event_type + for cls in _walk_event_subclasses(BaseEvent) + if (event_type := getattr(cls, "_event_type", None)) is not None + } + assert set(ALL_EVENT_TYPES) == discovered + assert len(ALL_EVENT_TYPES) == len(set(ALL_EVENT_TYPES)), "duplicate event types" + + +def test_high_volume_registry_matches_subclasses(): + """HIGH_VOLUME_EVENT_TYPES must equal the high_volume-classed subclasses.""" + discovered = { + event_type + for cls in _walk_event_subclasses(BaseEvent) + if (event_type := getattr(cls, "_event_type", None)) is not None + and getattr(cls, "_volume_class", None) == "high_volume" + } + assert set(HIGH_VOLUME_EVENT_TYPES) == discovered + assert set(HIGH_VOLUME_EVENT_TYPES) <= set(ALL_EVENT_TYPES) + + +def test_deriver_token_combos_are_valid_and_complete(): + """Every combo uses real enum values, and every DeriverComponent is covered. + + Fails if a DeriverComponent is added to the enum without deciding which + token_type it pairs with in _DERIVER_TOKEN_COMBOS. + """ + valid_token_types = {t.value for t in TokenTypes} + valid_components = {c.value for c in DeriverComponents} + for token_type, component in _DERIVER_TOKEN_COMBOS: + assert token_type in valid_token_types + assert component in valid_components + # every component appears in exactly one combo + combo_components = {comp for _, comp in _DERIVER_TOKEN_COMBOS} + assert combo_components == valid_components + # the cartesian product would be larger — we intentionally enumerate fewer + assert len(_DERIVER_TOKEN_COMBOS) < len(valid_token_types) * len(valid_components) + + +# --------------------------------------------------------------------------- +# API-process zero-init +# --------------------------------------------------------------------------- + + +@pytest.mark.usefixtures("metrics_enabled") +def test_api_init_materializes_event_type_children(): + prometheus_metrics.initialize_bounded_metrics(instance_type="api") + for event_type in ALL_EVENT_TYPES: + assert sample("telemetry_events_emitted_total", type=event_type) is not None + for event_type in HIGH_VOLUME_EVENT_TYPES: + assert sample("telemetry_events_sampled_out_total", type=event_type) is not None + + +@pytest.mark.usefixtures("metrics_enabled") +def test_api_init_materializes_dialectic_and_embed(): + prometheus_metrics.initialize_bounded_metrics(instance_type="api") + for token_type in TokenTypes: + for level in REASONING_LEVELS: + assert ( + sample( + "dialectic_tokens_processed_total", + token_type=token_type.value, + component=DialecticComponents.TOTAL.value, + reasoning_level=level, + ) + is not None + ) + assert sample("embed_now_tasks_shed_total") is not None + assert sample("embed_now_tasks_in_flight") == 0.0 # gauge, explicit .set(0) + + +@pytest.mark.usefixtures("metrics_enabled") +def test_sampled_out_excludes_ground_truth_event_types(): + """Ground-truth events can never be sampled out, so their sampled_out series + must NOT be pre-created (they'd be permanently misleading zeros).""" + prometheus_metrics.initialize_bounded_metrics(instance_type="api") + ground_truth = set(ALL_EVENT_TYPES) - set(HIGH_VOLUME_EVENT_TYPES) + for event_type in ground_truth: + assert sample("telemetry_events_sampled_out_total", type=event_type) is None + + +# --------------------------------------------------------------------------- +# Deriver-process zero-init +# --------------------------------------------------------------------------- + + +@pytest.mark.usefixtures("metrics_enabled") +def test_deriver_init_materializes_token_and_backlog(): + prometheus_metrics.initialize_bounded_metrics(instance_type="deriver") + for task_type in DeriverTaskTypes: + for token_type, component in _DERIVER_TOKEN_COMBOS: + assert ( + sample( + "deriver_tokens_processed_total", + task_type=task_type.value, + token_type=token_type, + component=component, + ) + is not None + ) + # dreamer specialists are derived from the concrete BaseSpecialist subclasses + for specialist_name in ("deduction", "induction"): + assert ( + sample( + "dreamer_tokens_processed_total", + specialist_name=specialist_name, + token_type=TokenTypes.INPUT.value, + ) + is not None + ) + assert sample("message_embeddings_pending") == 0.0 # gauge zero-init + + +@pytest.mark.usefixtures("metrics_enabled") +def test_deriver_init_omits_impossible_token_combos(): + """The cartesian product includes combos that never occur (e.g. output tokens + with an input component). Those must not be materialized.""" + prometheus_metrics.initialize_bounded_metrics(instance_type="deriver") + assert ( + sample( + "deriver_tokens_processed_total", + task_type=DeriverTaskTypes.INGESTION.value, + token_type=TokenTypes.OUTPUT.value, + component=DeriverComponents.PROMPT.value, + ) + is None + ) + # base specialist is abstract and never emits — must not be materialized + assert ( + sample( + "dreamer_tokens_processed_total", + specialist_name="base", + token_type=TokenTypes.INPUT.value, + ) + is None + ) + + +# --------------------------------------------------------------------------- +# High-cardinality counters are left open, and per-process isolation holds +# --------------------------------------------------------------------------- + + +@pytest.mark.usefixtures("metrics_enabled") +def test_high_cardinality_counters_not_materialized(): + prometheus_metrics.initialize_bounded_metrics(instance_type="api") + prometheus_metrics.initialize_bounded_metrics(instance_type="deriver") + # no endpoint/workspace_name series fabricated + assert ( + sample( + "api_requests_total", + method="GET", + endpoint="/v3/does-not-exist", + status_code="200", + ) + is None + ) + assert sample("messages_created_total", workspace_name="nope_ws") is None + + +@pytest.mark.usefixtures("metrics_enabled") +def test_api_init_does_not_touch_deriver_counters(): + """api-only init must not materialize or change a deriver-only counter. + + Delta-based (before == after) so it's robust to prior tests having + materialized the series. + """ + labels = dict( + task_type=DeriverTaskTypes.INGESTION.value, + token_type=TokenTypes.INPUT.value, + component=DeriverComponents.PROMPT.value, + ) + before = sample("deriver_tokens_processed_total", **labels) + prometheus_metrics.initialize_bounded_metrics(instance_type="api") + after = sample("deriver_tokens_processed_total", **labels) + assert before == after + + +# --------------------------------------------------------------------------- +# Dropped-counter backfill (#927 shipped without a test) +# --------------------------------------------------------------------------- + + +@pytest.mark.usefixtures("metrics_enabled") +def test_dropped_counter_children_materialized(): + prometheus_metrics.initialize_telemetry_dropped_metrics( + reasons=["buffer_full", "send_failed"] + ) + assert sample("telemetry_events_dropped_total", reason="buffer_full") is not None + assert sample("telemetry_events_dropped_total", reason="send_failed") is not None + + +def test_init_noop_when_metrics_disabled(monkeypatch: pytest.MonkeyPatch): + """With metrics disabled, init must not fabricate series for a fresh label.""" + monkeypatch.setattr("src.config.settings.METRICS.ENABLED", False) + monkeypatch.setattr("src.config.settings.METRICS.NAMESPACE", "disabled_ns") + prometheus_metrics.initialize_bounded_metrics(instance_type="api") + assert ( + REGISTRY.get_sample_value( + "telemetry_events_emitted_total", + {"namespace": "disabled_ns", "type": "message.created"}, + ) + is None + )