feat(observability): ingestion heartbeat and stale-data alerts (#758)

Author: ZacLou

astroml/ingestion/metrics.py

@@ -61,3 +61,9 @@
     "astroml_ingestion_batch_flush_seconds",
     "Time spent flushing a batch of models",
 )
+
+# Heartbeat / stale-data metric (Issue #758)
+INGESTION_LAST_PROCESSED_AT = Gauge(
+    "astroml_ingestion_last_processed_at_seconds",
+    "Unix timestamp of the most recently processed ledger",
+)

astroml/ingestion/state.py

@@ -3,6 +3,9 @@
 import json
 import os
 from dataclasses import dataclass
+from datetime import datetime, timezone
+
+from astroml.ingestion.metrics import INGESTION_LAST_PROCESSED_AT
 
 DEFAULT_STATE_DIR = os.path.join(os.getcwd(), ".astroml_state")
 DEFAULT_STATE_FILE = os.path.join(DEFAULT_STATE_DIR, "ingestion_state.json")
@@ -12,19 +15,22 @@
 class IngestionState:
     last_processed_ledger: int | None
     processed_ledgers: set[int]
+    last_processed_at: str | None = None
 
     def to_dict(self) -> dict:
         return {
             "last_processed_ledger": self.last_processed_ledger,
             # store as sorted list for readability
             "processed_ledgers": sorted(self.processed_ledgers),
+            "last_processed_at": self.last_processed_at,
         }
 
     @staticmethod
     def from_dict(data: dict) -> IngestionState:
         return IngestionState(
             last_processed_ledger=data.get("last_processed_ledger"),
             processed_ledgers=set(data.get("processed_ledgers", [])),
+            last_processed_at=data.get("last_processed_at"),
         )
 
 
@@ -60,6 +66,11 @@ def mark_processed(self, ledger_id: int) -> IngestionState:
             state.last_processed_ledger = ledger_id
         else:
             state.last_processed_ledger = max(state.last_processed_ledger, ledger_id)
+        state.last_processed_at = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
+        try:
+            INGESTION_LAST_PROCESSED_AT.set(datetime.now(timezone.utc).timestamp())
+        except Exception:
+            pass  # metrics registry may not be initialised in tests/CLI
         self.save(state)
         return state
 

astroml/observability/ingestion.py

@@ -0,0 +1,136 @@
+"""Ingestion heartbeat / stale-data alerts (Issue #758).
+
+Exposes a health check that reports when no new ledger has been ingested
+for longer than a configurable threshold.  The check is transport-agnostic
+so it can be reused by the FastAPI health router, CLI, or periodic probes.
+"""
+
+from __future__ import annotations
+
+import time
+from datetime import datetime, timedelta, timezone
+from typing import Final
+
+from astroml.ingestion.state import StateStore
+from astroml.observability.health import CheckResult, HealthStatus
+
+#: Default threshold at which ingestion is considered stale (seconds).
+DEFAULT_STALE_THRESHOLD_SECONDS: Final[int] = 300
+
+#: Threshold at which ingestion is considered critically stale (seconds).
+#: Used when the caller does not supply an explicit fail threshold.
+DEFAULT_FAIL_THRESHOLD_SECONDS: Final[int] = 600
+
+
+def _parse_timestamp(value: str | None) -> datetime | None:
+    """Parse an ISO-8601 timestamp produced by ``StateStore``.
+
+    Args:
+        value: ISO-8601 string, e.g. ``2024-01-01T00:00:00Z``.
+
+    Returns:
+        A timezone-aware UTC datetime, or ``None`` if parsing fails.
+    """
+    if not value:
+        return None
+    try:
+        # StateStore writes ``datetime.utcnow().isoformat() + "Z"``.
+        return datetime.fromisoformat(value.replace("Z", "+00:00"))
+    except ValueError:
+        return None
+
+
+def check_ingestion_heartbeat(
+    state_store: StateStore,
+    *,
+    stale_threshold_seconds: float = DEFAULT_STALE_THRESHOLD_SECONDS,
+    fail_threshold_seconds: float | None = None,
+    now: datetime | None = None,
+) -> CheckResult:
+    """Check whether the ingestion pipeline has processed a ledger recently.
+
+    Args:
+        state_store: Store that tracks ``last_processed_at``.
+        stale_threshold_seconds: Seconds without a new ledger before the
+            check becomes ``DEGRADED``.
+        fail_threshold_seconds: Seconds without a new ledger before the
+            check becomes ``FAIL``. Defaults to twice the stale threshold.
+        now: Optional reference time for tests. Defaults to UTC now.
+
+    Returns:
+        A :class:`CheckResult` named ``"ingestion_heartbeat"``.
+    """
+    started = time.perf_counter()
+    fail_threshold = fail_threshold_seconds or stale_threshold_seconds * 2
+    now = now or datetime.now(timezone.utc)
+
+    try:
+        state = state_store.load()
+    except OSError as exc:
+        return CheckResult(
+            name="ingestion_heartbeat",
+            status=HealthStatus.FAIL,
+            details={"state_path": state_store.path, "error": str(exc)},
+            remediation=(
+                "Cannot read ingestion state. Verify the state file path "
+                "and that the process user has read access."
+            ),
+            duration_ms=(time.perf_counter() - started) * 1000,
+        )
+
+    last_processed_at = _parse_timestamp(state.last_processed_at)
+
+    if last_processed_at is None:
+        return CheckResult(
+            name="ingestion_heartbeat",
+            status=HealthStatus.DEGRADED,
+            details={
+                "last_processed_ledger": state.last_processed_ledger,
+                "last_processed_at": state.last_processed_at,
+                "stale_threshold_seconds": stale_threshold_seconds,
+                "fail_threshold_seconds": fail_threshold,
+            },
+            remediation=(
+                "No ingestion timestamp has been recorded yet. "
+                "Run an ingestion batch or verify the state store is being updated."
+            ),
+            duration_ms=(time.perf_counter() - started) * 1000,
+        )
+
+    # Ensure comparison is timezone-aware.
+    if last_processed_at.tzinfo is None:
+        last_processed_at = last_processed_at.replace(tzinfo=timezone.utc)
+
+    elapsed_seconds = (now - last_processed_at).total_seconds()
+
+    if elapsed_seconds >= fail_threshold:
+        status = HealthStatus.FAIL
+        remediation = (
+            f"No ledger ingested for {elapsed_seconds:.0f}s "
+            f"(fail threshold {fail_threshold:.0f}s). "
+            "Investigate the ingestion worker, Horizon stream, and network connectivity."
+        )
+    elif elapsed_seconds >= stale_threshold_seconds:
+        status = HealthStatus.DEGRADED
+        remediation = (
+            f"No ledger ingested for {elapsed_seconds:.0f}s "
+            f"(stale threshold {stale_threshold_seconds:.0f}s). "
+            "Check the ingestion worker logs for stalls or rate-limiting."
+        )
+    else:
+        status = HealthStatus.OK
+        remediation = ""
+
+    return CheckResult(
+        name="ingestion_heartbeat",
+        status=status,
+        details={
+            "last_processed_ledger": state.last_processed_ledger,
+            "last_processed_at": state.last_processed_at,
+            "elapsed_seconds": round(elapsed_seconds, 2),
+            "stale_threshold_seconds": stale_threshold_seconds,
+            "fail_threshold_seconds": fail_threshold,
+        },
+        remediation=remediation,
+        duration_ms=(time.perf_counter() - started) * 1000,
+    )

docs/ingestion-monitoring.md

@@ -0,0 +1,42 @@
+# Ingestion monitoring
+
+The ingestion pipeline exposes a heartbeat check so operators can detect when
+no new ledgers have been processed for a configurable period.
+
+## 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).
+
+Example:
+
+```python
+from astroml.ingestion.state import StateStore
+from astroml.observability.ingestion import check_ingestion_heartbeat
+
+store = StateStore()
+result = check_ingestion_heartbeat(store, stale_threshold_seconds=300)
+print(result.status, result.remediation)
+```
+
+## Prometheus metric
+
+`astroml.ingestion.metrics.INGESTION_LAST_PROCESSED_AT` is a Gauge that
+records the Unix timestamp of the most recently processed ledger. Use it to
+build alerts such as:
+
+```promql
+(time() - astroml_ingestion_last_processed_at_seconds) > 300
+```
+
+## 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_observability_ingestion.py

@@ -0,0 +1,112 @@
+"""Tests for ingestion heartbeat / stale-data alerts (Issue #758)."""
+
+from __future__ import annotations
+
+import json
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+import pytest
+
+from astroml.ingestion.state import StateStore
+from astroml.observability.health import HealthStatus
+from astroml.observability.ingestion import check_ingestion_heartbeat
+
+
+class TestCheckIngestionHeartbeat:
+    def test_ok_when_recent_ingestion(self, tmp_path: Path) -> None:
+        state_path = tmp_path / "ingestion_state.json"
+        state_path.write_text(
+            json.dumps(
+                {
+                    "last_processed_ledger": 1000,
+                    "processed_ledgers": [1000],
+                    "last_processed_at": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
+                }
+            ),
+            encoding="utf-8",
+        )
+        store = StateStore(str(state_path))
+
+        result = check_ingestion_heartbeat(store, stale_threshold_seconds=300)
+
+        assert result.status is HealthStatus.OK
+        assert result.details["last_processed_ledger"] == 1000
+        assert result.remediation == ""
+
+    def test_degraded_when_stale(self, tmp_path: Path) -> None:
+        state_path = tmp_path / "ingestion_state.json"
+        stale_at = datetime.now(timezone.utc) - timedelta(seconds=400)
+        state_path.write_text(
+            json.dumps(
+                {
+                    "last_processed_ledger": 1000,
+                    "processed_ledgers": [1000],
+                    "last_processed_at": stale_at.isoformat().replace("+00:00", "Z"),
+                }
+            ),
+            encoding="utf-8",
+        )
+        store = StateStore(str(state_path))
+
+        result = check_ingestion_heartbeat(store, stale_threshold_seconds=300)
+
+        assert result.status is HealthStatus.DEGRADED
+        assert "stale threshold" in result.remediation
+
+    def test_fail_when_critically_stale(self, tmp_path: Path) -> None:
+        state_path = tmp_path / "ingestion_state.json"
+        stale_at = datetime.now(timezone.utc) - timedelta(seconds=700)
+        state_path.write_text(
+            json.dumps(
+                {
+                    "last_processed_ledger": 1000,
+                    "processed_ledgers": [1000],
+                    "last_processed_at": stale_at.isoformat().replace("+00:00", "Z"),
+                }
+            ),
+            encoding="utf-8",
+        )
+        store = StateStore(str(state_path))
+
+        result = check_ingestion_heartbeat(store, stale_threshold_seconds=300)
+
+        assert result.status is HealthStatus.FAIL
+        assert "fail threshold" in result.remediation
+
+    def test_degraded_when_no_timestamp_recorded(self, tmp_path: Path) -> None:
+        state_path = tmp_path / "ingestion_state.json"
+        state_path.write_text(
+            json.dumps(
+                {
+                    "last_processed_ledger": 1000,
+                    "processed_ledgers": [1000],
+                }
+            ),
+            encoding="utf-8",
+        )
+        store = StateStore(str(state_path))
+
+        result = check_ingestion_heartbeat(store)
+
+        assert result.status is HealthStatus.DEGRADED
+        assert "No ingestion timestamp" in result.remediation
+
+    def test_state_store_records_timestamp_on_mark_processed(self, tmp_path: Path) -> None:
+        state_path = tmp_path / "ingestion_state.json"
+        store = StateStore(str(state_path))
+
+        store.mark_processed(1000)
+        state = store.load()
+
+        assert state.last_processed_at is not None
+        assert state.last_processed_at.endswith("Z")
+
+    def test_state_store_round_trip_preserves_timestamp(self, tmp_path: Path) -> None:
+        state_path = tmp_path / "ingestion_state.json"
+        store = StateStore(str(state_path))
+        store.mark_processed(1000)
+
+        reloaded = StateStore(str(state_path)).load()
+
+        assert reloaded.last_processed_at == store.load().last_processed_at