Python manual instrumentation

Auto-instrumentation is great for HTTP requests, database calls, and common libraries. Manual instrumentation is how you capture your important business steps (for example: orders.handle, inventory.reserve, shipment.dispatched, pricing.recalculate-discount) with the attributes you care about.

Prerequisites

• Complete one of these setups first:
  • OpenTelemetry auto-instrumentation: Python (OpenTelemetry)
  • Middleware SDK: Python guide
• Install the OpenTelemetry API/SDK (if you don't already have them through another package):

pip install opentelemetry-api opentelemetry-sdk

1. Create custom spans

Use a tracer and wrap meaningful work inside spans. Child spans created inside a parent span automatically connect to the same trace.

from opentelemetry import trace

tracer = trace.get_tracer("orders-service")


def handle_order(order_id: str, customer_id: str) -> None:
    with tracer.start_as_current_span("orders.handle") as span:
        span.set_attribute("order.id", order_id)
        span.set_attribute("customer.id", customer_id)
        span.set_attribute("order.source", "web")

        validate_order(order_id)
        reserve_items(order_id)

from opentelemetry import trace

tracer = trace.get_tracer("orders-service")


def reserve_items(order_id: str) -> None:
    with tracer.start_as_current_span("inventory.reserve") as span:
        span.set_attribute("order.id", order_id)
        span.set_attribute("inventory.region", "eu-central-1")
        # ... call database / external services here ...

from opentelemetry import trace

tracer = trace.get_tracer("pricing")


@tracer.start_as_current_span("pricing.recalculate-discount")
def recalculate_discount(customer_tier: str, items_total: float) -> float:
    # ... custom discount logic ...
    return items_total * 0.9 if customer_tier == "gold" else items_total

Tips:

• Reuse tracer instances (create once per module), not per request.
• Pick span names that match business steps and stay stable over time.
• Keep span attributes low-cardinality (avoid IDs that explode the number of unique values unless needed for debugging).

2. Add attributes and events

Attributes become filter/group dimensions. Events capture notable moments inside a span (retries, cache misses, queue waits).

from opentelemetry import trace


def log_shipment(order_id: str, carrier: str, region: str) -> None:
    span = trace.get_current_span()

    span.set_attribute("shipment.order_id", order_id)
    span.set_attribute("shipment.carrier", carrier)
    span.set_attribute("shipment.region", region)

    span.add_event("shipment.label_created")

    # ... call shipping provider ...

    span.add_event(
        "shipment.dispatched",
        attributes={"status": "in_transit", "tracking.enabled": True},
    )

If you want to follow OpenTelemetry naming conventions, use semantic attribute keys where possible (for example, http.request.method, url.full, db.system).

3. Record errors

When something fails, record the exception and set the span status to ERROR. This makes failures easier to find in trace search and alerts.

from opentelemetry import trace
from opentelemetry.trace import Status, StatusCode


def fetch_user_profile(user_id: str) -> dict:
    span = trace.get_current_span()

    try:
        profile = load_profile_from_api(user_id)
        span.set_status(Status(StatusCode.OK))
        return profile
    except TimeoutError as exc:
        span.record_exception(exc)
        span.set_status(Status(StatusCode.ERROR, "profile service timeout"))
        raise
    except Exception as exc:
        span.record_exception(exc)
        span.set_status(Status(StatusCode.ERROR, "unexpected profile failure"))
        raise

4. Add span links (optional)

Span links are useful when work is related but not parent/child (for example, message queues where a consumer links back to a producer span context).

from opentelemetry import trace

tracer = trace.get_tracer("analytics")


def ingest_event(raw_event: dict) -> trace.SpanContext:
    with tracer.start_as_current_span("pipeline.ingest") as span:
        span.set_attribute("event.type", raw_event.get("type", "unknown"))
        return span.get_span_context()


def enrich_event(parent_ctx: trace.SpanContext, raw_event: dict) -> None:
    link = trace.Link(parent_ctx)
    with tracer.start_as_current_span("pipeline.enrich", links=[link]):
        # ... enrich and forward event ...
        pass

Validate

1. Trigger the code paths that create manual spans.
2. In Middleware, open APM > Traces and search by:
   • service.name, and/or
   • your custom span names (for example, orders.handle).
3. Open a trace and verify attributes, events, and error status are present.

Troubleshooting

Next steps

• If you're using OpenTelemetry auto-instrumentation, keep this handy: Python (OpenTelemetry)
• For Middleware SDK features (profiling, Middleware options, Host Agent), see: Python guide