Observability Guide

from svc_infra.api.fastapi.ease import easy_service_app

# Observability enabled by default
app = easy_service_app(name="MyAPI", release="1.0.0")

# Metrics available at /metrics
# Health probes at /_ops/live, /_ops/ready, /_ops/startup

from fastapi import FastAPI
from svc_infra.obs.add import add_observability

app = FastAPI()

# Add observability with options
add_observability(
    app,
    db_engines=[engine],  # Optional: SQLAlchemy engines for pool metrics
    metrics_path="/metrics",
    skip_metric_paths=["/metrics", "/health", "/healthz"],
)

curl http://localhost:8000/metrics

# Output:
# HELP http_server_requests_total Total HTTP requests
# TYPE http_server_requests_total counter
# http_server_requests_total{method="GET",route="/api/v1/users",code="200"} 42
# ...

from svc_infra.obs.metrics.base import counter, gauge, histogram

# Create custom metrics
api_calls = counter(
    "myapp_api_calls_total",
    "Total API calls by endpoint",
    labels=["endpoint", "status"],
)

queue_size = gauge(
    "myapp_queue_size",
    "Current queue size",
    labels=["queue_name"],
)

processing_time = histogram(
    "myapp_processing_seconds",
    "Processing time in seconds",
    labels=["operation"],
    buckets=(0.01, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0),
)

# Use in code
api_calls.labels(endpoint="/users", status="success").inc()
queue_size.labels(queue_name="emails").set(42)
with processing_time.labels(operation="parse").time():
    parse_document(doc)

# Default: seconds
(0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.0, 5.0, 10.0)

# Configure via environment
METRICS_DEFAULT_BUCKETS="0.01,0.05,0.1,0.25,0.5,1,2.5,5,10"

#  Good: Low cardinality, meaningful labels
http_requests_total{method="GET", status="2xx", endpoint="/users"}

#  Bad: High cardinality (unique IDs, timestamps)
http_requests_total{user_id="uuid-123", timestamp="..."}

from svc_infra.api.fastapi.ops.add import add_probes

add_probes(app, prefix="/_ops")

# Exposes:
# GET /_ops/live    - Liveness (always 200 if process running)
# GET /_ops/ready   - Readiness (can add dependency checks)
# GET /_ops/startup - Startup probe

from svc_infra.health import (
    HealthRegistry,
    HealthStatus,
    check_database,
    check_redis,
    check_url,
    check_tcp,
    add_health_routes,
)
import os

# Create registry
registry = HealthRegistry()

# Add checks
registry.add(
    "database",
    check_database(os.getenv("DATABASE_URL")),
    critical=True,  # Service is unhealthy if this fails
    timeout=5.0,
)

registry.add(
    "redis",
    check_redis(os.getenv("REDIS_URL")),
    critical=False,  # Service is degraded but functional
)

registry.add(
    "external_api",
    check_url("http://api.example.com/health", expected_status=200),
    critical=False,
)

registry.add(
    "postgres_port",
    check_tcp("db.example.com", 5432),
    critical=True,
)

# Add routes to FastAPI
add_health_routes(app, registry, prefix="/_health")

# Exposes:
# GET /_health/live   - Always 200
# GET /_health/ready  - Runs all checks
# GET /_health/startup - Runs critical checks

from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Wait up to 60 seconds for dependencies
    if not await registry.wait_until_healthy(timeout=60, interval=2):
        raise RuntimeError("Dependencies not ready")
    yield

app = FastAPI(lifespan=lifespan)

from svc_infra.health import HealthCheckResult, HealthStatus

async def check_custom_service() -> HealthCheckResult:
    try:
        # Your health check logic
        response = await my_service.ping()
        if response.ok:
            return HealthCheckResult(
                name="custom_service",
                status=HealthStatus.HEALTHY,
                latency_ms=response.latency_ms,
            )
        else:
            return HealthCheckResult(
                name="custom_service",
                status=HealthStatus.DEGRADED,
                latency_ms=response.latency_ms,
                message="Service responding slowly",
            )
    except Exception as e:
        return HealthCheckResult(
            name="custom_service",
            status=HealthStatus.UNHEALTHY,
            latency_ms=0,
            message=str(e),
        )

registry.add("custom_service", check_custom_service, critical=True)

# deployment.yaml
spec:
  containers:
    - name: api
      livenessProbe:
        httpGet:
          path: /_health/live
          port: 8000
        initialDelaySeconds: 5
        periodSeconds: 10
        failureThreshold: 3

      readinessProbe:
        httpGet:
          path: /_health/ready
          port: 8000
        initialDelaySeconds: 10
        periodSeconds: 5
        failureThreshold: 3

      startupProbe:
        httpGet:
          path: /_health/startup
          port: 8000
        initialDelaySeconds: 5
        periodSeconds: 5
        failureThreshold: 30  # 30 * 5s = 150s max startup

from svc_infra.app.logging import setup_logging

# Auto-detects format based on environment
# Production: JSON, Development: Plain
setup_logging()

# Explicit configuration
setup_logging(
    level="INFO",      # DEBUG, INFO, WARNING, ERROR
    fmt="json",        # "json" or "plain"
    drop_paths=["/metrics", "/healthz"],  # Don't log these
)

{
  "timestamp": "2025-01-15T10:30:45.123Z",
  "level": "INFO",
  "message": "Request completed",
  "logger": "uvicorn.access",
  "request_id": "req-abc123",
  "method": "GET",
  "path": "/api/v1/users",
  "status": 200,
  "duration_ms": 45.2
}

from svc_infra.api.fastapi.middleware.request_id import get_request_id

@app.get("/users")
async def list_users():
    request_id = get_request_id()
    logger.info("Listing users", extra={"request_id": request_id})
    return users

# Environment variable
LOG_DROP_PATHS="/metrics,/healthz,/_ops/live"

# Programmatic
setup_logging(drop_paths=["/metrics", "/healthz", "/_ops/live"])

# Set up environment
export GRAFANA_CLOUD_URL="https://your-stack.grafana.net"
export GRAFANA_CLOUD_TOKEN="glsa_..."

# Push dashboards
svc-infra obs dashboard push

# Start observability stack
svc-infra obs-up

# Open:
# - Grafana: http://localhost:3000 (admin/admin)
# - Prometheus: http://localhost:9090

# Stop when done
svc-infra obs-down

# Override rate window (default: \$__rate_interval)
SVC_INFRA_RATE_WINDOW="5m"

# Dashboard refresh rate
SVC_INFRA_DASHBOARD_REFRESH="10s"

# Default time range
SVC_INFRA_DASHBOARD_RANGE="now-24h"

from svc_infra.obs.cloud_dash import push_dashboards_from_pkg

push_dashboards_from_pkg(
    base_url=os.getenv("GRAFANA_CLOUD_URL"),
    token=os.getenv("GRAFANA_CLOUD_TOKEN"),
    folder_title="Service Infrastructure",
)

from svc_infra.obs.metrics.sqlalchemy import bind_sqlalchemy_pool_metrics

# Bind metrics to engine(s)
bind_sqlalchemy_pool_metrics(engine)

# Metrics exposed:
# db_pool_size - Configured pool size
# db_pool_checkedout - Currently checked out connections
# db_pool_overflow - Overflow connections in use
# db_pool_checkedout_total - Total checkouts (counter)
# db_pool_checkedin_total - Total check-ins (counter)

add_observability(
    app,
    db_engines=[engine1, engine2],  # Pass your engines
)

from svc_infra.obs.metrics.http import instrument_httpx, instrument_requests

# Instrument httpx (async)
instrument_httpx()

# Instrument requests (sync)
instrument_requests()

# Now all HTTP clients emit metrics:
# http_client_requests_total{method="GET", host="api.example.com", status="200"}
# http_client_request_duration_seconds{method="GET", host="api.example.com"}

# Set multiprocess directory
export PROMETHEUS_MULTIPROC_DIR=/tmp/prometheus_multiproc

# Ensure directory exists and is writable
mkdir -p \$PROMETHEUS_MULTIPROC_DIR

# svc-infra automatically detects and uses MultiProcessCollector
from svc_infra.obs.metrics.base import registry

# Registry handles multiprocess aggregation

#  Bad: User IDs create unbounded cardinality
http_requests_total{user_id="..."}

#  Good: Use bounded categories
http_requests_total{user_tier="free|pro|enterprise"}

import random

# Sample 10% of requests for detailed metrics
if random.random() < 0.1:
    detailed_metric.observe(value)

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

# Grafana Cloud retention is based on your plan
# Local Prometheus: configure storage.tsdb.retention.time

groups:
  - name: svc-infra
    rules:
      - alert: HighErrorRate
        expr: rate(http_server_requests_total{code=~"5.."}[5m]) > 0.1
        for: 5m
        labels:
          severity: critical
        annotations:
          summary: High error rate detected

      - alert: SlowResponses
        expr: histogram_quantile(0.99, rate(http_server_request_duration_seconds_bucket[5m])) > 2
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: P99 latency exceeds 2 seconds

      - alert: DatabasePoolExhausted
        expr: db_pool_checkedout / db_pool_size > 0.9
        for: 2m
        labels:
          severity: critical
        annotations:
          summary: Database connection pool nearly exhausted

curl http://localhost:8000/metrics returns 404

prometheus: high cardinality on metric http_server_requests_total

Metrics show stale values

/_health/ready returns 503 after long delay

Logs appearing in wrong format