Tracing Python API Calls with OpenTelemetry
Learn how to implement OpenTelemetry tracing in your Python API to debug performance issues, with step-by-step setup for Flask, custom spans, error capture, and context propagation.
Tracing Python API Calls with OpenTelemetry: A Practical Guide
If you've ever spent hours trying to figure out why a Python API endpoint is running slowly, you already understand the value of tracing. OpenTelemetry gives you a way to see exactly what happens during each API call - the database queries, external service calls, and everything in between.
Let me show you how to implement this in a way that actually helps you debug your applications.
Why Bother with Tracing?
Here's the thing - logs tell you what happened, metrics tell you how much, but traces tell you the complete journey. When a user calls your API endpoint, tracing shows you every step that took time. Without it, you're essentially debugging blindfolded.
Getting Started with the Setup
First, install what we need:
pip install opentelemetry-api opentelemetry-sdk
pip install opentelemetry-instrumentation-flask
pip install opentelemetry-exporter-otlp
If you're using FastAPI instead, swap the Flask package for opentelemetry-instrumentation-fastapi.
Setting Up Your Tracer Provider
Let's configure OpenTelemetry in your Python app. I'll use Flask for this example, but the pattern works the same across frameworks:
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.flask import FlaskInstrumentor
from flask import Flask
app = Flask(__name__)
# Set up the tracer provider
provider = TracerProvider()
processor = BatchSpanProcessor(OTLPSpanExporter(
endpoint="http://localhost:4317",
insecure=True # Use False in production
))
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)
# Auto-instrument Flask
FlaskInstrumentor().instrument_app(app)
@app.route('/api/users')
def get_users():
# Your API logic here
return {"users": []}
if __name__ == "__main__":
app.run()
Adding Custom Spans for Deeper Visibility
Automatic instrumentation is great, but you'll want custom spans for your business logic. Here's how you add them:
from opentelemetry import trace
tracer = trace.get_tracer(__name__)
@app.route('/api/orders/<order_id>')
def get_order(order_id):
with tracer.start_as_current_span("fetch_order") as span:
span.set_attribute("order.id", order_id)
# Simulate database call
order_data = query_database(order_id)
with tracer.start_as_current_span("process_payment") as payment_span:
payment_span.set_attribute("payment.method", "credit_card")
result = process_payment(order_data)
return {"order": order_data}
Capturing Errors in Traces
When things go wrong, you want traces to show those failures clearly:
@app.route('/api/checkout')
def checkout():
with tracer.start_as_current_span("checkout_process") as span:
try:
cart = get_user_cart()
if not cart:
span.set_status(trace.Status(trace.StatusCode.ERROR))
span.record_exception(Exception("Empty cart"))
return {"error": "Cart is empty"}, 400
process_checkout(cart)
return {"status": "success"}
except Exception as e:
span.set_status(trace.Status(trace.StatusCode.ERROR))
span.record_exception(e)
raise
Adding Context Propagation
For microservices, you need to pass trace context between services. Here's how:
import requests
from opentelemetry.propagate import inject
def call_payment_service(order_data):
headers = {}
inject(headers) # Adds trace context to headers
response = requests.post(
"http://payment-service/process",
json=order_data,
headers=headers
)
return response.json()
The payment service receives this context and continues the trace - everything connects properly.
Where to Send Your Traces
You need a backend to store and visualize these traces. Options include:
- Jaeger: Self-hosted, works great for testing
- Zipkin: Another solid open-source choice
- Datadog: Cloud-hosted with generous free tier
- Grafana Tempo: Integrates with your existing Grafana setup
For local development, Jaeger is easiest:
docker run -d -p 16686:16686 -p 4317:4317 jaegertracing/all-in-one:latest
Then point your exporter to localhost:4317.
Filtering What You Trace
Not every request needs tracing. For high-traffic endpoints, sample strategically:
from opentelemetry.sdk.trace.sampling import Sampler, Decision
class CustomSampler(Sampler):
def should_sample(self, parent_context, trace_id, name, kind, attributes, links):
if "health" in name: # Skip health checks
return Decision.DROP
if "payment" in name: # Always trace payments
return Decision.RECORD_AND_SAMPLE
return Decision.RECORD_AND_SAMPLE # Sample everything else
provider = TracerProvider(sampler=CustomSampler())
Real-World Example: Tracing an E-Commerce Checkout
Let me show you what this looks like in practice. Here's a complete checkout endpoint with tracing:
@app.route('/api/checkout', methods=['POST'])
def process_checkout():
tracer = trace.get_tracer(__name__)
with tracer.start_as_current_span("checkout") as root_span:
root_span.set_attribute("user.id", session.get('user_id'))
# Validate cart
with tracer.start_as_current_span("validate_cart"):
cart = get_cart()
if not cart or len(cart['items']) == 0:
root_span.set_status(trace.Status(trace.StatusCode.ERROR))
return {"error": "Empty cart"}, 400
# Process each item
for item in cart['items']:
with tracer.start_as_current_span("process_item") as item_span:
item_span.set_attribute("item.id", item['id'])
item_span.set_attribute("item.price", item['price'])
# Deduct inventory
deduct_inventory(item['id'], item['quantity'])
# Apply discount if applicable
if item.get('discount'):
apply_discount(item)
# Charge payment
with tracer.start_as_current_span("charge_payment") as payment_span:
payment_span.set_attribute("amount", cart['total'])
payment_result = charge_card(cart['total'])
if not payment_result['success']:
payment_span.set_status(trace.Status(trace.StatusCode.ERROR))
payment_span.record_exception(Exception(payment_result['error']))
return {"error": "Payment failed"}, 500
# Create order
with tracer.start_as_current_span("create_order"):
order_id = save_order(cart, payment_result['transaction_id'])
root_span.set_attribute("order.id", order_id)
return {"order_id": order_id, "status": "success"}
When you look at this trace in Jaeger, you'll see each span with its duration. If the checkout takes 5 seconds, you'll instantly know which step is slow.
Common Pitfalls to Avoid
Forgetting to flush spans - Your last few traces might not export if the app crashes. Add graceful shutdown:
import atexit
def shutdown():
trace.get_tracer_provider().shutdown()
atexit.register(shutdown)
Tracing too much data - Don't add user passwords or credit card numbers as span attributes. Use non-sensitive identifiers instead.
Not setting span status - Failed requests without error status in traces are confusing. Always mark failures properly.
Ignoring sampling - At high traffic levels, tracing everything becomes expensive. Use rate-based sampling for most endpoints.
Wrapping Up
OpenTelemetry tracing transforms how you debug API performance issues. Instead of guessing what's slow, you see the exact problem. Start with automatic instrumentation, add custom spans for your critical paths, and use a backend like Jaeger to visualize everything.
The key is to start small - instrument one endpoint, verify your traces appear in Jaeger, then expand from there. You'll wonder how you ever debugged without it.
Comments
Questions, corrections, and tips stay visible for everyone reading this page.
Join the discussion
No comments yet
Be the first to leave a note — it helps the next reader.