feat(observability): ingestion + storage metrics (measure before scaling)

README.md

@@ -131,7 +131,8 @@ open and unacknowledged:
 - **Maintenance windows** — one-shot or recurring (`cron`) alert silencing.
 - **Auth** — local accounts (argon2, JWT) and **OIDC SSO** (any OpenID Connect
   provider) on top, sharing the same session; SSO-only accounts need no password.
-- **Interfaces** — REST API (`/api`, OpenAPI at `/docs`), web UI, a `ghostmon` CLI, Prometheus metrics (`/metrics`), liveness/readiness (`/healthz`, `/readyz`).
+- **Network discovery** — scheduled CIDR scans (ping/TCP) that provision hosts for responsive addresses, applying a template; CIDR-capped and owner-scoped.
+- **Interfaces** — REST API (`/api`, OpenAPI at `/docs`), web UI, a `ghostmon` CLI, Prometheus metrics (`/metrics`, incl. ingestion throughput and time-series storage-size gauges), liveness/readiness (`/healthz`, `/readyz`).
 
 Stack: Python 3.12 · FastAPI · SQLAlchemy 2 (async) + asyncpg · PostgreSQL · APScheduler · Typer · Jinja2. Dependencies managed with [`uv`](https://docs.astral.sh/uv/). Containers built and run with **Podman**.
 

app/core/observability.py

@@ -0,0 +1,41 @@
+"""Storage/ingestion observability — the data needed to decide *when* the
+time-series store must scale (partitioning, a TSDB), rather than guessing now.
+
+Exposed on `/metrics` (default Prometheus registry):
+- `ghostmon_values_ingested_total` — counter of metric values written to history;
+  `rate()` of it is the ingestion throughput.
+- `ghostmon_table_rows_estimate{table=...}` — approximate row counts for the
+  time-series tables, sampled periodically from `pg_class.reltuples` (instant; no
+  expensive `count(*)`), so growth is visible over time.
+"""
+
+from __future__ import annotations
+
+from prometheus_client import Counter, Gauge
+from sqlalchemy import bindparam, text
+from sqlalchemy.ext.asyncio import AsyncSession
+
+VALUES_INGESTED = Counter(
+    "ghostmon_values_ingested_total", "Metric values appended to time-series history"
+)
+TABLE_ROWS = Gauge(
+    "ghostmon_table_rows_estimate", "Estimated row count of a time-series table", ["table"]
+)
+
+_TRACKED_TABLES = ("metric_values", "metric_trends", "monitor_results")
+
+
+def count_ingested(n: int = 1) -> None:
+    if n > 0:
+        VALUES_INGESTED.inc(n)
+
+
+async def update_storage_metrics(session: AsyncSession) -> None:
+    """Refresh the table-size gauges from Postgres' planner statistics."""
+    stmt = text(
+        "SELECT relname, reltuples::bigint FROM pg_class WHERE relkind = 'r' AND relname IN :tables"
+    ).bindparams(bindparam("tables", expanding=True))
+    rows = (await session.execute(stmt, {"tables": list(_TRACKED_TABLES)})).all()
+    seen = {relname: estimate for relname, estimate in rows}
+    for table in _TRACKED_TABLES:
+        TABLE_ROWS.labels(table=table).set(max(seen.get(table, 0), 0))

app/core/services/host_service.py

@@ -11,6 +11,7 @@
 from app.core.models.metric_trend import MetricTrend
 from app.core.models.metric_value import MetricValue
 from app.core.models.trigger import Severity, Trigger, TriggerState
+from app.core.observability import count_ingested
 from app.core.schemas.host import HostCreate, HostUpdate, ItemCreate, ItemUpdate
 from app.core.security.field_crypto import REDACTED, encrypt_secret
 
@@ -196,6 +197,7 @@ async def record_value(
         self._session.add(sample)
         await self._session.commit()
         await self._session.refresh(sample)
+        count_ingested()
         return sample
 
     async def list_values(self, item_id: uuid.UUID, limit: int = 100) -> Sequence[MetricValue]:

app/tasks/item_poller.py

@@ -20,6 +20,7 @@
 from app.core.db.session import SessionLocal
 from app.core.models.host import Host, Item, ItemSource, ItemValueType
 from app.core.models.metric_value import MetricValue
+from app.core.observability import count_ingested
 from app.core.security.field_crypto import decrypt_secret
 from app.core.services.trigger_service import TriggerService
 from app.tasks.notifications.dispatcher import schedule_item_trigger_alerts
@@ -115,6 +116,7 @@ async def _poll_one(target: _PollTarget, now: datetime) -> None:
             )
         )
         await session.commit()
+        count_ingested()
         fired = await TriggerService(session).evaluate_item(
             target.item_id,
             target.item_key,

app/tasks/scheduler.py

@@ -15,6 +15,7 @@
 from app.core.models.monitor import Monitor, MonitorStatus
 from app.core.models.monitor_result import MonitorResult, ProbeStatus
 from app.core.models.trigger import TriggerMetric
+from app.core.observability import count_ingested, update_storage_metrics
 from app.core.services.discovery_service import DiscoveryService
 from app.core.services.escalation_service import EscalationService
 from app.core.services.maintenance_service import MaintenanceService
@@ -34,17 +35,20 @@
 POLL_ITEMS_INTERVAL_SECONDS = 15
 ESCALATION_INTERVAL_SECONDS = 60
 DISCOVERY_INTERVAL_SECONDS = 60
+STORAGE_METRICS_INTERVAL_SECONDS = 300
 _RECONCILE_JOB_ID = "__reconcile__"
 _PRUNE_JOB_ID = "__prune__"
 _POLL_ITEMS_JOB_ID = "__poll_items__"
 _ESCALATION_JOB_ID = "__escalation__"
 _DISCOVERY_JOB_ID = "__discovery__"
+_STORAGE_METRICS_JOB_ID = "__storage_metrics__"
 _RESERVED_JOB_IDS = {
     _RECONCILE_JOB_ID,
     _PRUNE_JOB_ID,
     _POLL_ITEMS_JOB_ID,
     _ESCALATION_JOB_ID,
     _DISCOVERY_JOB_ID,
+    _STORAGE_METRICS_JOB_ID,
 }
 
 
@@ -82,6 +86,15 @@ async def start(self) -> None:
             max_instances=1,
             coalesce=True,
         )
+        self._scheduler.add_job(
+            _storage_metrics_job,
+            trigger=IntervalTrigger(seconds=STORAGE_METRICS_INTERVAL_SECONDS),
+            id=_STORAGE_METRICS_JOB_ID,
+            replace_existing=True,
+            next_run_time=datetime.now(UTC) + timedelta(seconds=20),
+            max_instances=1,
+            coalesce=True,
+        )
         self._scheduler.add_job(
             _discovery_job,
             trigger=IntervalTrigger(seconds=DISCOVERY_INTERVAL_SECONDS),
@@ -226,6 +239,7 @@ async def _run_probe_job(monitor_id: uuid.UUID) -> None:
         # Mirror the probe signals into the host/item time-series history (migrate
         # step): status (1=up/0=down) every probe, latency and error when present.
         backing = await ensure_backing_items(session, monitor)
+        mirrored = 1  # status is recorded every probe
         session.add(
             MetricValue(
                 item_id=backing.status.id,
@@ -241,11 +255,14 @@ async def _run_probe_job(monitor_id: uuid.UUID) -> None:
                     collected_at=now,
                 )
             )
+            mirrored += 1
         if final.error:
             session.add(
                 MetricValue(item_id=backing.error.id, value_text=final.error, collected_at=now)
             )
+            mirrored += 1
         await session.commit()
+        count_ingested(mirrored)
 
         trigger_alerts = [
             TriggerAlertEvent(
@@ -346,5 +363,11 @@ async def _discovery_job() -> None:
                 logger.exception("discovery scan failed for rule %s", rule.id)
 
 
+async def _storage_metrics_job() -> None:
+    """Refresh the time-series storage-size gauges (cheap planner estimates)."""
+    async with SessionLocal() as session:
+        await update_storage_metrics(session)
+
+
 def build_scheduler() -> ProbeScheduler:
     return ProbeScheduler()

docs/roadmap.md

@@ -132,10 +132,14 @@ GhostMonitor's reason to exist over a plain Zabbix clone is privacy (ghost-suite
   CIDR capped (≤ 1024 hosts), bounded scan concurrency, dedupe by address, owner-scoped.
   REST CRUD + scan-now under `/api/discovery-rules`, and a `/discovery` web page to
   manage rules (create, scan-now, delete).
-- *(deferred — measure first)* Distributed collection (proxy/agent fan-in) and history
-  storage scaling (TimescaleDB / a dedicated TSDB). Deliberately not built yet: it is
-  premature without real load data. Add ingestion-volume/throughput metrics first so
-  the decision is evidence-driven.
+- ✅ **Scale observability (measure first)**: Prometheus metrics quantify the
+  time-series load — `ghostmon_values_ingested_total` (throughput via `rate()`) and
+  `ghostmon_table_rows_estimate{table=…}` (history/trends/results size, sampled from
+  `pg_class.reltuples` so it's instant, no `count(*)`). These make the scaling
+  decision evidence-driven.
+- *(deferred — now measurable)* Distributed collection (proxy/agent fan-in) and history
+  storage scaling (TimescaleDB / a dedicated TSDB / partitioning). Still not built:
+  premature until the metrics above show it's needed.
 
 ---
 

tests/test_observability.py

@@ -0,0 +1,49 @@
+"""Storage/ingestion metrics — the data behind a future scaling decision."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from prometheus_client import REGISTRY
+from sqlalchemy import text
+
+from app.core.models.host import Host, Item, ItemValueType
+from app.core.observability import update_storage_metrics
+from app.core.services.host_service import ItemService
+
+
+def _ingested() -> float:
+    return REGISTRY.get_sample_value("ghostmon_values_ingested_total") or 0.0
+
+
+async def _item(session: Any, owner_id: Any) -> Item:
+    host = Host(name="srv", owner_id=owner_id)
+    session.add(host)
+    await session.flush()
+    item = Item(host_id=host.id, key="cpu", name="CPU", value_type=ItemValueType.FLOAT, interval=60)
+    session.add(item)
+    await session.commit()
+    await session.refresh(item)
+    return item
+
+
+async def test_record_value_increments_ingestion_counter(session: Any, user: Any) -> None:
+    item = await _item(session, user.id)
+    before = _ingested()
+    await ItemService(session).record_value(item, 42.0)
+    await ItemService(session).record_value(item, 43.0)
+    assert _ingested() == before + 2
+
+
+async def test_storage_metrics_estimate_table_rows(session: Any, user: Any) -> None:
+    item = await _item(session, user.id)
+    for v in range(5):
+        await ItemService(session).record_value(item, float(v))
+    # reltuples is maintained by ANALYZE — refresh it so the estimate is meaningful.
+    await session.execute(text("ANALYZE metric_values"))
+    await session.commit()
+
+    await update_storage_metrics(session)
+    rows = REGISTRY.get_sample_value("ghostmon_table_rows_estimate", {"table": "metric_values"})
+    assert rows is not None
+    assert rows >= 5