O Que é Observabilidade e Por Que Você Precisa Disso
Observabilidade é a capacidade de entender o estado interno de um sistema a partir de seus sinais externos. Diferentemente de monitoramento tradicional, que responde a perguntas predefinidas ("o servidor está up?"), observabilidade permite fazer perguntas arbitrárias sobre o comportamento da sua aplicação ("por que essa requisição levou 3 segundos?").
Em aplicações Python modernas, especialmente em arquiteturas de microsserviços, você precisa de três pilares: logs estruturados, métricas e traces distribuídos. Saber apenas que uma requisição falhou não é suficiente — você precisa rastrear por onde passou, quanto tempo levou em cada etapa e quais recursos consumiu. Este artigo cobre três ferramentas que formam a base de observabilidade profissional: OpenTelemetry (padrão da indústria), Sentry (tratamento de erros), e py-spy (profiling de performance).
OpenTelemetry: O Padrão de Observabilidade
O Que é OpenTelemetry
OpenTelemetry (OTel) é um projeto CNCF que padroniza como você coleta telemetria — traces, métricas e logs — sem ficar preso a um fornecedor específico. Você instrui seu código uma vez e pode enviar dados para Jaeger, Datadog, New Relic, ou qualquer backend compatível.
Existem três conceitos fundamentais: um span representa uma unidade de trabalho (como uma requisição HTTP), um trace é uma árvore de spans conectados mostrando o fluxo completo, e instrumentação é o processo de adicionar código que coleta esses dados.
Instalação e Configuração Básica
pip install opentelemetry-api opentelemetry-sdk opentelemetry-exporter-jaeger opentelemetry-instrumentation-flask opentelemetry-instrumentation-requests
Aqui está uma aplicação Flask completa com OpenTelemetry:
from flask import Flask, jsonify
from opentelemetry import trace, metrics
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.jaeger.thrift import JaegerExporter
from opentelemetry.instrumentation.flask import FlaskInstrumentor
from opentelemetry.instrumentation.requests import RequestsInstrumentor
import requests
# Configurar exporter para Jaeger (local, porta 6831)
jaeger_exporter = JaegerExporter(
agent_host_name="localhost",
agent_port=6831,
)
# Configurar provider e adicionar exporter
trace_provider = TracerProvider()
trace_provider.add_span_processor(BatchSpanProcessor(jaeger_exporter))
trace.set_tracer_provider(trace_provider)
# Criar aplicação Flask
app = Flask(__name__)
# Instrumentar automaticamente
FlaskInstrumentor().instrument_app(app)
RequestsInstrumentor().instrument()
# Obter tracer
tracer = trace.get_tracer(__name__)
@app.route("/api/process")
def process():
# Criar span manualmente para lógica customizada
with tracer.start_as_current_span("process_data") as span:
span.set_attribute("user.id", 123)
# Simular chamada externa
response = requests.get("https://api.example.com/data")
# Span filhos são criados automaticamente pela instrumentação
result = {
"status": "success",
"status_code": response.status_code
}
span.set_attribute("result.size", len(str(result)))
return jsonify(result)
if __name__ == "__main__":
app.run(debug=False)
Quando você executa essa aplicação e faz requisições, o OpenTelemetry coleta automaticamente spans para Flask e requests. Para visualizar, abra http://localhost:16686 (UI do Jaeger) — você verá toda a árvore de execução com timings.
Atributos e Eventos em Spans
Spans são mais poderosos quando você adiciona contexto. Use atributos para metadados persistentes e eventos para marcos de tempo específicos:
@app.route("/api/checkout", methods=["POST"])
def checkout():
with tracer.start_as_current_span("checkout_process") as span:
# Atributos: metadados da requisição
span.set_attribute("checkout.user_id", 456)
span.set_attribute("checkout.item_count", 5)
span.set_attribute("checkout.total_amount", 199.99)
# Simular processamento
try:
# Evento: algo importante aconteceu
span.add_event("inventory_check_started")
# ... verificar inventário
span.add_event("inventory_check_completed", attributes={"items_available": True})
span.add_event("payment_processing_started")
# ... processar pagamento
span.add_event("payment_processing_completed", attributes={"payment_id": "pay_xyz123"})
except Exception as e:
span.record_exception(e)
span.set_status(trace.Status(trace.StatusCode.ERROR))
raise
return jsonify({"order_id": "ORD-789"})
Isso permite visualizar no Jaeger não apenas quanto tempo cada etapa levou, mas também o contexto específico (qual usuário, quantos itens, qual ID de pagamento).
Sentry: Capturando e Rastreando Erros
O Problema que Sentry Resolve
Logs tradicionais são volumosos e fáceis de perder. Sentry diferencia-se ao: agrupar erros idênticos automaticamente, capturar contexto de usuário e ambiente, e fornecer alertas inteligentes. Você não monitora logs — você detecta problemas reais antes dos usuários reclamarem.
Integrando Sentry em uma Aplicação Python
pip install sentry-sdk
Configuração básica (qualquer aplicação):
import sentry_sdk
from sentry_sdk.integrations.flask import FlaskIntegration
from sentry_sdk.integrations.sqlalchemy import SqlalchemyIntegration
sentry_sdk.init(
dsn="https://seu_key@sentry.io/seu_project", # Obtenha em sentry.io
integrations=[
FlaskIntegration(),
SqlalchemyIntegration(),
],
traces_sample_rate=0.1, # Envie 10% dos traces
environment="production",
)
from flask import Flask, jsonify
import logging
app = Flask(__name__)
# Opcional: capturar logs também
logging.basicConfig(level=logging.WARNING)
@app.route("/api/divide")
def divide():
a = int(request.args.get("a", 10))
b = int(request.args.get("b", 0))
try:
result = a / b # Vai falhar se b=0
except ZeroDivisionError as e:
# Capturar contexto customizado
sentry_sdk.capture_exception(e)
return jsonify({"error": "Division by zero"}), 400
return jsonify({"result": result})
@app.route("/api/risky", methods=["POST"])
def risky_operation():
# Sentry captura automaticamente exceções não tratadas
data = request.json
user_id = data["user_id"] # Pode gerar KeyError
# Adicionar contexto de usuário
sentry_sdk.set_user({
"id": user_id,
"email": data.get("email"),
})
# Adicionar tags (facilita filtragem no dashboard)
sentry_sdk.set_tag("operation", "risky_operation")
sentry_sdk.set_tag("request_type", data.get("type"))
# Adicionar informações estruturadas
sentry_sdk.set_context("operation_details", {
"items_count": len(data.get("items", [])),
"priority": data.get("priority", "normal"),
})
# Seu código aqui
return jsonify({"status": "ok"})
if __name__ == "__main__":
app.run(debug=False)
Quando um erro ocorre, Sentry captura automaticamente: stack trace completo, variáveis locais de cada frame, IP do usuário, navegador (se web), variáveis de ambiente (sem valores sensíveis). No dashboard Sentry, você vê tendências — "esse erro apareceu 50 vezes hoje" — e pode configurar alertas.
Diferenciando Errors de Warnings
Nem tudo que vai errado é crítico. Use captura manual para informar Sentry sem disparar exceções:
import sentry_sdk
@app.route("/api/check-quota")
def check_quota():
user_id = request.args.get("user_id")
quota = get_user_quota(user_id)
if quota > 0.9: # 90% consumido
# Informar sem falhar
sentry_sdk.capture_message(
f"User {user_id} quota at {quota*100:.0f}%",
level="warning"
)
if quota >= 1.0:
# Algo realmente errado
sentry_sdk.capture_message(
f"User {user_id} exceeded quota",
level="error"
)
return jsonify({"error": "Quota exceeded"}), 429
return jsonify({"quota_remaining": 1.0 - quota})
Profiling com py-spy: Encontrando Gargalos
Por Que Profiling Importa
OpenTelemetry te diz quando algo é lento. Sentry te diz que errou. Mas e se uma função leva 5 segundos e ninguém sabe por quê? Profiling analisa onde o CPU gasta tempo. py-spy é um profiler que não requer instrumentação — você o roda e ele tira uma foto do que sua aplicação está fazendo.
Instalação e Uso Básico
pip install py-spy
Rodar um profiling de 30 segundos em uma aplicação já em execução:
# Se sua app Python roda com PID 12345
py-spy record -o profile.svg -d 30 12345
# Ou perfil de toda a sessão (até Ctrl+C)
py-spy record -o profile.svg python seu_script.py
Isso gera um arquivo SVG interativo. Áreas maiores = mais tempo de CPU. Clique para ver detalhes.
Exemplo Prático: Aplicação com Gargalo Intencional
import time
from flask import Flask, jsonify
app = Flask(__name__)
def inefficient_algorithm(n):
"""Algoritmo O(n²) — gargalo intencional"""
total = 0
for i in range(n):
for j in range(n):
total += i * j
return total
def optimized_algorithm(n):
"""Versão O(n) — muito mais rápida"""
return (n * (n - 1) // 2) ** 2
@app.route("/api/calculate/<int:size>")
def calculate(size):
# Use py-spy para medir qual versão é mais rápida
start = time.time()
result = inefficient_algorithm(size)
elapsed = time.time() - start
return jsonify({
"result": result,
"method": "inefficient",
"elapsed_seconds": elapsed
})
if __name__ == "__main__":
app.run(debug=False)
Execute assim:
# Terminal 1: rodar a app
python app.py
# Terminal 2: gerar profile enquanto faz requisições
py-spy record -o profile.svg $(pgrep -f "python app.py")
# Terminal 3: fazer requisições (em outro terminal)
for i in {1..10}; do curl http://localhost:5000/api/calculate/500; done
Abrindo profile.svg no navegador, você vê que inefficient_algorithm consume 80%+ do CPU. Mudança simples: substituir por optimized_algorithm.
Análise Estatística com py-spy dump
Para snapshot instantâneo sem gerar SVG:
py-spy dump --pid 12345
Mostra stack trace de cada thread naquele momento. Útil para descobrir se a app está travada em alguma operação.
Integrando Profiling em Ambiente de Produção
import os
from functools import wraps
import time
def profile_if_slow(threshold_ms=1000):
"""Decorator que alerta se função demorar demais"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
start = time.perf_counter()
result = func(*args, **kwargs)
elapsed_ms = (time.perf_counter() - start) * 1000
if elapsed_ms > threshold_ms:
import sentry_sdk
sentry_sdk.capture_message(
f"{func.__name__} took {elapsed_ms:.0f}ms (threshold: {threshold_ms}ms)",
level="warning"
)
return result
return wrapper
return decorator
@profile_if_slow(threshold_ms=500)
def fetch_user_data(user_id):
# Se isso demorar mais de 500ms, Sentry avisa
time.sleep(0.6)
return {"id": user_id, "name": "User"}
Integrando Tudo Junto: Um Exemplo Completo
Para solidificar, aqui está uma aplicação que combina os três:
import sentry_sdk
from sentry_sdk.integrations.flask import FlaskIntegration
from flask import Flask, request, jsonify
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.jaeger.thrift import JaegerExporter
from opentelemetry.instrumentation.flask import FlaskInstrumentor
import time
# Sentry
sentry_sdk.init(
dsn="https://key@sentry.io/project",
integrations=[FlaskIntegration()],
traces_sample_rate=0.1,
environment="production",
)
# OpenTelemetry
jaeger_exporter = JaegerExporter(agent_host_name="localhost", agent_port=6831)
trace_provider = TracerProvider()
trace_provider.add_span_processor(BatchSpanProcessor(jaeger_exporter))
trace.set_tracer_provider(trace_provider)
# Flask
app = Flask(__name__)
FlaskInstrumentor().instrument_app(app)
tracer = trace.get_tracer(__name__)
@app.route("/api/process-order", methods=["POST"])
def process_order():
with tracer.start_as_current_span("order_processing") as span:
data = request.json
order_id = data.get("order_id")
span.set_attribute("order.id", order_id)
sentry_sdk.set_context("order", {"id": order_id})
try:
span.add_event("validating_order")
validate_order(data) # Pode gerar erro
span.add_event("processing_payment")
# Simular operação lenta
time.sleep(0.1)
span.set_attribute("order.status", "completed")
return jsonify({"order_id": order_id, "status": "success"})
except ValueError as e:
span.record_exception(e)
sentry_sdk.capture_exception(e)
return jsonify({"error": str(e)}), 400
def validate_order(data):
if not data.get("items"):
raise ValueError("Order must have items")
if data.get("total") < 0:
raise ValueError("Total must be positive")
if __name__ == "__main__":
app.run(debug=False)
Com isso você tem:
- Jaeger mostra a árvore completa de execução (traces e spans)
- Sentry alertar sobre exceções em tempo real
- py-spy consegue profiles CPU quando necessário
- Tudo correlacionado: um erro em Sentry pode levar a um trace no Jaeger
Conclusão
Observabilidade em produção não é luxo — é necessidade. Os três pilares aprendidos resolvem problemas diferentes: OpenTelemetry oferece rastreamento de requisição ponta-a-ponta, mostrando exatamente onde o tempo é gasto em arquiteturas complexas. Sentry captura erros antes de se tornarem crises, diferenciando problemas críticos de warnings e agrupando automaticamente. py-spy identifica gargalos de performance sem overhead em produção, usando amostragem.
A chave é não escolher apenas um — use-os juntos. Quando um cliente reporta "a requisição demorou", você vai direto ao Jaeger, vê qual step foi lento, roda py-spy naquele endpoint, encontra a função problemática, corrige, e Sentry avisa se regredir. É observabilidade acionável.