Counter vs. gauge under lost scrapes

# /// script
# requires-python = ">=3.10"
# ///
# Copyright (c) 2026 Aayush Sharma · SPDX-License-Identifier: MIT
# https://thedataengineering.com/lab-code/LICENSE.txt
"""Counter against gauge under lost scrapes.

A counter rises at 100 events per second; a queue-length gauge spikes to
10,000 for 5 seconds once a minute. Both are scraped every 15 s and every
60 s for an hour, with 30% of scrapes dropped at random, over many random
seeds: the rate computed from the counter, and whether the gauge's spike was
ever seen.
"""
import argparse, json, random, statistics

p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
p.add_argument("--loss", type=float, default=0.3, help="Share of scrapes dropped.")
p.add_argument("--intervals", default="15,60", help="Scrape intervals (s).")
p.add_argument("--seeds", type=int, default=200, help="Simulated hours (seeds).")
p.add_argument("--json", help="Also write the results to this JSON file.")
args = p.parse_args()

def rate_from_counter(samples):                                        # samples: list of (t, value) that survived
    (t0, v0), (t1, v1) = samples[0], samples[-1]
    return (v1 - v0) / (t1 - t0)
counter = lambda t: 100 * t                                            # accumulator
gauge = lambda t: 10_000 if (t % 60) < 5 else 50                       # spike for 5 s each minute
rows = []
for interval in map(int, args.intervals.split(",")):
    rates, seen, seen_share = [], 0, []
    for seed in range(args.seeds):
        random.seed(seed); offset = random.randrange(interval)          # scrapes are not aligned with the spike
        ts = [t for t in range(offset, 3600, interval) if random.random() > args.loss]
        rates.append(rate_from_counter([(t, counter(t)) for t in ts]))
        hits = sum(gauge(t) > 50 for t in ts); seen += hits > 0; seen_share.append(hits / 60)
    rows.append({"interval_s": interval, "rate_mean": statistics.mean(rates), "rate_min": min(rates), "rate_max": max(rates),
                 "hours_spike_seen_pct": seen / args.seeds * 100, "spikes_seen_pct": statistics.mean(seen_share) * 100})
for r in rows:
    print(f"every {r['interval_s']:>2} s, {args.loss:.0%} lost: counter rate {r['rate_min']:.1f}-{r['rate_max']:.1f}/s; "
          f"gauge spike seen in {r['hours_spike_seen_pct']:.0f}% of hours, {r['spikes_seen_pct']:.1f}% of the 60 spikes per hour")

if args.json:
    with open(args.json, "w") as f:
        json.dump({"tables": {"intervals": rows}}, f)