diff --git a/README.md b/README.md index 3859311..b8c02d9 100644 --- a/README.md +++ b/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**. diff --git a/app/core/observability.py b/app/core/observability.py new file mode 100644 index 0000000..8c075fa --- /dev/null +++ b/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)) diff --git a/app/core/services/host_service.py b/app/core/services/host_service.py index f69fe07..b6379db 100644 --- a/app/core/services/host_service.py +++ b/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]: diff --git a/app/tasks/item_poller.py b/app/tasks/item_poller.py index 6b2d8cd..26a7e09 100644 --- a/app/tasks/item_poller.py +++ b/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, diff --git a/app/tasks/scheduler.py b/app/tasks/scheduler.py index 705674c..386c0f3 100644 --- a/app/tasks/scheduler.py +++ b/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() diff --git a/docs/roadmap.md b/docs/roadmap.md index 7b4ea58..98ad7c1 100644 --- a/docs/roadmap.md +++ b/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. --- diff --git a/tests/test_observability.py b/tests/test_observability.py new file mode 100644 index 0000000..dd0ac27 --- /dev/null +++ b/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