Mastering OpenTelemetry Context Propagation: A Deep Dive into Distributed Trace Continuity across Services

# app.py
from flask import Flask, request
from opentelemetry import trace, propagators
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.instrumentation.flask import FlaskInstrumentor
from opentelemetry.instrumentation.kafka import KafkaInstrumentor

app = Flask(__name__)

# 1️⃣ Set up tracer provider and exporter
resource = Resource(attributes={"service.name": "order-api"})
provider = TracerProvider(resource=resource)
trace.set_tracer_provider(provider)
otlp_exporter = OTLPSpanExporter(endpoint="http://otel-collector:4318/v1/traces")
provider.add_span_processor(BatchSpanProcessor(otlp_exporter))

# 2️⃣ Instrument Flask and Kafka
FlaskInstrumentor().instrument_app(app)
KafkaInstrumentor().instrument()

# 3️⃣ Example endpoint that forwards a message to Kafka
@app.route("/order", methods=["POST"])
def create_order():
    tracer = trace.get_tracer(__name__)
    with tracer.start_as_current_span("create_order_handler") as span:
        # Business logic here...
        # Produce a Kafka message with injected context
        from kafka import KafkaProducer
        producer = KafkaProducer(bootstrap_servers="kafka:9092")
        # Use OpenTelemetry's propagator to inject headers
        headers = {}
        propagators.inject(headers)
        producer.send(
            "orders",
            value=b'{"order_id": 42}',
            headers=[("traceparent", headers["traceparent"].encode())]
        )
        return {"status": "accepted"}, 202

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)

// build.gradle
plugins {
    id 'org.springframework.boot' version '3.2.0'
    id 'io.spring.dependency-management' version '1.1.0'
    id 'java'
}
dependencies {
    implementation 'org.springframework.boot:spring-boot-starter-web'
    implementation 'io.opentelemetry:opentelemetry-api:1.32.0'
    implementation 'io.opentelemetry:opentelemetry-sdk:1.32.0'
    implementation 'io.opentelemetry:opentelemetry-exporter-otlp:1.32.0'
    implementation 'io.opentelemetry.instrumentation:spring-webmvc-3.1:1.32.0'
    implementation 'io.opentelemetry.instrumentation:opentelemetry-spring-boot-starter:1.32.0'
}

// src/main/java/com/example/order/OrderApplication.java
package com.example.order;

import io.opentelemetry.api.GlobalOpenTelemetry;
import io.opentelemetry.api.trace.Tracer;
import io.opentelemetry.context.propagation.TextMapPropagator;
import io.opentemetry.exporter.otlp.trace.OtlpGrpcSpanExporter;
import io.opentelemetry.sdk.OpenTelemetrySdk;
import io.opentelemetry.sdk.resources.Resource;
import io.opentelemetry.sdk.trace.SdkTracerProvider;
import io.opentelemetry.sdk.trace.export.BatchSpanProcessor;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.web.bind.annotation.*;

import java.util.Collections;

@SpringBootApplication
public class OrderApplication {

    public static void main(String[] args) {
        // 1️⃣ Build OpenTelemetry SDK
        Resource serviceName = Resource.create(Collections.singletonMap("service.name", "order-service"));
        SdkTracerProvider tracerProvider = SdkTracerProvider.builder()
                .setResource(serviceName)
                .addSpanProcessor(BatchSpanProcessor.builder(
                        OtlpGrpcSpanExporter.builder()
                                .setEndpoint("http://otel-collector:4317")
                                .build()
                ).build())
                .build();

        OpenTelemetrySdk.builder()
                .setTracerProvider(tracerProvider)
                .buildAndRegisterGlobal();

        SpringApplication.run(OrderApplication.class, args);
    }
}

// src/main/java/com/example/order/OrderController.java
package com.example.order;

import io.opentelemetry.api.trace.Span;
import io.opentelemetry.api.trace.Tracer;
import io.opentelemetry.context.propagation.TextMapSetter;
import io.opentelemetry.context.Context;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;

import java.util.Properties;

@RestController
@RequestMapping("/order")
public class OrderController {

    private static final Tracer tracer = GlobalOpenTelemetry.getTracer("order-service");
    private static final KafkaProducer<String, String> producer = createProducer();

    @PostMapping
    public ResponseEntity<String> create(@RequestBody OrderPayload payload) {
        // 2️⃣ Start a new span for the request handler
        Span span = tracer.spanBuilder("createOrderHandler").startSpan();
        try (var scope = span.makeCurrent()) {
            // Business logic...

            // 3️⃣ Inject context into Kafka headers
            ProducerRecord<String, String> record = new ProducerRecord<>("orders", payload.toJson());
            TextMapSetter<ProducerRecord<String, String>> setter = (carrier, key, value) ->
                    carrier.headers().add(key, value.getBytes());

            GlobalOpenTelemetry.getPropagators()
                    .getTextMapPropagator()
                    .inject(Context.current(), record, setter);

            producer.send(record);
            return ResponseEntity.accepted().body("{\"status\":\"queued\"}");
        } finally {
            span.end();
        }
    }

    private static KafkaProducer<String, String> createProducer() {
        Properties props = new Properties();
        props.put("bootstrap.servers", "kafka:9092");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        return new KafkaProducer<>(props);
    }
}

// main.go
package main

import (
    "context"
    "log"
    "net"

    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp"
    "go.opentelemetry.io/otel/sdk/resource"
    "go.opentelemetry.io/otel/sdk/trace"
    "go.opentelemetry.io/otel/propagation"
    pb "myapp/proto"
    "google.golang.org/grpc"
)

func initTracer() {
    exporter, _ := otlptracehttp.New(context.Background(),
        otlptracehttp.WithEndpoint("otel-collector:4318"),
        otlptracehttp.WithInsecure(),
    )
    tp := trace.NewTracerProvider(
        trace.WithBatcher(exporter),
        trace.WithResource(resource.NewWithAttributes()),
    )
    otel.SetTracerProvider(tp)
    otel.SetTextMapPropagator(propagation.TraceContext{})
}

type server struct{ pb.UnimplementedOrderServiceServer }

func (s *server) CreateOrder(ctx context.Context, req *pb.OrderRequest) (*pb.OrderResponse, error) {
    tracer := otel.Tracer("order-service")
    ctx, span := tracer.Start(ctx, "CreateOrder")
    defer span.End()

    // Business logic...
    // Call downstream service with the same ctx, propagation handled by gRPC interceptor
    return &pb.OrderResponse{Status: "accepted"}, nil
}

func main() {
    initTracer()
    lis, _ := net.Listen("tcp", ":50051")
    grpcServer := grpc.NewServer(
        // OpenTelemetry gRPC interceptors automatically inject/extract context
    )
    pb.RegisterOrderServiceServer(grpcServer, &server{})
    log.Fatal(grpcServer.Serve(lis))
}

# collector-config.yaml
receivers:
  otlp:
    protocols:
      grpc:
      http:

exporters:
  otlp:
    endpoint: "tempo:4317"
  logging:
    loglevel: debug

service:
  pipelines:
    traces:
      receivers: [otlp]
      exporters: [otlp, logging]