Logging, Monitoramento e Observabilidade

Um sistema em produção que você não consegue observar é uma caixa preta. Quando algo dá errado — e algo sempre dá errado — você precisa de ferramentas para entender o que aconteceu, quando aconteceu e por quê. Observabilidade é a capacidade de entender o estado interno de um sistema a partir de suas saídas externas. Ela se apoia em três pilares: logs (o que aconteceu), métricas (quanto e com que frequência) e traces (como uma requisição percorreu o sistema).

O Módulo logging da Biblioteca Padrão

import logging

# Configuração básica — suficiente para scripts simples
logging.basicConfig(
    level=logging.DEBUG,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S"
)

logger = logging.getLogger(__name__)

logger.debug("Mensagem de debug — detalhes internos")
logger.info("Informação — fluxo normal")
logger.warning("Aviso — algo inesperado mas não crítico")
logger.error("Erro — funcionalidade comprometida")
logger.critical("Crítico — sistema em risco")

# Hierarquia de níveis
# DEBUG < INFO < WARNING < ERROR < CRITICAL

Configuração Profissional com Handlers

import logging
import logging.handlers
import sys
from pathlib import Path


def configurar_logging(
    nivel:       str  = "INFO",
    arquivo_log: str  = "logs/app.log",
    json_format: bool = False,
    rotacao:     bool = True
) -> logging.Logger:
    """Configura logging completo para produção."""

    Path(arquivo_log).parent.mkdir(parents=True, exist_ok=True)
    nivel_num = getattr(logging, nivel.upper(), logging.INFO)

    # Logger raiz
    logger = logging.getLogger()
    logger.setLevel(logging.DEBUG)   # captura tudo — handlers filtram
    logger.handlers.clear()          # chamar de novo não duplica as linhas

    # ── Handler 1: Console ─────────────────────────────
    handler_console = logging.StreamHandler(sys.stdout)
    handler_console.setLevel(nivel_num)

    fmt_console = logging.Formatter(
        fmt="%(asctime)s [%(levelname)-8s] %(name)s: %(message)s",
        datefmt="%H:%M:%S"
    )
    handler_console.setFormatter(fmt_console)

    # ── Handler 2: Arquivo com Rotação ─────────────────
    if rotacao:
        handler_arquivo = logging.handlers.RotatingFileHandler(
            filename=arquivo_log,
            maxBytes=10 * 1024 * 1024,   # 10 MB
            backupCount=5,
            encoding="utf-8"
        )
    else:
        handler_arquivo = logging.FileHandler(arquivo_log, encoding="utf-8")

    handler_arquivo.setLevel(logging.DEBUG)

    # ── Handler 3: Arquivo separado para erros ─────────
    handler_erros = logging.FileHandler("logs/erros.log", encoding="utf-8")
    handler_erros.setLevel(logging.ERROR)

    # ── Formatadores ───────────────────────────────────
    if json_format:
        fmt_arquivo = JsonFormatter()
    else:
        fmt_arquivo = logging.Formatter(
            fmt="%(asctime)s [%(levelname)-8s] %(name)s "
                "[%(filename)s:%(lineno)d] %(message)s",
            datefmt="%Y-%m-%d %H:%M:%S"
        )

    handler_arquivo.setFormatter(fmt_arquivo)
    handler_erros.setFormatter(fmt_arquivo)

    # ── Registrar handlers ─────────────────────────────
    logger.addHandler(handler_console)
    logger.addHandler(handler_arquivo)
    logger.addHandler(handler_erros)

    return logger


# Usando em módulos específicos
def get_logger(nome: str) -> logging.Logger:
    """Retorna logger configurado para um módulo."""
    return logging.getLogger(f"escola.{nome}")


logger_aluno    = get_logger("aluno")
logger_api      = get_logger("api")
logger_database = get_logger("database")

JSON Logging: Logs Estruturados

Logs em texto são difíceis de filtrar em escala. Logs JSON são indexáveis por ferramentas como Elasticsearch, Loki e Datadog:

import logging
import json
import traceback
from datetime import datetime, timezone


# Atributos que todo LogRecord já tem; o resto veio de extra={}
ATRIBUTOS_PADRAO = set(vars(logging.LogRecord("", 0, "", 0, "", None, None))) | {
    "message", "asctime",
}


class JsonFormatter(logging.Formatter):
    """Formata logs como JSON estruturado."""

    def __init__(self, servico: str = "escola-api", versao: str = "1.0.0"):
        super().__init__()
        self.servico = servico
        self.versao  = versao

    def format(self, record: logging.LogRecord) -> str:
        entrada = {
            # a hora do evento (record.created), não a da formatação
            "timestamp": datetime.fromtimestamp(record.created, timezone.utc).isoformat(),
            "nivel":     record.levelname,
            "logger":    record.name,
            "mensagem":  record.getMessage(),
            "modulo":    record.module,
            "funcao":    record.funcName,
            "linha":     record.lineno,
            "servico":   self.servico,
            "versao":    self.versao,
        }

        # Contexto extra — tudo o que veio via extra={} ou de um filtro:
        # qualquer atributo que um LogRecord comum não tem
        for chave, valor in vars(record).items():
            if chave not in ATRIBUTOS_PADRAO and chave not in entrada:
                entrada[chave] = valor

        # Stack trace para erros
        if record.exc_info:
            entrada["exception"] = {
                "tipo":      record.exc_info[0].__name__,
                "mensagem":  str(record.exc_info[1]),
                "traceback": traceback.format_exception(*record.exc_info)
            }

        return json.dumps(entrada, ensure_ascii=False, default=str)


# Configurando com JSON formatter
def configurar_logging_json(nivel: str = "INFO") -> None:
    logger = logging.getLogger()
    logger.setLevel(getattr(logging, nivel))

    handler = logging.StreamHandler()
    handler.setFormatter(JsonFormatter(servico="escola-api", versao="2.0.0"))
    logger.addHandler(handler)


# Uso com contexto extra
logger = logging.getLogger("escola.api")

# Adicionando contexto à mensagem
logger.info(
    "Requisição processada",
    extra={
        "request_id": "req-abc-123",
        "usuario_id": 42,
        "duracao_ms": 145.3
    }
)

Context Variables: Propagando Contexto

import logging
from contextvars import ContextVar
from uuid import uuid4
import functools


# Variáveis de contexto — propagam por corrotinas
request_id_var:  ContextVar[str] = ContextVar("request_id",  default="")
usuario_id_var:  ContextVar[int] = ContextVar("usuario_id",  default=0)


class ContextualFilter(logging.Filter):
    """Injeta variáveis de contexto em todos os logs."""

    def filter(self, record: logging.LogRecord) -> bool:
        record.request_id  = request_id_var.get()
        record.usuario_id  = usuario_id_var.get()
        return True


# Middleware FastAPI que injeta request_id
from fastapi import FastAPI, Request
import time

app    = FastAPI()
logger = logging.getLogger("escola.api")

# O filtro vai nos handlers, e eles precisam existir: chamar DEPOIS de
# configurar o logging. No import, a raiz ainda não tem handler nenhum.
def instalar_contexto_nos_handlers() -> None:
    for handler in logging.getLogger().handlers:
        handler.addFilter(ContextualFilter())


configurar_logging_json()
instalar_contexto_nos_handlers()


@app.middleware("http")
async def middleware_logging(request: Request, call_next):
    req_id  = str(uuid4())[:8]
    token   = request_id_var.set(req_id)
    inicio  = time.perf_counter()

    logger.info(f"→ {request.method} {request.url.path}")

    try:
        resposta = await call_next(request)
        duracao  = (time.perf_counter() - inicio) * 1000

        logger.info(
            f"← {resposta.status_code} em {duracao:.1f}ms",
            extra={"duracao_ms": round(duracao, 1)}
        )
        return resposta

    except Exception as e:
        logger.error(f"Erro na requisição: {e}", exc_info=True)
        raise
    finally:
        request_id_var.reset(token)

O filtro é o que faz o request_id aparecer em todos os logs da requisição, e não só nos do middleware. Na primeira versão, o laço que o instalava rodava no import, antes de o logging ser configurado. Medido: naquele momento a raiz tinha 0 handlers, o laço não fazia nada, e um log de dentro da rota saía sem request_id. Só as duas linhas do middleware o tinham, porque o passavam à mão em extra. O defeito é silencioso, e o sintoma aparece meses depois, quando alguém tenta seguir uma requisição pelos logs e descobre que não dá. Com o filtro instalado depois da configuração, o extra manual fica desnecessário.

Métricas com Prometheus

pip install prometheus-client

from prometheus_client import (
    Counter, Histogram, Gauge,
    Summary, generate_latest, CONTENT_TYPE_LATEST
)
from fastapi import FastAPI, Request, Response
import time


# ── Definindo Métricas ─────────────────────────────────
requisicoes_total = Counter(
    "http_requests_total",
    "Total de requisições HTTP",
    labelnames=["method", "endpoint", "status_code"]
)

duracao_requisicao = Histogram(
    "http_request_duration_seconds",
    "Duração das requisições HTTP em segundos",
    labelnames=["method", "endpoint"],
    buckets=[0.01, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0]
)

alunos_ativos = Gauge(
    "escola_alunos_ativos_total",
    "Total de alunos ativos no sistema"
)

operacoes_banco = Counter(
    "db_operations_total",
    "Total de operações no banco de dados",
    labelnames=["operacao", "tabela", "sucesso"]
)

# Summary no cliente Python expõe só _count e _sum (sem quantis);
# para percentis, use Histogram e histogram_quantile() no Prometheus
tempo_predicao_ml = Summary(
    "ml_prediction_duration_seconds",
    "Tempo de predição do modelo ML"
)


# ── Middleware para métricas HTTP ──────────────────────
@app.middleware("http")
async def middleware_metricas(request: Request, call_next):
    inicio = time.perf_counter()

    resposta = await call_next(request)

    duracao = time.perf_counter() - inicio
    # o molde da rota ("/alunos/{aluno_id}"), não a URL: um ID por série
    # faz a memória do Prometheus crescer sem limite
    rota     = request.scope.get("route")
    endpoint = rota.path if rota else "desconhecida"

    requisicoes_total.labels(
        method=       request.method,
        endpoint=     endpoint,
        status_code=  resposta.status_code
    ).inc()

    duracao_requisicao.labels(
        method=   request.method,
        endpoint= endpoint
    ).observe(duracao)

    return resposta


# ── Endpoint de métricas para Prometheus ──────────────
@app.get("/metrics")
async def metricas():
    return Response(
        content=generate_latest(),
        media_type=CONTENT_TYPE_LATEST
    )


# Medido: com request.url.path como rótulo, 1000 requisições a
# /alunos/0 … /alunos/999 criaram 1000 séries e um /metrics de 1,2 MB;
# com o molde da rota, 1 série e 4 KB.


# ── Usando métricas no código ──────────────────────────
def cadastrar_aluno(nome: str, email: str) -> int:
    try:
        # Simulação de cadastro
        aluno_id = 1
        operacoes_banco.labels(
            operacao="INSERT", tabela="alunos", sucesso="true"
        ).inc()
        alunos_ativos.inc()
        return aluno_id
    except Exception:
        operacoes_banco.labels(
            operacao="INSERT", tabela="alunos", sucesso="false"
        ).inc()
        raise


@tempo_predicao_ml.time()   # decorator que mede automaticamente
def prever_aprovacao(features):
    import time
    time.sleep(0.05)   # simulando predição
    return True

Sentry: Rastreamento de Erros em Produção

pip install "sentry-sdk[fastapi]"   # aspas: no zsh, colchetes são curinga

import sentry_sdk
from sentry_sdk.integrations.fastapi import FastApiIntegration
from sentry_sdk.integrations.sqlalchemy import SqlalchemyIntegration
from sentry_sdk.integrations.redis import RedisIntegration
import os


def configurar_sentry():
    sentry_sdk.init(
        dsn=os.getenv("SENTRY_DSN"),
        environment=os.getenv("AMBIENTE", "producao"),
        release=os.getenv("VERSAO", "1.0.0"),

        # Integrações automáticas
        integrations=[
            FastApiIntegration(transaction_style="endpoint"),
            SqlalchemyIntegration(),
            RedisIntegration(),
        ],

        # Taxa de amostragem para performance
        traces_sample_rate=0.1,        # 10% das transações
        profiles_sample_rate=0.1,

        # Filtragem de dados sensíveis
        send_default_pii=False,

        # Ignorar erros esperados
        ignore_errors=[
            "NotFound",
            "ValidationError",
        ],

        # Antes de enviar — pode modificar ou descartar
        before_send=filtrar_evento_sentry,
    )


def filtrar_evento_sentry(event, hint):
    """Filtra ou enriquece eventos antes de enviar ao Sentry."""
    # Remove dados sensíveis
    if "request" in event:
        headers = event["request"].get("headers", {})
        headers.pop("authorization", None)
        headers.pop("cookie", None)

    # Descarta erros de saúde (muito comuns e sem valor)
    url = event.get("request", {}).get("url", "")
    if "/health" in url or "/metrics" in url:
        return None

    return event


# Adicionando contexto ao Sentry
def processar_pedido(usuario_id: int, pedido_id: int):
    with sentry_sdk.new_scope() as scope:
        scope.set_user({"id": usuario_id})
        scope.set_tag("pedido_id", pedido_id)
        scope.set_context("pedido", {
            "id":       pedido_id,
            "usuario":  usuario_id,
            "ambiente": os.getenv("AMBIENTE")
        })

        try:
            # processamento...
            pass
        except Exception as e:
            sentry_sdk.capture_exception(e)
            raise


# Capturando mensagens manuais
sentry_sdk.capture_message(
    "Limite de rate limit atingido para usuário",
    level="warning"
)

Distributed Tracing com OpenTelemetry

pip install opentelemetry-api opentelemetry-sdk opentelemetry-instrumentation-fastapi \
    opentelemetry-instrumentation-sqlalchemy opentelemetry-exporter-otlp-proto-grpc

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import (
    BatchSpanProcessor, ConsoleSpanExporter
)
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor
import asyncio
import os


def configurar_tracing(servico: str = "escola-api") -> trace.Tracer:
    """Configura OpenTelemetry tracing."""

    provider = TracerProvider()

    # Console no desenvolvimento; com OTEL_EXPORTER_OTLP_ENDPOINT definida
    # (como no compose abaixo), os spans vão para o Tempo ou o Jaeger
    if os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT"):
        from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import (
            OTLPSpanExporter,
        )
        exporter = OTLPSpanExporter()
    else:
        exporter = ConsoleSpanExporter()
    provider.add_span_processor(BatchSpanProcessor(exporter))

    trace.set_tracer_provider(provider)

    # Instrumentação automática
    FastAPIInstrumentor.instrument_app(app)
    # SQLAlchemyInstrumentor().instrument(engine=engine)

    return trace.get_tracer(servico)


tracer = configurar_tracing()


# Criando spans customizados
async def buscar_e_processar_aluno(aluno_id: int) -> dict:
    with tracer.start_as_current_span("buscar_aluno") as span:
        span.set_attribute("aluno.id", aluno_id)

        # Simula busca no banco
        with tracer.start_as_current_span("db.query"):
            await asyncio.sleep(0.01)   # time.sleep travaria o event loop
            aluno = {"id": aluno_id, "nome": "Ana", "nota": 9.5}
            span.set_attribute("aluno.nome", aluno["nome"])

        # Simula processamento
        with tracer.start_as_current_span("processar"):
            await asyncio.sleep(0.005)
            resultado = {**aluno, "aprovado": aluno["nota"] >= 6}

        span.set_attribute("processamento.sucesso", True)
        return resultado

Dashboard de Observabilidade: Stack Completa

# docker-compose.observabilidade.yml
services:
  # ── API ───────────────────────────────────────────
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - SENTRY_DSN=${SENTRY_DSN}
      - OTEL_EXPORTER_OTLP_ENDPOINT=http://tempo:4317
    depends_on: [prometheus, loki]

  # ── Métricas ──────────────────────────────────────
  prometheus:
    image: prom/prometheus:latest
    ports:
      - "9090:9090"
    volumes:
      - ./config/prometheus.yml:/etc/prometheus/prometheus.yml
      - prometheus_data:/prometheus
    command:
      - --config.file=/etc/prometheus/prometheus.yml
      - --storage.tsdb.retention.time=15d

  # ── Logs ──────────────────────────────────────────
  loki:
    image: grafana/loki:latest
    ports:
      - "3100:3100"
    volumes:
      - loki_data:/loki

  # Promtail chegou ao fim de vida em 2/3/2026; o substituto é o Grafana Alloy
  # (a documentação do Loki traz um conversor de configuração)
  promtail:
    image: grafana/promtail:latest
    volumes:
      - /var/log:/var/log
      - ./config/promtail.yml:/etc/promtail/config.yml

  # ── Traces ────────────────────────────────────────
  tempo:
    image: grafana/tempo:latest
    ports:
      - "3200:3200"
      - "4317:4317"
    volumes:
      - tempo_data:/var/tempo

  # ── Visualização ──────────────────────────────────
  grafana:
    image: grafana/grafana:latest
    ports:
      - "3000:3000"
    environment:
      - GF_SECURITY_ADMIN_PASSWORD=${GRAFANA_ADMIN_PASSWORD}   # do .env, nunca no arquivo
    volumes:
      - grafana_data:/var/lib/grafana
      - ./config/grafana/dashboards:/var/lib/grafana/dashboards
    depends_on: [prometheus, loki, tempo]

  # Métricas da máquina (CPU, memória, disco) — o prometheus.yml coleta dele
  node-exporter:
    image: prom/node-exporter:latest
    ports:
      - "9100:9100"

volumes:
  prometheus_data:
  loki_data:
  tempo_data:
  grafana_data:

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

scrape_configs:
  - job_name: escola-api
    static_configs:
      - targets: ["api:8000"]
    metrics_path: /metrics

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

Exemplo Completo: Sistema de Logging para Produção

import logging
import logging.handlers
import json
import time
import sys
from contextlib import contextmanager
from contextvars import ContextVar
from pathlib import Path
from typing import Generator
from uuid import uuid4
from datetime import datetime, timezone


# Variáveis de contexto globais
_request_id: ContextVar[str] = ContextVar("request_id", default="")
_usuario_id: ContextVar[int] = ContextVar("usuario_id", default=0)
_operacao:   ContextVar[str] = ContextVar("operacao",   default="")


class LoggerSistema:
    """Sistema centralizado de logging para produção."""

    _instancia = None

    def __new__(cls):
        if cls._instancia is None:
            cls._instancia = super().__new__(cls)
            cls._instancia._configurado = False
        return cls._instancia

    def configurar(
        self,
        nivel:       str   = "INFO",
        servico:     str   = "escola-api",
        versao:      str   = "1.0.0",
        pasta_logs:  str   = "logs",
        json_output: bool  = True
    ) -> None:
        if self._configurado:
            return

        Path(pasta_logs).mkdir(exist_ok=True)
        self.servico = servico
        self.versao  = versao

        root = logging.getLogger()
        root.setLevel(logging.DEBUG)

        fmt_json = self._criar_formatter_json(servico, versao)
        fmt_text = logging.Formatter(
            "%(asctime)s [%(levelname)-8s] %(name)s: %(message)s",
            datefmt="%H:%M:%S"
        )

        # Console
        h_console = logging.StreamHandler(sys.stdout)
        h_console.setLevel(getattr(logging, nivel.upper()))
        h_console.setFormatter(fmt_json if json_output else fmt_text)
        h_console.addFilter(self._ContextFilter())
        root.addHandler(h_console)

        # Arquivo rotativo
        h_arquivo = logging.handlers.TimedRotatingFileHandler(
            filename=f"{pasta_logs}/{servico}.log",
            when="midnight",
            backupCount=30,
            encoding="utf-8"
        )
        h_arquivo.setLevel(logging.DEBUG)
        h_arquivo.setFormatter(fmt_json)
        h_arquivo.addFilter(self._ContextFilter())
        root.addHandler(h_arquivo)

        # Erros separados
        h_erros = logging.handlers.RotatingFileHandler(
            filename=f"{pasta_logs}/erros.log",
            maxBytes=5 * 1024 * 1024,
            backupCount=10,
            encoding="utf-8"
        )
        h_erros.setLevel(logging.ERROR)
        h_erros.setFormatter(fmt_json)
        root.addHandler(h_erros)

        self._configurado = True

    class _ContextFilter(logging.Filter):
        def filter(self, record):
            record.request_id = _request_id.get()
            record.usuario_id = _usuario_id.get()
            record.operacao   = _operacao.get()
            return True

    @staticmethod
    def _criar_formatter_json(servico, versao):
        class _JsonFmt(logging.Formatter):
            def format(self, record):
                entrada = {
                    "ts":         datetime.fromtimestamp(
                        record.created, timezone.utc
                    ).isoformat(),
                    "nivel":      record.levelname,
                    "logger":     record.name,
                    "msg":        record.getMessage(),
                    "modulo":     record.module,
                    "linha":      record.lineno,
                    "servico":    servico,
                    "versao":     versao,
                    "request_id": getattr(record, "request_id", ""),
                    "usuario_id": getattr(record, "usuario_id", 0),
                    "operacao":   getattr(record, "operacao",   ""),
                }
                # campos passados em extra={}, como o aluno_id abaixo
                padrao = set(vars(logging.LogRecord("", 0, "", 0, "", None, None)))
                for chave, valor in vars(record).items():
                    if chave not in padrao | {"message", "asctime"} and chave not in entrada:
                        entrada[chave] = valor
                if record.exc_info:
                    import traceback
                    entrada["erro"] = {
                        "tipo":  record.exc_info[0].__name__,
                        "msg":   str(record.exc_info[1]),
                        "trace": traceback.format_exception(*record.exc_info)
                    }
                return json.dumps(entrada, ensure_ascii=False, default=str)
        return _JsonFmt()

    @contextmanager
    def contexto_requisicao(
        self,
        usuario_id: int = 0,
        operacao:   str = ""
    ) -> Generator[str, None, None]:
        """Context manager que define contexto de logging."""
        req_id = str(uuid4())[:8]
        t1 = _request_id.set(req_id)
        t2 = _usuario_id.set(usuario_id)
        t3 = _operacao.set(operacao)
        inicio = time.perf_counter()
        logger = logging.getLogger("escola.contexto")
        logger.info(f"Iniciando operação: {operacao or 'desconhecida'}")
        try:
            yield req_id
        except Exception as e:
            logger.error(f"Erro na operação: {e}", exc_info=True)
            raise
        finally:
            duracao = (time.perf_counter() - inicio) * 1000
            logger.info(f"Operação concluída em {duracao:.1f}ms")
            _request_id.reset(t1)
            _usuario_id.reset(t2)
            _operacao.reset(t3)

    def get(self, nome: str) -> logging.Logger:
        return logging.getLogger(f"escola.{nome}")


# Singleton global
sistema_log = LoggerSistema()
sistema_log.configurar(
    nivel="INFO",
    servico="escola-api",
    versao="2.0.0",
    json_output=True
)


# Usando no código da aplicação
logger = sistema_log.get("aluno")

with sistema_log.contexto_requisicao(usuario_id=42, operacao="cadastro_aluno"):
    logger.info("Cadastrando aluno")
    logger.info("Aluno cadastrado com sucesso", extra={"aluno_id": 1})

Observabilidade se apoia em três fontes, e o Python tem ferramenta para cada uma. O módulo logging, com handlers, formatters e filtros, produz logs estruturados em JSON, e as ContextVar carregam o request_id por toda a requisição. O prometheus_client expõe contadores, histogramas e gauges num endpoint /metrics. O OpenTelemetry mede o caminho de cada requisição em spans. O Sentry junta a isso o rastreamento de erros com contexto, e a stack Prometheus, Loki, Tempo e Grafana reúne tudo num só painel — o mesmo tipo de vigilância que MLOps: Versionamento de Modelos, MLflow e Deploy de ML aplica a um modelo em produção.

A armadilha comum a quase todos os defeitos deste artigo é que eles não aparecem no dia em que o código é escrito. O filtro que não foi instalado só faz falta quando alguém tenta seguir uma requisição pelos logs; o rótulo com o ID do aluno só pesa quando o Prometheus já guarda milhares de séries; o campo passado em extra só some quando é procurado. Por isso vale testar a instrumentação como se testa o resto: fazer uma requisição, abrir o log e o /metrics, e conferir se o que se espera encontrar está lá. É o tipo de verificação que cabe no pipeline de CI/CD com GitHub Actions.

Fontes e leituras recomendadas

• logging — documentação oficial — https://docs.python.org/3/library/logging.html
• Logging HOWTO — https://docs.python.org/3/howto/logging.html
• prometheus-client Python — https://github.com/prometheus/client_python
• Sentry Python SDK — https://docs.sentry.io/platforms/python/
• OpenTelemetry Python — https://opentelemetry.io/docs/languages/python/
• Grafana — documentação — https://grafana.com/docs/
• BEYER, Betsy et al. Site Reliability Engineering. O'Reilly Media, 2016. — fundamentos de observabilidade e confiabilidade de sistemas em produção pelo time do Google.
• MAJORS, Charity; FONG-JONES, Liz; MIRANDA, George. Observability Engineering. O'Reilly Media, 2022. — referência definitiva sobre os três pilares de observabilidade em sistemas modernos.

Exercícios

Exercício 1

Um incidente em produção: a equipe quer ver tudo o que aconteceu na requisição 4490a2c1. A busca no Loki devolve só duas linhas, a de entrada e a de saída do middleware. Os logs do serviço de alunos, gravados no meio da requisição, estão lá, mas sem request_id. O código tem um ContextualFilter que lê a ContextVar, instalado com um laço sobre logging.getLogger().handlers logo depois da definição da classe. Por que ele não funcionou?

Exercício 2

Três semanas depois de ligar as métricas, o Prometheus começa a usar muita memória, e o /metrics da API demora a responder. O middleware registra http_request_duration_seconds com os rótulos method e endpoint=request.url.path. A API tem rotas como /alunos/{aluno_id} e 40 mil alunos. O que está acontecendo?

Exercício 3

O código registra logger.info("Aluno cadastrado com sucesso", extra={"aluno_id": 1}). O log sai em JSON, com mensagem, nível, request_id e usuário, mas não há aluno_id em lugar nenhum. O extra funciona? Por que o campo sumiu?

Exercício 4

Para acompanhar o tempo de predição do modelo, alguém usou Summary("ml_prediction_duration_seconds", ...) e quer montar no Grafana um painel com o percentil 95. No /metrics, porém, aparecem só ml_prediction_duration_seconds_count e ml_prediction_duration_seconds_sum. Onde estão os quantis?

Exercício 5

O compose de observabilidade define OTEL_EXPORTER_OTLP_ENDPOINT=http://tempo:4317 para a API, e o Tempo sobe sem erros. No Grafana, porém, não aparece nenhum trace. Os logs da API estão cheios de blocos JSON com "name": "buscar_aluno". A primeira versão da função configurar_tracing criava o exportador com exporter = ConsoleSpanExporter(). Explique.