How to Simulate Trace Sampling Head in Python

Simulate head-based probabilistic trace sampling on mock trace data with a configurable sample rate and optional seed for reproducibility.

Easy Python 3.9+ Aug 9, 2026 Observability & SRE 12 views 0 copies

Python code

46 lines
Python 3.9+
import random

def trace_sampling_head(mock_traces, sample_rate=0.5, seed=None):
    """Simulate probabilistic trace sampling (head-based) on mock data.
    
    Args:
        mock_traces: list of trace dictionaries with a unique 'trace_id'
        sample_rate: float 0.0-1.0, probability of keeping a trace
        seed: optional seed for reproducibility
    Returns:
        tuple of (sampled_traces, dropped_traces)
    """
    if seed is not None:
        random.seed(seed)

    sampled = []
    dropped = []
    for trace in mock_traces:
        if random.random() < sample_rate:
            sampled.append(trace)
        else:
            dropped.append(trace)
    return sampled, dropped


if __name__ == "__main__":
    mock_traces = [
        {"trace_id": "trace-1001", "service": "auth"},
        {"trace_id": "trace-1002", "service": "payment"},
        {"trace_id": "trace-1003", "service": "database"},
        {"trace_id": "trace-1004", "service": "api-gateway"},
        {"trace_id": "trace-1005", "service": "cache"},
        {"trace_id": "trace-1006", "service": "worker"},
        {"trace_id": "trace-1007", "service": "auth"},
        {"trace_id": "trace-1008", "service": "notification"},
    ]

    sampled, dropped = trace_sampling_head(mock_traces, sample_rate=0.4, seed=42)

    print(f"Total traces: {len(mock_traces)}")
    print(f"Sampled ({len(sampled)}):")
    for t in sampled:
        print(f"  {t['trace_id']} ({t['service']})")
    print(f"Dropped ({len(dropped)}):")
    for t in dropped:
        print(f"  {t['trace_id']} ({t['service']})")

Output

stdout
Total traces: 8
Sampled (3):
  trace-1001 (auth)
  trace-1003 (database)
  trace-1007 (auth)
Dropped (5):
  trace-1002 (payment)
  trace-1004 (api-gateway)
  trace-1005 (cache)
  trace-1006 (worker)
  trace-1008 (notification)

How it works

This function implements head-based sampling, where each trace is independently kept or dropped based on a random comparison against the sample_rate. Using random.seed() makes the selection reproducible across runs, which is useful for testing and debugging. Traces that pass the sampling check are added to sampled, while the rest go into dropped, keeping the two lists mutually exclusive. The function returns both lists so callers can inspect or forward the sampled traces and optionally log the dropped ones.

Common mistakes

  • Forgetting that `random.random()` returns values in [0, 1), so the sample_rate should be compared strictly with `<` to match typical probability semantics.
  • Not seeding the random generator when reproducibility is required, leading to flaky test results.
  • Mutating the original `mock_traces` list instead of building new lists for sampled and dropped traces.

Variations

  1. Use a sampling decision based on trace ID hash (e.g., `hash(trace_id) % 100 < sample_rate * 100`) for consistent sampling across distributed nodes.
  2. Return only the sampled traces and use a separate counter for dropped ones to reduce memory overhead when the drop list isn't needed.

Real-world use cases

  • Simulating distributed tracing sampling policies before rolling them out to production to estimate storage and cost impact.
  • Building a mock observability pipeline for local development or tests where you need deterministic, controlled trace volumes.
  • Benchmarking downstream analytics or alerting tools by feeding them a representative subset of traces under realistic sampling rates.

Sponsored

Run this sample

Open the browser IDE to tweak the example and see results without installing anything.

Open editor

More from Observability & SRE

Related tutorials and quizzes for this topic.