Real-Time Telemetry Analyzer

Turn Telemetry Questions into Requirements

A real-time telemetry analyzer is the control room for a busy transport network. Individual signals—requests, errors, device readings, and latency measurements—arrive continuously. The analyzer cannot keep every signal in fast memory, yet it must answer operational questions quickly: How many events arrived in the last minute? How many services were active? Which keys dominate errors? Is p99 latency breaching its objective? Is the answer complete enough to alert on?

Suppose five events arrive with event-time seconds and latency milliseconds: (0, api, 40), (8, api, 60), (12, db, 90), (18, api, 50), (25, cache, 20). At event time 30, the half-open window [10,30) contains three events, three distinct keys, api as a joint top key, and sorted latencies [20,50,90]. The nearest-rank p50 is 50 ms. If the event at time 12 arrives after the watermark has passed 17 with five seconds of allowed lateness, the system must apply its documented late-data policy rather than quietly rewriting history.

An event time is when the source says the event occurred; processing time is when the analyzer handles it. A watermark is a claim that events earlier than a point are unlikely to arrive, not a clock and not proof of completeness. A window groups event times; a rate divides an accepted count by a stated duration. Cardinality counts distinct keys, heavy hitters identify high-frequency keys, and a quantile answers a rank question about a distribution. An alert is a policy decision over a versioned query result, not simply a red chart.

Stakeholders include application owners, SREs, platform engineers, security/privacy reviewers, finance teams, and incident commanders. Acceptance requires bounded state, deterministic replay, defined late and duplicate handling, stable versioned schemas, calibrated sketch error, query p99 below the service target, no cross-tenant leakage, recoverable checkpoints, and alerts whose evidence can be audited.

Design the Ingestion, Window and Alert Architecture

The target workload is multi-tenant telemetry at 500,000 events per second, 5,000 active streams, 60-second tumbling and 5-minute sliding views, and queries below 100 ms p99. The event contract is {tenant_id, stream_id, event_id, event_time, key, value, attributes, schema_version}. The gateway authenticates, validates a 64 KiB size limit, redacts forbidden attributes, applies quotas, and writes to a replayable partition log. Partition workers use event time, a five-second allowed-lateness default, idempotent checkpoint epochs, and per-tenant memory budgets.

The ingestion path updates aligned bucketed counters, HLL registers for cardinality, a Space-Saving candidate summary backed by a Count-Min estimate for top keys, and KLL-style quantile summaries. Window queries merge only compatible bucket generations. Alert evaluation consumes query results only when watermark freshness, coverage, and error metadata satisfy policy. Operators change versioned configuration through a separate control plane. Snapshots record algorithm identifiers, dimensions, seed IDs, schema versions, bucket boundaries, watermark, checkpoint epoch, payload checksum, and encryption metadata.

The threat model includes malformed events, cardinality attacks, crafted hash collisions, replay duplication, clock skew, cross-tenant queries, snapshot tampering, poisoned alert rules, and compromised operator tokens. Privacy rules minimize raw attributes, hash or tokenize permitted keys, restrict drill-down, audit sensitive queries, and cap retention. The service explicitly excludes forensic raw-event search; durable object storage and a governed warehouse own that use case.

flowchart TD
    P[Telemetry producers] --> G[Auth validate redact]
    G --> L[Partitioned replay log]
    L --> W[Event-time workers]
    W --> C[Window counters]
    W --> H[HLL cardinality]
    W --> F[Heavy hitters]
    W --> Q[Quantile sketches]
    C --> A[Query and alert engine]
    H --> A
    F --> A
    Q --> A
    W --> S[Versioned snapshots]
    A --> N[Dashboards and notifications]
    O[Operators] --> R[Rules and rollout control]
    R --> A
    W --> M[Metrics traces audit]
    classDef navy fill:#17365D,color:#fff,stroke:#102844;
    classDef teal fill:#147D84,color:#fff,stroke:#0c5c61;
    classDef crimson fill:#C6284D,color:#fff,stroke:#8e1935;
    classDef purple fill:#6C3BAA,color:#fff,stroke:#482675;
    class P,O navy;
    class G,L,W,C teal;
    class H,F,Q,S purple;
    class A,N,R,M crimson;

Integrate Summaries into a Live Analyzer

Build vertically. Slice 1 accepts a typed event, assigns it to fixed event-time buckets, and answers exact count/rate queries with an exact replay oracle. Slice 2 adds watermarking, allowed lateness, deduplication, checkpoint/restore, and finalization. Slice 3 adds HLL, top-key summaries, and quantile sketches behind a shared result envelope. Slice 4 evaluates alert rules only from results that pass freshness and quality gates. Each slice has its own demo, memory ceiling, and rollback boundary.

The stable contracts are ingest(event), advance_watermark(time), query(metric, start, end), snapshot(), and restore(bytes). Configuration validates window alignment, lateness, bucket count, sketch dimensions, tenant budget, and seed identity before allocation. Snapshot format includes a checksum and an explicit version; adjacent migrations are named and tested, while unsupported versions trigger replay or a visible STATE_INCOMPATIBLE condition. Safe fallback may serve a labeled stale result, reduce optional dimensions, or reject work—never fabricate completeness.

from dataclasses import dataclass
from collections import Counter, defaultdict
from typing import Iterable

@dataclass(frozen=True)
class Config:
    bucket_seconds: int = 10
    retention_seconds: int = 60
    allowed_lateness: int = 5
    max_keys_per_bucket: int = 1000
    format_version: int = 1

    def validate(self) -> None:
        if self.bucket_seconds <= 0 or self.retention_seconds % self.bucket_seconds:
            raise ValueError("retention must be a positive multiple of bucket size")
        if not 0 <= self.allowed_lateness <= self.retention_seconds:
            raise ValueError("invalid allowed lateness")
        if not 1 <= self.max_keys_per_bucket <= 100_000:
            raise ValueError("unsafe key budget")
        if self.format_version != 1:
            raise ValueError("unsupported format version")

@dataclass(frozen=True)
class Event:
    event_id: str
    event_time: int
    key: str
    latency_ms: float

class Analyzer:
    def __init__(self, config: Config):
        config.validate()
        self.config = config
        self.watermark = -1
        self.buckets: dict[int, list[Event]] = defaultdict(list)
        self.seen: set[str] = set()
        self.late_events = 0

    def ingest(self, event: Event) -> str:
        if not event.event_id or not event.key or event.event_time < 0 or event.latency_ms < 0:
            raise ValueError("invalid event")
        if event.event_id in self.seen:
            return "duplicate"
        if event.event_time < self.watermark - self.config.allowed_lateness:
            self.late_events += 1
            return "too_late"
        start = event.event_time - event.event_time % self.config.bucket_seconds
        keys = {e.key for e in self.buckets[start]}
        if event.key not in keys and len(keys) >= self.config.max_keys_per_bucket:
            raise MemoryError("per-bucket key budget exceeded")
        self.seen.add(event.event_id)
        self.buckets[start].append(event)
        return "accepted"

    def advance_watermark(self, value: int) -> None:
        if value < self.watermark:
            raise ValueError("watermark must be monotonic")
        self.watermark = value
        cutoff = value - self.config.retention_seconds - self.config.allowed_lateness
        self.buckets = defaultdict(list, {k: v for k, v in self.buckets.items()
                                          if k + self.config.bucket_seconds > cutoff})

    def query(self, start: int, end: int) -> dict:
        if start >= end or start % self.config.bucket_seconds or end % self.config.bucket_seconds:
            raise ValueError("query must use aligned half-open boundaries")
        events: list[Event] = []
        for bucket in range(start, end, self.config.bucket_seconds):
            events.extend(e for e in self.buckets.get(bucket, []) if start <= e.event_time < end)
        counts = Counter(e.key for e in events)
        values = sorted(e.latency_ms for e in events)
        p50 = values[(len(values) - 1) // 2] if values else None
        return {"window": [start, end], "count": len(events), "rate": len(events) / (end-start),
                "cardinality": len(counts), "top": counts.most_common(3), "p50_ms": p50,
                "watermark": self.watermark, "complete": self.watermark >= end + self.config.allowed_lateness,
                "algorithm": "exact_reference", "format_version": self.config.format_version}

def replay(events: Iterable[Event], cfg: Config) -> Analyzer:
    analyzer = Analyzer(cfg)
    for event in events:
        analyzer.ingest(event)
    return analyzer

def self_test() -> None:
    cfg = Config()
    events = [Event("1",0,"api",40), Event("2",8,"api",60), Event("3",12,"db",90),
              Event("4",18,"api",50), Event("5",25,"cache",20)]
    a = replay(events, cfg)
    a.advance_watermark(35)
    result = a.query(10, 30)
    assert (result["count"], result["cardinality"], result["p50_ms"]) == (3, 3, 50)
    assert result["complete"] is True
    assert a.ingest(events[0]) == "duplicate"
    assert a.ingest(Event("6",20,"api",30)) == "too_late"
    try:
        a.advance_watermark(34)
    except ValueError:
        pass
    else:
        raise AssertionError("watermark regression accepted")
    try:
        Config(bucket_seconds=7, retention_seconds=60).validate()
    except ValueError:
        pass
    else:
        raise AssertionError("unsafe configuration accepted")
    print("integration checks passed", result)

if __name__ == "__main__":
    self_test()

flowchart TD
    C[Typed event contract] --> E[Exact bucket baseline]
    E --> W[Watermark and lateness]
    W --> S[Checkpoint and replay]
    S --> H[HLL and heavy hitters]
    H --> Q[Quantile summaries]
    Q --> A[Alert quality gate]
    A --> K[Shadow canary]
    K -->|SLO and oracle pass| P[Progressive promotion]
    K -->|Drift or invariant fail| R[Rollback generation]
    R --> E
    classDef navy fill:#17365D,color:#fff,stroke:#102844;
    classDef teal fill:#147D84,color:#fff,stroke:#0c5c61;
    classDef crimson fill:#C6284D,color:#fff,stroke:#8e1935;
    classDef purple fill:#6C3BAA,color:#fff,stroke:#482675;
    class C,E navy;
    class W,S,H,Q teal;
    class A,K purple;
    class P,R crimson;

Replay Incidents and Prove Operational Readiness

Verification separates algorithm error from pipeline error. Unit tests cover boundary assignment, watermark monotonicity, parameter calculators, serialization, and rule evaluation. Property tests generate streams to assert that accepted events land in exactly the documented windows, finalized windows do not change, HLL error is calibrated, heavy-hitter candidates satisfy the configured threshold contract, and observed quantile rank error stays within its stated envelope. Differential replay compares every approximate result with exact sets, maps, and sorted arrays.

The approachable scenario is the five-event trace plus one duplicate and one late arrival. The professional workload is 500,000 events/s across 5,000 streams, 60-second and 5-minute windows, Zipf-skewed keys, bimodal latency, and 5% out-of-order delivery. SLOs are 99.9% query availability, p99 under 100 ms, ingestion acceptance above 99.99% for valid in-quota events, alert evaluation within 10 seconds of window finalization, memory below 4 GiB per worker, and recovery below 10 minutes. The cost budget is tracked per million events and per active stream.

Failure injection kills workers during checkpoints, delays one partition, corrupts snapshot bytes, removes a notification sink, rotates a hash seed incorrectly, advances a source clock by two minutes, and floods unique keys. Metrics expose ingest lag, watermark delay, late/duplicate/rejected counts, state bytes, sketch saturation, estimate/oracle deltas, checkpoint age, alert suppression reason, and notification failures. Traces connect ingest, bucket update, query, rule, and notification. Audit logs preserve configuration, access, alert evidence, acknowledgement, rollback, and owner.

Ship the Analyzer with a Runbook

The release bundle includes the API and event schemas, window/lateness semantics, result-envelope examples, configuration calculators, a pinned environment, seeded replay datasets, benchmark commands and raw outputs, the correctness dossier, migration matrix, security/privacy review, dashboards, alert rules, and an operational runbook. Each artifact names its version, owner, generation command, and review date.

The runbook starts from symptoms—watermark delay, ingest rejection, state growth, oracle drift, stale checkpoints, notification failure—and gives diagnostic queries, decision thresholds, safe mitigations, escalation paths, rollback commands, and evidence-preservation steps. It distinguishes late source data from pipeline lag and true service degradation. The demonstration ingests the five-event trace, shows a complete result envelope, introduces a duplicate and late event, compares approximate summaries with exact oracles, blocks a stale alert, rejects incompatible state, restores a valid checkpoint, and ends on the professional workload dashboard.

Peer reviewers ask: Are half-open boundaries and finalization unambiguous? Can event-time and processing-time failures be distinguished? Do sketch guarantees survive bucket merge? Can a caller tell error, staleness, and partial coverage apart? Does high cardinality degrade safely? Are sampled attributes and key identities protected? Handoff requires a tagged build, checksums, one-command verification, benchmark provenance, signed dossier, rehearsed recovery/rollback, on-call ownership, and no undocumented exception.