03 - Distributed Tracing - Understanding Request Flows Across Services

Trace ID: 4bf92f3577b34da6a3ce929d0e0e4736

[Root Span] API Gateway - POST /api/orders (250ms)
  |
  +-- [Child Span] Order Service - validate-order (15ms)
  |
  +-- [Child Span] Order Service - check-inventory (45ms)
  |     |
  |     +-- [Child Span] Inventory Service - GET /stock (40ms)
  |           |
  |           +-- [Child Span] Database - SELECT stock (8ms)
  |
  +-- [Child Span] Order Service - process-payment (120ms)
  |     |
  |     +-- [Child Span] Payment Service - charge (115ms)
  |           |
  |           +-- [Child Span] External API - Stripe charge (100ms)
  |
  +-- [Child Span] Order Service - send-confirmation (30ms)
        |
        +-- [Child Span] Notification Service - send-email (25ms)
      

from opentelemetry import trace
from opentelemetry.trace import StatusCode, SpanKind

tracer = trace.get_tracer("order-service", "1.0.0")

def process_payment(order_id, amount, currency):
    # Create a span with all components
    with tracer.start_as_current_span(
        name="process-payment",
        kind=SpanKind.CLIENT,  # Span type
        attributes={          # Initial attributes
            "order.id": order_id,
            "payment.amount": amount,
            "payment.currency": currency,
            "payment.provider": "stripe"
        }
    ) as span:
        try:
            # Add an event (automatic timestamp)
            span.add_event("payment.validation.started", {
                "validation.rules": "amount,currency,card"
            })

            validate_payment(amount, currency)

            span.add_event("payment.validation.completed")

            # Call to payment provider
            result = stripe_client.charge(amount, currency)

            # Add attributes after execution
            span.set_attribute("payment.transaction_id", result.tx_id)
            span.set_attribute("payment.status", "success")

            # Set status
            span.set_status(StatusCode.OK)

            return result

        except PaymentDeclinedException as e:
            # Record exception as event
            span.record_exception(e)
            span.set_status(StatusCode.ERROR, "Payment declined")
            span.set_attribute("payment.decline_reason", str(e))
            raise

        except TimeoutError as e:
            span.record_exception(e)
            span.set_status(StatusCode.ERROR, "Payment gateway timeout")
            span.add_event("payment.retry.scheduled", {
                "retry.attempt": 1,
                "retry.delay_ms": 1000
            })
            raise
      

# W3C Trace Context Headers

# traceparent: contains trace_id, parent_span_id, trace_flags
traceparent: 00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01
             |  |                                |                |
             v  v                                v                v
          version  trace_id (32 hex)     parent_span_id (16 hex)  flags (sampled=01)

# tracestate: vendor-specific state (optional)
tracestate: congo=t61rcWkgMzE,rojo=00f067aa0ba902b7
      

from opentelemetry.trace import SpanKind

# SERVER: service receives a request from a remote client
with tracer.start_as_current_span("handle-request", kind=SpanKind.SERVER):
    # Handle incoming HTTP request
    pass

# CLIENT: service sends a request to a remote service
with tracer.start_as_current_span("call-payment-api", kind=SpanKind.CLIENT):
    # HTTP call to payment-service
    pass

# PRODUCER: service sends a message to a broker
with tracer.start_as_current_span("publish-order-event", kind=SpanKind.PRODUCER):
    # Publish to Kafka topic
    pass

# CONSUMER: service receives a message from a broker
with tracer.start_as_current_span("process-order-event", kind=SpanKind.CONSUMER):
    # Consume from Kafka topic
    pass

# INTERNAL: internal service operation (default)
with tracer.start_as_current_span("validate-order", kind=SpanKind.INTERNAL):
    # Internal validation logic
    pass
      

from opentelemetry import trace, context

# Scenario: batch processing messages from a queue
# Each message has its own original trace context

def process_batch(messages):
    # Create links to all messages in the batch
    links = []
    for msg in messages:
        # Extract trace context from message
        msg_context = extract_context(msg.headers)
        span_context = trace.get_current_span(msg_context).get_span_context()
        links.append(trace.Link(
            context=span_context,
            attributes={"messaging.message.id": msg.id}
        ))

    # Create a span with links to all original messages
    with tracer.start_as_current_span(
        name="process-message-batch",
        kind=SpanKind.CONSUMER,
        links=links,
        attributes={
            "messaging.batch.message_count": len(messages)
        }
    ) as span:
        for msg in messages:
            process_single_message(msg)