Observability

Validating Distributed Traces: Completeness, Attributes, and Jaeger Integration

HelpMeTest

HelpMeTest

— 7 min read
Validating Distributed Traces: Completeness, Attributes, and Jaeger Integration

A distributed trace is only useful if it's complete. Missing spans, incorrect parent-child relationships, dropped attributes — these aren't just observability gaps, they're signals that your system isn't behaving the way you think. Validating traces in tests catches these problems before they mislead you in production. This post covers the full toolkit: completeness assertions, attribute validation, sampling strategy, and running Jaeger or Zipkin in your test environment.

What "Trace Completeness" Actually Means

A complete trace has four properties:

  1. Every expected service contributes a span — if a user request touches an API gateway, an order service, a payment service, and a database, the trace should contain spans from all four.
  2. Parent-child relationships are correct — the payment span should be a child of the order span, not a root span floating independently.
  3. Spans have the expected attributes — service name, HTTP status, database statement, error code.
  4. No spans are orphaned — every non-root span has a parent that exists in the same trace.

Testing for these properties requires more than checking that spans exist — you need to validate the shape of the trace tree.

Setting Up Jaeger for Integration Tests

Jaeger ships an all-in-one container that stores traces in memory, making it ideal for integration test environments. It exposes both the OTLP receiver and a query API you can assert against.

# docker-compose.test.yml
services:
  jaeger:
    image: jaegertracing/all-in-one:1.54
    environment:
      COLLECTOR_OTLP_ENABLED: "true"
    ports:
      - "4317:4317"   # OTLP gRPC
      - "4318:4318"   # OTLP HTTP
      - "16686:16686" # Jaeger UI
      - "16685:16685" # Jaeger query gRPC

  app:
    build: .
    environment:
      OTEL_EXPORTER_OTLP_ENDPOINT: http://jaeger:4317
      OTEL_SERVICE_NAME: order-service
      OTEL_TRACES_SAMPLER: always_on
    depends_on:
      jaeger:
        condition: service_started

Wait for Jaeger to be ready before running tests:

#!/bin/bash
# wait-for-jaeger.sh
until curl -sf http://localhost:16686/api/services > /dev/null; do
  sleep 1
done
echo "Jaeger ready"

Querying Jaeger in Tests

Jaeger's HTTP API lets you fetch traces by service, operation, or trace ID. Wrap it in a small client for your tests:

# jaeger_client.py
import requests
import time
from dataclasses import dataclass
from typing import List, Optional

@dataclass
class Span:
    span_id: str
    parent_span_id: Optional[str]
    operation_name: str
    service_name: str
    tags: dict
    duration_us: int
    logs: list

@dataclass
class Trace:
    trace_id: str
    spans: List[Span]

    def span_by_operation(self, name: str) -> Optional[Span]:
        return next((s for s in self.spans if s.operation_name == name), None)

    def children_of(self, span: Span) -> List[Span]:
        return [s for s in self.spans if s.parent_span_id == span.span_id]

    def root_span(self) -> Optional[Span]:
        return next((s for s in self.spans if s.parent_span_id is None), None)


class JaegerClient:
    def __init__(self, base_url: str = "http://localhost:16686"):
        self.base_url = base_url

    def get_traces(self, service: str, operation: str = None,
                   limit: int = 20, lookback: str = "1h") -> List[Trace]:
        params = {"service": service, "limit": limit, "lookback": lookback}
        if operation:
            params["operation"] = operation

        resp = requests.get(f"{self.base_url}/api/traces", params=params)
        resp.raise_for_status()
        data = resp.json()

        traces = []
        for trace_data in data.get("data", []):
            process_map = {
                pid: proc["serviceName"]
                for pid, proc in trace_data.get("processes", {}).items()
            }
            spans = []
            for s in trace_data["spans"]:
                tags = {t["key"]: t["value"] for t in s.get("tags", [])}
                parent_id = None
                for ref in s.get("references", []):
                    if ref["refType"] == "CHILD_OF":
                        parent_id = ref["spanID"]
                        break
                spans.append(Span(
                    span_id=s["spanID"],
                    parent_span_id=parent_id,
                    operation_name=s["operationName"],
                    service_name=process_map.get(s["processID"], "unknown"),
                    tags=tags,
                    duration_us=s["duration"],
                    logs=s.get("logs", [])
                ))
            traces.append(Trace(trace_id=trace_data["traceID"], spans=spans))

        return traces

    def wait_for_trace(self, service: str, operation: str,
                       timeout: int = 10) -> Trace:
        deadline = time.time() + timeout
        while time.time() < deadline:
            traces = self.get_traces(service, operation, limit=1, lookback="5m")
            if traces:
                return traces[0]
            time.sleep(0.5)
        raise TimeoutError(f"No trace found for {service}/{operation} within {timeout}s")

Trace Completeness Assertions

# test_order_trace_completeness.py
import pytest
from jaeger_client import JaegerClient

jaeger = JaegerClient()

def test_order_request_produces_complete_trace(http_client):
    response = http_client.post("/api/orders", json={"item": "widget", "qty": 2})
    assert response.status_code == 201

    trace = jaeger.wait_for_trace("order-service", "POST /api/orders")

    # All expected services contributed spans
    services_in_trace = {s.service_name for s in trace.spans}
    expected_services = {"api-gateway", "order-service", "payment-service", "inventory-service"}
    missing = expected_services - services_in_trace
    assert not missing, f"Missing spans from services: {missing}"

def test_trace_has_no_orphaned_spans(http_client):
    http_client.get("/api/products/123")
    trace = jaeger.wait_for_trace("product-service", "GET /api/products/{id}")

    span_ids = {s.span_id for s in trace.spans}
    root = trace.root_span()
    assert root is not None, "Trace has no root span"

    for span in trace.spans:
        if span.span_id == root.span_id:
            continue
        assert span.parent_span_id in span_ids, (
            f"Span '{span.operation_name}' has parent {span.parent_span_id} "
            f"which does not exist in trace"
        )

def test_payment_span_is_child_of_order_span(http_client):
    http_client.post("/api/orders", json={"item": "gadget", "qty": 1})
    trace = jaeger.wait_for_trace("order-service", "POST /api/orders")

    order_span = trace.span_by_operation("process_order")
    payment_span = trace.span_by_operation("charge_payment")

    assert order_span is not None, "process_order span not found"
    assert payment_span is not None, "charge_payment span not found"
    assert payment_span.parent_span_id == order_span.span_id, (
        "charge_payment is not a child of process_order"
    )

Span Attribute Validation

Beyond completeness, you need to verify that spans carry the right attributes. This matters for alerting (your SLO dashboards read span attributes) and for debugging (missing attributes mean incomplete context in the trace).

def test_http_spans_carry_required_attributes(http_client):
    http_client.get("/api/products/42")
    trace = jaeger.wait_for_trace("product-service", "GET /api/products/{id}")

    http_spans = [s for s in trace.spans if "http.method" in s.tags]

    for span in http_spans:
        # OTel semantic conventions for HTTP spans
        assert "http.method" in span.tags, f"Span {span.operation_name} missing http.method"
        assert "http.status_code" in span.tags, f"Span {span.operation_name} missing http.status_code"
        assert "http.url" in span.tags or "http.target" in span.tags, (
            f"Span {span.operation_name} missing http.url/http.target"
        )
        assert "net.peer.name" in span.tags or "server.address" in span.tags, (
            f"Span {span.operation_name} missing peer address"
        )

def test_db_spans_include_statement_type(http_client):
    http_client.get("/api/users/5/orders")
    trace = jaeger.wait_for_trace("order-service", "GET /api/users/{id}/orders")

    db_spans = [s for s in trace.spans if s.tags.get("db.system")]

    assert len(db_spans) > 0, "No database spans found"

    for span in db_spans:
        assert span.tags.get("db.system") in ("postgresql", "mysql", "sqlite", "redis"), (
            f"Unknown db.system: {span.tags.get('db.system')}"
        )
        assert "db.operation" in span.tags, f"DB span missing db.operation"
        # db.statement should be present but may be redacted
        assert "db.statement" in span.tags or span.tags.get("db.statement.redacted") == "true"

def test_error_spans_include_exception_details(http_client):
    response = http_client.get("/api/products/99999")  # non-existent
    assert response.status_code == 404

    trace = jaeger.wait_for_trace("product-service", "GET /api/products/{id}")
    root = trace.root_span()

    assert root.tags.get("error") is True or root.tags.get("otel.status_code") == "ERROR"
    assert root.tags.get("http.status_code") == 404

    # Exception log event should be present
    error_logs = [log for log in root.logs if any(
        field["key"] == "event" and field["value"] == "exception"
        for field in log.get("fields", [])
    )]
    assert len(error_logs) > 0, "No exception event logged on error span"

Zipkin Integration

If your stack uses Zipkin, the query pattern is similar. Zipkin's API is REST-based and traces are queryable by service name and span name.

# zipkin_client.py
import requests
import time
from typing import List, Optional

class ZipkinClient:
    def __init__(self, base_url: str = "http://localhost:9411"):
        self.base_url = base_url

    def get_traces(self, service: str, span_name: str = None,
                   limit: int = 10, lookback_ms: int = 3_600_000) -> list:
        params = {
            "serviceName": service,
            "limit": limit,
            "lookback": lookback_ms,
            "endTs": int(time.time() * 1000)
        }
        if span_name:
            params["spanName"] = span_name

        resp = requests.get(f"{self.base_url}/api/v2/traces", params=params)
        resp.raise_for_status()
        return resp.json()

    def assert_span_exists(self, service: str, span_name: str, timeout: int = 10):
        deadline = time.time() + timeout
        while time.time() < deadline:
            traces = self.get_traces(service, span_name, limit=1, lookback_ms=60_000)
            if traces:
                return traces[0]
            time.sleep(0.5)
        raise AssertionError(f"Span '{span_name}' from '{service}' not found in Zipkin")
# docker-compose.test.yml (Zipkin variant)
services:
  zipkin:
    image: openzipkin/zipkin:3
    ports:
      - "9411:9411"

  app:
    environment:
      OTEL_EXPORTER_ZIPKIN_ENDPOINT: http://zipkin:9411/api/v2/spans
      OTEL_TRACES_EXPORTER: zipkin
      OTEL_TRACES_SAMPLER: always_on

Sampling Strategy for Tests

In production you sample to control volume. In tests you need 100% sampling — but you also need to avoid test traces polluting your production dashboards when running integration tests against staging.

Strategy 1: Separate collector endpoints per environment

# otel-collector-staging.yaml
exporters:
  jaeger:
    endpoint: jaeger-staging:14250
    tls:
      insecure: true

processors:
  filter:
    traces:
      span:
        # Drop traces tagged as test runs
        - 'attributes["test.run"] == true'

Strategy 2: Tag test traces and filter in dashboards

# conftest.py
from opentelemetry import baggage, context
from opentelemetry.baggage.propagation import W3CBaggagePropagator

@pytest.fixture(autouse=True)
def tag_test_traces():
    ctx = baggage.set_baggage("test.run", "true")
    token = context.attach(ctx)
    yield
    context.detach(token)

Strategy 3: Head-based sampling override in test config

from opentelemetry.sdk.trace.sampling import ParentBased, ALWAYS_ON

# Always sample in tests, regardless of parent decision
provider = TracerProvider(sampler=ALWAYS_ON)

Asserting on Trace Depth and Shape

Some bugs only manifest as incorrect trace structure — extra layers of indirection, missing branches for error paths, unexpected fan-out.

def test_trace_depth_does_not_exceed_budget(http_client):
    http_client.post("/api/checkout", json={"cart_id": "cart-1"})
    trace = jaeger.wait_for_trace("checkout-service", "POST /api/checkout")

    def depth(span_id: str, all_spans: list, current: int = 0) -> int:
        children = [s for s in all_spans if s.parent_span_id == span_id]
        if not children:
            return current
        return max(depth(c.span_id, all_spans, current + 1) for c in children)

    root = trace.root_span()
    max_depth = depth(root.span_id, trace.spans)

    # Checkout should not fan out more than 4 levels deep
    assert max_depth <= 4, f"Trace depth {max_depth} exceeds budget of 4"

def test_payment_failure_does_not_trigger_fulfillment_span(http_client, mock_payment_gateway):
    mock_payment_gateway.fail_next()
    response = http_client.post("/api/checkout", json={"cart_id": "cart-2"})
    assert response.status_code == 402

    trace = jaeger.wait_for_trace("checkout-service", "POST /api/checkout")

    fulfillment_span = trace.span_by_operation("create_fulfillment_order")
    assert fulfillment_span is None, (
        "Fulfillment span should not appear when payment fails"
    )

CI Pipeline Integration

# .github/workflows/integration-test.yml
name: Integration Tests

on: [push]

jobs:
  integration:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Start services
        run: docker compose -f docker-compose.test.yml up -d

      - name: Wait for Jaeger
        run: |
          timeout 30 bash -c 'until curl -sf http://localhost:16686/api/services; do sleep 1; done'

      - name: Run integration tests
        run: pytest tests/integration/ -v

      - name: Export failed traces
        if: failure()
        run: |
          curl -s "http://localhost:16686/api/traces?service=order-service&limit=50" \
            > /tmp/failed-traces.json

      - uses: actions/upload-artifact@v3
        if: failure()
        with:
          name: traces
          path: /tmp/failed-traces.json

Conclusion

Trace validation transforms observability from a passive post-incident tool into an active correctness guarantee. When you assert that a trace is complete, that spans carry the right attributes, and that the parent-child structure reflects your design, you catch a class of bugs that no unit test can reach: missing instrumentation, context propagation failures, and architectural drift. Setting up Jaeger or Zipkin in your test environment takes under an hour — the return is a suite of tests that verifies not just what your code returns, but how it got there.

Read more

Start now free