How to Implement Tail Sampling in Python

import random
import time
from collections import deque

class TailSampler:
    def __init__(self, tail_ratio=0.1, max_samples=100):
        self.tail_ratio = tail_ratio
        self.max_samples = max_samples
        self.samples = deque(maxlen=max_samples)
        self.total_calls = 0

    def record(self, latency_ms):
        self.total_calls += 1
        # Sample from the tail: the slowest tail_ratio of calls
        if random.random() < self.tail_ratio and self.total_calls > 10:
            self.samples.append((time.time(), latency_ms, self.total_calls))

    def get_samples(self):
        # Return the sampled tail calls, sorted by latency descending
        return sorted(self.samples, key=lambda x: x[1], reverse=True)

    def summary(self):
        if not self.samples:
            return "No tail samples yet"
        avg_latency = sum(s[1] for s in self.samples) / len(self.samples)
        return f"Tail samples: {len(self.samples)}, avg latency: {avg_latency:.2f}ms, total calls: {self.total_calls}"

# Mock usage
if __name__ == "__main__":
    sampler = TailSampler(tail_ratio=0.2, max_samples=5)
    # Simulate 50 calls with varying latency
    for i in range(50):
        latency = random.gauss(100, 30)  # base latency around 100ms
        if random.random() < 0.1:  # occasional slow calls
            latency *= random.uniform(2, 4)
        sampler.record(latency)
    print(sampler.summary())
    for sample in sampler.get_samples()[:3]:  # show top 3 slowest
        print(f"latency={sample[1]:.2f}ms, call#{sample[2]}")