diff --git a/src/telemetry/prometheus/metrics.py b/src/telemetry/prometheus/metrics.py index 319a6d18..d271b4f7 100644 --- a/src/telemetry/prometheus/metrics.py +++ b/src/telemetry/prometheus/metrics.py @@ -71,17 +71,27 @@ class DialecticComponents(Enum): # 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), -) +# Valid (token_type, component) pairs for deriver_tokens_processed, per task_type. +# NOT the cartesian product: input tokens only pair with input components, output +# only with OUTPUT_TOTAL, and PREVIOUS_SUMMARY occurs only for summary tasks +# (ingestion has no previous summary). Enumerating anything broader would +# fabricate impossible always-0 series (e.g. output/prompt, or +# ingestion/previous_summary). Explicit literal, drift-guarded by +# tests/telemetry/test_metric_zero_init.py. Sources: track_deriver_input_tokens +# (utils/tokens.py) + the OUTPUT_TOTAL sites in deriver.py / summarizer.py. +_DERIVER_TOKEN_COMBOS_BY_TASK: dict[str, tuple[tuple[str, str], ...]] = { + DeriverTaskTypes.INGESTION.value: ( + (TokenTypes.INPUT.value, DeriverComponents.PROMPT.value), + (TokenTypes.INPUT.value, DeriverComponents.MESSAGES.value), + (TokenTypes.OUTPUT.value, DeriverComponents.OUTPUT_TOTAL.value), + ), + DeriverTaskTypes.SUMMARY.value: ( + (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( @@ -366,6 +376,14 @@ class PrometheusMetrics: except Exception as e: self._handle_metric_error("_touch", e) + def _set_gauge_zero(self, gauge: NamespacedGauge) -> None: + """Materialize a (namespace-only) gauge at 0. Fail-soft, like ``_touch``: + a startup init must never propagate an exception into process boot.""" + try: + gauge.labels().set(0) + except Exception as e: + self._handle_metric_error("_set_gauge_zero", e) + def initialize_telemetry_dropped_metrics(self, *, reasons: list[str]) -> None: """Pre-create telemetry_events_dropped child series at 0. @@ -414,7 +432,7 @@ class PrometheusMetrics: 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) + self._set_gauge_zero(telemetry_buffer_size_gauge) if instance_type == "api": # dialectic tokens: token_type x component(total) x reasoning_level @@ -428,16 +446,17 @@ class PrometheusMetrics: ) # 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) + self._set_gauge_zero(embed_now_tasks_in_flight_gauge) 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: + # deriver tokens: only the VALID (token_type, component) tuples per + # task_type (see _DERIVER_TOKEN_COMBOS_BY_TASK) — never the cartesian + # product, which would fabricate impossible always-0 series. + for task_type_value, combos in _DERIVER_TOKEN_COMBOS_BY_TASK.items(): + for token_type_value, component_value in combos: self._touch( deriver_tokens_processed_counter, - task_type=task_type.value, + task_type=task_type_value, token_type=token_type_value, component=component_value, ) @@ -455,7 +474,7 @@ class PrometheusMetrics: ) # 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) + self._set_gauge_zero(message_embeddings_pending_gauge) def set_telemetry_buffer_size(self, *, size: int) -> None: try: diff --git a/tests/telemetry/test_metric_zero_init.py b/tests/telemetry/test_metric_zero_init.py index 0b68a6da..e71fa57f 100644 --- a/tests/telemetry/test_metric_zero_init.py +++ b/tests/telemetry/test_metric_zero_init.py @@ -19,7 +19,7 @@ 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] + _DERIVER_TOKEN_COMBOS_BY_TASK, # pyright: ignore[reportPrivateUsage] REASONING_LEVELS, DeriverComponents, DeriverTaskTypes, @@ -85,21 +85,35 @@ def test_high_volume_registry_matches_subclasses(): def test_deriver_token_combos_are_valid_and_complete(): - """Every combo uses real enum values, and every DeriverComponent is covered. + """Every combo uses real enum values; the union across tasks covers every + DeriverComponent; and no task enumerates an impossible pair. - Fails if a DeriverComponent is added to the enum without deciding which - token_type it pairs with in _DERIVER_TOKEN_COMBOS. + Fails if a DeriverComponent/DeriverTaskType is added without deciding which + task_type + token_type it pairs with in _DERIVER_TOKEN_COMBOS_BY_TASK. """ 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) + valid_task_types = {t.value for t in DeriverTaskTypes} + + assert set(_DERIVER_TOKEN_COMBOS_BY_TASK) == valid_task_types + all_components: set[str] = set() + for task_type, combos in _DERIVER_TOKEN_COMBOS_BY_TASK.items(): + assert task_type in valid_task_types + for token_type, component in combos: + assert token_type in valid_token_types + assert component in valid_components + # each task enumerates fewer than its cartesian product (no impossible pairs) + assert len(combos) < len(valid_token_types) * len(valid_components) + all_components.update(comp for _, comp in combos) + + # every component is reachable via some task + assert all_components == valid_components + # previous_summary is summary-only: ingestion must NOT enumerate it + ingestion = _DERIVER_TOKEN_COMBOS_BY_TASK[DeriverTaskTypes.INGESTION.value] + assert ( + TokenTypes.INPUT.value, + DeriverComponents.PREVIOUS_SUMMARY.value, + ) not in ingestion # --------------------------------------------------------------------------- @@ -152,12 +166,12 @@ def test_sampled_out_excludes_ground_truth_event_types(): @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: + for task_type, combos in _DERIVER_TOKEN_COMBOS_BY_TASK.items(): + for token_type, component in combos: assert ( sample( "deriver_tokens_processed_total", - task_type=task_type.value, + task_type=task_type, token_type=token_type, component=component, ) @@ -181,6 +195,7 @@ 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") + # output tokens never pair with an input component assert ( sample( "deriver_tokens_processed_total", @@ -190,6 +205,16 @@ def test_deriver_init_omits_impossible_token_combos(): ) is None ) + # previous_summary is summary-only — ingestion must not materialize it + assert ( + sample( + "deriver_tokens_processed_total", + task_type=DeriverTaskTypes.INGESTION.value, + token_type=TokenTypes.INPUT.value, + component=DeriverComponents.PREVIOUS_SUMMARY.value, + ) + is None + ) # base specialist is abstract and never emits — must not be materialized assert ( sample(