Observability

Testing OpenTelemetry Instrumentation: Validating Spans and Trace Propagation

HelpMeTest

HelpMeTest

— 6 min read
Testing OpenTelemetry Instrumentation: Validating Spans and Trace Propagation

OpenTelemetry instrumentation can silently break—spans stop being created, attributes go missing, or trace context fails to propagate across service boundaries. The OTel SDK ships with an InMemorySpanExporter specifically for testing: it captures spans in memory so you can assert their structure without running a collector.

Key Takeaways

InMemorySpanExporter is your test double for the OTel Collector. It captures all spans in process so you can assert span names, attributes, events, and parent-child relationships without any external dependencies.

Trace propagation is a contract. Test that W3C TraceContext headers flow correctly across HTTP boundaries—a missing traceparent header silently breaks distributed traces.

Test auto-instrumentation explicitly. Auto-instrumented libraries (HTTP clients, DB drivers) can silently stop producing spans after a library update. An integration test that asserts span count catches this early.

Why OTel Instrumentation Needs Tests

OpenTelemetry is the industry standard for distributed tracing, but the instrumentation layer has its own failure modes that don't surface as application errors:

  • A span is created but never finished (leaked spans, memory growth)
  • An attribute name has a typo—it gets silently dropped if it exceeds the attribute count limit
  • Trace context headers aren't propagated in an outbound HTTP request, breaking the distributed trace
  • A library update changes the auto-instrumented span names that your dashboards and alerts depend on
  • Baggage propagates into a downstream service but is never read

None of these cause an exception. The application works. The traces are just wrong—missing context, broken chains, or absent spans. Testing instrumentation makes these failures visible.

Setting Up InMemorySpanExporter

The OTel SDK ships with InMemorySpanExporter in the testing packages. It captures all finished spans in an in-process list.

Python

# tests/conftest.py
import pytest
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
from opentelemetry.sdk.trace.export import SimpleSpanProcessor

@pytest.fixture(autouse=True)
def otel_test_setup():
    """Configure OTel with in-memory exporter for each test."""
    exporter = InMemorySpanExporter()
    provider = TracerProvider()
    provider.add_span_processor(SimpleSpanProcessor(exporter))
    trace.set_tracer_provider(provider)

    yield exporter

    exporter.clear()
    # Reset to default provider to avoid cross-test contamination
    trace.set_tracer_provider(trace.ProxyTracerProvider())

JavaScript / TypeScript

// tests/helpers/otel-setup.ts
import {
  InMemorySpanExporter,
  SimpleSpanProcessor,
} from '@opentelemetry/sdk-trace-base';
import { NodeTracerProvider } from '@opentelemetry/sdk-trace-node';
import { context, trace } from '@opentelemetry/api';

let exporter: InMemorySpanExporter;
let provider: NodeTracerProvider;

export function setupOtelTest() {
  exporter = new InMemorySpanExporter();
  provider = new NodeTracerProvider();
  provider.addSpanProcessor(new SimpleSpanProcessor(exporter));
  provider.register();

  return exporter;
}

export function teardownOtelTest() {
  exporter.reset();
  provider.shutdown();
}

export function getSpans() {
  return exporter.getFinishedSpans();
}

Asserting Span Attributes and Events

With the exporter set up, write assertions against the captured spans:

# tests/test_payment_tracing.py
from myapp.payments import process_payment

def test_payment_span_has_required_attributes(otel_test_setup):
    exporter = otel_test_setup

    process_payment(amount=4900, currency='USD', user_id='u_123')

    spans = exporter.get_finished_spans()
    assert len(spans) > 0, "No spans were created"

    payment_span = next(
        (s for s in spans if s.name == 'payment.process'),
        None
    )
    assert payment_span is not None, \
        f"Expected 'payment.process' span, got: {[s.name for s in spans]}"

    attrs = payment_span.attributes
    assert attrs['payment.amount'] == 4900
    assert attrs['payment.currency'] == 'USD'
    assert attrs['user.id'] == 'u_123'
    assert attrs.get('payment.provider') is not None


def test_payment_span_records_error_event_on_failure(otel_test_setup):
    exporter = otel_test_setup

    with pytest.raises(ValueError):
        process_payment(amount=-1, currency='USD', user_id='u_123')

    spans = exporter.get_finished_spans()
    payment_span = next(s for s in spans if s.name == 'payment.process')

    # Span should be marked with ERROR status
    from opentelemetry.trace import StatusCode
    assert payment_span.status.status_code == StatusCode.ERROR

    # Should have recorded an exception event
    events = payment_span.events
    assert len(events) == 1
    assert events[0].name == 'exception'
    assert 'ValueError' in events[0].attributes['exception.type']


def test_payment_span_parent_child_relationship(otel_test_setup):
    exporter = otel_test_setup

    # process_payment should create a child span for the DB write
    process_payment(amount=1000, currency='EUR', user_id='u_456')

    spans = exporter.get_finished_spans()
    payment_span = next(s for s in spans if s.name == 'payment.process')
    db_span = next((s for s in spans if s.name == 'db.payment.insert'), None)

    assert db_span is not None, "DB span not created"
    assert db_span.parent.span_id == payment_span.context.span_id, \
        "DB span is not a child of payment span"

JavaScript example

// tests/checkout.test.ts
import { setupOtelTest, teardownOtelTest, getSpans } from './helpers/otel-setup';
import { processCheckout } from '../src/checkout';

beforeEach(() => setupOtelTest());
afterEach(() => teardownOtelTest());

test('checkout creates span with order attributes', async () => {
  await processCheckout({ cartId: 'cart_abc', userId: 'u_789' });

  const spans = getSpans();
  const checkoutSpan = spans.find(s => s.name === 'checkout.process');

  expect(checkoutSpan).toBeDefined();
  expect(checkoutSpan!.attributes['cart.id']).toBe('cart_abc');
  expect(checkoutSpan!.attributes['user.id']).toBe('u_789');
  expect(checkoutSpan!.status.code).toBe(SpanStatusCode.OK);
});

test('checkout span has exactly one db child span', async () => {
  await processCheckout({ cartId: 'cart_abc', userId: 'u_789' });

  const spans = getSpans();
  const checkoutSpan = spans.find(s => s.name === 'checkout.process')!;
  const dbSpans = spans.filter(
    s => s.parentSpanId === checkoutSpan.spanContext().spanId
      && s.name.startsWith('db.')
  );

  expect(dbSpans).toHaveLength(1);
  expect(dbSpans[0].attributes['db.statement']).toContain('INSERT INTO orders');
});

Testing Trace Context Propagation Across Service Boundaries

Distributed tracing only works if the traceparent header is correctly injected into outbound requests and extracted in downstream services. This is the most commonly broken part of OTel instrumentation.

Test HTTP propagation with a stub downstream server:

# tests/test_trace_propagation.py
import threading
from http.server import HTTPServer, BaseHTTPRequestHandler
from opentelemetry.propagate import extract
from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator

received_context = {}

class PropagationTestServer(BaseHTTPRequestHandler):
    def do_POST(self):
        # Extract the trace context from incoming headers
        carrier = dict(self.headers)
        ctx = extract(carrier)
        span = get_current_span(ctx)
        received_context['trace_id'] = format(
            span.get_span_context().trace_id, '032x'
        )
        self.send_response(200)
        self.end_headers()
        self.wfile.write(b'{}')

    def log_message(self, *args): pass

def test_outbound_http_request_propagates_trace_context(otel_test_setup):
    """Verify that traceparent is injected into outbound HTTP requests."""
    exporter = otel_test_setup

    # Start local stub server
    server = HTTPServer(('127.0.0.1', 19876), PropagationTestServer)
    thread = threading.Thread(target=server.handle_request)
    thread.start()

    tracer = trace.get_tracer('test')
    with tracer.start_as_current_span('test-root') as root_span:
        root_trace_id = format(root_span.get_span_context().trace_id, '032x')
        # Call a service method that makes an HTTP request to the stub
        call_downstream_service('http://127.0.0.1:19876/api/endpoint')

    thread.join(timeout=2)

    assert received_context.get('trace_id') == root_trace_id, \
        "Downstream service did not receive matching trace_id — propagation broken"

For services using the OTel auto-instrumentation for the requests library (Python) or undici/http (Node.js), the propagation happens automatically—but you still need this test to confirm it wasn't accidentally disabled.

Testing Auto-Instrumentation Correctness

Auto-instrumentation for HTTP frameworks, database clients, and message brokers can silently break after library updates. Test it explicitly:

# tests/test_auto_instrumentation.py
from opentelemetry.instrumentation.flask import FlaskInstrumentor
from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor

def test_flask_auto_instrumentation_creates_http_span(otel_test_setup, flask_app):
    exporter = otel_test_setup

    with flask_app.test_client() as client:
        client.get('/api/users/123')

    spans = exporter.get_finished_spans()
    http_spans = [s for s in spans if s.name.startswith('GET ')]

    assert len(http_spans) == 1, \
        "Flask auto-instrumentation did not create HTTP span"

    span = http_spans[0]
    assert span.attributes['http.method'] == 'GET'
    assert span.attributes['http.target'] == '/api/users/123'
    assert span.attributes['http.status_code'] == 200


def test_sqlalchemy_auto_instrumentation_creates_db_span(otel_test_setup, db_session):
    exporter = otel_test_setup

    db_session.execute("SELECT 1")

    spans = exporter.get_finished_spans()
    db_spans = [s for s in spans if s.attributes.get('db.system') == 'postgresql']

    assert len(db_spans) >= 1, "SQLAlchemy auto-instrumentation did not create DB span"
    assert db_spans[0].attributes['db.statement'] == 'SELECT 1'

Testing Baggage Propagation

OpenTelemetry Baggage allows key-value pairs to propagate across service boundaries—useful for tenant IDs, feature flags, and request-level context. Test that baggage is correctly set and read:

# tests/test_baggage_propagation.py
from opentelemetry import baggage, context

def test_baggage_is_set_and_readable():
    ctx = baggage.set_baggage('tenant.id', 'tenant_abc')
    with context.use_context(ctx):
        tenant_id = baggage.get_baggage('tenant.id')
        assert tenant_id == 'tenant_abc'


def test_baggage_is_encoded_in_outbound_headers():
    from opentelemetry.propagate import inject
    from opentelemetry import baggage, context

    ctx = baggage.set_baggage('tenant.id', 'tenant_xyz')
    headers = {}
    with context.use_context(ctx):
        inject(headers)

    assert 'baggage' in headers
    assert 'tenant.id=tenant_xyz' in headers['baggage']


def test_baggage_is_attached_as_span_attribute(otel_test_setup):
    """Verify that middleware reads baggage and attaches it to spans."""
    exporter = otel_test_setup

    # Make a request with baggage header set
    response = test_client.get(
        '/api/data',
        headers={'baggage': 'tenant.id=tenant_999'}
    )

    spans = exporter.get_finished_spans()
    http_span = next(s for s in spans if s.name.startswith('GET'))

    # Your middleware should copy baggage to span attributes
    assert http_span.attributes.get('tenant.id') == 'tenant_999'

Integration Testing with the OTel Collector

For end-to-end integration tests, run the OTel Collector in your test environment using Docker Compose and configure it with a file exporter so you can assert on actual exported telemetry:

# docker-compose.test.yaml
services:
  otel-collector:
    image: otel/opentelemetry-collector-contrib:0.99.0
    command: ["--config=/etc/otel/config.yaml"]
    volumes:
      - ./tests/otel-collector-test.yaml:/etc/otel/config.yaml
      - ./tests/otel-output:/tmp/otel-output
    ports:
      - "4317:4317"   # OTLP gRPC
      - "4318:4318"   # OTLP HTTP
# tests/otel-collector-test.yaml
receivers:
  otlp:
    protocols:
      grpc:
        endpoint: 0.0.0.0:4317
      http:
        endpoint: 0.0.0.0:4318

exporters:
  file:
    path: /tmp/otel-output/spans.json

service:
  pipelines:
    traces:
      receivers: [otlp]
      exporters: [file]

Then in your test:

import json, time, pathlib

def test_spans_exported_to_collector():
    output_file = pathlib.Path('tests/otel-output/spans.json')
    output_file.unlink(missing_ok=True)

    # Run your service action
    response = requests.post('http://localhost:8080/api/checkout', json={...})
    assert response.status_code == 200

    # Give the SDK time to flush to collector
    time.sleep(1)

    spans_raw = output_file.read_text().strip().split('\n')
    all_spans = [json.loads(line) for line in spans_raw if line]

    span_names = [
        s['name']
        for entry in all_spans
        for rs in entry.get('resourceSpans', [])
        for ss in rs.get('scopeSpans', [])
        for s in ss.get('spans', [])
    ]

    assert 'checkout.process' in span_names
    assert 'db.order.insert' in span_names

This approach tests the full pipeline—SDK, OTLP export, Collector processing—and catches configuration issues that only appear when the real exporter is active.

HelpMeTest can monitor your observability stack automatically — sign up free

Read more