Merge pull request #815 from ZacLou/issue-727-ingestion-throughput-metrics

feat(ingestion): per-batch progress and throughput metrics (#727)

Author: gelluisaac

astroml/ingestion/batch_metrics.py

@@ -0,0 +1,69 @@
+"""Per-batch ingestion progress/throughput metrics recorder (Issue #727)."""
+
+from __future__ import annotations
+
+import time
+from dataclasses import dataclass
+from typing import Any
+
+from astroml.ingestion.metrics import (
+    INGESTION_BATCH_DURATION_SECONDS,
+    INGESTION_BATCH_LEDGERS,
+    INGESTION_BATCH_THROUGHPUT,
+)
+
+
+@dataclass
+class BatchCounters:
+    """Mutable counters for a single batch window."""
+
+    processed: int = 0
+    skipped: int = 0
+    errors: int = 0
+
+    def observe(self, outcome: Any) -> None:
+        """Record one ledger outcome.
+
+        Args:
+            outcome: An object with a ``status`` attribute of either
+                ``"processed"``, ``"skipped"``, or ``"error"``.
+        """
+        status = getattr(outcome, "status", "unknown")
+        if status == "processed":
+            self.processed += 1
+        elif status == "skipped":
+            self.skipped += 1
+        elif status == "error":
+            self.errors += 1
+
+
+class BatchMetricsRecorder:
+    """Records per-batch ingestion metrics.
+
+    Call :meth:`start` at the beginning of a batch, :meth:`observe` for each
+    ledger outcome, and :meth:`finish` when the batch ends to publish Prometheus
+    metrics.
+    """
+
+    def __init__(self) -> None:
+        self._counters = BatchCounters()
+        self._start_time: float = 0.0
+
+    def start(self) -> None:
+        """Reset counters and start the batch timer."""
+        self._counters = BatchCounters()
+        self._start_time = time.perf_counter()
+
+    def observe(self, outcome: Any) -> None:
+        """Record one ledger outcome in the current batch."""
+        self._counters.observe(outcome)
+
+    def finish(self) -> None:
+        """Publish metrics for the current batch."""
+        elapsed = time.perf_counter() - self._start_time
+        INGESTION_BATCH_DURATION_SECONDS.observe(elapsed)
+        INGESTION_BATCH_LEDGERS.labels(status="processed").inc(self._counters.processed)
+        INGESTION_BATCH_LEDGERS.labels(status="skipped").inc(self._counters.skipped)
+        INGESTION_BATCH_LEDGERS.labels(status="error").inc(self._counters.errors)
+        total = self._counters.processed + self._counters.skipped + self._counters.errors
+        INGESTION_BATCH_THROUGHPUT.set(total / elapsed if elapsed > 0 else 0.0)

astroml/ingestion/metrics.py

@@ -61,3 +61,20 @@
     "astroml_ingestion_batch_flush_seconds",
     "Time spent flushing a batch of models",
 )
+
+# Per-batch ingestion progress / throughput metrics (Issue #727)
+INGESTION_BATCH_DURATION_SECONDS = Histogram(
+    "astroml_ingestion_batch_duration_seconds",
+    "Wall-clock time spent processing one batch of ledgers",
+)
+
+INGESTION_BATCH_LEDGERS = Counter(
+    "astroml_ingestion_batch_ledgers_total",
+    "Total number of ledgers handled in batch metrics",
+    ["status"],
+)
+
+INGESTION_BATCH_THROUGHPUT = Gauge(
+    "astroml_ingestion_batch_throughput_ledgers_per_second",
+    "Ledgers processed per second during the most recent batch",
+)

astroml/ingestion/service.py

@@ -25,6 +25,7 @@
 from astroml.core.abstracts import Ingestor
 from astroml.utils.validators import validate_positive_int, validate_range
 
+from .batch_metrics import BatchMetricsRecorder
 from .state import StateStore
 
 logger = logging.getLogger("astroml.ingestion.service")
@@ -221,17 +222,25 @@ def ingest_stream(
         from astroml.observability.metrics import track_active_job
 
         pending_flush = 0
+        batch_metrics = BatchMetricsRecorder()
+        batch_metrics.start()
         try:
             # Active ingestion jobs gauge (issue #567). Entering the context
             # here keeps the gauge balanced even if the caller abandons the
             # generator partway through — GeneratorExit unwinds this `with`.
             with track_active_job("ingestion"):
                 for offset, ledger_id in enumerate(range(start_ledger, end_ledger + 1), start=1):
                     if ledger_id in processed_set:
-                        yield ledger_id, LedgerOutcome(ledger_id=ledger_id, status="skipped")
+                        outcome = LedgerOutcome(ledger_id=ledger_id, status="skipped")
                     else:
-                        payload = fetch(ledger_id)
-                        process(ledger_id, payload)
+                        try:
+                            payload = fetch(ledger_id)
+                            process(ledger_id, payload)
+                        except Exception as exc:
+                            batch_metrics.observe(LedgerOutcome(ledger_id=ledger_id, status="error"))
+                            batch_metrics.finish()
+                            logger.error("Ingestion error for ledger %d: %s", ledger_id, exc)
+                            raise
                         processed_set.add(ledger_id)
                         state.last_processed_ledger = (
                             ledger_id
@@ -242,9 +251,14 @@ def ingest_stream(
                         if pending_flush >= batch_size:
                             self.state.save(state)
                             pending_flush = 0
-                        yield ledger_id, LedgerOutcome(ledger_id=ledger_id, status="processed")
+                        outcome = LedgerOutcome(ledger_id=ledger_id, status="processed")
+
+                    batch_metrics.observe(outcome)
+                    yield ledger_id, outcome
 
                     if offset % batch_size == 0:
+                        batch_metrics.finish()
+                        batch_metrics.start()
                         logger.info(
                             "ingest_stream progress: %d/%d ledgers (up to %d)",
                             offset,
@@ -254,6 +268,7 @@ def ingest_stream(
         finally:
             if pending_flush:
                 self.state.save(state)
+            batch_metrics.finish()
 
     def ingest_incremental(
         self,

docs/ingestion-monitoring.md

@@ -0,0 +1,38 @@
+# Ingestion monitoring
+
+The ingestion pipeline exposes metrics and health checks so operators can
+monitor backfills and detect stalls.
+
+## Heartbeat check
+
+`astroml.observability.ingestion.check_ingestion_heartbeat` compares the
+current time to the `last_processed_at` timestamp recorded in the ingestion
+state store and returns a `CheckResult`:
+
+- `OK` — a ledger was processed within `stale_threshold_seconds`.
+- `DEGRADED` — no ledger for `stale_threshold_seconds` (default 300s).
+- `FAIL` — no ledger for `fail_threshold_seconds` (default 2 × stale threshold).
+
+## Per-batch throughput metrics
+
+`astroml.ingestion.batch_metrics.BatchMetricsRecorder` emits progress metrics
+for each batch of ledgers handled by `IngestionService.ingest_stream`:
+
+- `astroml_ingestion_batch_duration_seconds` — wall-clock time per batch.
+- `astroml_ingestion_batch_ledgers_total{status="processed|skipped|error"}` —
+  ledgers handled per batch.
+- `astroml_ingestion_batch_throughput_ledgers_per_second` — throughput of the
+  most recent batch.
+
+Use these metrics to monitor and tune long backfills, for example:
+
+```promql
+rate(astroml_ingestion_batch_ledgers_total{status="processed"}[5m])
+```
+
+## State store timestamp
+
+`StateStore.mark_processed` records `last_processed_at` as an ISO-8601 UTC
+timestamp whenever a ledger is processed. Existing state files without the
+field are treated as having no recorded ingestion time and report
+`DEGRADED` until the next successful ingestion.

tests/test_ingestion_batch_metrics.py

@@ -0,0 +1,112 @@
+"""Tests for per-batch ingestion progress/throughput metrics (Issue #727)."""
+
+from __future__ import annotations
+
+import time
+from unittest.mock import MagicMock
+
+import pytest
+from prometheus_client import REGISTRY
+
+
+@pytest.fixture(autouse=True)
+def _reset_batch_metrics():
+    """Reset batch metric counters before each test so tests are order-independent."""
+    INGESTION_BATCH_LEDGERS.labels(status="processed")._value.set(0.0)
+    INGESTION_BATCH_LEDGERS.labels(status="skipped")._value.set(0.0)
+    INGESTION_BATCH_LEDGERS.labels(status="error")._value.set(0.0)
+    INGESTION_BATCH_THROUGHPUT._value.set(0.0)
+    INGESTION_BATCH_DURATION_SECONDS._sum.set(0.0)
+
+from astroml.ingestion.batch_metrics import BatchMetricsRecorder
+from astroml.ingestion.metrics import (
+    INGESTION_BATCH_DURATION_SECONDS,
+    INGESTION_BATCH_LEDGERS,
+    INGESTION_BATCH_THROUGHPUT,
+)
+
+
+class _Outcome:
+    def __init__(self, status: str) -> None:
+        self.status = status
+
+
+class TestBatchMetricsRecorder:
+    def test_records_processed_and_skipped(self) -> None:
+        recorder = BatchMetricsRecorder()
+        recorder.start()
+        recorder.observe(_Outcome("processed"))
+        recorder.observe(_Outcome("processed"))
+        recorder.observe(_Outcome("skipped"))
+        recorder.finish()
+
+        assert (
+            INGESTION_BATCH_LEDGERS.labels(status="processed")._value.get() == 2.0
+        )
+        assert INGESTION_BATCH_LEDGERS.labels(status="skipped")._value.get() == 1.0
+
+    def test_records_error_outcome(self) -> None:
+        recorder = BatchMetricsRecorder()
+        recorder.start()
+        recorder.observe(_Outcome("error"))
+        recorder.finish()
+
+        assert INGESTION_BATCH_LEDGERS.labels(status="error")._value.get() == 1.0
+
+    def test_throughput_is_non_negative(self) -> None:
+        recorder = BatchMetricsRecorder()
+        recorder.start()
+        recorder.observe(_Outcome("processed"))
+        recorder.finish()
+
+        throughput = INGESTION_BATCH_THROUGHPUT._value.get()
+        assert throughput >= 0.0
+
+    def test_duration_observed(self) -> None:
+        recorder = BatchMetricsRecorder()
+        recorder.start()
+        time.sleep(0.01)
+        recorder.finish()
+
+        # Histogram exposes samples through _sum and _count
+        assert INGESTION_BATCH_DURATION_SECONDS._sum.get() >= 0.01
+
+    def test_start_resets_counters(self) -> None:
+        recorder = BatchMetricsRecorder()
+        recorder.start()
+        recorder.observe(_Outcome("processed"))
+        recorder.finish()
+
+        recorder.start()
+        recorder.finish()
+
+        # The second batch has no new processed observations, so the counter
+        # should still be 1.0 (counters are monotonic).
+        assert INGESTION_BATCH_LEDGERS.labels(status="processed")._value.get() == 1.0
+
+
+class TestIngestionServiceBatchMetrics:
+    def test_emits_batch_metrics_after_batch_boundary(self, tmp_path) -> None:
+        from astroml.ingestion.service import IngestionService
+        from astroml.ingestion.state import StateStore
+
+        state_path = tmp_path / "state.json"
+        store = StateStore(str(state_path))
+        service = IngestionService(store)
+
+        list(service.ingest_stream(start_ledger=1, end_ledger=5, batch_size=5))
+
+        assert INGESTION_BATCH_LEDGERS.labels(status="processed")._value.get() == 5.0
+        assert INGESTION_BATCH_THROUGHPUT._value.get() >= 0.0
+
+    def test_emits_batch_metrics_for_partial_final_batch(self, tmp_path) -> None:
+        from astroml.ingestion.service import IngestionService
+        from astroml.ingestion.state import StateStore
+
+        state_path = tmp_path / "state.json"
+        store = StateStore(str(state_path))
+        service = IngestionService(store)
+
+        list(service.ingest_stream(start_ledger=1, end_ledger=3, batch_size=5))
+
+        assert INGESTION_BATCH_LEDGERS.labels(status="processed")._value.get() == 3.0