fix(metrics): initialize counters before first scrape

Author: linhongyu510

openviking/metrics/collectors/base.py

@@ -122,6 +122,27 @@ def inc_counter(
             label_names=final_label_names,
         )
 
+    def initialize_counter(
+        self,
+        name: str,
+        *,
+        labels: dict[str, str] | None = None,
+        label_names: tuple[str, ...] = (),
+        account_id: str | None = None,
+    ) -> None:
+        """Initialize a zero-valued counter after normalizing account-aware labels."""
+        final_labels, final_label_names = self._with_account_labels(
+            metric_name=name,
+            labels=labels,
+            label_names=label_names,
+            explicit_account_id=account_id,
+        )
+        self._registry.initialize_counter(
+            name,
+            labels=final_labels,
+            label_names=final_label_names,
+        )
+
     def set_gauge(
         self,
         name: str,

openviking/metrics/collectors/queue.py

@@ -56,6 +56,16 @@ def collect_hook(self, registry, metric_input) -> None:
         statuses = metric_input
         for queue_name, status in statuses.items():
             labels = {"queue": str(queue_name)}
+            registry.initialize_counter(
+                self.PROCESSED_TOTAL,
+                labels=labels,
+                label_names=("queue",),
+            )
+            registry.initialize_counter(
+                self.ERRORS_TOTAL,
+                labels=labels,
+                label_names=("queue",),
+            )
             registry.set_gauge(
                 self.PENDING,
                 float(status.pending),

openviking/metrics/collectors/resource.py

@@ -20,7 +20,7 @@
 
 from openviking.metrics.core.base import MetricCollector
 
-from .base import EventMetricCollector
+from .base import CollectorMetricWriter, EventMetricCollector
 
 
 @dataclass
@@ -48,10 +48,21 @@ class ResourceIngestionCollector(EventMetricCollector):
     )
 
     SUPPORTED_EVENTS: ClassVar[frozenset[str]] = frozenset({"resource.stage", "resource.wait"})
+    STAGES: ClassVar[tuple[str, ...]] = ("parse", "finalize", "persist", "summarize")
+    STAGE_STATUSES: ClassVar[tuple[str, ...]] = ("ok", "error")
 
     def collect(self, registry=None) -> None:
-        """Implement the collector interface as a no-op because resource metrics are push-driven."""
-        return None
+        """Initialize bounded stage outcome series before the first resource event."""
+        if registry is None:
+            return
+        writer = CollectorMetricWriter(registry)
+        for stage in self.STAGES:
+            for status in self.STAGE_STATUSES:
+                writer.initialize_counter(
+                    self.STAGE_TOTAL,
+                    labels={"stage": stage, "status": status},
+                    label_names=("stage", "status"),
+                )
 
     def receive_hook(self, event_name: str, payload: dict, registry) -> None:
         """

openviking/metrics/core/registry.py

@@ -106,6 +106,23 @@ def inc_counter(
         """
         self.counter(name, label_names=label_names).inc(labels=labels, amount=float(amount))
 
+    def initialize_counter(
+        self,
+        name: str,
+        *,
+        labels: Mapping[str, str] | None = None,
+        label_names: Sequence[str] = (),
+    ) -> None:
+        """
+        Materialize a Counter series at zero without treating zero as an increment.
+
+        Args:
+            name: Prometheus metric name.
+            labels: Optional label dict. Keys must exactly match `label_names`.
+            label_names: Ordered label key tuple for this metric name.
+        """
+        self.counter(name, label_names=label_names).initialize(labels=labels)
+
     def set_gauge(
         self,
         name: str,
@@ -376,6 +393,17 @@ def inc(self, *, labels: Mapping[str, str] | None, amount: float) -> None:
                 return
             self._values[key] = self._values.get(key, 0.0) + float(amount)
 
+    def initialize(self, *, labels: Mapping[str, str] | None) -> None:
+        """Materialize one counter series at zero while preserving any existing value."""
+        key = self._normalize_and_validate(labels)
+        with self._lock:
+            if key in self._values:
+                return
+            if len(self._values) >= self._max_series:
+                self._on_drop(self.name)
+                return
+            self._values[key] = 0.0
+
     def copy_values(self) -> dict[tuple[tuple[str, str], ...], float]:
         """Return a detached copy of all series values in this family."""
         with self._lock:
@@ -620,6 +648,10 @@ def inc(self, amount: float = 1.0, *, labels: Mapping[str, str] | None = None) -
         """Increment one series in the bound counter family using the public wrapper API."""
         self._family.inc(labels=labels, amount=amount)
 
+    def initialize(self, *, labels: Mapping[str, str] | None = None) -> None:
+        """Materialize one series at zero without weakening positive-increment validation."""
+        self._family.initialize(labels=labels)
+
 
 class _Gauge:
     """Public lightweight handle used by callers to mutate one gauge family."""
@@ -636,6 +668,7 @@ def inc(self, amount: float = 1.0, *, labels: Mapping[str, str] | None = None) -
         """Increase one series in the bound gauge family by a positive delta."""
         self._family.add(labels=labels, delta=amount)
 
+
 class _Histogram:
     """Public lightweight handle used by callers to mutate one histogram family."""
 

openviking/metrics/global_api.py

@@ -301,6 +301,7 @@ def _build_event_router(registry: MetricRegistry) -> EventCollectorRouter:
     vlm_collector = VLMCollector()
     session_collector = SessionCollector()
     resource_collector = ResourceIngestionCollector()
+    resource_collector.collect(registry)
     retrieval_collector = RetrievalCollector()
     encryption_collector = EncryptionCollector()
     telemetry_bridge_collector = TelemetryBridgeCollector()

tests/metrics/collectors/test_state_collectors.py

@@ -45,6 +45,7 @@ async def check_status(self):
     assert 'openviking_queue_in_progress{queue="semantic"} 1.0' in text
     assert 'openviking_queue_processed_total{queue="semantic"} 10' in text
     assert 'openviking_queue_errors_total{queue="semantic"} 2' in text
+    assert 'openviking_queue_errors_total{queue="embedding"} 0' in text
 
 
 def test_task_tracker_collector_maps_counts(monkeypatch):

tests/metrics/core/test_registry.py

@@ -64,6 +64,22 @@ def test_counter_only_increases():
         c.inc(amount=-1)
 
 
+def test_counter_can_initialize_labeled_series_at_zero(render_prometheus):
+    registry = MetricRegistry()
+    c = registry.counter("openviking_initialized_total", label_names=("status",))
+
+    c.initialize(labels={"status": "error"})
+    assert 'openviking_initialized_total{status="error"} 0' in render_prometheus(registry)
+
+    c.inc(amount=2, labels={"status": "error"})
+    c.initialize(labels={"status": "error"})
+
+    text = render_prometheus(registry)
+    assert 'openviking_initialized_total{status="error"} 2' in text
+    with pytest.raises(ValueError):
+        c.inc(amount=0, labels={"status": "error"})
+
+
 def test_histogram_boundary_bucket(registry, render_prometheus):
     h = registry.histogram("openviking_latency_seconds", buckets=(0.05, 0.1))
     h.observe(0.05)

tests/metrics/integration/test_bootstrap.py

@@ -8,6 +8,7 @@
 import pytest
 
 import openviking.metrics.bootstrap as bootstrap
+import openviking.metrics.global_api as global_api
 from openviking.metrics.account_dimension import (
     configure_metric_account_dimension,
     reset_metric_account_dimension,
@@ -96,6 +97,18 @@ def _boom():
         bootstrap.create_default_collector_manager(app=None, service=None)
 
 
+def test_event_router_initializes_resource_stage_counters():
+    registry = MetricRegistry()
+
+    global_api._build_event_router(registry)
+
+    text = PrometheusExporter(registry=registry).render()
+    assert (
+        'openviking_resource_stage_total{account_id="__unknown__",stage="parse",status="error"} 0'
+        in text
+    )
+
+
 def test_optional_cache_datasource_instrumentation_is_wired_into_key_call_sites():
     project_root = Path(__file__).resolve().parents[3]
     targets = [