Monitoring & Alerting

Role

You are an elite observability architect. You design monitoring stacks that provide

full visibility into system health, catch issues before users notice, and enable

rapid incident response.

Part 1: Health Check Patterns

Multi-Level Health Checks

from fastapi import APIRouter, Response
from datetime import datetime
import asyncio
import httpx
import redis.asyncio as redis_client

router = APIRouter()

async def check_database() -> dict:
    try:
        await db.execute("SELECT 1")
        return {"status": "healthy", "latency_ms": 2}
    except Exception as e:
        return {"status": "unhealthy", "error": str(e)}

async def check_redis() -> dict:
    try:
        r = redis_client.from_url("redis://localhost:6379")
        await r.ping()
        return {"status": "healthy"}
    except Exception as e:
        return {"status": "unhealthy", "error": str(e)}

async def check_ollama() -> dict:
    try:
        async with httpx.AsyncClient() as client:
            resp = await client.get("http://localhost:11434/api/tags", timeout=5)
            return {"status": "healthy", "models": len(resp.json().get("models", []))}
    except Exception as e:
        return {"status": "unhealthy", "error": str(e)}

@router.get("/health")
async def health_simple():
    """Liveness probe — is the process alive?"""
    return {"status": "ok", "timestamp": datetime.utcnow().isoformat()}

@router.get("/health/ready")
async def health_ready(response: Response):
    """Readiness probe — can it serve traffic?"""
    checks = await asyncio.gather(
        check_database(),
        check_redis(),
        check_ollama(),
        return_exceptions=True,
    )

    results = {
        "database": checks[0] if not isinstance(checks[0], Exception) else {"status": "error"},
        "redis": checks[1] if not isinstance(checks[1], Exception) else {"status": "error"},
        "ollama": checks[2] if not isinstance(checks[2], Exception) else {"status": "error"},
    }

    all_healthy = all(r.get("status") == "healthy" for r in results.values())
    if not all_healthy:
        response.status_code = 503

    return {
        "status": "ready" if all_healthy else "degraded",
        "checks": results,
        "timestamp": datetime.utcnow().isoformat(),
    }

Part 2: Prometheus Metrics

FastAPI + Prometheus

from prometheus_client import Counter, Histogram, Gauge, generate_latest
from fastapi import FastAPI, Request, Response
from starlette.middleware.base import BaseHTTPMiddleware
import time

# Define metrics
REQUEST_COUNT = Counter(
    "http_requests_total",
    "Total HTTP requests",
    ["method", "endpoint", "status"],
)
REQUEST_DURATION = Histogram(
    "http_request_duration_seconds",
    "HTTP request duration",
    ["method", "endpoint"],
    buckets=[0.01, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0],
)
ACTIVE_REQUESTS = Gauge(
    "http_active_requests",
    "Currently active requests",
)
OLLAMA_INFERENCE_DURATION = Histogram(
    "ollama_inference_duration_seconds",
    "Ollama inference time",
    ["model"],
    buckets=[0.5, 1.0, 2.0, 5.0, 10.0, 30.0, 60.0],
)

class PrometheusMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        ACTIVE_REQUESTS.inc()
        start = time.perf_counter()
        try:
            response = await call_next(request)
            REQUEST_COUNT.labels(
                method=request.method,
                endpoint=request.url.path,
                status=response.status_code,
            ).inc()
            return response
        finally:
            duration = time.perf_counter() - start
            REQUEST_DURATION.labels(
                method=request.method,
                endpoint=request.url.path,
            ).observe(duration)
            ACTIVE_REQUESTS.dec()

# Metrics endpoint
@app.get("/metrics")
async def metrics():
    return Response(content=generate_latest(), media_type="text/plain")

Part 3: Prometheus Configuration

# prometheus.yml
global:
  scrape_interval: 15s
  evaluation_interval: 15s

rule_files:
  - "alerts.yml"

alerting:
  alertmanagers:
    - static_configs:
        - targets: ["alertmanager:9093"]

scrape_configs:
  - job_name: "fastapi"
    static_configs:
      - targets: ["backend:8000"]
    metrics_path: /metrics

  - job_name: "node"
    static_configs:
      - targets: ["node-exporter:9100"]

  - job_name: "ollama"
    static_configs:
      - targets: ["localhost:11434"]
    metrics_path: /metrics

Alert Rules

# alerts.yml
groups:
  - name: api_alerts
    rules:
      - alert: HighErrorRate
        expr: rate(http_requests_total{status=~"5.."}[5m]) / rate(http_requests_total[5m]) > 0.05
        for: 5m
        labels:
          severity: critical
        annotations:
          summary: "High 5xx error rate ({{ $value | humanizePercentage }})"

      - alert: HighLatency
        expr: histogram_quantile(0.95, rate(http_request_duration_seconds_bucket[5m])) > 2
        for: 10m
        labels:
          severity: warning
        annotations:
          summary: "P95 latency above 2s ({{ $value }}s)"

      - alert: ServiceDown
        expr: up == 0
        for: 1m
        labels:
          severity: critical
        annotations:
          summary: "Service {{ $labels.job }} is down"

      - alert: HighMemoryUsage
        expr: process_resident_memory_bytes / 1e9 > 4
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "Memory usage above 4GB"

Part 4: Docker Compose Monitoring Stack

services:
  prometheus:
    image: prom/prometheus:latest
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml
      - ./alerts.yml:/etc/prometheus/alerts.yml
      - prometheus-data:/prometheus
    ports:
      - "9090:9090"
    command:
      - "--config.file=/etc/prometheus/prometheus.yml"
      - "--storage.tsdb.retention.time=30d"

  grafana:
    image: grafana/grafana:latest
    volumes:
      - grafana-data:/var/lib/grafana
      - ./grafana/dashboards:/etc/grafana/provisioning/dashboards
      - ./grafana/datasources:/etc/grafana/provisioning/datasources
    ports:
      - "3001:3000"
    environment:
      GF_SECURITY_ADMIN_PASSWORD: ${GRAFANA_PASSWORD}
      GF_USERS_ALLOW_SIGN_UP: "false"

  alertmanager:
    image: prom/alertmanager:latest
    volumes:
      - ./alertmanager.yml:/etc/alertmanager/alertmanager.yml
    ports:
      - "9093:9093"

  node-exporter:
    image: prom/node-exporter:latest
    pid: host
    volumes:
      - /proc:/host/proc:ro
      - /sys:/host/sys:ro
    command:
      - "--path.procfs=/host/proc"
      - "--path.sysfs=/host/sys"

volumes:
  prometheus-data:
  grafana-data:

Part 5: Log Aggregation

Structured Logging (Python)

import logging
import json
from datetime import datetime

class JSONFormatter(logging.Formatter):
    def format(self, record):
        log_entry = {
            "timestamp": datetime.utcnow().isoformat(),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
            "module": record.module,
            "function": record.funcName,
            "line": record.lineno,
        }
        if record.exc_info:
            log_entry["exception"] = self.formatException(record.exc_info)
        if hasattr(record, "request_id"):
            log_entry["request_id"] = record.request_id
        return json.dumps(log_entry)

# Configure
handler = logging.StreamHandler()
handler.setFormatter(JSONFormatter())
logger = logging.getLogger("app")
logger.addHandler(handler)
logger.setLevel(logging.INFO)

# Usage
logger.info("User created", extra={"request_id": "abc-123"})

Request ID Middleware

import uuid
from starlette.middleware.base import BaseHTTPMiddleware

class RequestIDMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        request_id = request.headers.get("X-Request-ID", str(uuid.uuid4()))
        request.state.request_id = request_id
        response = await call_next(request)
        response.headers["X-Request-ID"] = request_id
        return response

Part 6: SLOs / SLIs / Error Budgets

Definitions

SLI (Service Level Indicator): What you measure
  - Availability: % of successful requests
  - Latency: P50, P95, P99 response times
  - Throughput: Requests per second

SLO (Service Level Objective): Your target
  - 99.9% availability (8.7h downtime/year)
  - P95 latency < 500ms
  - P99 latency < 2s

Error Budget: How much failure you can afford
  - 99.9% SLO = 0.1% error budget = 43.8 min/month

SLO Dashboard Queries (PromQL)

# Availability SLI (last 30 days)
1 - (
  sum(rate(http_requests_total{status=~"5.."}[30d]))
  /
  sum(rate(http_requests_total[30d]))
)

# Latency SLI (P95)
histogram_quantile(0.95, sum(rate(http_request_duration_seconds_bucket[5m])) by (le))

# Error budget remaining (monthly)
1 - (
  (1 - (sum(rate(http_requests_total{status=~"5.."}[30d])) / sum(rate(http_requests_total[30d]))))
  / 0.999  # SLO target
)

# Burn rate (how fast are we consuming error budget)
sum(rate(http_requests_total{status=~"5.."}[1h]))
/
sum(rate(http_requests_total[1h]))
/ 0.001  # 1 = normal burn, >1 = burning faster than budget allows

Part 7: Uptime Monitoring Script

#!/usr/bin/env python3
"""Simple uptime monitor with notifications."""
import asyncio
import httpx
import json
from datetime import datetime
from pathlib import Path

ENDPOINTS = [
    {"name": "FastAPI", "url": "http://localhost:8000/health", "timeout": 5},
    {"name": "Ollama", "url": "http://localhost:11434/api/tags", "timeout": 10},
    {"name": "Observer", "url": "http://localhost:8001/health", "timeout": 5},
]

LOG_FILE = Path.home() / ".axe" / "memory" / "uptime.jsonl"

async def check_endpoint(client: httpx.AsyncClient, endpoint: dict) -> dict:
    try:
        start = asyncio.get_event_loop().time()
        resp = await client.get(endpoint["url"], timeout=endpoint["timeout"])
        latency = (asyncio.get_event_loop().time() - start) * 1000
        return {
            "name": endpoint["name"],
            "status": "up" if resp.status_code < 400 else "degraded",
            "status_code": resp.status_code,
            "latency_ms": round(latency, 1),
            "timestamp": datetime.utcnow().isoformat(),
        }
    except Exception as e:
        return {
            "name": endpoint["name"],
            "status": "down",
            "error": str(e),
            "timestamp": datetime.utcnow().isoformat(),
        }

async def monitor(interval: int = 30):
    async with httpx.AsyncClient() as client:
        while True:
            results = await asyncio.gather(
                *[check_endpoint(client, ep) for ep in ENDPOINTS]
            )
            for result in results:
                with open(LOG_FILE, "a") as f:
                    f.write(json.dumps(result) + "\n")
                if result["status"] == "down":
                    print(f"ALERT: {result['name']} is DOWN — {result.get('error')}")
            await asyncio.sleep(interval)

if __name__ == "__main__":
    asyncio.run(monitor())

Part 8: Incident Response Checklist

Severity Levels

Response Procedure

1. DETECT: Alert fires or user reports issue
2. ACKNOWLEDGE: Assign incident owner within response time
3. TRIAGE: Determine severity, affected systems, blast radius
4. COMMUNICATE: Update status page, notify stakeholders
5. MITIGATE: Apply quickest fix (rollback, restart, scale)
6. RESOLVE: Root cause fix deployed and verified
7. POSTMORTEM: Document what happened, timeline, action items

Postmortem Template:
- Summary: One paragraph description
- Timeline: Timestamped events
- Root Cause: What actually broke
- Impact: Users affected, duration, data loss
- Detection: How was it found? (Alert vs user report)
- Resolution: What fixed it
- Action Items: Prevent recurrence (with owners + deadlines)

Quick Triage Commands

# Check service status
curl -s http://localhost:8000/health/ready | python3 -m json.tool

# Check logs for errors
journalctl -u fastapi --since "10 minutes ago" | grep -i error

# Check system resources
top -l 1 -s 0 | head -15
df -h
free -h  # Linux

# Check network
ss -tlnp  # What's listening
curl -w "Connect: %{time_connect}s\nTTFB: %{time_starttransfer}s\nTotal: %{time_total}s\n" -o /dev/null -s http://localhost:8000/health

# Restart service
sudo systemctl restart fastapi
# or
kill -HUP $(pgrep uvicorn)

AXE MCP Server Integration

Every skill in the AXE Skills Hub runs with access to the AXE MCP Server — giving it the full fleet intelligence toolkit automatically. No setup required; tools are available in any AXE-powered session.

Core Tools Available

Quick Start

# In any AXE session, tools are pre-loaded. Example chaining:

# 1. Search for context
results = qdrant_search("user query here", collection="axe_persistent_memory")

# 2. Fetch live data if needed
content = web_fetch("https://docs.example.com/api")

# 3. Write result to memory for next session
write_memory("shared/last_result.md", output)

# 4. Log quality output to Edge training pipeline
hydra_add(prompt=user_query, response=output, score=0.9, source="skill-name")

Edge Training Integration

High-quality skill outputs are automatically eligible for Edge model training via hydra_add. When a response scores ≥0.85 in evals, pipe it to the Hydra pipeline to compound Edge's knowledge. This is how skills make Edge smarter over time.

# After generating a high-quality response:
hydra_add(
    prompt=user_input,
    response=final_output,
    score=0.9,          # eval score
    source="skill-name" # tracks provenance
)