Quick Start¶

Get obskit wired up in a fresh service in under five minutes. Each step is independent — copy the ones that apply to your service.

Install¶

Bash

# Typical microservice stack
pip install "obskit[prometheus,otlp,fastapi]"

Just want one thing?

obskit uses optional extras. pip install obskit is all you need for structured logging. Add extras only for what you need.

Recommended: One-Call Setup (v1.0.0+)¶

Starting with v1.0.0, you can bootstrap logging, tracing, and metrics in a single call using configure_observability(). This is the recommended approach for new services.

Python

from obskit import configure_observability, instrument_fastapi
from fastapi import FastAPI

# One call sets up logging, tracing, and metrics
obs = configure_observability(
    service_name="order-service",
    environment="production",
    version="1.0.0",
    otlp_endpoint="http://tempo:4317",
    trace_sample_rate=0.1,
)

app = FastAPI(title="Order Service")

# One call adds correlation ID, request logging, and RED metrics middleware
instrument_fastapi(app)

@app.get("/")
async def root():
    obs.logger.info("root_called")
    return {"hello": "world"}

The returned Observability object gives you access to all configured components:

Attribute	Description
`obs.logger`	Structured logger (structlog) with trace correlation
`obs.tracer`	Configured OpenTelemetry tracer
`obs.metrics`	RED metrics recorder
`obs.config`	Frozen `ObservabilityConfig` dataclass
`obs.shutdown()`	Graceful shutdown of all components

You can also retrieve the singleton later from anywhere in your code:

Python

from obskit import get_observability

obs = get_observability()  # returns the same instance

The old API still works

The individual-module approach shown below (get_logger(), setup_tracing(), REDMetrics, ObskitMiddleware, etc.) continues to work without any deprecation warnings. Both APIs are fully supported.

Individual Module Approach¶

The sections below show how to set up each observability component individually. This gives you fine-grained control and works with all obskit versions.

Step 1 — Structured Logging with Trace Correlation¶

obskit wraps structlog under the hood and emits machine-readable JSON by default. Every log record is automatically enriched with trace_id, span_id, service, and environment when a span is active.

CodeJSON output (production)Console output (development)

Python

from obskit.logging import get_logger

log = get_logger(__name__)

# Keyword arguments become top-level JSON fields
log.info("user_logged_in", user_id="u-123", ip="10.0.0.1")
log.warning("rate_limit_approaching", user_id="u-123", current_rpm=95, limit_rpm=100)
log.error("payment_failed", order_id="ord-789", reason="card_declined", amount=49.99)

JSON

{
  "event": "user_logged_in",
  "user_id": "u-123",
  "ip": "10.0.0.1",
  "trace_id": "4bf92f3577b34da6a3ce929d0e0e4736",
  "span_id": "00f067aa0ba902b7",
  "service": "order-service",
  "environment": "production",
  "level": "info",
  "timestamp": "2026-02-28T09:12:34.567Z"
}

JSON

{
  "event": "rate_limit_approaching",
  "user_id": "u-123",
  "current_rpm": 95,
  "limit_rpm": 100,
  "trace_id": "4bf92f3577b34da6a3ce929d0e0e4736",
  "span_id": "00f067aa0ba902b7",
  "service": "order-service",
  "environment": "production",
  "level": "warning",
  "timestamp": "2026-02-28T09:12:34.890Z"
}

Set OBSKIT_LOG_FORMAT=console for a human-friendly coloured format:

Text Only

2026-02-28 09:12:34 [info     ] user_logged_in   user_id=u-123 ip=10.0.0.1 trace_id=4bf92f35...
2026-02-28 09:12:34 [warning  ] rate_limit_approaching user_id=u-123 current_rpm=95 limit_rpm=100
2026-02-28 09:12:34 [error    ] payment_failed   order_id=ord-789 reason=card_declined amount=49.99

Logging Configuration¶

Environment variable	Effect
`OBSKIT_LOG_LEVEL=DEBUG`	Emit debug messages
`OBSKIT_LOG_FORMAT=console`	Human-readable output (local dev)
`OBSKIT_LOG_FORMAT=json`	Machine-readable JSON (default, production)
`OBSKIT_SERVICE_NAME=order-service`	Sets `service` field on every record
`OBSKIT_ENVIRONMENT=production`	Sets `environment` field on every record

Trace IDs are injected automatically

When obskit[otlp] is also installed, trace_id and span_id come from the active OpenTelemetry span context. If no span is active, those fields are omitted cleanly. No manual plumbing required.

Step 2 — Distributed Tracing¶

Call setup_tracing() once at application startup, before your framework is imported. It installs the OTel SDK and auto-patches every detected library.

Basic setupLocal developmentSelective instrumentationManual spansBaggage propagation

Python

from obskit.tracing import setup_tracing

# Auto-detects FastAPI, SQLAlchemy, Redis, httpx, Celery, …
setup_tracing(
    exporter_endpoint="http://tempo:4317",
    sample_rate=0.1,   # keep 10 % of traces in production
)

Python

from obskit.tracing import setup_tracing

# Prints spans to stdout — no collector required
setup_tracing(debug=True)

Python

from obskit.tracing import setup_tracing

# Explicit list — faster startup, predictable behaviour
setup_tracing(
    exporter_endpoint="http://tempo:4317",
    sample_rate=0.1,
    instrument=["fastapi", "sqlalchemy", "redis"],
)

Python

from obskit.tracing import trace_span, async_trace_span

# Synchronous span
with trace_span("process_order", attributes={"order_id": "ord-789", "user_id": "u-123"}):
    result = process_order(order_id="ord-789")

# Asynchronous span
async def fetch_user(uid: str):
    async with async_trace_span("fetch_user", attributes={"user_id": uid}):
        return await db.get_user(uid)

Python

from obskit.tracing import set_baggage, get_baggage

# Baggage flows to every downstream service in the same trace
set_baggage("tenant_id", "acme-corp")
set_baggage("feature_flag", "new_checkout_v2")

tenant = get_baggage("tenant_id")   # → "acme-corp"

span output (debug=True)¶

Text Only

[obskit] SPAN  process_order
  trace_id  : 4bf92f3577b34da6a3ce929d0e0e4736
  span_id   : 00f067aa0ba902b7
  attributes: {order_id: ord-789, user_id: u-123}
  duration  : 45.3 ms
  status    : OK

Step 3 — RED Metrics with Trace Exemplars¶

obskit provides a high-level REDMetrics class and a low-level observe_with_exemplar() helper that links a Prometheus data-point to the active OTel trace — enabling metric-to-trace drill-down in Grafana.

REDMetrics (recommended)observe_with_exemplar (low-level)Prometheus /metrics output

Python

from obskit.metrics.red import REDMetrics

red = REDMetrics(service="order-service")

# Record one request — increments rate counter, error counter (if status>=400),
# and observes duration histogram.
red.record_request(
    endpoint="/orders",
    method="POST",
    status=200,
    duration=0.123,
)

Python

from prometheus_client import Histogram, Counter
from obskit.metrics import observe_with_exemplar

REQUEST_DURATION = Histogram(
    "http_request_duration_seconds",
    "Request latency",
    ["method", "path", "status"],
)

# The exemplar carries {trace_id: "4bf92f..."} so Grafana can jump
# from the histogram bucket straight to the matching Tempo trace.
observe_with_exemplar(
    REQUEST_DURATION.labels(method="GET", path="/orders", status="200"),
    0.045,
)

Text Only

# HELP http_request_duration_seconds Request latency
# TYPE http_request_duration_seconds histogram
http_request_duration_seconds_bucket{method="GET",path="/orders",status="200",le="0.05"} 1 # {trace_id="4bf92f3577b34da6"} 0.045
http_request_duration_seconds_bucket{method="GET",path="/orders",status="200",le="0.1"} 1
http_request_duration_seconds_bucket{method="GET",path="/orders",status="200",le="0.25"} 1
http_request_duration_seconds_bucket{method="GET",path="/orders",status="200",le="+Inf"} 1
http_request_duration_seconds_sum{method="GET",path="/orders",status="200"} 0.045
http_request_duration_seconds_count{method="GET",path="/orders",status="200"} 1

Exemplars require Prometheus ≥ 2.27 with --enable-feature=exemplar-storage

In Grafana Cloud and most managed Prometheus offerings this is already enabled. See the Trace Exemplars guide for Prometheus configuration details.

Step 4 — Health Checks¶

obskit implements the Kubernetes liveness/readiness/health pattern. The HealthChecker reads OBSKIT_SERVICE_NAME and OBSKIT_VERSION from the environment automatically.

Basic health checkerJSON responseDegraded responseCustom async check

Python

import asyncio
from obskit.health import HealthChecker, create_http_check

checker = HealthChecker()

# Add dependency checks (used for readiness)
checker.add_check("database", create_http_check("http://postgres:5432/ping"))
checker.add_check("redis",    create_http_check("http://redis:6379/ping"))
checker.add_check("payments", create_http_check("https://api.stripe.com/v1/ping"))

# Run all checks
result = asyncio.run(checker.check_health())
print(result.to_dict())

JSON

{
  "status": "healthy",
  "service": "order-service",
  "version": "2.0.0",
  "environment": "production",
  "trace_id": "4bf92f3577b34da6a3ce929d0e0e4736",
  "span_id": "00f067aa0ba902b7",
  "checks": {
    "database": {"status": "healthy", "latency_ms": 2.1},
    "redis":    {"status": "healthy", "latency_ms": 0.8},
    "payments": {"status": "healthy", "latency_ms": 45.2}
  },
  "timestamp": "2026-02-28T09:12:34.567Z"
}

JSON

{
  "status": "degraded",
  "service": "order-service",
  "version": "2.0.0",
  "checks": {
    "database": {"status": "healthy", "latency_ms": 2.1},
    "redis":    {"status": "unhealthy", "error": "connection refused"},
    "payments": {"status": "healthy", "latency_ms": 45.2}
  }
}

Python

from obskit.health import HealthChecker

checker = HealthChecker()

@checker.add_readiness_check("database")
async def check_db():
    """Returns True if the DB is reachable."""
    try:
        await db.execute("SELECT 1")
        return True
    except Exception:
        return False

trace_id / span_id in health responses

When obskit[otlp] is active, every /health call is automatically wrapped in a span. The trace_id and span_id fields in the JSON response let you correlate health check results with distributed traces in Grafana.

Step 5 — Run obskit diagnose¶

After wiring everything up, run the built-in diagnostic to confirm the full stack is operational.

CommandWith environment variablesExpected output

Bash

python -m obskit.core.diagnose

Bash

OBSKIT_SERVICE_NAME=order-service \
OBSKIT_OTLP_ENDPOINT=http://localhost:4317 \
OBSKIT_LOG_FORMAT=console \
  python -m obskit.core.diagnose

Text Only

obskit v1.0.0 — Diagnostic Report
══════════════════════════════════════════════════════════════
  Component          Status
  ─────────────────────────────────────────────────────────
  obskit             1.0.0     OK
  prometheus         OK
  otlp               OK
  fastapi            OK

  Environment
  ─────────────────────────────────────────────────────────
  OBSKIT_SERVICE_NAME    order-service
  OBSKIT_ENVIRONMENT     development
  OBSKIT_OTLP_ENDPOINT   http://localhost:4317   (reachable)
  OBSKIT_LOG_LEVEL       INFO
  OBSKIT_LOG_FORMAT      console

  Auto-instrumentors detected: fastapi, sqlalchemy, redis, httpx
══════════════════════════════════════════════════════════════
  All checks passed.

OTLP endpoint unreachable?

The diagnose tool tries a gRPC health-check against OBSKIT_OTLP_ENDPOINT. If it times out, traces will be dropped silently (OTel's default behaviour). Start the observability stack with docker compose up -d or override the endpoint.

Complete Minimal Example¶

Here is the smallest possible FastAPI service with the full observability stack wired in.

v1.0.0+ (recommended)Individual modules

Python

# main.py
from obskit import configure_observability, instrument_fastapi
from fastapi import FastAPI
from obskit.health import HealthChecker

obs = configure_observability(
    service_name="demo",
    environment="development",
    debug=True,
)

checker = HealthChecker()
app = FastAPI(title="Demo")
instrument_fastapi(app)


@app.get("/")
async def root():
    obs.logger.info("root_called")
    return {"hello": "world"}


@app.get("/health")
async def health():
    result = await checker.check_health()
    return result.to_dict()

Python

# main.py
import os
os.environ.setdefault("OBSKIT_SERVICE_NAME", "demo")
os.environ.setdefault("OBSKIT_LOG_FORMAT", "console")

# 1. Tracing MUST be set up before FastAPI is imported
from obskit.tracing import setup_tracing
setup_tracing(debug=True)

from fastapi import FastAPI
from obskit.logging import get_logger
from obskit.metrics.red import REDMetrics
from obskit.health import HealthChecker
from obskit.middleware.fastapi import ObskitMiddleware

log = get_logger(__name__)
red = REDMetrics(service="demo")
checker = HealthChecker()

app = FastAPI(title="Demo")
app.add_middleware(ObskitMiddleware)


@app.get("/")
async def root():
    log.info("root_called")
    return {"hello": "world"}


@app.get("/health")
async def health():
    result = await checker.check_health()
    return result.to_dict()

Bash

uvicorn main:app --reload
# curl http://localhost:8000/
# curl http://localhost:8000/health

What's Next¶

Your First Observable App

Build a complete Order Service with Docker Compose, Grafana, Tempo, and Prometheus.

Upgrading an existing service? Read the breaking-changes summary and import mapping.
Configuration Reference

Every OBSKIT_* environment variable, its default, and valid values.
Trace-Log Correlation

How trace_id / span_id flow from spans into log records.
Trace Exemplars

Link Prometheus histogram buckets to Tempo traces for one-click drill-down.