OpenTelemetry in Practice: Instrumentation, the Collector, and Connecting to Your Observability Stack

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

# docker-compose.yml — full observability stack for development
services:
  # OpenTelemetry Collector
  otel-collector:
    image: otel/opentelemetry-collector-contrib:latest
    volumes:
      - ./otel-collector.yaml:/etc/otelcol-contrib/config.yaml
    ports:
      - "4317:4317"   # OTLP gRPC
      - "4318:4318"   # OTLP HTTP
      - "8888:8888"   # Collector metrics (self-monitoring)
      - "8889:8889"   # Prometheus exporter (scrape endpoint)
      - "13133:13133" # Health check
    depends_on:
      - jaeger
      - prometheus

  # Jaeger — trace backend
  jaeger:
    image: jaegertracing/all-in-one:latest
    environment:
      - COLLECTOR_OTLP_ENABLED=true
    ports:
      - "16686:16686"  # Jaeger UI
      - "4317"         # OTLP gRPC (internal)

  # Prometheus — metrics backend
  prometheus:
    image: prom/prometheus:latest
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml
    ports:
      - "9090:9090"

  # Grafana — visualization
  grafana:
    image: grafana/grafana:latest
    environment:
      - GF_AUTH_ANONYMOUS_ENABLED=true
      - GF_AUTH_ANONYMOUS_ORG_ROLE=Admin
    volumes:
      - ./grafana/datasources.yml:/etc/grafana/provisioning/datasources/datasources.yml
      - ./grafana/dashboards:/etc/grafana/provisioning/dashboards
    ports:
      - "3000:3000"

  # Loki — log backend
  loki:
    image: grafana/loki:latest
    ports:
      - "3100:3100"
    command: -config.file=/etc/loki/local-config.yaml

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131

# otel-collector.yaml
receivers:
  # Accept OTLP from applications
  otlp:
    protocols:
      grpc:
        endpoint: 0.0.0.0:4317
      http:
        endpoint: 0.0.0.0:4318

  # Scrape Prometheus metrics from services that expose /metrics
  prometheus:
    config:
      scrape_configs:
        - job_name: 'otel-collector'
          scrape_interval: 10s
          static_configs:
            - targets: ['0.0.0.0:8888']

processors:
  # Add resource attributes to all telemetry
  resource:
    attributes:
      - action: insert
        key: deployment.environment
        value: development

  # Batch telemetry to reduce export overhead
  batch:
    timeout: 1s
    send_batch_size: 1024
    send_batch_max_size: 2048

  # Memory limiter — prevent OOM if traffic spikes
  memory_limiter:
    check_interval: 1s
    limit_mib: 512
    spike_limit_mib: 128

  # Filter out noisy health check spans
  filter/drop_health_checks:
    error_mode: ignore
    traces:
      span:
        - 'attributes["http.target"] == "/health"'
        - 'attributes["http.target"] == "/metrics"'
        - 'attributes["http.route"] == "/health"'

  # Add k8s metadata (when running in Kubernetes)
  k8sattributes:
    auth_type: "serviceAccount"
    passthrough: false
    extract:
      metadata:
        - k8s.namespace.name
        - k8s.deployment.name
        - k8s.pod.name
        - k8s.node.name
      labels:
        - tag_name: app.version
          key: app.kubernetes.io/version
          from: pod

  # Tail-based sampling — only keep interesting traces
  tail_sampling:
    decision_wait: 10s
    num_traces: 50000
    expected_new_traces_per_sec: 100
    policies:
      # Always keep errors
      - name: errors
        type: status_code
        status_code: { status_codes: [ERROR] }
      # Always keep slow traces (>1s)
      - name: slow-traces
        type: latency
        latency: { threshold_ms: 1000 }
      # Sample 5% of everything else
      - name: probabilistic
        type: probabilistic
        probabilistic: { sampling_percentage: 5 }

exporters:
  # Send traces to Jaeger
  otlp/jaeger:
    endpoint: jaeger:4317
    tls:
      insecure: true

  # Send metrics to Prometheus (via scrape endpoint)
  prometheus:
    endpoint: "0.0.0.0:8889"
    namespace: otel
    send_timestamps: true
    metric_expiration: 180m

  # Send logs to Loki
  loki:
    endpoint: http://loki:3100/loki/api/v1/push
    labels:
      resource:
        - service.name
        - service.namespace

  # Debug exporter — logs telemetry to stdout (development only)
  debug:
    verbosity: detailed
    sampling_initial: 5
    sampling_thereafter: 200

extensions:
  health_check:
    endpoint: 0.0.0.0:13133
  pprof:
    endpoint: 0.0.0.0:1777

service:
  extensions: [health_check, pprof]
  pipelines:
    traces:
      receivers: [otlp]
      processors: [memory_limiter, filter/drop_health_checks, resource, batch, tail_sampling]
      exporters: [otlp/jaeger]
    metrics:
      receivers: [otlp, prometheus]
      processors: [memory_limiter, resource, batch]
      exporters: [prometheus]
    logs:
      receivers: [otlp]
      processors: [memory_limiter, resource, batch]
      exporters: [loki]

1
2

pip install opentelemetry-distro opentelemetry-exporter-otlp
opentelemetry-bootstrap -a install  # Installs framework-specific instrumentation

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

# app.py — FastAPI application
from fastapi import FastAPI, HTTPException
import httpx
import asyncio

app = FastAPI()

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    async with httpx.AsyncClient() as client:
        resp = await client.get(f"http://user-service/api/users/{user_id}")
        if resp.status_code == 404:
            raise HTTPException(status_code=404, detail="User not found")
        return resp.json()

1
2
3
4
5
6
7

# Run with auto-instrumentation — no code changes needed
OTEL_SERVICE_NAME=api-gateway \
OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4317 \
OTEL_RESOURCE_ATTRIBUTES="deployment.environment=production,service.version=1.2.3" \
OTEL_TRACES_SAMPLER=parentbased_traceidratio \
OTEL_TRACES_SAMPLER_ARG=0.1 \
opentelemetry-instrument uvicorn app:app --host 0.0.0.0 --port 8080

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120

from opentelemetry import trace, metrics, baggage
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter
from opentelemetry.sdk.resources import Resource, SERVICE_NAME, SERVICE_VERSION
from opentelemetry.trace import Status, StatusCode
import time
import logging

# --- Setup ---
resource = Resource(attributes={
    SERVICE_NAME: "order-service",
    SERVICE_VERSION: "2.1.0",
    "deployment.environment": "production",
    "host.name": "web-01",
})

# Trace provider
tracer_provider = TracerProvider(resource=resource)
tracer_provider.add_span_processor(
    BatchSpanProcessor(OTLPSpanExporter(endpoint="http://localhost:4317", insecure=True))
)
trace.set_tracer_provider(tracer_provider)
tracer = trace.get_tracer(__name__)

# Metric provider
metric_reader = PeriodicExportingMetricReader(
    OTLPMetricExporter(endpoint="http://localhost:4317", insecure=True),
    export_interval_millis=15000,  # Export every 15 seconds
)
meter_provider = MeterProvider(resource=resource, metric_readers=[metric_reader])
metrics.set_meter_provider(meter_provider)
meter = metrics.get_meter(__name__)

# --- Define metrics instruments ---
order_counter = meter.create_counter(
    "orders.created.total",
    unit="1",
    description="Total number of orders created",
)

order_value = meter.create_histogram(
    "orders.value.dollars",
    unit="USD",
    description="Distribution of order values in dollars",
)

active_orders = meter.create_up_down_counter(
    "orders.active",
    unit="1",
    description="Number of currently active orders",
)

db_query_duration = meter.create_histogram(
    "db.query.duration",
    unit="ms",
    description="Duration of database queries",
)


# --- Application code ---
async def create_order(user_id: str, items: list, db) -> dict:
    with tracer.start_as_current_span("order.create") as span:
        # Add semantic attributes
        span.set_attribute("user.id", user_id)
        span.set_attribute("order.item_count", len(items))

        # Validate inventory
        with tracer.start_as_current_span("inventory.check") as inv_span:
            for item in items:
                inv_span.add_event("checking_item", {"item.sku": item["sku"]})
                available = await check_inventory(item["sku"], item["qty"])
                if not available:
                    inv_span.set_status(Status(StatusCode.ERROR, "Item out of stock"))
                    raise ValueError(f"Item {item['sku']} is out of stock")

        # Calculate total
        total = sum(item["price"] * item["qty"] for item in items)
        span.set_attribute("order.total_usd", total)

        # Write to database
        order_id = await write_order_to_db(user_id, items, total, db)
        span.set_attribute("order.id", order_id)

        # Record metrics
        order_counter.add(1, {"payment.method": "card", "region": "us-east"})
        order_value.record(total, {"product.category": items[0].get("category", "other")})
        active_orders.add(1)

        span.add_event("order_created", {"order.id": order_id})
        return {"order_id": order_id, "total": total}


async def write_order_to_db(user_id, items, total, db):
    start = time.monotonic()
    with tracer.start_as_current_span("db.insert") as span:
        span.set_attribute("db.system", "postgresql")
        span.set_attribute("db.name", "orders")
        span.set_attribute("db.operation", "INSERT")
        span.set_attribute("db.sql.table", "orders")
        try:
            order_id = await db.execute(
                "INSERT INTO orders (user_id, items, total) VALUES ($1, $2, $3) RETURNING id",
                user_id, items, total
            )
            return order_id
        except Exception as e:
            span.record_exception(e)
            span.set_status(Status(StatusCode.ERROR, str(e)))
            raise
        finally:
            duration_ms = (time.monotonic() - start) * 1000
            db_query_duration.record(duration_ms, {
                "db.operation": "INSERT",
                "db.table": "orders",
                "db.status": "ok" if not span.status.is_ok else "error",
            })

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

import logging
import json
from opentelemetry import trace

class OTelJsonFormatter(logging.Formatter):
    """Inject trace context into log records for Loki correlation."""
    def format(self, record):
        span = trace.get_current_span()
        ctx = span.get_span_context()

        log_entry = {
            "timestamp": self.formatTime(record),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
        }

        if ctx.is_valid:
            log_entry["trace_id"] = format(ctx.trace_id, "032x")
            log_entry["span_id"] = format(ctx.span_id, "016x")
            log_entry["trace_flags"] = ctx.trace_flags

        if record.exc_info:
            log_entry["exception"] = self.formatException(record.exc_info)

        return json.dumps(log_entry)

# Configure logging
handler = logging.StreamHandler()
handler.setFormatter(OTelJsonFormatter())
logging.basicConfig(level=logging.INFO, handlers=[handler])
logger = logging.getLogger(__name__)

# Now logs automatically include trace_id and span_id
logger.info("Processing order", extra={"order_id": "ord_123"})
# Output: {"timestamp": "...", "level": "INFO", "message": "Processing order",
#           "trace_id": "4bf92f3577b34da6a3ce929d0e0e4736", "span_id": "00f067aa0ba902b7"}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97

// telemetry/setup.go
package telemetry

import (
    "context"
    "fmt"
    "time"

    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
    "go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc"
    "go.opentelemetry.io/otel/metric"
    "go.opentelemetry.io/otel/propagation"
    sdkmetric "go.opentelemetry.io/otel/sdk/metric"
    "go.opentelemetry.io/otel/sdk/resource"
    sdktrace "go.opentelemetry.io/otel/sdk/trace"
    semconv "go.opentelemetry.io/otel/semconv/v1.26.0"
    "go.opentelemetry.io/otel/trace"
    "google.golang.org/grpc"
    "google.golang.org/grpc/credentials/insecure"
)

type SDK struct {
    TracerProvider *sdktrace.TracerProvider
    MeterProvider  *sdkmetric.MeterProvider
    Shutdown       func(context.Context) error
}

func NewSDK(ctx context.Context, collectorAddr string) (*SDK, error) {
    conn, err := grpc.NewClient(collectorAddr,
        grpc.WithTransportCredentials(insecure.NewCredentials()),
    )
    if err != nil {
        return nil, fmt.Errorf("grpc dial: %w", err)
    }

    res, err := resource.New(ctx,
        resource.WithAttributes(
            semconv.ServiceName("payment-service"),
            semconv.ServiceVersion("3.0.1"),
            semconv.DeploymentEnvironment("production"),
        ),
    )
    if err != nil {
        return nil, fmt.Errorf("resource: %w", err)
    }

    // Trace exporter
    traceExp, err := otlptracegrpc.New(ctx, otlptracegrpc.WithGRPCConn(conn))
    if err != nil {
        return nil, fmt.Errorf("trace exporter: %w", err)
    }

    tp := sdktrace.NewTracerProvider(
        sdktrace.WithBatcher(traceExp,
            sdktrace.WithBatchTimeout(1*time.Second),
            sdktrace.WithMaxExportBatchSize(512),
        ),
        sdktrace.WithSampler(sdktrace.ParentBased(
            sdktrace.TraceIDRatioBased(0.1), // 10% head sampling
        )),
        sdktrace.WithResource(res),
    )

    // Metric exporter
    metricExp, err := otlpmetricgrpc.New(ctx, otlpmetricgrpc.WithGRPCConn(conn))
    if err != nil {
        return nil, fmt.Errorf("metric exporter: %w", err)
    }

    mp := sdkmetric.NewMeterProvider(
        sdkmetric.WithReader(sdkmetric.NewPeriodicReader(metricExp,
            sdkmetric.WithInterval(15*time.Second),
        )),
        sdkmetric.WithResource(res),
    )

    // Set global providers
    otel.SetTracerProvider(tp)
    otel.SetMeterProvider(mp)
    otel.SetTextMapPropagator(propagation.NewCompositeTextMapPropagator(
        propagation.TraceContext{}, // W3C Trace Context (traceparent header)
        propagation.Baggage{},
    ))

    sdk := &SDK{
        TracerProvider: tp,
        MeterProvider:  mp,
        Shutdown: func(ctx context.Context) error {
            if err := tp.Shutdown(ctx); err != nil {
                return err
            }
            return mp.Shutdown(ctx)
        },
    }
    return sdk, nil
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152

// payment/handler.go
package payment

import (
    "context"
    "fmt"
    "net/http"
    "time"

    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/attribute"
    "go.opentelemetry.io/otel/codes"
    "go.opentelemetry.io/otel/metric"
    semconv "go.opentelemetry.io/otel/semconv/v1.26.0"
    "go.opentelemetry.io/otel/trace"
)

var (
    tracer = otel.Tracer("payment-service")
    meter  = otel.Meter("payment-service")
)

type PaymentHandler struct {
    paymentProcessed metric.Int64Counter
    paymentLatency   metric.Float64Histogram
    paymentsFailed   metric.Int64Counter
}

func NewPaymentHandler() (*PaymentHandler, error) {
    processed, err := meter.Int64Counter("payments.processed.total",
        metric.WithDescription("Total number of payments processed"),
        metric.WithUnit("1"),
    )
    if err != nil {
        return nil, err
    }

    latency, err := meter.Float64Histogram("payments.duration.seconds",
        metric.WithDescription("Payment processing duration"),
        metric.WithUnit("s"),
        metric.WithExplicitBucketBoundaries(0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5),
    )
    if err != nil {
        return nil, err
    }

    failed, err := meter.Int64Counter("payments.failed.total",
        metric.WithDescription("Total number of failed payments"),
        metric.WithUnit("1"),
    )
    if err != nil {
        return nil, err
    }

    return &PaymentHandler{
        paymentProcessed: processed,
        paymentLatency:   latency,
        paymentsFailed:   failed,
    }, nil
}

func (h *PaymentHandler) ProcessPayment(ctx context.Context, req PaymentRequest) (*PaymentResult, error) {
    ctx, span := tracer.Start(ctx, "payment.process",
        trace.WithAttributes(
            attribute.String("payment.method", req.Method),
            attribute.String("payment.currency", req.Currency),
            attribute.Float64("payment.amount", req.Amount),
            attribute.String("customer.id", req.CustomerID),
        ),
    )
    defer span.End()

    start := time.Now()
    attrs := []attribute.KeyValue{
        attribute.String("payment.method", req.Method),
        attribute.String("payment.currency", req.Currency),
    }

    // Validate
    ctx, validationSpan := tracer.Start(ctx, "payment.validate")
    if err := validatePayment(req); err != nil {
        validationSpan.RecordError(err)
        validationSpan.SetStatus(codes.Error, "validation failed")
        validationSpan.End()

        h.paymentsFailed.Add(ctx, 1, metric.WithAttributes(
            append(attrs, attribute.String("failure.reason", "validation"))...
        ))
        return nil, fmt.Errorf("validation: %w", err)
    }
    validationSpan.End()

    // Charge via payment gateway
    ctx, chargeSpan := tracer.Start(ctx, "payment_gateway.charge",
        trace.WithSpanKind(trace.SpanKindClient),
        trace.WithAttributes(
            semconv.RPCSystem("grpc"),
            semconv.RPCService("PaymentGateway"),
            semconv.RPCMethod("Charge"),
        ),
    )
    result, err := h.chargeGateway(ctx, req)
    if err != nil {
        chargeSpan.RecordError(err)
        chargeSpan.SetStatus(codes.Error, err.Error())
        chargeSpan.End()

        h.paymentsFailed.Add(ctx, 1, metric.WithAttributes(
            append(attrs, attribute.String("failure.reason", "gateway"))...
        ))
        return nil, fmt.Errorf("gateway charge: %w", err)
    }
    chargeSpan.SetAttributes(attribute.String("payment.transaction_id", result.TransactionID))
    chargeSpan.End()

    duration := time.Since(start).Seconds()
    h.paymentProcessed.Add(ctx, 1, metric.WithAttributes(attrs...))
    h.paymentLatency.Record(ctx, duration, metric.WithAttributes(attrs...))

    span.SetAttributes(attribute.String("payment.transaction_id", result.TransactionID))
    span.SetStatus(codes.Ok, "")
    return result, nil
}

// HTTP middleware: extract trace context from incoming requests
func OTelMiddleware(next http.Handler) http.Handler {
    propagator := otel.GetTextMapPropagator()
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        // Extract W3C traceparent from incoming headers
        ctx := propagator.Extract(r.Context(), propagation.HeaderCarrier(r.Header))

        ctx, span := tracer.Start(ctx, r.Method+" "+r.URL.Path,
            trace.WithSpanKind(trace.SpanKindServer),
            trace.WithAttributes(
                semconv.HTTPMethod(r.Method),
                semconv.HTTPTarget(r.URL.Path),
                semconv.HTTPScheme(r.URL.Scheme),
                semconv.NetHostName(r.Host),
                attribute.String("http.client_ip", r.RemoteAddr),
            ),
        )
        defer span.End()

        rw := &responseWriter{ResponseWriter: w, statusCode: 200}
        next.ServeHTTP(rw, r.WithContext(ctx))

        span.SetAttributes(semconv.HTTPStatusCode(rw.statusCode))
        if rw.statusCode >= 500 {
            span.SetStatus(codes.Error, http.StatusText(rw.statusCode))
        }
    })
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70

// main.go
package main

import (
    "context"
    "log"
    "net/http"
    "os"
    "os/signal"
    "syscall"
    "time"

    "myapp/telemetry"
    "myapp/payment"
    "go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
)

func main() {
    ctx := context.Background()

    collectorAddr := os.Getenv("OTEL_EXPORTER_OTLP_ENDPOINT")
    if collectorAddr == "" {
        collectorAddr = "localhost:4317"
    }

    sdk, err := telemetry.NewSDK(ctx, collectorAddr)
    if err != nil {
        log.Fatalf("failed to initialize OTel: %v", err)
    }
    defer func() {
        ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
        defer cancel()
        if err := sdk.Shutdown(ctx); err != nil {
            log.Printf("error shutting down OTel: %v", err)
        }
    }()

    handler, err := payment.NewPaymentHandler()
    if err != nil {
        log.Fatalf("failed to create handler: %v", err)
    }

    mux := http.NewServeMux()
    mux.Handle("/api/payments", otelhttp.NewHandler(
        http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
            // handler logic
        }),
        "payment.api",
    ))

    srv := &http.Server{
        Addr:    ":8080",
        Handler: mux,
    }

    // Graceful shutdown
    sigCh := make(chan os.Signal, 1)
    signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
    go func() {
        <-sigCh
        ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
        defer cancel()
        srv.Shutdown(ctx)
    }()

    log.Println("Starting server on :8080")
    if err := srv.ListenAndServe(); err != http.ErrServerClosed {
        log.Fatalf("server error: %v", err)
    }
}

1
2
3
4
5
6

npm install @opentelemetry/sdk-node \
            @opentelemetry/auto-instrumentations-node \
            @opentelemetry/exporter-trace-otlp-grpc \
            @opentelemetry/exporter-metrics-otlp-grpc \
            @opentelemetry/resources \
            @opentelemetry/semantic-conventions

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

// tracing.js — loaded before your app via --require
const { NodeSDK } = require('@opentelemetry/sdk-node');
const { getNodeAutoInstrumentations } = require('@opentelemetry/auto-instrumentations-node');
const { OTLPTraceExporter } = require('@opentelemetry/exporter-trace-otlp-grpc');
const { OTLPMetricExporter } = require('@opentelemetry/exporter-metrics-otlp-grpc');
const { PeriodicExportingMetricReader } = require('@opentelemetry/sdk-metrics');
const { Resource } = require('@opentelemetry/resources');
const { SEMRESATTRS_SERVICE_NAME, SEMRESATTRS_SERVICE_VERSION } = require('@opentelemetry/semantic-conventions');
const { TraceIdRatioBasedSampler, ParentBasedSampler } = require('@opentelemetry/sdk-trace-base');

const sdk = new NodeSDK({
  resource: new Resource({
    [SEMRESATTRS_SERVICE_NAME]: 'notification-service',
    [SEMRESATTRS_SERVICE_VERSION]: '1.5.2',
    'deployment.environment': process.env.NODE_ENV || 'development',
  }),

  traceExporter: new OTLPTraceExporter({
    url: process.env.OTEL_EXPORTER_OTLP_ENDPOINT || 'http://localhost:4317',
  }),

  metricReader: new PeriodicExportingMetricReader({
    exporter: new OTLPMetricExporter({
      url: process.env.OTEL_EXPORTER_OTLP_ENDPOINT || 'http://localhost:4317',
    }),
    exportIntervalMillis: 15000,
  }),

  sampler: new ParentBasedSampler({
    root: new TraceIdRatioBasedSampler(0.1),
  }),

  instrumentations: [
    getNodeAutoInstrumentations({
      // Auto-instruments: http, express, pg, redis, mongodb, grpc, aws-sdk, etc.
      '@opentelemetry/instrumentation-http': {
        ignoreIncomingRequestHook: (req) => {
          // Don't trace health checks
          return req.url === '/health' || req.url === '/metrics';
        },
      },
      '@opentelemetry/instrumentation-express': { enabled: true },
      '@opentelemetry/instrumentation-pg': { dbStatementSerializer: (query) => query.text },
      '@opentelemetry/instrumentation-redis': { enabled: true },
    }),
  ],
});

sdk.start();

// Graceful shutdown
process.on('SIGTERM', () => {
  sdk.shutdown().then(() => process.exit(0)).catch(() => process.exit(1));
});

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63

// app.js — manual spans and metrics
const { trace, metrics, context, propagation } = require('@opentelemetry/api');

const tracer = trace.getTracer('notification-service');
const meter = metrics.getMeter('notification-service');

// Define metric instruments
const emailsSent = meter.createCounter('notifications.emails.sent.total', {
  description: 'Total emails sent',
  unit: '1',
});

const notificationDuration = meter.createHistogram('notifications.send.duration.seconds', {
  description: 'Time to send notifications',
  unit: 's',
  boundaries: [0.01, 0.05, 0.1, 0.5, 1, 2, 5],
});

async function sendNotification(userId, template, channel) {
  return tracer.startActiveSpan(`notification.send.${channel}`, async (span) => {
    span.setAttributes({
      'notification.channel': channel,
      'notification.template': template,
      'user.id': userId,
    });

    const start = Date.now();
    try {
      const user = await fetchUser(userId);
      span.setAttributes({ 'user.email': user.email, 'user.locale': user.locale });

      const rendered = await renderTemplate(template, user);
      const result = await deliverViaChannel(channel, user, rendered);

      span.setStatus({ code: 1 }); // SpanStatusCode.OK
      emailsSent.add(1, { channel, template, status: 'success' });
      return result;
    } catch (err) {
      span.recordException(err);
      span.setStatus({ code: 2, message: err.message }); // SpanStatusCode.ERROR
      emailsSent.add(1, { channel, template, status: 'error' });
      throw err;
    } finally {
      const duration = (Date.now() - start) / 1000;
      notificationDuration.record(duration, { channel, template });
      span.end();
    }
  });
}

// Context propagation for outgoing HTTP calls
async function callDownstreamService(url, payload, parentCtx) {
  const headers = {};
  propagation.inject(parentCtx || context.active(), headers);
  // headers now contains: { traceparent: '00-4bf92f...', tracestate: '' }

  const response = await fetch(url, {
    method: 'POST',
    headers: { 'Content-Type': 'application/json', ...headers },
    body: JSON.stringify(payload),
  });
  return response.json();
}

1
2

# Start with auto-instrumentation loaded
node --require ./tracing.js app.js

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

from opentelemetry import trace, propagate
from opentelemetry.trace import SpanKind

# Producer: inject trace context into message headers
def publish_event(topic: str, payload: dict, producer):
    with tracer.start_as_current_span(f"kafka.send {topic}",
                                      kind=SpanKind.PRODUCER) as span:
        span.set_attribute("messaging.system", "kafka")
        span.set_attribute("messaging.destination", topic)

        headers = {}
        propagate.inject(headers)  # Adds traceparent to headers dict
        # headers = {"traceparent": "00-4bf9...", "tracestate": ""}

        producer.send(topic, value=payload, headers=list(headers.items()))

# Consumer: extract trace context from message headers
def consume_event(message, consumer_group: str):
    # Reconstruct the context from the producer's span
    headers = dict(message.headers)
    ctx = propagate.extract(headers)  # Recovers the producer's trace context

    with tracer.start_as_current_span(
        f"kafka.receive {message.topic}",
        context=ctx,
        kind=SpanKind.CONSUMER,
    ) as span:
        span.set_attribute("messaging.system", "kafka")
        span.set_attribute("messaging.source", message.topic)
        span.set_attribute("messaging.consumer.group", consumer_group)
        span.set_attribute("messaging.message.offset", message.offset)
        process_event(message.value)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

from opentelemetry.sdk.trace.sampling import (
    ParentBased,
    TraceIdRatioBased,
    ALWAYS_ON,
    ALWAYS_OFF,
)

# Sample 10% of new traces, follow parent decision for child spans
sampler = ParentBased(root=TraceIdRatioBased(0.10))

# For high-volume services, sample more aggressively
sampler = ParentBased(root=TraceIdRatioBased(0.01))  # 1%

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

# In otel-collector.yaml (shown above) — tail_sampling processor
tail_sampling:
  decision_wait: 10s      # Wait up to 10s for all spans of a trace to arrive
  num_traces: 100000      # Max traces held in memory awaiting decision
  expected_new_traces_per_sec: 1000
  policies:
    # 100% of traces with errors
    - name: errors
      type: status_code
      status_code: {status_codes: [ERROR]}

    # 100% of traces slower than 500ms
    - name: slow
      type: latency
      latency: {threshold_ms: 500}

    # 100% of traces from specific high-value users
    - name: priority-users
      type: string_attribute
      string_attribute:
        key: user.tier
        values: [enterprise, trial]

    # 1% of everything else
    - name: base-rate
      type: probabilistic
      probabilistic: {sampling_percentage: 1}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

from opentelemetry.sdk.metrics.export import AggregationTemporality
from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter

# Use cumulative temporality for Prometheus compatibility
exporter = OTLPMetricExporter(
    endpoint="http://localhost:4317",
    preferred_temporality={
        Counter: AggregationTemporality.CUMULATIVE,
        UpDownCounter: AggregationTemporality.CUMULATIVE,
        Histogram: AggregationTemporality.CUMULATIVE,
    }
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

from opentelemetry.sdk.metrics._internal.exemplar import AlwaysOnExemplarFilter

# Enable exemplars in the MeterProvider
meter_provider = MeterProvider(
    resource=resource,
    metric_readers=[reader],
    views=[
        View(
            instrument_type=Histogram,
            exemplar_filter=AlwaysOnExemplarFilter(),
        )
    ]
)

1
2

# Histogram query with exemplar display enabled
histogram_quantile(0.99, sum(rate(http_request_duration_seconds_bucket[5m])) by (le, service))

1
2
3
4
5

# Install cert-manager (required by the operator)
kubectl apply -f https://github.com/cert-manager/cert-manager/releases/latest/download/cert-manager.yaml

# Install the OTel Operator
kubectl apply -f https://github.com/open-telemetry/opentelemetry-operator/releases/latest/download/opentelemetry-operator.yaml

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

apiVersion: opentelemetry.io/v1beta1
kind: OpenTelemetryCollector
metadata:
  name: otel-daemonset
  namespace: observability
spec:
  mode: daemonset
  resources:
    requests:
      memory: 256Mi
      cpu: 100m
    limits:
      memory: 512Mi
      cpu: 500m
  config:
    receivers:
      otlp:
        protocols:
          grpc:
            endpoint: 0.0.0.0:4317
          http:
            endpoint: 0.0.0.0:4318

    processors:
      batch: {}
      memory_limiter:
        check_interval: 1s
        limit_mib: 400
      k8sattributes:
        auth_type: serviceAccount
        extract:
          metadata: [k8s.namespace.name, k8s.pod.name, k8s.deployment.name, k8s.node.name]

    exporters:
      otlp:
        endpoint: otel-gateway-collector:4317
        tls:
          insecure: true

    service:
      pipelines:
        traces:
          receivers: [otlp]
          processors: [memory_limiter, k8sattributes, batch]
          exporters: [otlp]
        metrics:
          receivers: [otlp]
          processors: [memory_limiter, k8sattributes, batch]
          exporters: [otlp]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

apiVersion: opentelemetry.io/v1alpha1
kind: Instrumentation
metadata:
  name: auto-instrumentation
  namespace: default
spec:
  exporter:
    endpoint: http://otel-daemonset-collector:4317

  propagators:
    - tracecontext
    - baggage

  sampler:
    type: parentbased_traceidratio
    argument: "0.1"

  python:
    env:
      - name: OTEL_LOGS_EXPORTER
        value: otlp

  java:
    image: ghcr.io/open-telemetry/opentelemetry-operator/autoinstrumentation-java:latest

  nodejs:
    env:
      - name: OTEL_NODEJS_DEBUG
        value: "false"

  dotnet:
    env: []

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

# Annotate a pod to get auto-instrumentation
apiVersion: apps/v1
kind: Deployment
metadata:
  name: my-python-app
  namespace: default
spec:
  template:
    metadata:
      annotations:
        instrumentation.opentelemetry.io/inject-python: "true"
        # The operator injects OTEL_EXPORTER_OTLP_ENDPOINT, OTEL_SERVICE_NAME,
        # PYTHONPATH, and runs opentelemetry-instrument automatically
    spec:
      containers:
      - name: app
        image: my-python-app:latest
        env:
        - name: OTEL_SERVICE_NAME
          value: "my-python-app"
        - name: OTEL_RESOURCE_ATTRIBUTES
          value: "service.version=1.2.3,deployment.environment=production"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

# grafana/provisioning/datasources/datasources.yml
apiVersion: 1
datasources:
  - name: Prometheus
    type: prometheus
    url: http://prometheus:9090
    isDefault: true
    jsonData:
      exemplarTraceIdDestinations:
        - name: trace_id
          datasourceUid: tempo  # Link exemplars to Tempo traces

  - name: Tempo
    type: tempo
    uid: tempo
    url: http://tempo:3200
    jsonData:
      tracesToLogsV2:
        datasourceUid: loki
        filterByTraceID: true
        filterBySpanID: true
        tags: [{key: "service.name", value: "service"}]
      tracesToMetrics:
        datasourceUid: prometheus
        tags: [{key: "service.name", value: "job"}]
        queries:
          - name: Request rate
            query: rate(http_requests_total{$__tags}[5m])
      serviceMap:
        datasourceUid: prometheus  # Use Prometheus for service map
      nodeGraph:
        enabled: true

  - name: Loki
    type: loki
    url: http://loki:3100
    jsonData:
      derivedFields:
        - name: TraceID
          matcherRegex: '"trace_id":"(\w+)"'
          url: "${__value.raw}"
          datasourceUid: tempo  # Click trace_id in logs → open trace in Tempo

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

# Request rate by service
sum(rate(http_server_duration_milliseconds_count[5m])) by (service_name)

# P99 latency by service
histogram_quantile(0.99,
  sum(rate(http_server_duration_milliseconds_bucket[5m])) by (le, service_name)
)

# Error rate
sum(rate(http_server_duration_milliseconds_count{http_status_code=~"5.."}[5m])) by (service_name)
/
sum(rate(http_server_duration_milliseconds_count[5m])) by (service_name)

# Apdex score (satisfied <100ms, tolerating <500ms)
(
  sum(rate(http_server_duration_milliseconds_bucket{le="100"}[5m])) by (service_name) +
  sum(rate(http_server_duration_milliseconds_bucket{le="500"}[5m])) by (service_name) / 2
) /
sum(rate(http_server_duration_milliseconds_count[5m])) by (service_name)

1
2

# Without this, all your telemetry shows up as "unknown_service"
OTEL_SERVICE_NAME=my-service  # Always set this

1
2
3
4
5
6

# At process exit, flush buffered spans/metrics
tracer_provider.shutdown()
meter_provider.shutdown()
# Or use atexit:
import atexit
atexit.register(lambda: tracer_provider.shutdown())

1
2
3
4
5
6
7

# BAD: user_id as a metric attribute → millions of time series
request_counter.add(1, {"user_id": user_id})

# GOOD: put user_id on spans (traces can handle high cardinality)
span.set_attribute("user.id", user_id)
# Metrics use low-cardinality dimensions only
request_counter.add(1, {"region": region, "plan": user.plan})

1
2
3
4
5
6

# WRONG: pure ratio sampling drops errors too
sampler = TraceIdRatioBased(0.001)  # 0.1% — you'll miss most errors

# RIGHT: always keep errors, sample the rest
sampler = ParentBased(root=TraceIdRatioBased(0.001))
# Plus tail sampling in the Collector for error policy

1
2
3
4
5
6
7

# BAD: synchronous exporter blocks on network I/O
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
provider.add_span_processor(SimpleSpanProcessor(exporter))

# GOOD: batch processor is asynchronous
from opentelemetry.sdk.trace.export import BatchSpanProcessor
provider.add_span_processor(BatchSpanProcessor(exporter))