Building a Monitoring Stack

WITH hourly_health AS (
    SELECT
        service_name,
        DATE_TRUNC('hour', recorded_at) AS hour,
        COUNT(*) AS total_checks,
        COUNT(*) FILTER (WHERE status = 'healthy') AS healthy_checks,
        ROUND(
            100.0 * COUNT(*) FILTER (WHERE status = 'healthy') / COUNT(*),
            2
        ) AS uptime_pct
    FROM service_metrics
    WHERE recorded_at >= NOW() - INTERVAL '24 hours'
    GROUP BY service_name, DATE_TRUNC('hour', recorded_at)
),
service_summary AS (
    SELECT
        service_name,
        MIN(uptime_pct) AS min_uptime,
        AVG(uptime_pct) AS avg_uptime,
        COUNT(*) FILTER (WHERE uptime_pct < 99.9) AS degraded_hours
    FROM hourly_health
    GROUP BY service_name
)
SELECT
    service_name,
    ROUND(avg_uptime, 2) AS avg_uptime_pct,
    min_uptime AS worst_hour_pct,
    degraded_hours,
    CASE
        WHEN avg_uptime >= 99.9 THEN 'HEALTHY'
        WHEN avg_uptime >= 99.0 THEN 'DEGRADED'
        ELSE 'CRITICAL'
    END AS status
FROM service_summary
ORDER BY avg_uptime ASC;

import logging
import time
from dataclasses import dataclass, field
from datetime import datetime, timezone

import httpx
import psycopg2
from psycopg2.extras import execute_values

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


@dataclass
class ServiceEndpoint:
    name: str
    url: str
    timeout: float = 5.0


@dataclass
class MetricsCollector:
    dsn: str
    endpoints: list[ServiceEndpoint] = field(default_factory=list)
    interval: int = 60

    def probe(self, endpoint: ServiceEndpoint) -> dict:
        """Send a health check request and measure response time."""
        try:
            start = time.monotonic()
            resp = httpx.get(endpoint.url, timeout=endpoint.timeout)
            elapsed_ms = round((time.monotonic() - start) * 1000, 2)
            return {
                "service_name": endpoint.name,
                "status": "healthy" if resp.status_code < 500 else "unhealthy",
                "response_ms": elapsed_ms,
                "status_code": resp.status_code,
                "recorded_at": datetime.now(timezone.utc),
            }
        except httpx.RequestError as exc:
            logger.warning("Probe failed for %s: %s", endpoint.name, exc)
            return {
                "service_name": endpoint.name,
                "status": "unhealthy",
                "response_ms": None,
                "status_code": None,
                "recorded_at": datetime.now(timezone.utc),
            }

    def collect(self) -> list[dict]:
        """Probe all endpoints and return a batch of results."""
        return [self.probe(ep) for ep in self.endpoints]

    def store(self, results: list[dict]) -> int:
        """Persist metric results to Postgres."""
        rows = [
            (r["service_name"], r["status"], r["response_ms"],
             r["status_code"], r["recorded_at"])
            for r in results
        ]
        insert_sql = """
            INSERT INTO service_metrics
                (service_name, status, response_ms, status_code, recorded_at)
            VALUES %s
        """
        with psycopg2.connect(self.dsn) as conn:
            with conn.cursor() as cur:
                execute_values(cur, insert_sql, rows)
            conn.commit()
        return len(rows)

    def run(self):
        """Main loop: collect and store metrics at a fixed interval."""
        logger.info("Starting collector with %d endpoints", len(self.endpoints))
        while True:
            results = self.collect()
            count = self.store(results)
            healthy = sum(1 for r in results if r["status"] == "healthy")
            logger.info(
                "Stored %d metrics (%d/%d healthy)",
                count, healthy, len(results),
            )
            time.sleep(self.interval)


if __name__ == "__main__":
    endpoints = [
        ServiceEndpoint("api-gateway", "https://api.example.com/health"),
        ServiceEndpoint("auth-service", "https://auth.example.com/health"),
        ServiceEndpoint("payments", "https://payments.example.com/health"),
        ServiceEndpoint("notifications", "https://notify.example.com/health"),
    ]
    collector = MetricsCollector(
        dsn="postgresql://monitor:pass@localhost:5432/monitoring_db",
        endpoints=endpoints,
        interval=60,
    )
    collector.run()