RomanNum3ral
/
honcho
mirror of https://github.com/plastic-labs/honcho.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

telemetry: generalize counter zero-init to all bounded-label counters 

Extends #927 (which zero-inited telemetry_events_dropped) to every counter
whose label domain is bounded and known at startup, so metrics are present in
Prometheus before their first event — a missing series then signals a broken
scrape rather than "nothing happened yet".

- add initialize_bounded_metrics(instance_type) on PrometheusMetrics; call it
  per-process from main.py (api) and deriver/__main__.py (deriver).
- extract a shared _touch() helper; refactor initialize_telemetry_dropped_metrics
  onto it (that one stays per-emitter in start() — it's prefix-dependent).
- explicit ALL_EVENT_TYPES / HIGH_VOLUME_EVENT_TYPES registry in telemetry.events,
  drift-guarded by tests that walk BaseEvent subclasses.
- only VALID (task_type, token_type, component) tuples for deriver_tokens (the
  cartesian product would fabricate impossible always-0 series); only high-volume
  event types for sampled_out; high-cardinality labels (endpoint, workspace_name)
  left open.
- gauges: zero-init embed_now_tasks_in_flight + telemetry_buffer_size; add a new
  message_embeddings_pending backlog gauge, set each reconciliation cycle and
  zero-inited at deriver startup (Rajat's pending/in-flight ask).
- backfills the tests #927 shipped without.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Phil 2026-07-23 09:41:29 -04:00
parent 0f47e84993
commit fb8543a0c2
6 changed files with 491 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
src/deriver/__main__.py
View File

@ -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")

4
src/main.py
View File

@ -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

27
src/reconciler/sync_vectors.py
View File

@ -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

55
src/telemetry/events/__init__.py
View File

@ -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.

145
src/telemetry/prometheus/metrics.py
View File

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

269
tests/telemetry/test_metric_zero_init.py Normal file
View File

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