Artigo 46 — Logging, Monitoramento e Observabilidade
Prof. Ricardo Matos Módulo 8 · Testes, Qualidade e Boas Práticas · Artigo 46 de 52
Introdução
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("logs").mkdir(exist_ok=True)
nivel_num = getattr(logging, nivel.upper(), logging.INFO)
# Logger raiz
logger = logging.getLogger()
logger.setLevel(logging.DEBUG) # captura tudo — handlers filtram
# ── 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
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 = {
"timestamp": datetime.now(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 — passado via extra={}
if hasattr(record, "request_id"):
entrada["request_id"] = record.request_id
if hasattr(record, "usuario_id"):
entrada["usuario_id"] = record.usuario_id
if hasattr(record, "duracao_ms"):
entrada["duracao_ms"] = record.duracao_ms
# 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)
# 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")
# Adicionando o filtro contextual
for handler in logging.getLogger().handlers:
handler.addFilter(ContextualFilter())
@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}",
extra={"request_id": req_id}
)
try:
resposta = await call_next(request)
duracao = (time.perf_counter() - inicio) * 1000
logger.info(
f"← {resposta.status_code} em {duracao:.1f}ms",
extra={"request_id": req_id, "duracao_ms": duracao}
)
return resposta
except Exception as e:
logger.error(
f"Erro na requisição: {e}",
exc_info=True,
extra={"request_id": req_id}
)
raise
finally:
request_id_var.reset(token)
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"]
)
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
endpoint = request.url.path
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
)
# ── 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]
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
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 time
def configurar_tracing(servico: str = "escola-api") -> trace.Tracer:
"""Configura OpenTelemetry tracing."""
provider = TracerProvider()
# Exportador para console (dev) ou Jaeger/Tempo (prod)
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"):
time.sleep(0.01)
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"):
time.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
version: "3.9"
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:
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=admin123
volumes:
- grafana_data:/var/lib/grafana
- ./config/grafana/dashboards:/var/lib/grafana/dashboards
depends_on: [prometheus, loki, tempo]
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.now(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", ""),
}
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)
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})
Resumo
- Logs têm cinco níveis: DEBUG, INFO, WARNING, ERROR, CRITICAL — configure o mínimo por handler
- Handlers definem o destino dos logs — console, arquivo, rede — cada um com seu nível e formato
- Logs JSON estruturados são indexáveis e consultáveis em ferramentas como Loki e Elasticsearch
- ContextVar propaga request_id e usuario_id por toda a cadeia de chamadas sem passar parâmetros
- Prometheus coleta métricas numéricas — contadores, histogramas e gauges — via endpoint /metrics
- Sentry captura erros em produção com stack trace, contexto e agrupamento automático
- OpenTelemetry padroniza distributed tracing — visualize no Jaeger ou Grafana Tempo
- A stack Prometheus + Loki + Tempo + Grafana cobre os três pilares de observabilidade
Referências e Leituras Complementares
- 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.
Prof. Ricardo Matos — Dominando o Python em 1 Ano · Artigo 46 de 52 Próximo: Artigo 47 — Design Patterns em Python
you asked
Sim