<h2>O Que é Observabilidade e Por Que Você Precisa Disso</h2>
<p>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?").</p>
<p>Em aplicações Python modernas, especialmente em arquiteturas de microsserviços, você precisa de três pilares: <strong>logs estruturados</strong>, <strong>métricas</strong> e <strong>traces distribuídos</strong>. 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).</p>
<h2>OpenTelemetry: O Padrão de Observabilidade</h2>
<h3>O Que é OpenTelemetry</h3>
<p>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.</p>
<p>Existem três conceitos fundamentais: um <strong>span</strong> representa uma unidade de trabalho (como uma requisição HTTP), um <strong>trace</strong> é uma árvore de spans conectados mostrando o fluxo completo, e <strong>instrumentação</strong> é o processo de adicionar código que coleta esses dados.</p>
<h3>Instalação e Configuração Básica</h3>
<pre><code class="language-bash">pip install opentelemetry-api opentelemetry-sdk opentelemetry-exporter-jaeger opentelemetry-instrumentation-flask opentelemetry-instrumentation-requests</code></pre>
<p>Aqui está uma aplicação Flask completa com OpenTelemetry:</p>
<pre><code class="language-python">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)</code></pre>
<p>Quando você executa essa aplicação e faz requisições, o OpenTelemetry coleta automaticamente spans para Flask e requests. Para visualizar, abra <code>http://localhost:16686</code> (UI do Jaeger) — você verá toda a árvore de execução com timings.</p>
<h3>Atributos e Eventos em Spans</h3>
<p>Spans são mais poderosos quando você adiciona contexto. Use atributos para metadados persistentes e eventos para marcos de tempo específicos:</p>
<pre><code class="language-python">@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"})</code></pre>
<p>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).</p>
<h2>Sentry: Capturando e Rastreando Erros</h2>
<h3>O Problema que Sentry Resolve</h3>
<p>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.</p>
<h3>Integrando Sentry em uma Aplicação Python</h3>
<pre><code class="language-bash">pip install sentry-sdk</code></pre>
<p>Configuração básica (qualquer aplicação):</p>
<pre><code class="language-python">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)</code></pre>
<p>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.</p>
<h3>Diferenciando Errors de Warnings</h3>
<p>Nem tudo que vai errado é crítico. Use captura manual para informar Sentry sem disparar exceções:</p>
<pre><code class="language-python">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})</code></pre>
<h2>Profiling com py-spy: Encontrando Gargalos</h2>
<h3>Por Que Profiling Importa</h3>
<p>OpenTelemetry te diz <em>quando</em> algo é lento. Sentry te diz <em>que</em> errou. Mas e se uma função leva 5 segundos e ninguém sabe por quê? Profiling analisa <em>onde</em> 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.</p>
<h3>Instalação e Uso Básico</h3>
<pre><code class="language-bash">pip install py-spy</code></pre>
<p>Rodar um profiling de 30 segundos em uma aplicação já em execução:</p>
<pre><code class="language-bash"># 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</code></pre>
<p>Isso gera um arquivo SVG interativo. Áreas maiores = mais tempo de CPU. Clique para ver detalhes.</p>
<h3>Exemplo Prático: Aplicação com Gargalo Intencional</h3>
<pre><code class="language-python">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)</code></pre>
<p>Execute assim:</p>
<pre><code class="language-bash"># 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</code></pre>
<p>Abrindo <code>profile.svg</code> no navegador, você vê que <code>inefficient_algorithm</code> consume 80%+ do CPU. Mudança simples: substituir por <code>optimized_algorithm</code>.</p>
<h3>Análise Estatística com py-spy dump</h3>
<p>Para snapshot instantâneo sem gerar SVG:</p>
<pre><code class="language-bash">py-spy dump --pid 12345</code></pre>
<p>Mostra stack trace de cada thread naquele momento. Útil para descobrir se a app está travada em alguma operação.</p>
<h3>Integrando Profiling em Ambiente de Produção</h3>
<pre><code class="language-python">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"}</code></pre>
<h2>Integrando Tudo Junto: Um Exemplo Completo</h2>
<p>Para solidificar, aqui está uma aplicação que combina os três:</p>
<pre><code class="language-python">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)</code></pre>
<p>Com isso você tem:</p>
<ul>
<li><strong>Jaeger</strong> mostra a árvore completa de execução (traces e spans)</li>
<li><strong>Sentry</strong> alertar sobre exceções em tempo real</li>
<li><strong>py-spy</strong> consegue profiles CPU quando necessário</li>
<li>Tudo correlacionado: um erro em Sentry pode levar a um trace no Jaeger</li>
</ul>
<h2>Conclusão</h2>
<p>Observabilidade em produção não é luxo — é necessidade. Os três pilares aprendidos resolvem problemas diferentes: <strong>OpenTelemetry oferece rastreamento de requisição ponta-a-ponta</strong>, mostrando exatamente onde o tempo é gasto em arquiteturas complexas. <strong>Sentry captura erros antes de se tornarem crises</strong>, diferenciando problemas críticos de warnings e agrupando automaticamente. <strong>py-spy identifica gargalos de performance</strong> sem overhead em produção, usando amostragem.</p>
<p>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.</p>
<h2>Referências</h2>
<ul>
<li><a href="https://opentelemetry.io/docs/instrumentation/python/" target="_blank" rel="noopener noreferrer">OpenTelemetry Python Documentation</a></li>
<li><a href="https://docs.sentry.io/platforms/python/" target="_blank" rel="noopener noreferrer">Sentry Python SDK Guide</a></li>
<li><a href="https://github.com/benfred/py-spy" target="_blank" rel="noopener noreferrer">py-spy GitHub Repository</a></li>
<li><a href="https://www.jaegertracing.io/docs/" target="_blank" rel="noopener noreferrer">Jaeger Tracing Documentation</a></li>
<li><a href="https://landscape.cncf.io/guide" target="_blank" rel="noopener noreferrer">CNCF Observability Landscape</a></li>
</ul>
<p><!-- FIM --></p>