# Swe Observability

> Diretrizes de observabilidade para projetos Python. Use quando o usuário pedir para configurar logs estruturados, métricas (Prometheus), rastreamento distribuído (OpenTelemetry), alertas, dashboards, health checks, configurar structlog, integrar com Datadog/Grafana/Jaeger, depurar problemas em produção, ou qualquer tarefa relacionada ao monitoramento e observabilidade de aplicações Python.

- Skill: `nxs-cafi/swe-observability` (Agent Skill)
- Install (CLI): `npx skillmds@latest add nxs-cafi/swe-observability`
- Raw SKILL.md: https://api.skillmd.com/api/skills/nxs-cafi/swe-observability/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Coding & Dev Tools
- Author: nxs-cafi (https://skillmd.com/u/nxs-cafi)
- Updated: 2026-09-21
- Page: https://skillmd.com/skills/nxs-cafi/swe-observability

---


# Observabilidade — Python SWE Agent

## Os Três Pilares

```
┌────────────┐    ┌────────────┐    ┌────────────┐
│    LOGS    │    │  MÉTRICAS  │    │   TRACES   │
│            │    │            │    │            │
│ O que      │    │ Quantas    │    │ Por onde   │
│ aconteceu? │    │ vezes?     │    │ passou?    │
│            │    │ Quanto?    │    │ Quanto     │
│ Texto      │    │ Tendência? │    │ demorou?   │
│ estruturado│    │ Contador   │    │ Span tree  │
└────────────┘    └────────────┘    └────────────┘
```

## Logging Estruturado com structlog

```python
# infrastructure/config/logging.py
import structlog
import logging
import sys

def configure_logging(log_level: str = "INFO", json_output: bool = True) -> None:
    shared_processors = [
        structlog.contextvars.merge_contextvars,
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso", utc=True),
        structlog.stdlib.add_logger_name,
    ]

    if json_output:
        renderer = structlog.processors.JSONRenderer()
    else:
        renderer = structlog.dev.ConsoleRenderer(colors=True)

    structlog.configure(
        processors=shared_processors + [
            structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
        ],
        wrapper_class=structlog.make_filtering_bound_logger(
            logging.getLevelName(log_level)
        ),
        context_class=dict,
        logger_factory=structlog.PrintLoggerFactory(),
    )

# USO:
logger = structlog.get_logger(__name__)

# ✅ Log estruturado com contexto
logger.info(
    "order_created",
    order_id=str(order.id),
    customer_id=str(order.customer_id),
    total_amount=str(order.total.amount),
    currency=order.total.currency,
    items_count=len(order.items),
)

# ✅ Log de erro com contexto de exceção
try:
    result = await payment_service.charge(order)
except PaymentProviderError as e:
    logger.error(
        "payment_failed",
        order_id=str(order.id),
        error_code=e.code,
        error_message=e.message,
        provider=e.provider,
        exc_info=True,  # inclui stack trace
    )
    raise

# ❌ Log sem estrutura e sem contexto
logger.info(f"Order {order.id} created")  # difícil de filtrar/agregar
logger.error("Payment failed")            # sem contexto de diagnóstico
```

## Contexto de Requisição — Propagação

```python
# middleware que injeta trace_id em todos os logs da requisição
import uuid
from structlog.contextvars import bind_contextvars, clear_contextvars

@app.middleware("http")
async def request_context_middleware(request: Request, call_next):
    clear_contextvars()

    request_id = request.headers.get("X-Request-ID") or str(uuid.uuid4())
    bind_contextvars(
        request_id=request_id,
        method=request.method,
        path=request.url.path,
    )

    response = await call_next(request)

    # Adiciona request_id no response para rastreabilidade
    response.headers["X-Request-ID"] = request_id
    return response
```

## Métricas com Prometheus

```python
# infrastructure/metrics.py
from prometheus_client import Counter, Histogram, Gauge, CollectorRegistry

# Contadores
http_requests_total = Counter(
    "http_requests_total",
    "Total de requisições HTTP",
    ["method", "endpoint", "status_code"],
)

business_events_total = Counter(
    "business_events_total",
    "Eventos de negócio processados",
    ["event_type", "status"],
)

# Histogramas (latência)
http_request_duration_seconds = Histogram(
    "http_request_duration_seconds",
    "Latência das requisições HTTP",
    ["method", "endpoint"],
    buckets=[0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0],
)

order_processing_duration_seconds = Histogram(
    "order_processing_duration_seconds",
    "Tempo de processamento de pedido",
    buckets=[0.1, 0.5, 1.0, 2.0, 5.0, 10.0],
)

# Gauges (valores instantâneos)
active_connections = Gauge(
    "active_connections",
    "Conexões ativas com o banco de dados",
)

# Middleware de métricas HTTP
import time

@app.middleware("http")
async def metrics_middleware(request: Request, call_next):
    start = time.perf_counter()
    response = await call_next(request)
    duration = time.perf_counter() - start

    endpoint = request.url.path
    http_requests_total.labels(
        method=request.method,
        endpoint=endpoint,
        status_code=response.status_code,
    ).inc()

    http_request_duration_seconds.labels(
        method=request.method,
        endpoint=endpoint,
    ).observe(duration)

    return response

# Endpoint de métricas
from prometheus_client import generate_latest, CONTENT_TYPE_LATEST

@app.get("/metrics", include_in_schema=False)
async def metrics():
    return Response(generate_latest(), media_type=CONTENT_TYPE_LATEST)
```

## Rastreamento Distribuído com OpenTelemetry

```python
# infrastructure/tracing.py
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor
from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor

def configure_tracing(service_name: str, otlp_endpoint: str) -> None:
    provider = TracerProvider(
        resource=Resource.create({
            "service.name": service_name,
            "service.version": os.environ.get("APP_VERSION", "unknown"),
            "deployment.environment": os.environ.get("ENVIRONMENT", "development"),
        })
    )

    provider.add_span_processor(
        BatchSpanProcessor(OTLPSpanExporter(endpoint=otlp_endpoint))
    )

    trace.set_tracer_provider(provider)

    # Auto-instrumentação
    FastAPIInstrumentor.instrument()
    SQLAlchemyInstrumentor.instrument()
    HTTPXClientInstrumentor.instrument()

# Spans customizados para operações de negócio
tracer = trace.get_tracer(__name__)

class CreateOrderUseCase:
    async def execute(self, dto: CreateOrderDTO) -> Order:
        with tracer.start_as_current_span("create_order") as span:
            span.set_attributes({
                "order.customer_id": str(dto.customer_id),
                "order.items_count": len(dto.items),
            })

            try:
                order = await self._process(dto)
                span.set_attribute("order.id", str(order.id))
                span.set_status(trace.StatusCode.OK)
                return order
            except Exception as e:
                span.record_exception(e)
                span.set_status(trace.StatusCode.ERROR, str(e))
                raise
```

## Health Checks

```python
# interfaces/api/health.py
from fastapi import APIRouter
from enum import Enum

router = APIRouter(tags=["health"])

class HealthStatus(str, Enum):
    HEALTHY = "healthy"
    DEGRADED = "degraded"
    UNHEALTHY = "unhealthy"

@router.get("/health/live")  # Kubernetes liveness probe
async def liveness():
    """Verifica se a aplicação está rodando."""
    return {"status": "alive"}

@router.get("/health/ready")  # Kubernetes readiness probe
async def readiness(
    db: AsyncSession = Depends(get_db),
    redis: Redis = Depends(get_redis),
):
    """Verifica se a aplicação está pronta para receber tráfego."""
    checks = {}

    # Verifica banco de dados
    try:
        await db.execute(text("SELECT 1"))
        checks["database"] = HealthStatus.HEALTHY
    except Exception:
        checks["database"] = HealthStatus.UNHEALTHY

    # Verifica Redis
    try:
        await redis.ping()
        checks["cache"] = HealthStatus.HEALTHY
    except Exception:
        checks["cache"] = HealthStatus.UNHEALTHY

    overall = (
        HealthStatus.HEALTHY
        if all(v == HealthStatus.HEALTHY for v in checks.values())
        else HealthStatus.UNHEALTHY
    )

    status_code = 200 if overall == HealthStatus.HEALTHY else 503
    return JSONResponse(
        content={"status": overall, "checks": checks},
        status_code=status_code,
    )
```

## Convenções de Log — O que SEMPRE Logar

| Evento | Nível | Campos Obrigatórios |
|--------|-------|---------------------|
| Request recebido | DEBUG | request_id, method, path |
| Operação de negócio concluída | INFO | event_name, entity_id, duração |
| Erro de negócio (ex: estoque) | WARNING | event_name, reason, entity_id |
| Erro inesperado / exceção | ERROR | event_name, exc_info=True, contexto |
| Erro crítico (sistema down) | CRITICAL | event_name, componente afetado |

## O que NUNCA Logar

```python
# ❌ PII — dados pessoais identificáveis
logger.info("user_login", email=user.email)       # proibido
logger.info("payment", card_number=card_number)   # proibido
logger.info("user_created", cpf=user.cpf)         # proibido

# ✅ Use IDs opacos
logger.info("user_login", user_id=str(user.id))
logger.info("payment_processed", payment_id=str(payment.id))
```

## Stack de Observabilidade Recomendada

```
Coleta    → OpenTelemetry Collector
Logs      → Loki + Grafana (ou CloudWatch / Datadog)
Métricas  → Prometheus + Grafana
Traces    → Jaeger / Tempo + Grafana
Alertas   → Alertmanager / PagerDuty
```