Testing

Unit Testing OpenTelemetry Instrumentation: In-Memory Exporters

HelpMeTest

30 Jul 2026 — 7 min read

Custom OpenTelemetry instrumentation is code like any other code — it can be wrong, and wrong instrumentation is worse than no instrumentation. If your span names are wrong, your dashboards show garbage. If context propagation is broken, your distributed traces have gaps. If attributes are missing, your SLOs can't be calculated.

Testing instrumentation with in-memory exporters lets you verify exactly what spans and metrics your code produces, without spinning up a Jaeger or Prometheus instance. This post covers how to do this in Python and Go, and how to write tests for custom OpenTelemetry plugins.

Why Test Your Instrumentation

Three categories of instrumentation bugs that tests catch:

Wrong span naming: Your traces show HTTP POST instead of payments.process. Dashboards break, queries break, alerting rules break.

Missing attributes: You're emitting spans but not adding the attributes your queries depend on. user.id, order.total, payment.method are all absent. SLO calculation fails silently.

Broken context propagation: Services emit spans, but the spans aren't connected — no parent/child relationships. You get isolated spans instead of traces. Distributed tracing is useless.

Wrong span status: Your code marks a failed payment as OK instead of ERROR. Error rate metrics are wrong. You're flying blind.

In-Memory Exporters in Python

The OpenTelemetry Python SDK includes an in-memory span exporter and a metric exporter for testing:

# Install: pip install opentelemetry-sdk pytest
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
from opentelemetry.trace import SpanKind, StatusCode

Set up a test TracerProvider that uses the in-memory exporter:

# 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
def span_exporter():
    """Returns a fresh in-memory span exporter for each test"""
    return InMemorySpanExporter()

@pytest.fixture
def tracer_provider(span_exporter):
    """Returns a TracerProvider wired to the in-memory exporter"""
    provider = TracerProvider()
    provider.add_span_processor(SimpleSpanProcessor(span_exporter))
    
    # Set as global provider so instrumented code uses it
    trace.set_tracer_provider(provider)
    
    yield provider
    
    # Cleanup: reset to no-op provider after each test
    trace.set_tracer_provider(trace.NoOpTracerProvider())

Now write tests for instrumented code:

# payments/processor.py — the code under test
from opentelemetry import trace
from opentelemetry.trace import SpanKind, StatusCode

tracer = trace.get_tracer("payments.processor", "1.0.0")

def process_payment(user_id: str, amount_cents: int, payment_method: str) -> dict:
    with tracer.start_as_current_span(
        "payments.process",
        kind=SpanKind.INTERNAL,
    ) as span:
        span.set_attribute("user.id", user_id)
        span.set_attribute("payment.amount_cents", amount_cents)
        span.set_attribute("payment.method", payment_method)
        
        try:
            result = _charge_card(user_id, amount_cents)
            span.set_attribute("payment.transaction_id", result["transaction_id"])
            return result
        except PaymentDeclinedException as e:
            span.set_status(StatusCode.ERROR, str(e))
            span.record_exception(e)
            raise

# tests/test_payment_instrumentation.py
import pytest
from opentelemetry.trace import SpanKind, StatusCode
from payments.processor import process_payment, PaymentDeclinedException
from unittest.mock import patch

class TestPaymentInstrumentation:
    def test_successful_payment_emits_correct_span(self, tracer_provider, span_exporter):
        with patch('payments.processor._charge_card') as mock_charge:
            mock_charge.return_value = {"transaction_id": "txn_123"}
            
            process_payment("user-42", 1999, "visa")
        
        spans = span_exporter.get_finished_spans()
        assert len(spans) == 1, f"Expected 1 span, got {len(spans)}"
        
        span = spans[0]
        
        # Verify span name
        assert span.name == "payments.process"
        
        # Verify span kind
        assert span.kind == SpanKind.INTERNAL
        
        # Verify attributes
        assert span.attributes["user.id"] == "user-42"
        assert span.attributes["payment.amount_cents"] == 1999
        assert span.attributes["payment.method"] == "visa"
        assert span.attributes["payment.transaction_id"] == "txn_123"
        
        # Verify status
        assert span.status.status_code == StatusCode.UNSET  # OK/Unset for success

    def test_declined_payment_marks_span_error(self, tracer_provider, span_exporter):
        with patch('payments.processor._charge_card') as mock_charge:
            mock_charge.side_effect = PaymentDeclinedException("Insufficient funds")
            
            with pytest.raises(PaymentDeclinedException):
                process_payment("user-42", 99999, "visa")
        
        spans = span_exporter.get_finished_spans()
        assert len(spans) == 1
        
        span = spans[0]
        
        # Span should be marked ERROR
        assert span.status.status_code == StatusCode.ERROR
        assert "Insufficient funds" in span.status.description
        
        # Exception should be recorded as an event
        events = span.events
        assert len(events) == 1
        assert events[0].name == "exception"
        assert "Insufficient funds" in events[0].attributes["exception.message"]

    def test_span_includes_tracer_name_and_version(self, tracer_provider, span_exporter):
        with patch('payments.processor._charge_card') as mock_charge:
            mock_charge.return_value = {"transaction_id": "txn_1"}
            process_payment("user-1", 100, "visa")
        
        span = span_exporter.get_finished_spans()[0]
        
        assert span.instrumentation_scope.name == "payments.processor"
        assert span.instrumentation_scope.version == "1.0.0"

Testing Metrics in Python

For custom metrics, use the in-memory metric reader:

# tests/conftest.py (additions)
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import InMemoryMetricReader
from opentelemetry import metrics

@pytest.fixture
def metric_reader():
    return InMemoryMetricReader()

@pytest.fixture
def meter_provider(metric_reader):
    provider = MeterProvider(metric_readers=[metric_reader])
    metrics.set_meter_provider(provider)
    yield provider
    metrics.set_meter_provider(metrics.NoOpMeterProvider())

# payments/metrics.py
from opentelemetry import metrics

meter = metrics.get_meter("payments", "1.0.0")

payment_counter = meter.create_counter(
    name="payments.processed",
    description="Number of payments processed",
    unit="1"
)

payment_amount_histogram = meter.create_histogram(
    name="payments.amount",
    description="Payment amounts in cents",
    unit="cents"
)

def record_payment(amount_cents: int, method: str, status: str):
    payment_counter.add(1, {"payment.method": method, "payment.status": status})
    payment_amount_histogram.record(amount_cents, {"payment.method": method})

# tests/test_payment_metrics.py
from payments.metrics import record_payment

class TestPaymentMetrics:
    def test_successful_payment_increments_counter(self, meter_provider, metric_reader):
        record_payment(1999, "visa", "success")
        record_payment(2999, "mastercard", "success")
        record_payment(999, "visa", "declined")
        
        # Collect metrics
        metrics_data = metric_reader.get_metrics_data()
        
        # Find the payments.processed counter
        resource_metrics = metrics_data.resource_metrics
        assert len(resource_metrics) > 0
        
        scope_metrics = resource_metrics[0].scope_metrics
        payment_metrics = [
            m for sm in scope_metrics 
            for m in sm.metrics 
            if m.name == "payments.processed"
        ]
        assert len(payment_metrics) == 1
        
        counter = payment_metrics[0]
        data_points = counter.data.data_points
        
        # Find visa/success data point
        visa_success = next(
            dp for dp in data_points
            if dp.attributes.get("payment.method") == "visa" 
            and dp.attributes.get("payment.status") == "success"
        )
        assert visa_success.value == 1
        
        # Find mastercard/success data point
        mc_success = next(
            dp for dp in data_points
            if dp.attributes.get("payment.method") == "mastercard"
            and dp.attributes.get("payment.status") == "success"
        )
        assert mc_success.value == 1

In-Memory Exporters in Go

Go's OpenTelemetry SDK provides similar in-memory exporters:

// go.mod dependencies:
// go.opentelemetry.io/otel/sdk v1.21.0
// go.opentelemetry.io/otel/exporters/stdout/stdouttrace (for debugging)
// go.opentelemetry.io/otel/sdk/metric

package payments_test

import (
    "context"
    "testing"
    
    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/attribute"
    sdktrace "go.opentelemetry.io/otel/sdk/trace"
    "go.opentelemetry.io/otel/sdk/trace/tracetest"
    "go.opentelemetry.io/otel/codes"
    
    "github.com/example/payments"
)

func setupTracer(t *testing.T) (*tracetest.SpanRecorder, func()) {
    t.Helper()
    
    recorder := tracetest.NewSpanRecorder()
    provider := sdktrace.NewTracerProvider(
        sdktrace.WithSpanProcessor(recorder),
    )
    
    otel.SetTracerProvider(provider)
    
    cleanup := func() {
        otel.SetTracerProvider(otel.GetTracerProvider()) // reset
    }
    
    return recorder, cleanup
}

func TestProcessPaymentSpan(t *testing.T) {
    recorder, cleanup := setupTracer(t)
    defer cleanup()
    
    // Call the instrumented function
    processor := payments.NewProcessor()
    _, err := processor.ProcessPayment(context.Background(), payments.PaymentRequest{
        UserID:        "user-42",
        AmountCents:   1999,
        PaymentMethod: "visa",
    })
    if err != nil {
        t.Fatalf("unexpected error: %v", err)
    }
    
    // Get finished spans
    spans := recorder.Ended()
    if len(spans) != 1 {
        t.Fatalf("expected 1 span, got %d", len(spans))
    }
    
    span := spans[0]
    
    // Verify span name
    if span.Name() != "payments.process" {
        t.Errorf("expected span name 'payments.process', got %q", span.Name())
    }
    
    // Verify attributes
    attrs := span.Attributes()
    assertAttr(t, attrs, "user.id", attribute.StringValue("user-42"))
    assertAttr(t, attrs, "payment.amount_cents", attribute.Int64Value(1999))
    assertAttr(t, attrs, "payment.method", attribute.StringValue("visa"))
    
    // Verify status
    if span.Status().Code != codes.Unset {
        t.Errorf("expected status Unset for success, got %v", span.Status().Code)
    }
}

func TestProcessPaymentSpanErrorStatus(t *testing.T) {
    recorder, cleanup := setupTracer(t)
    defer cleanup()
    
    processor := payments.NewProcessor()
    processor.SetCardSimulator(payments.DeclineAll) // force decline
    
    _, err := processor.ProcessPayment(context.Background(), payments.PaymentRequest{
        UserID:      "user-42",
        AmountCents: 99999,
    })
    if err == nil {
        t.Fatal("expected error for declined payment")
    }
    
    spans := recorder.Ended()
    if len(spans) == 0 {
        t.Fatal("no spans recorded")
    }
    
    span := spans[0]
    
    if span.Status().Code != codes.Error {
        t.Errorf("expected status Error for declined payment, got %v", span.Status().Code)
    }
    
    // Check that exception was recorded as an event
    events := span.Events()
    if len(events) == 0 {
        t.Error("expected at least one event (exception), got none")
    }
    if events[0].Name != "exception" {
        t.Errorf("expected event name 'exception', got %q", events[0].Name)
    }
}

func assertAttr(t *testing.T, attrs []attribute.KeyValue, key string, expected attribute.Value) {
    t.Helper()
    for _, attr := range attrs {
        if string(attr.Key) == key {
            if attr.Value != expected {
                t.Errorf("attribute %q: expected %v, got %v", key, expected, attr.Value)
            }
            return
        }
    }
    t.Errorf("attribute %q not found in span", key)
}

Testing Context Propagation

Context propagation tests verify that parent-child span relationships are correctly established across service calls:

# tests/test_context_propagation.py
from opentelemetry import trace
from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator

def test_http_call_propagates_trace_context(tracer_provider, span_exporter):
    """Test that outbound HTTP calls include trace context headers"""
    from payments.http_client import TracedHttpClient  # custom instrumented client
    
    tracer = trace.get_tracer("test")
    propagator = TraceContextTextMapPropagator()
    
    captured_headers = {}
    
    with tracer.start_as_current_span("parent-span") as parent:
        parent_trace_id = format(parent.get_span_context().trace_id, '032x')
        
        # The traced HTTP client should inject trace context
        client = TracedHttpClient()
        client.on_request = lambda headers: captured_headers.update(headers)
        client.get("http://catalog-service/products")
    
    # The outbound request should have carried traceparent header
    assert "traceparent" in captured_headers, "Missing traceparent header in outbound request"
    
    # The traceparent should reference our parent trace ID
    traceparent = captured_headers["traceparent"]
    assert parent_trace_id in traceparent

def test_incoming_context_creates_child_span(tracer_provider, span_exporter):
    """Test that incoming trace context is correctly accepted as parent"""
    from payments.app import create_app
    import flask.testing
    
    app = create_app()
    client = app.test_client()
    
    # Inject a fake traceparent header simulating an upstream caller
    parent_trace_id = "4bf92f3577b34da6a3ce929d0e0e4736"
    parent_span_id = "00f067aa0ba902b7"
    traceparent = f"00-{parent_trace_id}-{parent_span_id}-01"
    
    response = client.post(
        "/api/payments",
        json={"userId": "user-1", "amountCents": 1000},
        headers={"traceparent": traceparent}
    )
    
    assert response.status_code == 200
    
    spans = span_exporter.get_finished_spans()
    assert len(spans) > 0
    
    # The request span should be a child of the incoming trace
    request_span = spans[0]
    span_context = request_span.get_span_context()
    
    # Same trace ID as the incoming traceparent
    assert format(span_context.trace_id, '032x') == parent_trace_id
    
    # Parent span ID should be the one we sent
    assert format(request_span.parent.span_id, '016x') == parent_span_id

Testing Custom OpenTelemetry Plugins

If you're writing a custom OpenTelemetry instrumentation library (e.g., to instrument a database driver or message broker), test the plugin with the in-memory exporter as well:

# my_kafka_instrumentation.py
class InstrumentedKafkaProducer:
    def __init__(self, producer, tracer=None):
        self._producer = producer
        self._tracer = tracer or trace.get_tracer("kafka.producer")

    def produce(self, topic: str, value: bytes, key: bytes = None):
        with self._tracer.start_as_current_span(
            f"kafka.produce {topic}",
            kind=trace.SpanKind.PRODUCER,
        ) as span:
            span.set_attribute("messaging.system", "kafka")
            span.set_attribute("messaging.destination", topic)
            span.set_attribute("messaging.destination_kind", "topic")
            if key:
                span.set_attribute("messaging.kafka.message.key", key.decode())
            
            # Inject trace context into message headers
            headers = {}
            TraceContextTextMapPropagator().inject(headers)
            
            self._producer.produce(topic, value=value, key=key, headers=headers)

# tests/test_kafka_instrumentation.py
def test_kafka_produce_emits_producer_span(tracer_provider, span_exporter):
    mock_producer = MagicMock()
    producer = InstrumentedKafkaProducer(mock_producer)
    
    producer.produce("order-events", b'{"orderId": "ord-1"}', key=b"ord-1")
    
    spans = span_exporter.get_finished_spans()
    assert len(spans) == 1
    
    span = spans[0]
    assert span.name == "kafka.produce order-events"
    assert span.kind == trace.SpanKind.PRODUCER
    assert span.attributes["messaging.system"] == "kafka"
    assert span.attributes["messaging.destination"] == "order-events"
    assert span.attributes["messaging.kafka.message.key"] == "ord-1"

def test_kafka_produce_injects_trace_context_in_headers(tracer_provider, span_exporter):
    mock_producer = MagicMock()
    producer = InstrumentedKafkaProducer(mock_producer)
    
    tracer = trace.get_tracer("test")
    with tracer.start_as_current_span("parent"):
        producer.produce("events", b'{}')
    
    # Check that the Kafka message was sent with traceparent header
    call_args = mock_producer.produce.call_args
    headers = call_args.kwargs.get("headers", {})
    
    assert "traceparent" in headers, "Trace context not injected into Kafka message headers"

Running Instrumentation Tests in CI

Instrumentation tests are fast (in-memory, no external dependencies) and should run in every PR:

# .github/workflows/otel-tests.yml
- name: Run instrumentation tests
  run: |
    pytest tests/test_payment_instrumentation.py \
           tests/test_context_propagation.py \
           tests/test_kafka_instrumentation.py \
           -v --tb=short

These tests take milliseconds and catch instrumentation regressions before they silently corrupt your observability data in production.

Unit Testing OpenTelemetry Instrumentation: In-Memory Exporters

HelpMeTest

Why Test Your Instrumentation

In-Memory Exporters in Python

Testing Metrics in Python

In-Memory Exporters in Go

Testing Context Propagation

Testing Custom OpenTelemetry Plugins

Running Instrumentation Tests in CI

Read more

Test Result Reporting and Failing Fast in CI Pipelines

SOC 2 Evidence Collection Automation: Stop Emailing Screenshots to Your Auditor

PCI-DSS 4.0 Penetration Testing: Scoping, Methodology, and What Assessors Actually Check

CI/CD Pipeline Testing: GitHub Actions, GitLab CI, and Jenkins Patterns