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.
Python code
46 linesimport 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
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
- Use a sampling decision based on trace ID hash (e.g., `hash(trace_id) % 100 < sample_rate * 100`) for consistent sampling across distributed nodes.
- 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
More from Observability & SRE
- Adding a Correlation ID to Log Context in Python medium
- Calculate Error Rate from Log Stream in Python easy
- Check if a Timestamp Falls in a Daily Maintenance Window in Python easy
- Export Metrics with OTLP Mock in Python medium
- Generate Mock CPU and Memory Metrics in Python easy
- Generate Prometheus Text Exposition Format in Python easy
Keep learning
Related tutorials and quizzes for this topic.