OpenTelemetry Python Tracing API
This document teaches you how to use the OpenTelemetry Python API. To learn how to install and configure the OpenTelemetry Python SDK, see Getting started with OpenTelemetry Python.
Installation
OpenTelemetry-Python is the Python implementation of OpenTelemetry. It provides the OpenTelemetry Python API, which you can use to instrument your application with OpenTelemetry traces.
Install the required packages:
pip install opentelemetry-api opentelemetry-sdk
Quickstart
Step 1. Let's instrument the following function:
def insert_user(**kwargs):
return User.objects.create(**kwargs)
Step 2. Wrap the operation with a span:
from opentelemetry import trace
tracer = trace.get_tracer("app_or_package_name", "1.0.0")
def insert_user(**kwargs):
with tracer.start_as_current_span("insert-user") as span:
return User.objects.create(**kwargs)
Step 3. Record exceptions and set status code:
from opentelemetry.trace import Status, StatusCode
def insert_user(**kwargs):
with tracer.start_as_current_span("insert-user") as span:
try:
return User.objects.create(**kwargs)
except Exception as exc:
span.record_exception(exc)
span.set_status(Status(StatusCode.ERROR, str(exc)))
raise # Re-throw the exception
Step 4. Record contextual information with attributes:
def insert_user(**kwargs):
with tracer.start_as_current_span("insert-user") as span:
# Check if span is being recorded to avoid unnecessary work
if span.is_recording():
span.set_attribute("user.id", kwargs.get("id"))
span.set_attribute("user.email", kwargs.get("email"))
span.set_attribute("user.type", kwargs.get("user_type", "regular"))
try:
result = User.objects.create(**kwargs)
span.set_attribute("operation.result", "success")
span.set_attribute("user.created_id", result.id)
return result
except Exception as exc:
span.record_exception(exc)
span.set_status(Status(StatusCode.ERROR, str(exc)))
raise
That's it! The operation is now fully instrumented with proper error handling and contextual information.
Tracer
To start creating spans, you need a tracer. You can create a tracer by providing the name and version of the library or application doing the instrumentation:
from opentelemetry import trace
tracer = trace.get_tracer("app_or_package_name", "1.0.0")
You can have as many tracers as you want, but usually you need only one tracer per application or library. Later, you can use tracer names to identify the instrumentation that produces the spans.
Tracer Configuration
When creating a tracer, you can provide additional metadata:
tracer = trace.get_tracer(
"app_or_package_name", # Instrumentation name
"1.0.0", # Instrumentation version
"https://example.com/schema", # Schema URL (optional)
)
Creating Spans
Once you have a tracer, creating spans is straightforward:
from opentelemetry import trace
from opentelemetry.trace import SpanKind, Status, StatusCode
# Create a span with name "operation-name" and kind="server"
with tracer.start_as_current_span(
"operation-name",
kind=SpanKind.SERVER
) as span:
try:
do_some_work()
except Exception as exc:
span.record_exception(exc)
span.set_status(Status(StatusCode.ERROR, str(exc)))
raise
Span Kinds
Specify the type of span using span kinds:
from opentelemetry.trace import SpanKind
# For incoming requests (server-side)
with tracer.start_as_current_span(
"handle-request",
kind=SpanKind.SERVER
) as span:
pass
# For outgoing requests (client-side)
with tracer.start_as_current_span(
"http-request",
kind=SpanKind.CLIENT
) as span:
pass
# For async operations (producer/consumer)
with tracer.start_as_current_span(
"publish-message",
kind=SpanKind.PRODUCER
) as span:
pass
# For internal operations (default)
with tracer.start_as_current_span(
"internal-operation",
kind=SpanKind.INTERNAL
) as span:
pass
Span Configuration Options
Configure spans with additional options:
from opentelemetry.trace import SpanKind
from opentelemetry import context
# Create span with custom configuration
with tracer.start_as_current_span(
"complex-operation",
kind=SpanKind.INTERNAL,
context=parent_context, # Set explicit parent
attributes={"operation.type": "batch"}, # Set initial attributes
links=[span_link], # Link to other spans
start_time=start_timestamp, # Custom start time
record_exception=True, # Auto-record exceptions
set_status_on_exception=True, # Auto-set error status
) as span:
pass
Adding Span Attributes
To record contextual information, you can annotate spans with attributes. For example, an HTTP endpoint may have attributes such as http.method = GET
and http.route = /projects/:id
.
# Check if span is being recorded to avoid expensive computations
if span.is_recording():
span.set_attribute("http.method", "GET")
span.set_attribute("http.route", "/projects/:id")
span.set_attribute("http.status_code", 200)
span.set_attribute("user.id", user_id)
Setting Multiple Attributes
You can set multiple attributes at once:
span.set_attributes({
"http.method": "POST",
"http.route": "/api/users",
"http.status_code": 201,
"user.role": "admin",
"request.size": len(request_body)
})
Semantic Conventions
Use semantic conventions for consistent attribute naming:
from opentelemetry.semconv.trace import SpanAttributes
span.set_attributes({
SpanAttributes.HTTP_METHOD: "GET",
SpanAttributes.HTTP_ROUTE: "/projects/:id",
SpanAttributes.HTTP_STATUS_CODE: 200,
SpanAttributes.USER_ID: user_id
})
Adding Span Events
You can annotate spans with events that represent significant moments during the span's lifetime:
import time
# Simple event
span.add_event("cache.miss")
# Event with attributes
span.add_event("user.login", {
"user.id": "12345",
"user.role": "admin",
"login.method": "oauth"
})
# Event with timestamp
span.add_event(
"database.query.start",
{
"db.statement": "SELECT * FROM users WHERE id = ?",
"db.operation": "SELECT"
},
timestamp=time.time_ns()
)
Logging with Events
Events are particularly useful for structured logging:
span.add_event("log", {
"log.severity": "error",
"log.message": "User not found",
"user.id": "123",
"error.code": "USER_NOT_FOUND"
})
Setting Span Status
Use status codes to indicate the outcome of operations:
from opentelemetry.trace import Status, StatusCode
# Successful operation (default)
span.set_status(Status(StatusCode.OK))
# Operation with error
span.set_status(Status(StatusCode.ERROR, "Database connection failed"))
# Unset status (let the system decide)
span.set_status(Status(StatusCode.UNSET))
Status with Exception Handling
try:
perform_database_operation()
span.set_status(Status(StatusCode.OK))
except Exception as exc:
span.set_status(Status(StatusCode.ERROR, str(exc)))
span.record_exception(exc)
raise
Recording Exceptions
OpenTelemetry provides a convenient method to record exceptions:
try:
risky_operation()
except Exception as exc:
span.record_exception(exc)
span.set_status(Status(StatusCode.ERROR, str(exc)))
raise # Re-throw if needed
Exception with Additional Context
try:
connect_to_database(host, port)
except ConnectionError as exc:
span.record_exception(exc, attributes={
"db.host": host,
"db.port": port,
"retry.count": retry_count
})
span.set_status(Status(StatusCode.ERROR, "Database connection failed"))
raise
Current Span and Context
OpenTelemetry stores the active span in a context and saves the context in thread-local storage. You can nest contexts inside each other, and OpenTelemetry will automatically activate the parent span context when you end the span.
Getting the Current Span
from opentelemetry import trace
# Get the currently active span
current_span = trace.get_current_span()
if current_span.is_recording():
current_span.set_attribute("additional.info", "added from nested function")
Manual Context Management
from opentelemetry import trace, context
# Create span without auto-activation
span = tracer.start_span("manual-span")
# Manually activate the span
token = context.attach(context.set_value("current_span", span))
try:
# Span is now active
do_work()
finally:
# Always detach to restore previous context
context.detach(token)
span.end()
Context Nesting
Contexts can be nested to create parent-child relationships:
# Parent span
with tracer.start_as_current_span("parent-operation") as parent_span:
# Child span - automatically becomes a child of the parent
with tracer.start_as_current_span("child-operation") as child_span:
# Perform child work
do_child_work()
# Both spans are properly nested
assert child_span.parent == parent_span.get_span_context()
Advanced Span Features
Span Links
Link spans to show relationships between operations:
from opentelemetry.trace import Link
# Create a span with links to other operations
with tracer.start_span(
"batch-process",
links=[
Link(related_span_context, {"link.type": "follows"}),
Link(another_span_context, {"link.type": "caused_by"})
]
) as span:
pass
Custom Start and End Times
Set explicit timestamps for spans:
import time
start_time = time.time_ns()
# Start span with custom timestamp
span = tracer.start_span("timed-operation", start_time=start_time)
# Perform operation
do_work()
# End with custom timestamp
end_time = time.time_ns()
span.end(end_time)
Span Name Updates
Update span names based on operation results:
with tracer.start_as_current_span("dynamic-operation") as span:
try:
result = perform_operation()
span.update_name(f"successful-operation-{result.type}")
except Exception as e:
span.update_name("failed-operation")
raise
Best Practices
Error Handling Pattern
Use proper exception handling patterns:
with tracer.start_as_current_span("operation") as span:
try:
result = perform_operation()
span.set_status(Status(StatusCode.OK))
span.set_attribute("operation.result", "success")
return result
except Exception as e:
span.record_exception(e)
span.set_status(Status(StatusCode.ERROR, str(e)))
span.set_attribute("operation.result", "failure")
raise
Context Manager Helper
Create a helper for automatic span management:
from contextlib import contextmanager
from opentelemetry.trace import SpanKind, Status, StatusCode
@contextmanager
def traced_operation(tracer, name, **attributes):
"""Context manager for traced operations with automatic error handling."""
with tracer.start_as_current_span(name) as span:
if span.is_recording():
span.set_attributes(attributes)
try:
yield span
span.set_status(Status(StatusCode.OK))
except Exception as e:
span.record_exception(e)
span.set_status(Status(StatusCode.ERROR, str(e)))
raise
# Usage
with traced_operation(tracer, "database-query", db_system="postgresql") as span:
span.set_attribute("db.statement", "SELECT * FROM users")
result = execute_query("SELECT * FROM users")
Attribute Optimization
Only set attributes when the span is being recorded:
if span.is_recording():
expensive_attribute = calculate_expensive_value()
span.set_attribute("expensive.attribute", expensive_attribute)
Span Naming
Use descriptive, low-cardinality span names:
# Good: Low cardinality
with tracer.start_as_current_span("GET /users/{id}") as span:
pass
# Bad: High cardinality
with tracer.start_as_current_span(f"GET /users/{user_id}") as span:
pass
Integration Examples
HTTP Client Instrumentation
import requests
from opentelemetry.trace import SpanKind, Status, StatusCode
from opentelemetry.semconv.trace import SpanAttributes
def make_http_request(url, method="GET", **kwargs):
"""Make an HTTP request with automatic tracing."""
with tracer.start_as_current_span(
f"HTTP {method}",
kind=SpanKind.CLIENT
) as span:
if span.is_recording():
span.set_attributes({
SpanAttributes.HTTP_METHOD: method,
SpanAttributes.HTTP_URL: url,
SpanAttributes.HTTP_USER_AGENT: "MyApp/1.0"
})
try:
response = requests.request(method, url, **kwargs)
if span.is_recording():
span.set_attribute(SpanAttributes.HTTP_STATUS_CODE, response.status_code)
span.set_attribute("http.response.size", len(response.content))
# Set status based on HTTP status code
if response.status_code >= 400:
span.set_status(Status(StatusCode.ERROR, f"HTTP {response.status_code}"))
else:
span.set_status(Status(StatusCode.OK))
return response
except Exception as e:
span.record_exception(e)
span.set_status(Status(StatusCode.ERROR, "HTTP request failed"))
raise
Database Query Instrumentation
import sqlite3
from opentelemetry.trace import SpanKind, Status, StatusCode
from opentelemetry.semconv.trace import SpanAttributes
def execute_query(query, params=None):
"""Execute database query with automatic tracing."""
with tracer.start_as_current_span(
"db.query",
kind=SpanKind.CLIENT
) as span:
if span.is_recording():
span.set_attributes({
SpanAttributes.DB_SYSTEM: "sqlite",
SpanAttributes.DB_STATEMENT: query,
SpanAttributes.DB_NAME: "app.db"
})
try:
start_time = time.time()
# Execute query
conn = sqlite3.connect("app.db")
cursor = conn.cursor()
if params:
cursor.execute(query, params)
else:
cursor.execute(query)
result = cursor.fetchall()
conn.close()
duration = time.time() - start_time
if span.is_recording():
span.set_attributes({
"db.rows_affected": len(result),
"db.duration": duration
})
span.set_status(Status(StatusCode.OK))
return result
except Exception as e:
span.record_exception(e)
span.set_status(Status(StatusCode.ERROR, "Database query failed"))
raise
Async Function Instrumentation
import asyncio
from opentelemetry.trace import SpanKind, Status, StatusCode
async def async_operation():
"""Example async operation with tracing."""
with tracer.start_as_current_span(
"async-operation",
kind=SpanKind.INTERNAL
) as span:
if span.is_recording():
span.set_attribute("operation.type", "async")
try:
# Simulate async work
await asyncio.sleep(0.1)
# Add event
span.add_event("async.work.completed")
span.set_status(Status(StatusCode.OK))
return "success"
except Exception as e:
span.record_exception(e)
span.set_status(Status(StatusCode.ERROR, str(e)))
raise
# Usage
async def main():
result = await async_operation()
print(result)
# Run with asyncio
asyncio.run(main())
Performance Considerations
Conditional Instrumentation
For high-performance scenarios, consider conditional instrumentation:
def conditional_instrumentation(operation, enable_tracing=True):
"""Conditionally instrument operations based on configuration."""
if not enable_tracing:
return operation()
with tracer.start_as_current_span("conditional-operation") as span:
return operation()
Sampling Awareness
Check if spans are being sampled to avoid unnecessary work:
with tracer.start_as_current_span("operation") as span:
# Only do expensive operations if the span is being recorded
if span.is_recording():
span.set_attribute("expensive.computation", compute_expensive_value())
Batch Attribute Setting
Set multiple attributes efficiently:
# Efficient: Set all attributes at once
attributes = {
"http.method": "GET",
"http.route": "/api/users",
"http.status_code": 200,
"user.id": user_id,
"user.role": user_role
}
if span.is_recording():
span.set_attributes(attributes)
Auto-instrumentation
When possible, you should prefer using explicit instrumentation, for example, auto-instrumentation does not work well with Flask in debug mode.
OpenTelemetry Python allows you to automatically instrument any Python app using opentelemetry-instrument
utility from opentelemetry-instrumentation
package.
First, you need to install the opentelemetry-instrument
executable:
pip install opentelemetry-instrumentation
Then, install instrumentations that should be applied automatically by opentelemetry-instrument
:
pip install opentelemetry-instrumentation-flask
pip install opentelemetry-instrumentation-requests
pip install opentelemetry-instrumentation-urllib3
pip install opentelemetry-instrumentation-psycopg2
And run your app using opentelemetry-instrument
wrapper:
UPTRACE_DSN="your_dsn_here" opentelemetry-instrument python your_app.py
Supported Libraries
Auto-instrumentation is available for:
- HTTP clients: requests, urllib3, httpx, aiohttp
- Web frameworks: Flask, Django, FastAPI, Pyramid, Starlette
- Databases: psycopg2, SQLAlchemy, pymongo, redis
- Message queues: celery, kombu
- Other: boto3, elasticsearch, grpc
Enabling Auto-Instrumentation
Set the following environment variables:
export OTEL_PYTHON_LOGGING_AUTO_INSTRUMENTATION_ENABLED=true
export OTEL_TRACES_EXPORTER=otlp
export OTEL_METRICS_EXPORTER=otlp
export OTEL_LOGS_EXPORTER=otlp
Or configure it programmatically:
from opentelemetry.instrumentation.auto_instrumentation import sitecustomize
# Enable all available instrumentations
sitecustomize.instrument()
# Or enable specific instrumentations
from opentelemetry.instrumentation.requests import RequestsInstrumentor
from opentelemetry.instrumentation.urllib3 import URLLib3Instrumentor
from opentelemetry.instrumentation.flask import FlaskInstrumentor
RequestsInstrumentor().instrument()
URLLib3Instrumentor().instrument()
FlaskInstrumentor().instrument()
How auto-instrumentation works?
uptrace-python registers an OpenTelemetry distro using an entry point in setup.cfg. Instrumentations register themselves using the same mechanism. OpenTelemetry then just calls the code specified in entry points to configure OpenTelemetry SDK and instrument the code.
OpenTelemetry APM
Uptrace is a OpenTelemetry APM that supports distributed tracing, metrics, and logs. You can use it to monitor applications and troubleshoot issues.
Uptrace comes with an intuitive query builder, rich dashboards, alerting rules with notifications, and integrations for most languages and frameworks.
Uptrace can process billions of spans and metrics on a single server and allows you to monitor your applications at 10x lower cost.
In just a few minutes, you can try Uptrace by visiting the cloud demo (no login required) or running it locally with Docker. The source code is available on GitHub.
What's Next?
Now that you understand the OpenTelemetry Python Tracing API, explore these related topics: