Compare sketch, exact and store-and-query in the cardinality demo as traffic grows

.gitignore

@@ -22,3 +22,6 @@ target
 /otel-arrow/
 .cache/
 benchmark-results/
+
+# Python bytecode
+__pycache__/

deploy/telemetry-functions/README.md

@@ -30,7 +30,7 @@ of scenario time, so all three pipelines receive identical points.
 | exact-merge | `exact_merge`: adds both sources' exact windows and emits each as soon as both have sent it (5 s timeout) |
 | exact-detector | `exact_detector`: the same answers and events from the exact windows (`method="exact"`): the ground truth |
 | raw-sink-a/b | `raw_sink`, the only stage per source: writes every raw point as one log record to `loki-raw` (4 requests in flight) and reports when `loki-raw` acknowledged each window's last point (`source` label), as soon as its source has sent a later point and every message of the window is acknowledged (5 s timeout if the source idles) |
-| query-job | `query_job.py`: once per window, times each approach's backend query (`asap_query_seconds`, labelled `method` and `backend`); in the top-k use case it also runs the LogQL top-20 over `loki-raw` and reports that answer and its heavy-hitter accuracy against exact (`method="raw"`) |
+| query-job | `query_job.py`: once per window, times each approach's backend query (`asap_query_seconds`, labelled `method` and `backend`); it also answers from `loki-raw` with LogQL (`method="raw"`): in the top-k use case the top-20 and its heavy-hitter accuracy against exact, in the cardinality use case the distinct count of every attribute key (`asap_query_failed` when Loki refuses a query) |
 | hh-accuracy | `heavy_hitter_accuracy`, outside the pipelines: in the top-k use case both detectors send it a copy of their answers; per window it reports heavy-hitter precision, recall, F1 and average relative count error |
 
 **Heavy-hitter accuracy** (top-k use case). A heavy hitter is a route with at
@@ -46,7 +46,9 @@ a window whose exact top 20 is all above the threshold only gets
 
 Each dashboard also shows every pipeline's throughput: the points per window
 that reached the end of the pipeline (`asap_points`; for raw, the points
-`loki-raw` acknowledged) divided by the window length.
+`loki-raw` acknowledged within 50 s of the window's end, after which raw-sink
+reports the window anyway) divided by the window length. A pipeline that stops
+reporting shows a gap after 2 minutes rather than its last value.
 
 **Latency breakdown.** Merges and detectors report, per window, when they
 received its last input and when they emitted it, in seconds after the window's
@@ -83,14 +85,19 @@ reports each point's time from its event time to that acknowledgement
 (`asap_raw_transit`, p50 and p99) and points `loki-raw` rejected
 (`asap_raw_failed`).
 
+The top-k and cardinality dashboards are generated by
+`grafana/gen_dashboards.py` (`python3 grafana/gen_dashboards.py topk
+grafana/dashboards/topk.json`, likewise `cardinality`); edit the script, not
+the JSON.
+
 ## Use cases
 
 `USE_CASE` selects the stage configs in the directory of the same name.
 
 | `USE_CASE` | Sketch | Scenario | Dashboard |
 | --- | --- | --- | --- |
 | `latency` | KLL per route | latency over 20 routes | `/d/asap-latency`: p50/p99 with bound vs exact |
-| `cardinality` | HLL per attribute key (`per_attribute: ["*"]`) | `user.id` unique, `k8s.pod.name` growing 5%/min, `http.route` and `http.method` fixed | `/d/asap-cardinality`: distinct values with bound, class per key (stable, growing, identifier-like), events |
+| `cardinality` | HLL per attribute key (`per_attribute: ["*"]`) | `user.id` unique, `k8s.pod.name` growing 5%/min, `http.route` and `http.method` fixed; the rate per source doubles every 5 minutes (5k to 80k points/s) and `http.route` explodes from 20 to 20,000 values at minute 10 | `/d/asap-cardinality`: distinct values per key of all three approaches side by side (with error against exact and each detector's class), class events, relative error over time, throughput and distinct values over time, freshness breakdown, CPU/memory/bits per component of each approach |
 | `drift` | KLL per route | `http.route-3`'s median triples at minute 10 | `/d/asap-drift`: KS distance from the last 5 windows, drift events |
 | `topk` | Count-Min (4 x 512) + top-20 heap per attribute key | Zipf routes; the rate per source doubles every 5 minutes (5k to 80k points/s), a new route takes 15% of traffic at minute 10, routes grow from 100,000 to 1,000,000 at minute 15 | `/d/asap-topk`: top-10 of all three approaches side by side (one table: rank, count and error against exact per approach), heavy-hitter precision/recall/F1/ARE, throughput and distinct routes over time, freshness breakdown, CPU/memory per component and per approach, bits per second into each stage |
 
@@ -135,7 +142,7 @@ fails. The checks:
 | --- | --- |
 | all | every stage up; points per window equal the offered rate (no backpressure); sketch and exact windows hold the same points; every pipeline reports its window completion delay (the raw pipeline's points per window are reported, not required); merges and detectors report their timing; the query job times the sketch and exact backend queries |
 | latency | exact p50/p99 inside the sketch's bound (routes with at least 2,000 points) |
-| cardinality | sketch and exact give every key the same class; the generated classes are found; distinct-count error at most 2%; class events in Loki |
+| cardinality | sketch and exact give every key the same class; the generated classes are found; distinct-count error at most 2%; store and query answers every key from `loki-raw` within 1% of exact; class events in Loki |
 | drift | the changed route drifts in both methods; no other route does; drift event in Loki |
 | topk | the new route enters both top-10s; sketch recall@10 at least 0.9; heavy-hitter precision and recall at least 0.9; store and query answers from `loki-raw` with heavy-hitter recall at least 0.9; novelty event in Loki |
 

deploy/telemetry-functions/acceptance.py

@@ -59,6 +59,8 @@ def scaled_copy(directory, into):
         # The 5 s timeout for a missing source is not scaled: merges emit as
         # soon as both sources sent a window, and shared CI runners can be slow.
         text = re.sub(r"\n\s*- \{after_seconds: \d+, (signals_per_second|attribute: [^,]+, cardinality): \d+\}", "", text)
+        # A `changes:` left with no entries would read as null, not an empty list.
+        text = re.sub(r"\n\s*changes:[ \t]*(?=\n(?!\s*-))", "", text)
         text = re.sub(r"after_seconds: (\d+)", lambda m: f"after_seconds: {int(m[1]) // SCALE}", text)
         text = re.sub(r"growth_per_minute: ([0-9.]+)",
                       lambda m: f"growth_per_minute: {float(m[1]) * SCALE:g}", text)
@@ -197,6 +199,17 @@ def cardinality_checks(stack, t):
             for k in keys if (k, "sketch") in distinct and (k, "exact") in distinct}
     yield check("distinct-count error <= 2% for every key", errs and max(errs.values()) <= 0.02,
                 ", ".join(f"{k} {e:.2%}" for k, e in errs.items()))
+    # Store and query answers later than exact: compare the latest window in
+    # which it answered every key exact did.
+    raw = {m["attribute_key"]: series for m, series in stack.matrix('asap_distinct{method="raw"}')}
+    exact = {m["attribute_key"]: series for m, series in stack.matrix('asap_distinct{method="exact"}')}
+    end = max((e for e in set.intersection(*[set(v) for v in exact.values()])
+               if all(e in raw.get(k, {}) for k in exact)), default=None) if exact else None
+    raw_errs = {k: abs(raw[k][end] - exact[k][end]) / exact[k][end] for k in exact} if end else {}
+    yield check("store and query: distinct counts from loki-raw, error <= 1% for every key",
+                raw_errs and max(raw_errs.values()) <= 0.01,
+                ", ".join(f"{k} {e:.2%}" for k, e in sorted(raw_errs.items()))
+                or f"no window with every key answered; never answered: {sorted(set(exact) - set(raw))}")
     events = stack.events('| event_kind="cardinality" | method="sketch"')
     yield check("class events in Loki", any("identifier-like" in e for e in events),
                 f"{len(events)} sketch events")

deploy/telemetry-functions/cardinality/generator-a.yaml

@@ -1,4 +1,8 @@
-# Source a: four attributes with known cardinality behaviour.
+# Source a: four attributes with known cardinality behaviour. Traffic grows
+# over 30 minutes so the three pipelines can be compared as load rises: the rate
+# per source doubles every 5 minutes (5,000 to 80,000 points/s), and at minute 10
+# http.route explodes from 20 to 20,000 values (as when an ID leaks into the
+# route).
 version: otel_dataflow/v1
 engine:
   http_admin:
@@ -17,7 +21,7 @@ groups:
             type: urn:asap:receiver:scenario_generator
             config:
               seed: 1
-              signals_per_second: 20000
+              signals_per_second: 5000
               metric_name: http.server.request.duration
               resource_attributes: {service.name: checkout, source: "a"}
               attributes:
@@ -26,6 +30,12 @@ groups:
                 - {key: k8s.pod.name, cardinality: 50, growth_per_minute: 0.05}  # growing
                 - {key: user.id, unique: true}                        # identifier-like
               value: {lognormal: {median: 50.0, sigma: 0.6}}
+              changes:
+                - {after_seconds: 300, signals_per_second: 10000}
+                - {after_seconds: 600, signals_per_second: 20000}
+                - {after_seconds: 900, signals_per_second: 40000}
+                - {after_seconds: 1200, signals_per_second: 80000}
+                - {after_seconds: 600, attribute: http.route, cardinality: 20000}
           export:
             type: exporter:otlp_http
             config:

deploy/telemetry-functions/cardinality/generator-b.yaml

@@ -1,4 +1,8 @@
-# Source b: four attributes with known cardinality behaviour.
+# Source b: four attributes with known cardinality behaviour. Traffic grows
+# over 30 minutes so the three pipelines can be compared as load rises: the rate
+# per source doubles every 5 minutes (5,000 to 80,000 points/s), and at minute 10
+# http.route explodes from 20 to 20,000 values (as when an ID leaks into the
+# route).
 version: otel_dataflow/v1
 engine:
   http_admin:
@@ -17,7 +21,7 @@ groups:
             type: urn:asap:receiver:scenario_generator
             config:
               seed: 2
-              signals_per_second: 20000
+              signals_per_second: 5000
               metric_name: http.server.request.duration
               resource_attributes: {service.name: checkout, source: "b"}
               attributes:
@@ -26,6 +30,12 @@ groups:
                 - {key: k8s.pod.name, cardinality: 50, growth_per_minute: 0.05}  # growing
                 - {key: user.id, unique: true}                        # identifier-like
               value: {lognormal: {median: 50.0, sigma: 0.6}}
+              changes:
+                - {after_seconds: 300, signals_per_second: 10000}
+                - {after_seconds: 600, signals_per_second: 20000}
+                - {after_seconds: 900, signals_per_second: 40000}
+                - {after_seconds: 1200, signals_per_second: 80000}
+                - {after_seconds: 600, attribute: http.route, cardinality: 20000}
           export:
             type: exporter:otlp_http
             config: