diff --git a/.gitignore b/.gitignore index ea0197d..b862d35 100644 --- a/.gitignore +++ b/.gitignore @@ -22,3 +22,6 @@ target /otel-arrow/ .cache/ benchmark-results/ + +# Python bytecode +__pycache__/ diff --git a/deploy/telemetry-functions/README.md b/deploy/telemetry-functions/README.md index edf098b..5dc2f9c 100644 --- a/deploy/telemetry-functions/README.md +++ b/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,6 +85,11 @@ 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. @@ -90,7 +97,7 @@ reports each point's time from its event time to that acknowledgement | `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 | diff --git a/deploy/telemetry-functions/acceptance.py b/deploy/telemetry-functions/acceptance.py index 678e013..0942e24 100755 --- a/deploy/telemetry-functions/acceptance.py +++ b/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") diff --git a/deploy/telemetry-functions/cardinality/generator-a.yaml b/deploy/telemetry-functions/cardinality/generator-a.yaml index 165b1d8..9c73023 100644 --- a/deploy/telemetry-functions/cardinality/generator-a.yaml +++ b/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: diff --git a/deploy/telemetry-functions/cardinality/generator-b.yaml b/deploy/telemetry-functions/cardinality/generator-b.yaml index 1ff2ac7..535afc6 100644 --- a/deploy/telemetry-functions/cardinality/generator-b.yaml +++ b/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: diff --git a/deploy/telemetry-functions/grafana/dashboards/cardinality.json b/deploy/telemetry-functions/grafana/dashboards/cardinality.json index c06d9f0..d6c3adf 100644 --- a/deploy/telemetry-functions/grafana/dashboards/cardinality.json +++ b/deploy/telemetry-functions/grafana/dashboards/cardinality.json @@ -1,116 +1,183 @@ { "uid": "asap-cardinality", - "title": "Attribute cardinality: sketch vs exact (real collector containers)", + "title": "Cardinality: sketch vs exact vs store and query (real collector containers)", "schemaVersion": 39, "version": 1, "refresh": "10s", "time": { - "from": "now-30m", + "from": "now-35m", "to": "now" }, - "templating": { - "list": [ - { - "name": "attribute", - "type": "query", - "datasource": { - "type": "prometheus", - "uid": "asap-prometheus" - }, - "query": { - "query": "label_values(asap_distinct, attribute_key)", - "refId": "attribute" - }, - "refresh": 2, - "current": { - "text": "k8s.pod.name", - "value": "k8s.pod.name" - } - } - ] - }, "panels": [ { "type": "text", - "title": "Pipeline", - "description": "Three pipelines that share no processor, fed identical points by their own generators (same seed and scenario start). Each box is one collector container pinned to its own CPU; arrows are OTLP/HTTP between containers.", + "title": "Three approaches to the same question: how many distinct values does each attribute have?", + "options": { + "mode": "html", + "content": "
generator-a/b
sources A and B
→
branch-a/b
HLL per attribute key per 1-min window
→
merge
both sources' HLLs per window
→
detector
distinct counts with bounds; class per key
→
Prometheus
query: method="sketch"
exact-generator-a/b
same points
→
exact-branch-a/b
value → count map per key per 1-min window
→
exact-merge
adds both sources' maps: every value
→
exact-detector
exact distinct counts and classes
→
Prometheus
query: method="exact"
raw-generator-a/b
same points
→
raw-sink-a/b
the only stage: every point into the log store
→
loki-raw
stores every raw point
→
query-job
LogQL distinct count per key once per window
→
Prometheus
answer: method="raw"

Every stage is one pinned container. Four attributes: http.route and http.method stable, k8s.pod.name growing 5%/min, user.id unique per point. Traffic grows over 30 minutes: the rate per source doubles every 5 minutes (5k to 80k points/s), and at minute 10 http.route explodes from 20 to 20,000 values. query-job compares store and query with exact and times every approach's backend query.

" + }, "gridPos": { "h": 10, "w": 24, "x": 0, "y": 0 }, - "options": { - "mode": "html", - "content": "
generator-a/b
sources A and B
→
branch-a/b
Stage 1 per source: HLL per attribute key per 1-min window
→
merge
Stage 2: merges both sources' sketches per window
→
detector
distinct counts with bounds; classes
→
Prometheus + Loki
method="sketch"
exact-generator-a/b
same points
→
exact-branch-a/b
Stage 1 per source: value → count map per key per 1-min window
→
exact-merge
Stage 2: adds both sources' exact windows
→
exact-detector
exact distinct counts; classes
→
Prometheus + Loki
method="exact" (ground truth)
raw-generator-a/b
same points
→
raw-sink-a/b
the only stage per source: writes every raw point; measures when the window's last point is stored
→
loki-raw
one log record per raw point
" - }, "id": 1 }, { - "type": "timeseries", - "title": "Distinct values of $attribute per minute: sketch with 95% bound vs exact", - "datasource": { - "type": "prometheus", - "uid": "asap-prometheus" - }, + "type": "row", + "title": "Answers", + "collapsed": false, + "panels": [], "gridPos": { - "h": 8, - "w": 12, + "h": 1, + "w": 24, "x": 0, "y": 10 }, + "id": 2 + }, + { + "type": "table", + "title": "Distinct values per attribute key, latest window store and query answered", + "description": "All from the latest window store and query has answered (query-job republishes sketch and exact for it), which lags behind when the log store is slow; the panel on the right shows which window. Error is (count - exact) / exact. Class: the detector's verdict from the last 10 windows (stable, growing, identifier-like). Store+query refused: 1 when Loki refused or timed out the key's LogQL query, which leaves its count empty.", + "datasource": { + "type": "prometheus", + "uid": "asap-prometheus" + }, "targets": [ { "refId": "A", - "expr": "asap_distinct{attribute_key=\"$attribute\",method=\"sketch\",bound=\"estimate\"}", - "legendFormat": "sketch" + "expr": "max by (attribute_key) (asap_compare_distinct{method=\"exact\"})", + "format": "table", + "instant": true }, { "refId": "B", - "expr": "asap_distinct{attribute_key=\"$attribute\",method=\"sketch\",bound=\"lower\"}", - "legendFormat": "sketch lower" + "expr": "max by (attribute_key) (asap_compare_distinct{method=\"sketch\"})", + "format": "table", + "instant": true }, { "refId": "C", - "expr": "asap_distinct{attribute_key=\"$attribute\",method=\"sketch\",bound=\"upper\"}", - "legendFormat": "sketch upper" + "expr": "max by (attribute_key) (asap_compare_distinct{method=\"raw\"})", + "format": "table", + "instant": true }, { "refId": "D", - "expr": "asap_distinct{attribute_key=\"$attribute\",method=\"exact\"}", - "legendFormat": "exact" + "expr": "100 * (max by (attribute_key) (asap_compare_distinct{method=\"sketch\"}) - on (attribute_key) max by (attribute_key) (asap_compare_distinct{method=\"exact\"})) / on (attribute_key) max by (attribute_key) (asap_compare_distinct{method=\"exact\"})", + "format": "table", + "instant": true + }, + { + "refId": "E", + "expr": "100 * (max by (attribute_key) (asap_compare_distinct{method=\"raw\"}) - on (attribute_key) max by (attribute_key) (asap_compare_distinct{method=\"exact\"})) / on (attribute_key) max by (attribute_key) (asap_compare_distinct{method=\"exact\"})", + "format": "table", + "instant": true + }, + { + "refId": "F", + "expr": "max by (attribute_key, exact_class) (label_replace(asap_compare_class{method=\"exact\"}, \"exact_class\", \"$1\", \"state\", \"(.*)\"))", + "format": "table", + "instant": true + }, + { + "refId": "G", + "expr": "max by (attribute_key, sketch_class) (label_replace(asap_compare_class{method=\"sketch\"}, \"sketch_class\", \"$1\", \"state\", \"(.*)\"))", + "format": "table", + "instant": true + }, + { + "refId": "H", + "expr": "max by (attribute_key) (asap_compare_failed)", + "format": "table", + "instant": true + } + ], + "transformations": [ + { + "id": "joinByField", + "options": { + "byField": "attribute_key", + "mode": "outer" + } + }, + { + "id": "organize", + "options": { + "excludeByName": { + "Time": true, + "Value #F": true, + "Value #G": true, + "Time 1": true, + "Time 2": true, + "Time 3": true, + "Time 4": true, + "Time 5": true, + "Time 6": true, + "Time 7": true, + "Time 8": true + }, + "indexByName": { + "attribute_key": 0, + "Value #A": 1, + "exact_class": 2, + "Value #B": 3, + "Value #D": 4, + "sketch_class": 5, + "Value #C": 6, + "Value #E": 7, + "Value #H": 8 + }, + "renameByName": { + "attribute_key": "Attribute key", + "Value #A": "Exact distinct", + "exact_class": "Exact class", + "Value #B": "Sketch estimate", + "Value #D": "Sketch error", + "sketch_class": "Sketch class", + "Value #C": "Store+query", + "Value #E": "Store+query error", + "Value #H": "Store+query refused" + } + } } ], + "options": { + "showHeader": true, + "sortBy": [ + { + "displayName": "Exact distinct", + "desc": true + } + ] + }, "fieldConfig": { "defaults": { + "decimals": 0, "custom": { - "spanNulls": true + "align": "auto" } }, "overrides": [ { "matcher": { "id": "byName", - "options": "sketch upper" + "options": "Exact distinct" }, "properties": [ - { - "id": "custom.fillBelowTo", - "value": "sketch lower" - }, - { - "id": "custom.lineWidth", - "value": 0 - }, - { - "id": "custom.fillOpacity", - "value": 20 - }, { "id": "color", "value": { "mode": "fixed", - "fixedColor": "blue" + "fixedColor": "#ff9830" + } + }, + { + "id": "custom.cellOptions", + "value": { + "type": "color-text" } } ] @@ -118,18 +185,20 @@ { "matcher": { "id": "byName", - "options": "sketch lower" + "options": "Sketch estimate" }, "properties": [ - { - "id": "custom.lineWidth", - "value": 0 - }, { "id": "color", "value": { "mode": "fixed", - "fixedColor": "blue" + "fixedColor": "#5794f2" + } + }, + { + "id": "custom.cellOptions", + "value": { + "type": "color-text" } } ] @@ -137,14 +206,20 @@ { "matcher": { "id": "byName", - "options": "sketch" + "options": "Store+query" }, "properties": [ { "id": "color", "value": { "mode": "fixed", - "fixedColor": "blue" + "fixedColor": "#8e8e8e" + } + }, + { + "id": "custom.cellOptions", + "value": { + "type": "color-text" } } ] @@ -152,132 +227,93 @@ { "matcher": { "id": "byName", - "options": "exact" + "options": "Sketch error" }, "properties": [ { - "id": "color", - "value": { - "mode": "fixed", - "fixedColor": "orange" - } + "id": "unit", + "value": "percent" }, { - "id": "custom.lineStyle", - "value": { - "fill": "dash", - "dash": [ - 6, - 4 - ] - } + "id": "decimals", + "value": 2 } ] - } - ] - }, - "id": 2 - }, - { - "type": "table", - "title": "Class of every attribute key (latest window)", - "datasource": { - "type": "prometheus", - "uid": "asap-prometheus" - }, - "gridPos": { - "h": 10, - "w": 12, - "x": 12, - "y": 10 - }, - "targets": [ - { - "refId": "A", - "expr": "asap_cardinality_state == 1", - "format": "table", - "instant": true - } - ], - "transformations": [ - { - "id": "organize", - "options": { - "excludeByName": { - "Time": true, - "Value": true, - "__name__": true, - "job": true, - "instance": true, - "metric": true + }, + { + "matcher": { + "id": "byName", + "options": "Store+query error" }, - "renameByName": { - "attribute_key": "attribute", - "method": "method", - "state": "class" - } - } - }, - { - "id": "sortBy", - "options": { - "sort": [ + "properties": [ { - "field": "class", - "desc": false + "id": "unit", + "value": "percent" }, { - "field": "attribute" + "id": "decimals", + "value": 2 } ] } - } - ], + ] + }, + "gridPos": { + "h": 7, + "w": 20, + "x": 0, + "y": 11 + }, "id": 3 }, { - "type": "timeseries", - "title": "Distinct / points per attribute key (1.0 = every point new)", + "type": "stat", + "title": "Window compared (end)", "datasource": { "type": "prometheus", "uid": "asap-prometheus" }, - "gridPos": { - "h": 8, - "w": 12, - "x": 0, - "y": 20 - }, + "description": "End of the window the table compares: the latest one store and query has answered.", "targets": [ { "refId": "A", - "expr": "asap_distinct_ratio", - "legendFormat": "{{attribute_key}} {{method}}" + "expr": "asap_compare_window_end", + "instant": true } ], + "options": { + "reduceOptions": { + "calcs": [ + "lastNotNull" + ], + "fields": "", + "values": false + }, + "colorMode": "none", + "graphMode": "none", + "textMode": "value" + }, "fieldConfig": { "defaults": { - "unit": "percentunit", - "max": 1, - "min": 0 + "unit": "dateTimeAsIsoNoDateIfToday" }, "overrides": [] }, + "gridPos": { + "h": 7, + "w": 4, + "x": 20, + "y": 11 + }, "id": 4 }, { "type": "logs", - "title": "Cardinality events (Loki)", + "title": "Class changes (Loki events from both detectors)", "datasource": { "type": "loki", "uid": "asap-loki" }, - "gridPos": { - "h": 8, - "w": 12, - "x": 12, - "y": 20 - }, "targets": [ { "refId": "A", @@ -289,108 +325,83 @@ "wrapLogMessage": true, "sortOrder": "Descending" }, + "gridPos": { + "h": 8, + "w": 24, + "x": 0, + "y": 18 + }, "id": 5 }, + { + "type": "row", + "title": "Accuracy against exact", + "collapsed": false, + "panels": [], + "gridPos": { + "h": 1, + "w": 24, + "x": 0, + "y": 26 + }, + "id": 6 + }, { "type": "timeseries", - "id": 10, - "title": "Throughput per pipeline: points per second each window delivered to the end of its pipeline", - "description": "Points in each 1-minute window that reached the end of the pipeline, divided by 60 s: for sketch and exact the points their detector answered for, for raw the points loki-raw acknowledged. Offered: 2 sources x 20,000 points/s. A pipeline below the offered line is dropping or falling behind.", + "title": "Sketch (HLL): relative error of the distinct count per key", "datasource": { "type": "prometheus", "uid": "asap-prometheus" }, - "gridPos": { - "h": 8, - "w": 24, - "x": 0, - "y": 28 - }, + "description": "|estimate - exact| / exact per window.", "targets": [ { "refId": "A", - "expr": "sum by (method) (asap_points{method=~\"sketch|exact\",attribute_key=\"http.route\"}) / 60", - "legendFormat": "{{method}}" - }, - { - "refId": "B", - "expr": "sum by (method) (asap_points{method=\"raw\"}) / 60", - "legendFormat": "{{method}}" - }, - { - "refId": "C", - "expr": "vector(40000)", - "legendFormat": "offered" + "expr": "(abs(max by (attribute_key) (asap_distinct{method=\"sketch\",bound=\"estimate\"}) - on (attribute_key) max by (attribute_key) (asap_distinct{method=\"exact\"})) / on (attribute_key) max by (attribute_key) (asap_distinct{method=\"exact\"})) and on (attribute_key) (max by (attribute_key) (timestamp(asap_distinct{method=\"sketch\"})) == on (attribute_key) max by (attribute_key) (timestamp(asap_distinct{method=\"exact\"})))", + "legendFormat": "{{attribute_key}}" } ], "fieldConfig": { "defaults": { - "unit": "short", - "decimals": 0, - "min": 0, "custom": { - "spanNulls": true, - "showPoints": "always", - "pointSize": 6 - } + "spanNulls": true + }, + "unit": "percentunit", + "min": 0 }, - "overrides": [ - { - "matcher": { - "id": "byName", - "options": "offered" - }, - "properties": [ - { - "id": "custom.lineStyle", - "value": { - "fill": "dash", - "dash": [ - 6, - 4 - ] - } - }, - { - "id": "color", - "value": { - "mode": "fixed", - "fixedColor": "#8e8e8e" - } - }, - { - "id": "custom.showPoints", - "value": "never" - } - ] - } - ] + "overrides": [] }, "options": { "legend": { "displayMode": "list", "placement": "bottom" + }, + "tooltip": { + "mode": "multi", + "sort": "none" } - } + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 27 + }, + "id": 7 }, { "type": "timeseries", - "title": "Window completion delay: seconds from a window's end until its result is complete", + "title": "Store and query: relative error of the distinct count per key", "datasource": { "type": "prometheus", "uid": "asap-prometheus" }, - "gridPos": { - "h": 8, - "w": 12, - "x": 0, - "y": 36 - }, + "description": "|estimate - exact| / exact per window. Store and query is exact while every point reaches the log store.", "targets": [ { "refId": "A", - "expr": "avg by (method) (asap_answer_delay)", - "legendFormat": "{{method}}" + "expr": "(abs(max by (attribute_key) (asap_distinct{method=\"raw\",bound=\"estimate\"}) - on (attribute_key) max by (attribute_key) (asap_distinct{method=\"exact\"})) / on (attribute_key) max by (attribute_key) (asap_distinct{method=\"exact\"})) and on (attribute_key) (max by (attribute_key) (timestamp(asap_distinct{method=\"raw\"})) == on (attribute_key) max by (attribute_key) (timestamp(asap_distinct{method=\"exact\"})))", + "legendFormat": "{{attribute_key}}" } ], "fieldConfig": { @@ -398,77 +409,60 @@ "custom": { "spanNulls": true }, - "unit": "s" + "unit": "percentunit", + "min": 0 }, - "overrides": [ - { - "matcher": { - "id": "byName", - "options": "sketch" - }, - "properties": [ - { - "id": "color", - "value": { - "mode": "fixed", - "fixedColor": "#5794f2" - } - } - ] - }, - { - "matcher": { - "id": "byName", - "options": "exact" - }, - "properties": [ - { - "id": "color", - "value": { - "mode": "fixed", - "fixedColor": "#ff9830" - } - } - ] - }, - { - "matcher": { - "id": "byName", - "options": "raw" - }, - "properties": [ - { - "id": "color", - "value": { - "mode": "fixed", - "fixedColor": "#8e8e8e" - } - } - ] - } - ] + "overrides": [] }, - "description": "sketch and exact: the detector has emitted the window's answers. raw: loki-raw has acknowledged the window's last raw point.", - "id": 6 + "options": { + "legend": { + "displayMode": "list", + "placement": "bottom" + }, + "tooltip": { + "mode": "multi", + "sort": "none" + } + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 27 + }, + "id": 8 + }, + { + "type": "row", + "title": "Traffic over time: volume and shape", + "collapsed": false, + "panels": [], + "gridPos": { + "h": 1, + "w": 24, + "x": 0, + "y": 35 + }, + "id": 9 }, { "type": "timeseries", - "title": "CPU cores per pipeline (all stages but the generators)", + "title": "End-to-end throughput: points per second each window delivered to the end of its pipeline", "datasource": { "type": "prometheus", "uid": "asap-prometheus" }, - "gridPos": { - "h": 8, - "w": 12, - "x": 12, - "y": 36 - }, + "description": "Both sources together. The generators skip batches a pipeline cannot take, so a pipeline that falls behind delivers less. Store and query: points loki-raw acknowledged within 50 s of the window's end. A pipeline that stops reporting shows a gap after 2 minutes.", "targets": [ { "refId": "A", - "expr": "sum by (pipeline) (label_replace(label_replace(rate(cpu_time_total{job=\"collectors\",stage!~\".*generator.*\",otel_scope_pipeline_id=\"main\"}[1m]), \"pipeline\", \"sketch\", \"stage\", \".*\"), \"pipeline\", \"$1\", \"stage\", \"(exact|raw)-.*\"))", - "legendFormat": "{{pipeline}}" + "expr": "sum by (method) (last_over_time(asap_points{method=~\"sketch|exact\",attribute_key=\"http.route\"}[2m])) / 60", + "legendFormat": "{{method}}" + }, + { + "refId": "B", + "expr": "sum(last_over_time(asap_points{method=\"raw\"}[2m])) / 60", + "legendFormat": "store and query" } ], "fieldConfig": { @@ -476,7 +470,8 @@ "custom": { "spanNulls": true }, - "unit": "short" + "unit": "short", + "min": 0 }, "overrides": [ { @@ -512,7 +507,7 @@ { "matcher": { "id": "byName", - "options": "raw" + "options": "store and query" }, "properties": [ { @@ -526,38 +521,115 @@ } ] }, - "id": 7 + "options": { + "legend": { + "displayMode": "list", + "placement": "bottom" + }, + "tooltip": { + "mode": "multi", + "sort": "none" + } + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 36 + }, + "id": 10 }, { "type": "timeseries", - "title": "Bytes received per pipeline (all stages but the generators), log scale", + "title": "Data shape: distinct values per key per window (exact), log scale", "datasource": { "type": "prometheus", "uid": "asap-prometheus" }, + "description": "", + "targets": [ + { + "refId": "A", + "expr": "max by (attribute_key) (asap_distinct{method=\"exact\"})", + "legendFormat": "{{attribute_key}}" + } + ], + "fieldConfig": { + "defaults": { + "custom": { + "spanNulls": true, + "scaleDistribution": { + "type": "log", + "log": 10 + } + }, + "unit": "short" + }, + "overrides": [] + }, + "options": { + "legend": { + "displayMode": "list", + "placement": "bottom" + }, + "tooltip": { + "mode": "multi", + "sort": "none" + } + }, "gridPos": { "h": 8, "w": 12, + "x": 12, + "y": 36 + }, + "id": 11 + }, + { + "type": "row", + "title": "Freshness: from a sample's generation to the queryable answer", + "collapsed": false, + "panels": [], + "gridPos": { + "h": 1, + "w": 24, "x": 0, "y": 44 }, + "id": 12 + }, + { + "type": "timeseries", + "title": "Total freshness of a window's first sample", + "datasource": { + "type": "prometheus", + "uid": "asap-prometheus" + }, + "description": "Window wait (up to 60 s for the first sample) + the pipeline until the answer is stored + the backend query that returns it.", "targets": [ { "refId": "A", - "expr": "sum by (pipeline) (label_replace(label_replace(rate(payload_size_bytes_total{job=\"collectors\",stage!~\".*generator.*\",otel_scope_node_type=\"receiver\"}[1m]), \"pipeline\", \"sketch\", \"stage\", \".*\"), \"pipeline\", \"$1\", \"stage\", \"(exact|raw)-.*\")) * 8", - "legendFormat": "{{pipeline}}" + "expr": "60 + max(asap_answer_delay{method=\"sketch\"}) + max(asap_query_seconds{method=\"sketch\"})", + "legendFormat": "sketch" + }, + { + "refId": "B", + "expr": "60 + max(asap_answer_delay{method=\"exact\"}) + max(asap_query_seconds{method=\"exact\"})", + "legendFormat": "exact" + }, + { + "refId": "C", + "expr": "60 + max(asap_answer_delay{method=\"raw\"}) + max(asap_query_seconds{method=\"raw\"})", + "legendFormat": "store and query" } ], "fieldConfig": { "defaults": { "custom": { - "spanNulls": true, - "scaleDistribution": { - "type": "log", - "log": 10 - } + "spanNulls": true }, - "unit": "bps" + "unit": "s", + "min": 0 }, "overrides": [ { @@ -593,7 +665,7 @@ { "matcher": { "id": "byName", - "options": "raw" + "options": "store and query" }, "properties": [ { @@ -607,84 +679,782 @@ } ] }, - "id": 8 + "options": { + "legend": { + "displayMode": "list", + "placement": "bottom" + }, + "tooltip": { + "mode": "multi", + "sort": "none" + } + }, + "gridPos": { + "h": 7, + "w": 24, + "x": 0, + "y": 45 + }, + "id": 13 }, { "type": "timeseries", - "title": "Resident memory per pipeline (all stages but the generators)", + "title": "Sketch: latency breakdown per window", "datasource": { "type": "prometheus", "uid": "asap-prometheus" }, + "description": "Seconds per segment. branch + to merge: the window's end until the last source's sketch reached the merge. merge: waiting for the other source's sketch (5 s timeout).", + "targets": [ + { + "refId": "A", + "expr": "0 * max(asap_answer_delay{method=\"sketch\"}) + 60", + "legendFormat": "window wait" + }, + { + "refId": "B", + "expr": "max(asap_stage_received{method=\"sketch\",stage=\"merge\"})", + "legendFormat": "branch + to merge" + }, + { + "refId": "C", + "expr": "max(asap_stage_emitted{method=\"sketch\",stage=\"merge\"}) - max(asap_stage_received{method=\"sketch\",stage=\"merge\"})", + "legendFormat": "merge (waits for both sources)" + }, + { + "refId": "D", + "expr": "max(asap_stage_received{method=\"sketch\",stage=\"detector\"}) - max(asap_stage_emitted{method=\"sketch\",stage=\"merge\"})", + "legendFormat": "to detector" + }, + { + "refId": "E", + "expr": "max(asap_stage_emitted{method=\"sketch\",stage=\"detector\"}) - max(asap_stage_received{method=\"sketch\",stage=\"detector\"})", + "legendFormat": "detector" + }, + { + "refId": "F", + "expr": "max(asap_query_seconds{method=\"sketch\"})", + "legendFormat": "Prometheus query" + } + ], + "fieldConfig": { + "defaults": { + "custom": { + "spanNulls": true, + "stacking": { + "mode": "normal" + }, + "fillOpacity": 60, + "lineInterpolation": "stepAfter", + "lineWidth": 1 + }, + "unit": "s", + "min": 0 + }, + "overrides": [] + }, + "options": { + "legend": { + "displayMode": "list", + "placement": "bottom" + }, + "tooltip": { + "mode": "multi", + "sort": "none" + } + }, "gridPos": { "h": 8, - "w": 12, - "x": 12, - "y": 44 + "w": 8, + "x": 0, + "y": 52 + }, + "id": 14 + }, + { + "type": "timeseries", + "title": "Exact: latency breakdown per window", + "datasource": { + "type": "prometheus", + "uid": "asap-prometheus" }, + "description": "Seconds per segment. branch + to merge: the window's end until the last source's sketch reached the merge. merge: waiting for the other source's sketch (5 s timeout).", "targets": [ { "refId": "A", - "expr": "sum by (pipeline) (label_replace(label_replace(max by (stage) (memory_rss{job=\"collectors\",stage!~\".*generator.*\"}), \"pipeline\", \"sketch\", \"stage\", \".*\"), \"pipeline\", \"$1\", \"stage\", \"(exact|raw)-.*\"))", - "legendFormat": "{{pipeline}}" + "expr": "0 * max(asap_answer_delay{method=\"exact\"}) + 60", + "legendFormat": "window wait" + }, + { + "refId": "B", + "expr": "max(asap_stage_received{method=\"exact\",stage=\"merge\"})", + "legendFormat": "branch + to merge" + }, + { + "refId": "C", + "expr": "max(asap_stage_emitted{method=\"exact\",stage=\"merge\"}) - max(asap_stage_received{method=\"exact\",stage=\"merge\"})", + "legendFormat": "merge (waits for both sources)" + }, + { + "refId": "D", + "expr": "max(asap_stage_received{method=\"exact\",stage=\"detector\"}) - max(asap_stage_emitted{method=\"exact\",stage=\"merge\"})", + "legendFormat": "to detector" + }, + { + "refId": "E", + "expr": "max(asap_stage_emitted{method=\"exact\",stage=\"detector\"}) - max(asap_stage_received{method=\"exact\",stage=\"detector\"})", + "legendFormat": "detector" + }, + { + "refId": "F", + "expr": "max(asap_query_seconds{method=\"exact\"})", + "legendFormat": "Prometheus query" } ], "fieldConfig": { "defaults": { "custom": { - "spanNulls": true + "spanNulls": true, + "stacking": { + "mode": "normal" + }, + "fillOpacity": 60, + "lineInterpolation": "stepAfter", + "lineWidth": 1 }, - "unit": "bytes" + "unit": "s", + "min": 0 }, - "overrides": [ - { - "matcher": { - "id": "byName", - "options": "sketch" + "overrides": [] + }, + "options": { + "legend": { + "displayMode": "list", + "placement": "bottom" + }, + "tooltip": { + "mode": "multi", + "sort": "none" + } + }, + "gridPos": { + "h": 8, + "w": 8, + "x": 8, + "y": 52 + }, + "id": 15 + }, + { + "type": "timeseries", + "title": "Store and query: latency breakdown per window", + "datasource": { + "type": "prometheus", + "uid": "asap-prometheus" + }, + "description": "Seconds per segment. into loki-raw: the window's end until loki-raw acknowledged its last point (or raw-sink's 50 s timeout). The query time includes refused and timed-out queries; with several queries per window it is their sum.", + "targets": [ + { + "refId": "A", + "expr": "0 * max(asap_answer_delay{method=\"raw\"}) + 60", + "legendFormat": "window wait" + }, + { + "refId": "B", + "expr": "max(asap_answer_delay{method=\"raw\"})", + "legendFormat": "into loki-raw (last point acknowledged)" + }, + { + "refId": "C", + "expr": "max(asap_query_seconds{method=\"raw\"})", + "legendFormat": "LogQL distinct counts (one query per key)" + } + ], + "fieldConfig": { + "defaults": { + "custom": { + "spanNulls": true, + "stacking": { + "mode": "normal" }, - "properties": [ - { - "id": "color", - "value": { - "mode": "fixed", - "fixedColor": "#5794f2" - } - } - ] + "fillOpacity": 60, + "lineInterpolation": "stepAfter", + "lineWidth": 1 }, - { - "matcher": { - "id": "byName", - "options": "exact" - }, - "properties": [ - { - "id": "color", - "value": { - "mode": "fixed", - "fixedColor": "#ff9830" - } - } - ] + "unit": "s", + "min": 0 + }, + "overrides": [] + }, + "options": { + "legend": { + "displayMode": "list", + "placement": "bottom" + }, + "tooltip": { + "mode": "multi", + "sort": "none" + } + }, + "gridPos": { + "h": 8, + "w": 8, + "x": 16, + "y": 52 + }, + "id": 16 + }, + { + "type": "row", + "title": "Resources per approach (generators and observers excluded)", + "collapsed": false, + "panels": [], + "gridPos": { + "h": 1, + "w": 24, + "x": 0, + "y": 60 + }, + "id": 17 + }, + { + "type": "timeseries", + "title": "Sketch: CPU cores per component", + "datasource": { + "type": "prometheus", + "uid": "asap-prometheus" + }, + "description": "Pipeline stages from the engine's CPU time; loki-raw and query-job from their process CPU time. total: the whole approach.", + "targets": [ + { + "refId": "A", + "expr": "sum by (stage) (rate(cpu_time_total{job=\"collectors\",stage=~\"branch-.|merge|detector\",otel_scope_pipeline_id=\"main\"}[1m]))", + "legendFormat": "{{stage}}" + }, + { + "refId": "B", + "expr": "sum(sum by (stage) (rate(cpu_time_total{job=\"collectors\",stage=~\"branch-.|merge|detector\",otel_scope_pipeline_id=\"main\"}[1m])))", + "legendFormat": "total" + } + ], + "fieldConfig": { + "defaults": { + "custom": { + "spanNulls": true }, - { - "matcher": { - "id": "byName", - "options": "raw" - }, - "properties": [ - { - "id": "color", - "value": { - "mode": "fixed", - "fixedColor": "#8e8e8e" - } - } - ] - } - ] + "unit": "short", + "min": 0 + }, + "overrides": [] }, - "id": 9 + "options": { + "legend": { + "displayMode": "table", + "placement": "bottom", + "calcs": [ + "lastNotNull" + ] + }, + "tooltip": { + "mode": "multi", + "sort": "none" + } + }, + "gridPos": { + "h": 12, + "w": 8, + "x": 0, + "y": 61 + }, + "id": 18 + }, + { + "type": "timeseries", + "title": "Exact: CPU cores per component", + "datasource": { + "type": "prometheus", + "uid": "asap-prometheus" + }, + "description": "Pipeline stages from the engine's CPU time; loki-raw and query-job from their process CPU time. total: the whole approach.", + "targets": [ + { + "refId": "A", + "expr": "sum by (stage) (rate(cpu_time_total{job=\"collectors\",stage=~\"exact-(branch-.|merge|detector)\",otel_scope_pipeline_id=\"main\"}[1m]))", + "legendFormat": "{{stage}}" + }, + { + "refId": "B", + "expr": "sum(sum by (stage) (rate(cpu_time_total{job=\"collectors\",stage=~\"exact-(branch-.|merge|detector)\",otel_scope_pipeline_id=\"main\"}[1m])))", + "legendFormat": "total" + } + ], + "fieldConfig": { + "defaults": { + "custom": { + "spanNulls": true + }, + "unit": "short", + "min": 0 + }, + "overrides": [] + }, + "options": { + "legend": { + "displayMode": "table", + "placement": "bottom", + "calcs": [ + "lastNotNull" + ] + }, + "tooltip": { + "mode": "multi", + "sort": "none" + } + }, + "gridPos": { + "h": 12, + "w": 8, + "x": 8, + "y": 61 + }, + "id": 19 + }, + { + "type": "timeseries", + "title": "Store and query: CPU cores per component", + "datasource": { + "type": "prometheus", + "uid": "asap-prometheus" + }, + "description": "Pipeline stages from the engine's CPU time; loki-raw and query-job from their process CPU time. total: the whole approach.", + "targets": [ + { + "refId": "A", + "expr": "sum by (stage) (rate(cpu_time_total{job=\"collectors\",stage=~\"raw-sink-.\",otel_scope_pipeline_id=\"main\"}[1m])) or sum by (stage) (rate(process_cpu_seconds_total{job=~\"loki-raw|query-job\"}[1m]))", + "legendFormat": "{{stage}}" + }, + { + "refId": "B", + "expr": "sum(sum by (stage) (rate(cpu_time_total{job=\"collectors\",stage=~\"raw-sink-.\",otel_scope_pipeline_id=\"main\"}[1m])) or sum by (stage) (rate(process_cpu_seconds_total{job=~\"loki-raw|query-job\"}[1m])))", + "legendFormat": "total" + } + ], + "fieldConfig": { + "defaults": { + "custom": { + "spanNulls": true + }, + "unit": "short", + "min": 0 + }, + "overrides": [] + }, + "options": { + "legend": { + "displayMode": "table", + "placement": "bottom", + "calcs": [ + "lastNotNull" + ] + }, + "tooltip": { + "mode": "multi", + "sort": "none" + } + }, + "gridPos": { + "h": 12, + "w": 8, + "x": 16, + "y": 61 + }, + "id": 20 + }, + { + "type": "timeseries", + "title": "Sketch: resident memory per component", + "datasource": { + "type": "prometheus", + "uid": "asap-prometheus" + }, + "description": "total: the whole approach.", + "targets": [ + { + "refId": "A", + "expr": "max by (stage) (memory_rss{job=\"collectors\",stage=~\"branch-.|merge|detector\"})", + "legendFormat": "{{stage}}" + }, + { + "refId": "B", + "expr": "sum(max by (stage) (memory_rss{job=\"collectors\",stage=~\"branch-.|merge|detector\"}))", + "legendFormat": "total" + } + ], + "fieldConfig": { + "defaults": { + "custom": { + "spanNulls": true + }, + "unit": "bytes", + "min": 0 + }, + "overrides": [] + }, + "options": { + "legend": { + "displayMode": "table", + "placement": "bottom", + "calcs": [ + "lastNotNull" + ] + }, + "tooltip": { + "mode": "multi", + "sort": "none" + } + }, + "gridPos": { + "h": 12, + "w": 8, + "x": 0, + "y": 73 + }, + "id": 21 + }, + { + "type": "timeseries", + "title": "Exact: resident memory per component", + "datasource": { + "type": "prometheus", + "uid": "asap-prometheus" + }, + "description": "total: the whole approach.", + "targets": [ + { + "refId": "A", + "expr": "max by (stage) (memory_rss{job=\"collectors\",stage=~\"exact-(branch-.|merge|detector)\"})", + "legendFormat": "{{stage}}" + }, + { + "refId": "B", + "expr": "sum(max by (stage) (memory_rss{job=\"collectors\",stage=~\"exact-(branch-.|merge|detector)\"}))", + "legendFormat": "total" + } + ], + "fieldConfig": { + "defaults": { + "custom": { + "spanNulls": true + }, + "unit": "bytes", + "min": 0 + }, + "overrides": [] + }, + "options": { + "legend": { + "displayMode": "table", + "placement": "bottom", + "calcs": [ + "lastNotNull" + ] + }, + "tooltip": { + "mode": "multi", + "sort": "none" + } + }, + "gridPos": { + "h": 12, + "w": 8, + "x": 8, + "y": 73 + }, + "id": 22 + }, + { + "type": "timeseries", + "title": "Store and query: resident memory per component", + "datasource": { + "type": "prometheus", + "uid": "asap-prometheus" + }, + "description": "total: the whole approach.", + "targets": [ + { + "refId": "A", + "expr": "max by (stage) (memory_rss{job=\"collectors\",stage=~\"raw-sink-.\"}) or max by (stage) (process_resident_memory_bytes{job=~\"loki-raw|query-job\"})", + "legendFormat": "{{stage}}" + }, + { + "refId": "B", + "expr": "sum(max by (stage) (memory_rss{job=\"collectors\",stage=~\"raw-sink-.\"}) or max by (stage) (process_resident_memory_bytes{job=~\"loki-raw|query-job\"}))", + "legendFormat": "total" + } + ], + "fieldConfig": { + "defaults": { + "custom": { + "spanNulls": true + }, + "unit": "bytes", + "min": 0 + }, + "overrides": [] + }, + "options": { + "legend": { + "displayMode": "table", + "placement": "bottom", + "calcs": [ + "lastNotNull" + ] + }, + "tooltip": { + "mode": "multi", + "sort": "none" + } + }, + "gridPos": { + "h": 12, + "w": 8, + "x": 16, + "y": 73 + }, + "id": 23 + }, + { + "type": "timeseries", + "title": "Sketch: bits per second into each stage", + "datasource": { + "type": "prometheus", + "uid": "asap-prometheus" + }, + "description": "into branch-a/b, exact-branch-a/b, raw-sink-a/b: raw points from the generators. into merge / exact-merge: sketches / exact maps. into detector / exact-detector: merged windows. into loki-raw: raw log records.", + "targets": [ + { + "refId": "A", + "expr": "sum by (stage) (rate(payload_size_bytes_total{job=\"collectors\",stage=~\"branch-.|merge|detector\",otel_scope_node_type=\"receiver\"}[2m])) * 8", + "legendFormat": "{{stage}}" + } + ], + "fieldConfig": { + "defaults": { + "custom": { + "spanNulls": true, + "scaleDistribution": { + "type": "log", + "log": 10 + } + }, + "unit": "bps" + }, + "overrides": [] + }, + "options": { + "legend": { + "displayMode": "table", + "placement": "bottom", + "calcs": [ + "lastNotNull" + ] + }, + "tooltip": { + "mode": "multi", + "sort": "none" + } + }, + "gridPos": { + "h": 12, + "w": 8, + "x": 0, + "y": 85 + }, + "id": 24 + }, + { + "type": "timeseries", + "title": "Exact: bits per second into each stage", + "datasource": { + "type": "prometheus", + "uid": "asap-prometheus" + }, + "description": "into branch-a/b, exact-branch-a/b, raw-sink-a/b: raw points from the generators. into merge / exact-merge: sketches / exact maps. into detector / exact-detector: merged windows. into loki-raw: raw log records.", + "targets": [ + { + "refId": "A", + "expr": "sum by (stage) (rate(payload_size_bytes_total{job=\"collectors\",stage=~\"exact-(branch-.|merge|detector)\",otel_scope_node_type=\"receiver\"}[2m])) * 8", + "legendFormat": "{{stage}}" + } + ], + "fieldConfig": { + "defaults": { + "custom": { + "spanNulls": true, + "scaleDistribution": { + "type": "log", + "log": 10 + } + }, + "unit": "bps" + }, + "overrides": [] + }, + "options": { + "legend": { + "displayMode": "table", + "placement": "bottom", + "calcs": [ + "lastNotNull" + ] + }, + "tooltip": { + "mode": "multi", + "sort": "none" + } + }, + "gridPos": { + "h": 12, + "w": 8, + "x": 8, + "y": 85 + }, + "id": 25 + }, + { + "type": "timeseries", + "title": "Store and query: bits per second into each stage", + "datasource": { + "type": "prometheus", + "uid": "asap-prometheus" + }, + "description": "into branch-a/b, exact-branch-a/b, raw-sink-a/b: raw points from the generators. into merge / exact-merge: sketches / exact maps. into detector / exact-detector: merged windows. into loki-raw: raw log records.", + "targets": [ + { + "refId": "A", + "expr": "sum by (stage) (rate(payload_size_bytes_total{job=\"collectors\",stage=~\"raw-sink-.\",otel_scope_node_type=\"receiver\"}[2m])) * 8 or label_replace(sum(rate(loki_distributor_bytes_received_total[2m])) * 8, \"stage\", \"loki-raw\", \"\", \"\")", + "legendFormat": "{{stage}}" + } + ], + "fieldConfig": { + "defaults": { + "custom": { + "spanNulls": true, + "scaleDistribution": { + "type": "log", + "log": 10 + } + }, + "unit": "bps" + }, + "overrides": [] + }, + "options": { + "legend": { + "displayMode": "table", + "placement": "bottom", + "calcs": [ + "lastNotNull" + ] + }, + "tooltip": { + "mode": "multi", + "sort": "none" + } + }, + "gridPos": { + "h": 12, + "w": 8, + "x": 16, + "y": 85 + }, + "id": 26 + }, + { + "type": "timeseries", + "title": "Shared by all three: Prometheus CPU cores", + "datasource": { + "type": "prometheus", + "uid": "asap-prometheus" + }, + "description": "Prometheus holds the answers of all three approaches and the monitoring data.", + "targets": [ + { + "refId": "A", + "expr": "sum(rate(process_cpu_seconds_total{job=\"prometheus\"}[1m]))", + "legendFormat": "Prometheus" + } + ], + "fieldConfig": { + "defaults": { + "custom": { + "spanNulls": true + }, + "unit": "short", + "min": 0 + }, + "overrides": [] + }, + "options": { + "legend": { + "displayMode": "list", + "placement": "bottom" + }, + "tooltip": { + "mode": "multi", + "sort": "none" + } + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 97 + }, + "id": 27 + }, + { + "type": "timeseries", + "title": "Shared by all three: Prometheus resident memory", + "datasource": { + "type": "prometheus", + "uid": "asap-prometheus" + }, + "description": "", + "targets": [ + { + "refId": "A", + "expr": "max(process_resident_memory_bytes{job=\"prometheus\"})", + "legendFormat": "Prometheus" + } + ], + "fieldConfig": { + "defaults": { + "custom": { + "spanNulls": true + }, + "unit": "bytes", + "min": 0 + }, + "overrides": [] + }, + "options": { + "legend": { + "displayMode": "list", + "placement": "bottom" + }, + "tooltip": { + "mode": "multi", + "sort": "none" + } + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 97 + }, + "id": 28 } ] } diff --git a/deploy/telemetry-functions/grafana/dashboards/drift.json b/deploy/telemetry-functions/grafana/dashboards/drift.json index 81c8075..eecd360 100644 --- a/deploy/telemetry-functions/grafana/dashboards/drift.json +++ b/deploy/telemetry-functions/grafana/dashboards/drift.json @@ -397,7 +397,7 @@ "type": "timeseries", "id": 10, "title": "Throughput per pipeline: points per second each window delivered to the end of its pipeline", - "description": "Points in each 1-minute window that reached the end of the pipeline, divided by 60 s: for sketch and exact the points their detector answered for, for raw the points loki-raw acknowledged. Offered: 2 sources x 20,000 points/s. A pipeline below the offered line is dropping or falling behind.", + "description": "Points in each 1-minute window that reached the end of the pipeline, divided by 60 s: for sketch and exact the points their detector answered for, for raw the points loki-raw acknowledged within 50 s of the window's end. Offered: 2 sources x 20,000 points/s. A pipeline below the offered line is dropping or falling behind.", "datasource": { "type": "prometheus", "uid": "asap-prometheus" @@ -411,12 +411,12 @@ "targets": [ { "refId": "A", - "expr": "sum by (method) (asap_points{method=~\"sketch|exact\",attribute_key=\"\"}) / 60", + "expr": "sum by (method) (last_over_time(asap_points{method=~\"sketch|exact\",attribute_key=\"\"}[2m])) / 60", "legendFormat": "{{method}}" }, { "refId": "B", - "expr": "sum by (method) (asap_points{method=\"raw\"}) / 60", + "expr": "sum by (method) (last_over_time(asap_points{method=\"raw\"}[2m])) / 60", "legendFormat": "{{method}}" }, { diff --git a/deploy/telemetry-functions/grafana/dashboards/latency.json b/deploy/telemetry-functions/grafana/dashboards/latency.json index 1f608e5..b6b4dba 100644 --- a/deploy/telemetry-functions/grafana/dashboards/latency.json +++ b/deploy/telemetry-functions/grafana/dashboards/latency.json @@ -347,7 +347,7 @@ "type": "timeseries", "id": 9, "title": "Throughput per pipeline: points per second each window delivered to the end of its pipeline", - "description": "Points in each 1-minute window that reached the end of the pipeline, divided by 60 s: for sketch and exact the points their detector answered for, for raw the points loki-raw acknowledged. Offered: 2 sources x 20,000 points/s. A pipeline below the offered line is dropping or falling behind.", + "description": "Points in each 1-minute window that reached the end of the pipeline, divided by 60 s: for sketch and exact the points their detector answered for, for raw the points loki-raw acknowledged within 50 s of the window's end. Offered: 2 sources x 20,000 points/s. A pipeline below the offered line is dropping or falling behind.", "datasource": { "type": "prometheus", "uid": "asap-prometheus" @@ -361,12 +361,12 @@ "targets": [ { "refId": "A", - "expr": "sum by (method) (asap_points{method=~\"sketch|exact\",attribute_key=\"\"}) / 60", + "expr": "sum by (method) (last_over_time(asap_points{method=~\"sketch|exact\",attribute_key=\"\"}[2m])) / 60", "legendFormat": "{{method}}" }, { "refId": "B", - "expr": "sum by (method) (asap_points{method=\"raw\"}) / 60", + "expr": "sum by (method) (last_over_time(asap_points{method=\"raw\"}[2m])) / 60", "legendFormat": "{{method}}" }, { diff --git a/deploy/telemetry-functions/grafana/dashboards/topk.json b/deploy/telemetry-functions/grafana/dashboards/topk.json index 83432cc..1692eda 100644 --- a/deploy/telemetry-functions/grafana/dashboards/topk.json +++ b/deploy/telemetry-functions/grafana/dashboards/topk.json @@ -39,8 +39,8 @@ }, { "type": "table", - "title": "Top-10 routes of the three approaches, latest window (union of their top-10s)", - "description": "One row per route in any approach's top-10. Error is (count - exact) / exact; empty when the route is outside that approach's reported top-20.", + "title": "Top-10 routes of the three approaches, latest window store and query answered (union of their top-10s)", + "description": "One row per route in any approach's top-10, all from the latest window store and query has answered (query-job republishes sketch and exact for it), which lags behind when the log store is slow; the panel on the right shows which window. Error is (count - exact) / exact; empty when the route is outside that approach's reported top-20.", "datasource": { "type": "prometheus", "uid": "asap-prometheus" @@ -48,31 +48,31 @@ "targets": [ { "refId": "A", - "expr": "max by (key, exact_rank) (label_replace(asap_topk{method=\"exact\",bound=\"estimate\",rank=~\"[1-9]|10\"} and (timestamp(asap_topk{method=\"exact\",bound=\"estimate\",rank=~\"[1-9]|10\"}) == scalar(max(timestamp(asap_topk{method=\"exact\",bound=\"estimate\",rank=~\"[1-9]|10\"})))), \"exact_rank\", \"$1\", \"rank\", \"(.*)\"))", + "expr": "max by (key, exact_rank) (label_replace(asap_compare_topk{method=\"exact\",rank=~\"[1-9]|10\"}, \"exact_rank\", \"$1\", \"rank\", \"(.*)\"))", "format": "table", "instant": true }, { "refId": "B", - "expr": "max by (key, sketch_rank) (label_replace(asap_topk{method=\"sketch\",bound=\"estimate\",rank=~\"[1-9]|10\"} and (timestamp(asap_topk{method=\"sketch\",bound=\"estimate\",rank=~\"[1-9]|10\"}) == scalar(max(timestamp(asap_topk{method=\"sketch\",bound=\"estimate\",rank=~\"[1-9]|10\"})))), \"sketch_rank\", \"$1\", \"rank\", \"(.*)\"))", + "expr": "max by (key, sketch_rank) (label_replace(asap_compare_topk{method=\"sketch\",rank=~\"[1-9]|10\"}, \"sketch_rank\", \"$1\", \"rank\", \"(.*)\"))", "format": "table", "instant": true }, { "refId": "C", - "expr": "max by (key, raw_rank) (label_replace(asap_topk{method=\"raw\",bound=\"estimate\",rank=~\"[1-9]|10\"} and (timestamp(asap_topk{method=\"raw\",bound=\"estimate\",rank=~\"[1-9]|10\"}) == scalar(max(timestamp(asap_topk{method=\"raw\",bound=\"estimate\",rank=~\"[1-9]|10\"})))), \"raw_rank\", \"$1\", \"rank\", \"(.*)\"))", + "expr": "max by (key, raw_rank) (label_replace(asap_compare_topk{method=\"raw\",rank=~\"[1-9]|10\"}, \"raw_rank\", \"$1\", \"rank\", \"(.*)\"))", "format": "table", "instant": true }, { "refId": "D", - "expr": "100 * (max by (key) (asap_topk{method=\"sketch\",bound=\"estimate\"} and (timestamp(asap_topk{method=\"sketch\",bound=\"estimate\"}) == scalar(max(timestamp(asap_topk{method=\"sketch\",bound=\"estimate\"}))))) - on (key) max by (key) (asap_topk{method=\"exact\",bound=\"estimate\"} and (timestamp(asap_topk{method=\"exact\",bound=\"estimate\"}) == scalar(max(timestamp(asap_topk{method=\"exact\",bound=\"estimate\"})))))) / on (key) max by (key) (asap_topk{method=\"exact\",bound=\"estimate\"} and (timestamp(asap_topk{method=\"exact\",bound=\"estimate\"}) == scalar(max(timestamp(asap_topk{method=\"exact\",bound=\"estimate\"}))))) and on (key) (max by (key, exact_rank) (label_replace(asap_topk{method=\"exact\",bound=\"estimate\",rank=~\"[1-9]|10\"} and (timestamp(asap_topk{method=\"exact\",bound=\"estimate\",rank=~\"[1-9]|10\"}) == scalar(max(timestamp(asap_topk{method=\"exact\",bound=\"estimate\",rank=~\"[1-9]|10\"})))), \"exact_rank\", \"$1\", \"rank\", \"(.*)\")) or max by (key, sketch_rank) (label_replace(asap_topk{method=\"sketch\",bound=\"estimate\",rank=~\"[1-9]|10\"} and (timestamp(asap_topk{method=\"sketch\",bound=\"estimate\",rank=~\"[1-9]|10\"}) == scalar(max(timestamp(asap_topk{method=\"sketch\",bound=\"estimate\",rank=~\"[1-9]|10\"})))), \"sketch_rank\", \"$1\", \"rank\", \"(.*)\")) or max by (key, raw_rank) (label_replace(asap_topk{method=\"raw\",bound=\"estimate\",rank=~\"[1-9]|10\"} and (timestamp(asap_topk{method=\"raw\",bound=\"estimate\",rank=~\"[1-9]|10\"}) == scalar(max(timestamp(asap_topk{method=\"raw\",bound=\"estimate\",rank=~\"[1-9]|10\"})))), \"raw_rank\", \"$1\", \"rank\", \"(.*)\")))", + "expr": "100 * (max by (key) (asap_compare_topk{method=\"sketch\"}) - on (key) max by (key) (asap_compare_topk{method=\"exact\"})) / on (key) max by (key) (asap_compare_topk{method=\"exact\"}) and on (key) (max by (key, exact_rank) (label_replace(asap_compare_topk{method=\"exact\",rank=~\"[1-9]|10\"}, \"exact_rank\", \"$1\", \"rank\", \"(.*)\")) or max by (key, sketch_rank) (label_replace(asap_compare_topk{method=\"sketch\",rank=~\"[1-9]|10\"}, \"sketch_rank\", \"$1\", \"rank\", \"(.*)\")) or max by (key, raw_rank) (label_replace(asap_compare_topk{method=\"raw\",rank=~\"[1-9]|10\"}, \"raw_rank\", \"$1\", \"rank\", \"(.*)\")))", "format": "table", "instant": true }, { "refId": "E", - "expr": "100 * (max by (key) (asap_topk{method=\"raw\",bound=\"estimate\"} and (timestamp(asap_topk{method=\"raw\",bound=\"estimate\"}) == scalar(max(timestamp(asap_topk{method=\"raw\",bound=\"estimate\"}))))) - on (key) max by (key) (asap_topk{method=\"exact\",bound=\"estimate\"} and (timestamp(asap_topk{method=\"exact\",bound=\"estimate\"}) == scalar(max(timestamp(asap_topk{method=\"exact\",bound=\"estimate\"})))))) / on (key) max by (key) (asap_topk{method=\"exact\",bound=\"estimate\"} and (timestamp(asap_topk{method=\"exact\",bound=\"estimate\"}) == scalar(max(timestamp(asap_topk{method=\"exact\",bound=\"estimate\"}))))) and on (key) (max by (key, exact_rank) (label_replace(asap_topk{method=\"exact\",bound=\"estimate\",rank=~\"[1-9]|10\"} and (timestamp(asap_topk{method=\"exact\",bound=\"estimate\",rank=~\"[1-9]|10\"}) == scalar(max(timestamp(asap_topk{method=\"exact\",bound=\"estimate\",rank=~\"[1-9]|10\"})))), \"exact_rank\", \"$1\", \"rank\", \"(.*)\")) or max by (key, sketch_rank) (label_replace(asap_topk{method=\"sketch\",bound=\"estimate\",rank=~\"[1-9]|10\"} and (timestamp(asap_topk{method=\"sketch\",bound=\"estimate\",rank=~\"[1-9]|10\"}) == scalar(max(timestamp(asap_topk{method=\"sketch\",bound=\"estimate\",rank=~\"[1-9]|10\"})))), \"sketch_rank\", \"$1\", \"rank\", \"(.*)\")) or max by (key, raw_rank) (label_replace(asap_topk{method=\"raw\",bound=\"estimate\",rank=~\"[1-9]|10\"} and (timestamp(asap_topk{method=\"raw\",bound=\"estimate\",rank=~\"[1-9]|10\"}) == scalar(max(timestamp(asap_topk{method=\"raw\",bound=\"estimate\",rank=~\"[1-9]|10\"})))), \"raw_rank\", \"$1\", \"rank\", \"(.*)\")))", + "expr": "100 * (max by (key) (asap_compare_topk{method=\"raw\"}) - on (key) max by (key) (asap_compare_topk{method=\"exact\"})) / on (key) max by (key) (asap_compare_topk{method=\"exact\"}) and on (key) (max by (key, exact_rank) (label_replace(asap_compare_topk{method=\"exact\",rank=~\"[1-9]|10\"}, \"exact_rank\", \"$1\", \"rank\", \"(.*)\")) or max by (key, sketch_rank) (label_replace(asap_compare_topk{method=\"sketch\",rank=~\"[1-9]|10\"}, \"sketch_rank\", \"$1\", \"rank\", \"(.*)\")) or max by (key, raw_rank) (label_replace(asap_compare_topk{method=\"raw\",rank=~\"[1-9]|10\"}, \"raw_rank\", \"$1\", \"rank\", \"(.*)\")))", "format": "table", "instant": true } @@ -268,12 +268,53 @@ }, "gridPos": { "h": 12, - "w": 24, + "w": 20, "x": 0, "y": 11 }, "id": 3 }, + { + "type": "stat", + "title": "Window compared (end)", + "datasource": { + "type": "prometheus", + "uid": "asap-prometheus" + }, + "description": "End of the window the table compares: the latest one store and query has answered.", + "targets": [ + { + "refId": "A", + "expr": "asap_compare_window_end", + "instant": true + } + ], + "options": { + "reduceOptions": { + "calcs": [ + "lastNotNull" + ], + "fields": "", + "values": false + }, + "colorMode": "none", + "graphMode": "none", + "textMode": "value" + }, + "fieldConfig": { + "defaults": { + "unit": "dateTimeAsIsoNoDateIfToday" + }, + "overrides": [] + }, + "gridPos": { + "h": 12, + "w": 4, + "x": 20, + "y": 11 + }, + "id": 4 + }, { "type": "bargauge", "title": "Sketch (CMS + heap): top-10 routes, latest window", @@ -316,7 +357,7 @@ "x": 0, "y": 23 }, - "id": 4 + "id": 5 }, { "type": "bargauge", @@ -360,7 +401,7 @@ "x": 8, "y": 23 }, - "id": 5 + "id": 6 }, { "type": "bargauge", @@ -404,7 +445,7 @@ "x": 16, "y": 23 }, - "id": 6 + "id": 7 }, { "type": "logs", @@ -430,7 +471,7 @@ "x": 0, "y": 33 }, - "id": 7 + "id": 8 }, { "type": "row", @@ -443,7 +484,7 @@ "x": 0, "y": 41 }, - "id": 8 + "id": 9 }, { "type": "timeseries", @@ -510,7 +551,7 @@ "x": 0, "y": 42 }, - "id": 9 + "id": 10 }, { "type": "timeseries", @@ -589,7 +630,7 @@ "x": 12, "y": 42 }, - "id": 10 + "id": 11 }, { "type": "row", @@ -602,7 +643,7 @@ "x": 0, "y": 50 }, - "id": 11 + "id": 12 }, { "type": "timeseries", @@ -611,16 +652,16 @@ "type": "prometheus", "uid": "asap-prometheus" }, - "description": "Both sources together. The generators skip batches a pipeline cannot take, so a pipeline that falls behind delivers less.", + "description": "Both sources together. The generators skip batches a pipeline cannot take, so a pipeline that falls behind delivers less. Store and query: points loki-raw acknowledged within 50 s of the window's end. A pipeline that stops reporting shows a gap after 2 minutes.", "targets": [ { "refId": "A", - "expr": "sum by (method) (asap_points{method=~\"sketch|exact\",attribute_key=\"http.route\"}) / 60", + "expr": "sum by (method) (last_over_time(asap_points{method=~\"sketch|exact\",attribute_key=\"http.route\"}[2m])) / 60", "legendFormat": "{{method}}" }, { "refId": "B", - "expr": "sum(asap_points{method=\"raw\"}) / 60", + "expr": "sum(last_over_time(asap_points{method=\"raw\"}[2m])) / 60", "legendFormat": "store and query" } ], @@ -696,7 +737,7 @@ "x": 0, "y": 51 }, - "id": 12 + "id": 13 }, { "type": "timeseries", @@ -739,7 +780,7 @@ "x": 12, "y": 51 }, - "id": 13 + "id": 14 }, { "type": "row", @@ -752,7 +793,7 @@ "x": 0, "y": 59 }, - "id": 14 + "id": 15 }, { "type": "timeseries", @@ -851,7 +892,7 @@ "x": 0, "y": 60 }, - "id": 15 + "id": 16 }, { "type": "timeseries", @@ -925,7 +966,7 @@ "x": 0, "y": 67 }, - "id": 16 + "id": 17 }, { "type": "timeseries", @@ -999,7 +1040,7 @@ "x": 8, "y": 67 }, - "id": 17 + "id": 18 }, { "type": "timeseries", @@ -1008,7 +1049,7 @@ "type": "prometheus", "uid": "asap-prometheus" }, - "description": "", + "description": "Seconds per segment. into loki-raw: the window's end until loki-raw acknowledged its last point (or raw-sink's 50 s timeout). The query time includes refused and timed-out queries; with several queries per window it is their sum.", "targets": [ { "refId": "A", @@ -1058,11 +1099,11 @@ "x": 16, "y": 67 }, - "id": 18 + "id": 19 }, { "type": "row", - "title": "Resources per approach (generators and the hh-accuracy observer excluded)", + "title": "Resources per approach (generators and observers excluded)", "collapsed": false, "panels": [], "gridPos": { @@ -1071,7 +1112,7 @@ "x": 0, "y": 75 }, - "id": 19 + "id": 20 }, { "type": "timeseries", @@ -1122,7 +1163,7 @@ "x": 0, "y": 76 }, - "id": 20 + "id": 21 }, { "type": "timeseries", @@ -1173,7 +1214,7 @@ "x": 8, "y": 76 }, - "id": 21 + "id": 22 }, { "type": "timeseries", @@ -1224,7 +1265,7 @@ "x": 16, "y": 76 }, - "id": 22 + "id": 23 }, { "type": "timeseries", @@ -1275,7 +1316,7 @@ "x": 0, "y": 88 }, - "id": 23 + "id": 24 }, { "type": "timeseries", @@ -1326,7 +1367,7 @@ "x": 8, "y": 88 }, - "id": 24 + "id": 25 }, { "type": "timeseries", @@ -1377,7 +1418,7 @@ "x": 16, "y": 88 }, - "id": 25 + "id": 26 }, { "type": "timeseries", @@ -1426,7 +1467,7 @@ "x": 0, "y": 100 }, - "id": 26 + "id": 27 }, { "type": "timeseries", @@ -1475,7 +1516,7 @@ "x": 8, "y": 100 }, - "id": 27 + "id": 28 }, { "type": "timeseries", @@ -1524,7 +1565,7 @@ "x": 16, "y": 100 }, - "id": 28 + "id": 29 }, { "type": "timeseries", @@ -1567,7 +1608,7 @@ "x": 0, "y": 112 }, - "id": 29 + "id": 30 }, { "type": "timeseries", @@ -1610,7 +1651,7 @@ "x": 12, "y": 112 }, - "id": 30 + "id": 31 } ] } diff --git a/deploy/telemetry-functions/grafana/gen_dashboards.py b/deploy/telemetry-functions/grafana/gen_dashboards.py new file mode 100644 index 0000000..4e6b751 --- /dev/null +++ b/deploy/telemetry-functions/grafana/gen_dashboards.py @@ -0,0 +1,441 @@ +"""Builds the top-k and cardinality dashboards in grafana/dashboards: + python3 grafana/gen_dashboards.py topk grafana/dashboards/topk.json + python3 grafana/gen_dashboards.py cardinality grafana/dashboards/cardinality.json +""" +import json +import sys + +USE_CASE, OUT = sys.argv[1], sys.argv[2] +PROM = {"type": "prometheus", "uid": "asap-prometheus"} +LOKI = {"type": "loki", "uid": "asap-loki"} +COLORS = {"sketch": "#5794f2", "exact": "#ff9830", "store and query": "#8e8e8e", + "raw": "#8e8e8e", "Prometheus (shared)": "#73bf69"} +panels = [] +y = 0 + + +def color_overrides(names): + return [{"matcher": {"id": "byName", "options": n}, + "properties": [{"id": "color", "value": {"mode": "fixed", "fixedColor": COLORS[n]}}]} + for n in names if n in COLORS] + + +def add(panel, w, h, x): + panel["gridPos"] = {"h": h, "w": w, "x": x, "y": y} + panel["id"] = len(panels) + 1 + panels.append(panel) + + +def row(title): + global y + add({"type": "row", "title": title, "collapsed": False, "panels": []}, 24, 1, 0) + y += 1 + + +def ts(title, targets, unit=None, desc="", stack=False, log=False, w=12, x=0, h=8, + min0=True, names=(), last=False): + custom = {"spanNulls": True} + if stack: + custom.update({"stacking": {"mode": "normal"}, "fillOpacity": 60, + "lineInterpolation": "stepAfter", "lineWidth": 1}) + if log: + custom["scaleDistribution"] = {"type": "log", "log": 10} + defaults = {"custom": custom} + if unit: + defaults["unit"] = unit + if min0 and not log: + defaults["min"] = 0 + add({"type": "timeseries", "title": title, "datasource": PROM, "description": desc, + "targets": [dict(refId=chr(65 + i), expr=e, legendFormat=l) for i, (e, l) in enumerate(targets)], + "fieldConfig": {"defaults": defaults, "overrides": color_overrides(names)}, + "options": {"legend": {"displayMode": "table", "placement": "bottom", "calcs": ["lastNotNull"]} + if last else {"displayMode": "list", "placement": "bottom"}, + "tooltip": {"mode": "multi", "sort": "none"}}}, w, h, x) + + +def latest(sel): + """Only the series of the latest window (older keys stay within the + lookback otherwise).""" + return f"{sel} and (timestamp({sel}) == scalar(max(timestamp({sel}))))" + + +def box(name, small, color, bg): + return (f'
{name}
{small}
') + + +ARROW = '→' + + +def flow(cells): + return "" + ARROW.join(cells) + "" + + +backend = ("#6e7681", "none") +blue, orange, grey = ("#5794f2", "rgba(87,148,242,0.12)"), ("#ff9830", "rgba(255,152,48,0.10)"), ("#8e8e8e", "rgba(142,142,142,0.12)") + + +def by_name(name, *properties): + return {"matcher": {"id": "byName", "options": name}, "properties": list(properties)} + + +def colored(name, method): + return by_name(name, {"id": "color", "value": {"mode": "fixed", "fixedColor": COLORS[method]}}, + {"id": "custom.cellOptions", "value": {"type": "color-text"}}) + + +def window_compared(h): + add({"type": "stat", "title": "Window compared (end)", "datasource": PROM, + "description": "End of the window the table compares: the latest one store and query has answered.", + "targets": [{"refId": "A", "expr": "asap_compare_window_end", "instant": True}], + "options": {"reduceOptions": {"calcs": ["lastNotNull"], "fields": "", "values": False}, + "colorMode": "none", "graphMode": "none", "textMode": "value"}, + "fieldConfig": {"defaults": {"unit": "dateTimeAsIsoNoDateIfToday"}, "overrides": []}}, 4, h, 20) + + +def topk(): + global y + html = ('' + + flow([box("generator-a/b", "sources A and B", *blue), + box("branch-a/b", "Count-Min 4×512 + top-20 heap per 1-min window", *blue), + box("merge", "both sources' sketches per window", *blue), + box("detector", "top-20 with bounds; new heavy hitters", *blue), + box("Prometheus", "query: method="sketch"", *backend)]) + + flow([box("exact-generator-a/b", "same points", *orange), + box("exact-branch-a/b", "route → count map per 1-min window", *orange), + box("exact-merge", "adds both sources' maps: every key", *orange), + box("exact-detector", "exact top-20; new heavy hitters", *orange), + box("Prometheus", "query: method="exact"", *backend)]) + + flow([box("raw-generator-a/b", "same points", *grey), + box("raw-sink-a/b", "the only stage: every point into the log store", *grey), + box("loki-raw", "stores every raw point", *grey), + box("query-job", "LogQL top-20 once per window", *grey), + box("Prometheus", "answer: method="raw"", *backend)]) + + "
" + "

Every stage is one pinned container. Traffic grows over 30 minutes: " + "the rate per source doubles every 5 minutes (5k to 80k points/s), /new/hot-route takes 15% " + "of traffic at minute 10, and the number of routes grows from 100,000 to 1,000,000 at minute 15. " + "hh-accuracy compares the sketch with exact; query-job compares store and query with exact and times " + "every approach's backend query.

") + add({"type": "text", "title": "Three approaches to the same question: which routes are heavy hitters, and which are new?", + "options": {"mode": "html", "content": html}}, 24, 10, 0) + y += 10 + + row("Answers") + TOP10 = 'rank=~"[1-9]|10"' + + + def top10(method, label): + sel = f'asap_compare_topk{{method="{method}",{TOP10}}}' + return f'max by (key, {label}) (label_replace({sel}, "{label}", "$1", "rank", "(.*)"))' + + + def count(method): + return f'asap_compare_topk{{method="{method}"}}' + + + def error(method): + return (f'100 * (max by (key) ({count(method)}) - on (key) max by (key) ({count("exact")}))' + f' / on (key) max by (key) ({count("exact")})' + f' and on (key) ({" or ".join(top10(m, m + "_rank") for m in ("exact", "sketch", "raw"))})') + + + compare = [(top10("exact", "exact_rank"), "exact"), (top10("sketch", "sketch_rank"), "sketch"), + (top10("raw", "raw_rank"), "raw"), (error("sketch"), "sketch_error"), (error("raw"), "raw_error")] + columns = {"key": "Route", "exact_rank": "Exact rank", "Value #A": "Exact count", + "sketch_rank": "Sketch rank", "Value #B": "Sketch count", "Value #D": "Sketch error", + "raw_rank": "Store+query rank", "Value #C": "Store+query count", + "Value #E": "Store+query error"} + + + + + add({"type": "table", "title": "Top-10 routes of the three approaches, latest window store and query answered (union of their top-10s)", + "description": "One row per route in any approach's top-10, all from the latest window store and query has " + "answered (query-job republishes sketch and exact for it), which lags behind when the log " + "store is slow; the panel on the right shows which window. Error is (count - exact) / exact; " + "empty when the route is outside that approach's reported top-20.", + "datasource": PROM, + "targets": [{"refId": chr(65 + i), "expr": e, "format": "table", "instant": True} + for i, (e, _) in enumerate(compare)], + "transformations": [ + {"id": "joinByField", "options": {"byField": "key", "mode": "outer"}}, + {"id": "organize", "options": { + "excludeByName": {f: True for f in ["Time", "Time 1", "Time 2", "Time 3", "Time 4", "Time 5"]}, + "indexByName": {name: i for i, name in enumerate(columns)}, + "renameByName": columns}}, + {"id": "convertFieldType", "options": {"conversions": [ + {"targetField": c, "destinationType": "number"} + for c in ["Exact rank", "Sketch rank", "Store+query rank"]]}}], + "options": {"showHeader": True, "sortBy": [{"displayName": "Exact count", "desc": True}]}, + "fieldConfig": {"defaults": {"decimals": 0, "custom": {"align": "auto"}}, + "overrides": [ + by_name("Sketch count", {"id": "color", "value": {"mode": "fixed", "fixedColor": COLORS["sketch"]}}, + {"id": "custom.cellOptions", "value": {"type": "color-text"}}), + by_name("Exact count", {"id": "color", "value": {"mode": "fixed", "fixedColor": COLORS["exact"]}}, + {"id": "custom.cellOptions", "value": {"type": "color-text"}}), + by_name("Store+query count", + {"id": "color", "value": {"mode": "fixed", "fixedColor": COLORS["raw"]}}, + {"id": "custom.cellOptions", "value": {"type": "color-text"}}), + by_name("Route", {"id": "custom.width", "value": 150}), + by_name("Sketch error", {"id": "unit", "value": "percent"}, {"id": "decimals", "value": 1}), + by_name("Store+query error", {"id": "unit", "value": "percent"}, + {"id": "decimals", "value": 1})]}}, 20, 12, 0) + window_compared(12) + y += 12 + for i, (method, title) in enumerate([("sketch", "Sketch (CMS + heap): top-10 routes, latest window"), + ("exact", "Exact: top-10 routes, latest window"), + ("raw", "Store and query (LogQL over loki-raw): top-10 routes, latest window")]): + sel = f'asap_topk{{method="{method}",bound="estimate",rank=~"[1-9]|10"}}' + add({"type": "bargauge", "title": title, "datasource": PROM, + "targets": [{"refId": "A", "expr": f"sort_desc({latest(sel)})", "legendFormat": "#{{rank}} {{key}}", + "instant": True}], + "options": {"orientation": "horizontal", "displayMode": "basic", + "reduceOptions": {"calcs": ["lastNotNull"], "fields": "", "values": False}}, + "fieldConfig": {"defaults": {"decimals": 0, "color": {"mode": "fixed", "fixedColor": COLORS[method]}}, + "overrides": []}}, 8, 10, 8 * i) + y += 10 + add({"type": "logs", "title": "New heavy hitters (Loki events from both detectors)", "datasource": LOKI, + "targets": [{"refId": "A", "expr": '{service_name="asap-detector"} | event_kind="novelty"'}], + "options": {"showTime": True, "wrapLogMessage": True, "sortOrder": "Descending"}}, 24, 8, 0) + y += 8 + + row("Accuracy against exact") + ts("Heavy hitters (routes with >= 1% of the window's points): precision, recall, F1", + [('asap_hh_precision{method="sketch"}', "sketch precision"), ('asap_hh_recall{method="sketch"}', "sketch recall"), + ('asap_hh_f1{method="sketch"}', "sketch F1"), ('asap_hh_precision{method="raw"}', "store and query precision"), + ('asap_hh_recall{method="raw"}', "store and query recall"), ('asap_hh_f1{method="raw"}', "store and query F1")], + unit="percentunit", desc="Truth: exact counts >= 1% of the window's points. Store and query loses accuracy " + "only when points never reached the log store.", min0=False) + ts("Average relative count error of the heavy hitters found (ARE)", + [('asap_hh_are{method="sketch"}', "sketch"), ('asap_hh_are{method="raw"}', "store and query")], + unit="percentunit", x=12, names=("sketch", "store and query")) + y += 8 + + row("Traffic over time: volume and shape") + ts("End-to-end throughput: points per second each window delivered to the end of its pipeline", + [('sum by (method) (last_over_time(asap_points{method=~"sketch|exact",attribute_key="http.route"}[2m])) / 60', + "{{method}}"), + ('sum(last_over_time(asap_points{method="raw"}[2m])) / 60', "store and query")], + unit="short", desc="Both sources together. The generators skip batches a pipeline cannot take, so a " + "pipeline that falls behind delivers less. Store and query: points loki-raw acknowledged within 50 s of the " + "window's end. A pipeline that stops reporting shows a gap after 2 minutes.", names=("sketch", "exact", "store and query")) + ts("Data shape: distinct routes per window (exact)", + [('asap_distinct{method="exact",attribute_key="http.route"}', "distinct routes")], unit="short", x=12) + y += 8 + + + +def freshness(RAW_QUERY): + global y + row("Freshness: from a sample's generation to the queryable answer") + ts("Total freshness of a window's first sample", + [('60 + max(asap_answer_delay{method="sketch"}) + max(asap_query_seconds{method="sketch"})', "sketch"), + ('60 + max(asap_answer_delay{method="exact"}) + max(asap_query_seconds{method="exact"})', "exact"), + ('60 + max(asap_answer_delay{method="raw"}) + max(asap_query_seconds{method="raw"})', "store and query")], + unit="s", w=24, h=7, names=("sketch", "exact", "store and query"), + desc="Window wait (up to 60 s for the first sample) + the pipeline until the answer is stored + the backend " + "query that returns it.") + y += 7 + for i, method in enumerate(("sketch", "exact")): + st = lambda metric, stage: f'max(asap_stage_{metric}{{method="{method}",stage="{stage}"}})' + ts(f"{'Sketch' if method == 'sketch' else 'Exact'}: latency breakdown per window", + [('0 * max(asap_answer_delay{method="%s"}) + 60' % method, "window wait"), + (st("received", "merge"), "branch + to merge"), + (f'{st("emitted", "merge")} - {st("received", "merge")}', "merge (waits for both sources)"), + (f'{st("received", "detector")} - {st("emitted", "merge")}', "to detector"), + (f'{st("emitted", "detector")} - {st("received", "detector")}', "detector"), + (f'max(asap_query_seconds{{method="{method}"}})', "Prometheus query")], + unit="s", stack=True, w=8, x=8 * i, + desc="Seconds per segment. branch + to merge: the window's end until the last source's sketch reached the " + "merge. merge: waiting for the other source's sketch (5 s timeout).") + ts("Store and query: latency breakdown per window", + [('0 * max(asap_answer_delay{method="raw"}) + 60', "window wait"), + ('max(asap_answer_delay{method="raw"})', "into loki-raw (last point acknowledged)"), + ('max(asap_query_seconds{method="raw"})', RAW_QUERY)], + unit="s", stack=True, w=8, x=16, + desc="Seconds per segment. into loki-raw: the window's end until loki-raw acknowledged its last point (or " + "raw-sink's 50 s timeout). The query time includes refused and timed-out queries; with several queries per " + "window it is their sum.") + y += 8 + + + +APPROACHES = [ + # (title, collector stages, process jobs scraped directly) + ("Sketch", "branch-.|merge|detector", None), + ("Exact", "exact-(branch-.|merge|detector)", None), + ("Store and query", "raw-sink-.", "loki-raw|query-job"), +] + + +def cpu(sel, jobs): + return (f'sum by (stage) (rate(cpu_time_total{{job="collectors",stage=~"{sel}",otel_scope_pipeline_id="main"}}[1m]))' + + (f' or sum by (stage) (rate(process_cpu_seconds_total{{job=~"{jobs}"}}[1m]))' if jobs else "")) + + +def mem(sel, jobs): + return (f'max by (stage) (memory_rss{{job="collectors",stage=~"{sel}"}})' + + (f' or max by (stage) (process_resident_memory_bytes{{job=~"{jobs}"}})' if jobs else "")) + + +def bits(sel, jobs): + return (f'sum by (stage) (rate(payload_size_bytes_total{{job="collectors",stage=~"{sel}",' + f'otel_scope_node_type="receiver"}}[2m])) * 8' + + (' or label_replace(sum(rate(loki_distributor_bytes_received_total[2m])) * 8, ' + '"stage", "loki-raw", "", "")' if jobs else "")) + + +def per_approach(title, query, unit, desc, log=False): + global y + for i, (name, sel, jobs) in enumerate(APPROACHES): + expr = query(sel, jobs) + targets = [(expr, "{{stage}}")] + if query is not bits: + targets.append((f"sum({expr})", "total")) + ts(f"{name}: {title}", targets, unit=unit, log=log, w=8, x=8 * i, h=12, desc=desc, last=True) + y += 12 + + +def resources(): + global y + row("Resources per approach (generators and observers excluded)") + per_approach("CPU cores per component", cpu, "short", + "Pipeline stages from the engine's CPU time; loki-raw and query-job from their process CPU time. " + "total: the whole approach.") + per_approach("resident memory per component", mem, "bytes", "total: the whole approach.") + per_approach("bits per second into each stage", bits, "bps", + "into branch-a/b, exact-branch-a/b, raw-sink-a/b: raw points from the generators. into merge / " + "exact-merge: sketches / exact maps. into detector / exact-detector: merged windows. into loki-raw: " + "raw log records.", log=True) + ts("Shared by all three: Prometheus CPU cores", + [('sum(rate(process_cpu_seconds_total{job="prometheus"}[1m]))', "Prometheus")], unit="short", + desc="Prometheus holds the answers of all three approaches and the monitoring data.") + ts("Shared by all three: Prometheus resident memory", + [('max(process_resident_memory_bytes{job="prometheus"})', "Prometheus")], unit="bytes", x=12) + y += 8 + + +KEYS = ("http.route", "http.method", "k8s.pod.name", "user.id") + + +def cardinality(): + global y + html = ('' + + flow([box("generator-a/b", "sources A and B", *blue), + box("branch-a/b", "HLL per attribute key per 1-min window", *blue), + box("merge", "both sources' HLLs per window", *blue), + box("detector", "distinct counts with bounds; class per key", *blue), + box("Prometheus", "query: method="sketch"", *backend)]) + + flow([box("exact-generator-a/b", "same points", *orange), + box("exact-branch-a/b", "value → count map per key per 1-min window", *orange), + box("exact-merge", "adds both sources' maps: every value", *orange), + box("exact-detector", "exact distinct counts and classes", *orange), + box("Prometheus", "query: method="exact"", *backend)]) + + flow([box("raw-generator-a/b", "same points", *grey), + box("raw-sink-a/b", "the only stage: every point into the log store", *grey), + box("loki-raw", "stores every raw point", *grey), + box("query-job", "LogQL distinct count per key once per window", *grey), + box("Prometheus", "answer: method="raw"", *backend)]) + + "
" + "

Every stage is one pinned container. Four attributes: " + "http.route and http.method stable, k8s.pod.name growing 5%/min, " + "user.id unique per point. Traffic grows over 30 minutes: the rate per source doubles every " + "5 minutes (5k to 80k points/s), and at minute 10 http.route explodes from 20 to 20,000 " + "values. query-job compares store and query with exact and times every approach's backend " + "query.

") + add({"type": "text", "title": "Three approaches to the same question: how many distinct values does each attribute have?", + "options": {"mode": "html", "content": html}}, 24, 10, 0) + y += 10 + + row("Answers") + + def distinct(method): + sel = f'asap_compare_distinct{{method="{method}"}}' + return f"max by (attribute_key) ({sel})" + + def error(method): + return f'100 * ({distinct(method)} - on (attribute_key) {distinct("exact")}) / on (attribute_key) {distinct("exact")}' + + def state(method, label): + sel = f'asap_compare_class{{method="{method}"}}' + return f'max by (attribute_key, {label}) (label_replace({sel}, "{label}", "$1", "state", "(.*)"))' + + compare = [distinct("exact"), distinct("sketch"), distinct("raw"), error("sketch"), error("raw"), + state("exact", "exact_class"), state("sketch", "sketch_class"), + "max by (attribute_key) (asap_compare_failed)"] + columns = {"attribute_key": "Attribute key", "Value #A": "Exact distinct", "exact_class": "Exact class", + "Value #B": "Sketch estimate", "Value #D": "Sketch error", "sketch_class": "Sketch class", + "Value #C": "Store+query", "Value #E": "Store+query error", "Value #H": "Store+query refused"} + hidden = ["Time", "Value #F", "Value #G"] + [f"Time {i}" for i in range(1, 9)] + add({"type": "table", "title": "Distinct values per attribute key, latest window store and query answered", + "description": "All from the latest window store and query has answered (query-job republishes sketch and " + "exact for it), which lags behind when the log store is slow; the panel on the right shows " + "which window. Error is (count - exact) / exact. Class: the detector's verdict from the last 10 windows " + "(stable, growing, identifier-like). Store+query refused: 1 when Loki refused or timed out " + "the key's LogQL query, which leaves its count empty.", + "datasource": PROM, + "targets": [{"refId": chr(65 + i), "expr": e, "format": "table", "instant": True} + for i, e in enumerate(compare)], + "transformations": [ + {"id": "joinByField", "options": {"byField": "attribute_key", "mode": "outer"}}, + {"id": "organize", "options": { + "excludeByName": {f: True for f in hidden}, + "indexByName": {name: i for i, name in enumerate(columns)}, + "renameByName": columns}}], + "options": {"showHeader": True, "sortBy": [{"displayName": "Exact distinct", "desc": True}]}, + "fieldConfig": {"defaults": {"decimals": 0, "custom": {"align": "auto"}}, + "overrides": [colored("Exact distinct", "exact"), colored("Sketch estimate", "sketch"), + colored("Store+query", "raw"), + by_name("Sketch error", {"id": "unit", "value": "percent"}, + {"id": "decimals", "value": 2}), + by_name("Store+query error", {"id": "unit", "value": "percent"}, + {"id": "decimals", "value": 2})]}}, 20, 7, 0) + window_compared(7) + y += 7 + add({"type": "logs", "title": "Class changes (Loki events from both detectors)", "datasource": LOKI, + "targets": [{"refId": "A", "expr": '{service_name="asap-detector"} | event_kind="cardinality"'}], + "options": {"showTime": True, "wrapLogMessage": True, "sortOrder": "Descending"}}, 24, 8, 0) + y += 8 + + row("Accuracy against exact") + ex = 'max by (attribute_key) (asap_distinct{method="exact"})' + + def same_window(expr, method): + # Prometheus's lookback would otherwise pair a stale answer with a newer exact window. + stamp = lambda m: f'max by (attribute_key) (timestamp(asap_distinct{{method="{m}"}}))' + return f"({expr}) and on (attribute_key) ({stamp(method)} == on (attribute_key) {stamp('exact')})" + + for i, method in enumerate(("sketch", "raw")): + est = f'max by (attribute_key) (asap_distinct{{method="{method}",bound="estimate"}})' + ts(f"{'Sketch (HLL)' if method == 'sketch' else 'Store and query'}: relative error of the distinct count per key", + [(same_window(f'abs({est} - on (attribute_key) {ex}) / on (attribute_key) {ex}', method), + "{{attribute_key}}")], + unit="percentunit", x=12 * i, + desc="|estimate - exact| / exact per window." + ( + " Store and query is exact while every point reaches the log store." if method == "raw" else "")) + y += 8 + + row("Traffic over time: volume and shape") + ts("End-to-end throughput: points per second each window delivered to the end of its pipeline", + [('sum by (method) (last_over_time(asap_points{method=~"sketch|exact",attribute_key="http.route"}[2m])) / 60', + "{{method}}"), + ('sum(last_over_time(asap_points{method="raw"}[2m])) / 60', "store and query")], + unit="short", desc="Both sources together. The generators skip batches a pipeline cannot take, so a " + "pipeline that falls behind delivers less. Store and query: points loki-raw acknowledged within 50 s of the " + "window's end. A pipeline that stops reporting shows a gap after 2 minutes.", names=("sketch", "exact", "store and query")) + ts("Data shape: distinct values per key per window (exact), log scale", + [('max by (attribute_key) (asap_distinct{method="exact"})', "{{attribute_key}}")], unit="short", x=12, + log=True) + y += 8 + + +{"topk": topk, "cardinality": cardinality}[USE_CASE]() +freshness({"topk": "LogQL top-k query", "cardinality": "LogQL distinct counts (one query per key)"}[USE_CASE]) +resources() +TITLES = {"topk": "Heavy hitters: sketch vs exact vs store and query (real collector containers)", + "cardinality": "Cardinality: sketch vs exact vs store and query (real collector containers)"} +json.dump({"uid": f"asap-{USE_CASE}", "title": TITLES[USE_CASE], + "schemaVersion": 39, "version": 1, "refresh": "10s", "time": {"from": "now-35m", "to": "now"}, + "panels": panels}, open(OUT, "w"), indent=2) +open(OUT, "a").write("\n") diff --git a/deploy/telemetry-functions/loki-raw.yaml b/deploy/telemetry-functions/loki-raw.yaml index efc3c82..99e52db 100644 --- a/deploy/telemetry-functions/loki-raw.yaml +++ b/deploy/telemetry-functions/loki-raw.yaml @@ -33,8 +33,11 @@ limits_config: per_stream_rate_limit_burst: 2GB reject_old_samples: false max_structured_metadata_entries_count: 512 - # The query job's top-k groups by route: up to a million series per query. - max_query_series: 2000000 + # The query job groups by an attribute: up to a million routes (top-k) or + # every user.id of a window (cardinality). Even so, Loki refuses the user.id + # distinct count (HTTP 413) from about 2.4 million values per window, so + # store and query has no user.id answer from 20,000 points/s per source. + max_query_series: 20000000 query_timeout: 5m # Only the latest windows are queried; keep 2 h so a long-running demo does # not fill the disk (about 3 GB per hour at the top rate). diff --git a/deploy/telemetry-functions/query_job.py b/deploy/telemetry-functions/query_job.py index d453c44..68a6e03 100644 --- a/deploy/telemetry-functions/query_job.py +++ b/deploy/telemetry-functions/query_job.py @@ -4,23 +4,35 @@ Once per window, after the window is complete in each pipeline: -* sketch and exact: times the Prometheus query that reads the window's top-k +* sketch and exact: times the Prometheus query that reads the window's answer (`asap_query_seconds{backend="prometheus"}`). * raw (store and query): once both raw sinks report the window stored in - loki-raw, computes the top-k from the raw log records with one LogQL query, - times it (`asap_query_seconds{backend="loki"}`), and reports the answer - (`asap_topk{method="raw"}`) and its heavy-hitter accuracy against the exact - pipeline (`asap_hh_*{method="raw"}`, same definition as the hh-accuracy node). + loki-raw and the exact answer is in, computes the answer from the raw log + records with LogQL and times it (`asap_query_seconds{backend="loki"}`; the + whole answer when it takes several queries). + - topk: one top-k query; reports `asap_topk{method="raw"}` and its + heavy-hitter accuracy against exact (`asap_hh_*{method="raw"}`, same + definition as the hh-accuracy node). + - cardinality: one distinct-count query per attribute key exact counted; + reports `asap_distinct{method="raw"}`. A query Loki refuses or answers + with no result is reported as `asap_query_failed{method="raw"}` instead. + +With each raw answer it also republishes the sketch and exact answers of the +same window (`asap_compare_topk`, `asap_compare_distinct`, +`asap_compare_class`, labelled `method`; `asap_compare_failed` for refused +keys) and that window's end (`asap_compare_window_end`, ms). These carry no +timestamp, so a dashboard always finds the latest compared window however far +store and query lags behind. The raw answer runs once exact has answered the same window, so a slow exact pipeline delays it; freshness comes from the pipelines' own gauges and is not affected. A window raw-sink reported with points still unacknowledged (its 50 s timeout) is answered anyway and flagged with `asap_raw_incomplete`. -Gauges carry the window end as their timestamp, so Prometheus stores them at +The other gauges carry the window end as their timestamp, so Prometheus stores them at the same time as the pipelines' own per-window gauges. The job also exposes -its own CPU time and resident memory. Only the top-k use case has a raw answer -here; for other use cases it only times the backend queries. +its own CPU time and resident memory. Only the top-k and cardinality use cases +have a raw answer here; for the others it only times the backend queries. Configuration (environment): USE_CASE, WINDOW_SECONDS (60), K (20), PHI (0.01), RAW_SOURCES (2), PROMETHEUS, LOKI_RAW, PORT (9100). @@ -47,6 +59,9 @@ LOKI_LABEL = ATTRIBUTE.replace(".", "_") LOKI_STREAM = '{service_name="checkout"}' TOPK_SELECTOR = f'attribute_key="{ATTRIBUTE}",bound="estimate"' +# What a dashboard reads for each use case's answer. +ANSWER_METRIC = {"topk": "asap_topk", "cardinality": "asap_distinct", "latency": "asap_quantile", + "drift": "asap_drift_ks"}.get(USE_CASE, "asap_points") # Latest sample per series: (name, sorted labels) -> (value, timestamp ms). samples = {} @@ -84,8 +99,19 @@ def at_window(expr, end): def put(name, labels, value, end_s): + """Sets a gauge stamped at window end `end_s`; None leaves it unstamped + (Prometheus stamps it at the scrape).""" with lock: - samples[(name, tuple(sorted(labels.items())))] = (value, end_s * 1000) + samples[(name, tuple(sorted(labels.items())))] = (value, None if end_s is None else end_s * 1000) + + +def replace(name, rows, end_s): + """Replaces every sample of `name` with `rows` ((labels, value) pairs).""" + with lock: + for key in [k for k in samples if k[0] == name]: + del samples[key] + for labels, value in rows: + put(name, labels, value, end_s) def heavy_hitter_accuracy(points, exact, reported_list): @@ -107,7 +133,7 @@ def heavy_hitter_accuracy(points, exact, reported_list): def time_prometheus_answer(method, end): t0 = time.monotonic() - at_window(f'asap_topk{{method="{method}",{TOPK_SELECTOR}}}', end) + at_window(f'{ANSWER_METRIC}{{method="{method}"}}', end) put("asap_query_seconds", {"method": method, "backend": "prometheus"}, time.monotonic() - t0, end) @@ -116,8 +142,54 @@ def flag_incomplete(end): put("asap_raw_incomplete", {"method": "raw"}, 1.0 if unacknowledged > 0 else 0.0, end) -def raw_answer(end): - """Top-k of window (end - WINDOW_S, end] from loki-raw; True when done.""" +def loki_at_window_end(query, end): + """Instant LogQL query over window [end - WINDOW_S, end): LogQL ranges are + (t - W, t], so evaluate at the window's last millisecond.""" + return get(LOKI_RAW, "/loki/api/v1/query", {"query": query, "time": str(end * 1_000_000_000 - 1_000_000)}) + + +def raw_distinct(end): + """Distinct values per attribute key of the window from loki-raw; True when done.""" + exact = at_window('asap_distinct{method="exact"}', end) + if not exact: + return False + flag_incomplete(end) + failed = [] + compared = [({"method": m, "attribute_key": r["attribute_key"]}, v) for m in ("sketch", "exact") + for r, v in at_window(f'asap_distinct{{method="{m}",bound="estimate"}}', end)] + classes = [({"method": m, "attribute_key": r["attribute_key"], "state": r["state"]}, 1.0) + for m in ("sketch", "exact") + for r, v in at_window(f'asap_cardinality_state{{method="{m}"}}', end) if v == 1] + total = 0.0 + for key in sorted({r["attribute_key"] for r, _ in exact}): + label = key.replace(".", "_") + query = f"count(sum by ({label}) (count_over_time({LOKI_STREAM}[{WINDOW_S}s])))" + labels = {"method": "raw", "attribute_key": key} + t0 = time.monotonic() + try: + rows = loki_at_window_end(query, end) + if not rows: + raise ValueError("no result") + except Exception as e: # refused, timed out or empty: no answer for this key + total += time.monotonic() - t0 + print(f"query_job: {key} at {end}: {e}", flush=True) + put("asap_query_failed", labels, 1.0, end) + failed.append(({"attribute_key": key}, 1.0)) + continue + total += time.monotonic() - t0 + distinct = float(rows[0]["value"][1]) + put("asap_distinct", dict(labels, bound="estimate"), distinct, end) + compared.append((labels, distinct)) + put("asap_query_seconds", {"method": "raw", "backend": "loki"}, total, end) + replace("asap_compare_distinct", compared, None) + replace("asap_compare_class", classes, None) + replace("asap_compare_failed", failed, None) + put("asap_compare_window_end", {}, end * 1000.0, None) + return True + + +def raw_topk(end): + """Top-k of window [end - WINDOW_S, end) from loki-raw; True when done.""" exact_rows = at_window(f'asap_topk{{method="exact",{TOPK_SELECTOR}}}', end) exact_points = sum(v for _, v in at_window( f'asap_points{{method="exact",attribute_key="{ATTRIBUTE}"}}', end)) @@ -126,20 +198,21 @@ def raw_answer(end): flag_incomplete(end) query = (f"topk({K}, sum by ({LOKI_LABEL}) " f"(count_over_time({LOKI_STREAM}[{WINDOW_S}s])))") - # The window is [end - W, end) in ms; LogQL ranges are (t - W, t]. - at_ns = end * 1_000_000_000 - 1_000_000 t0 = time.monotonic() - rows = get(LOKI_RAW, "/loki/api/v1/query", {"query": query, "time": str(at_ns)}) + rows = loki_at_window_end(query, end) latency = time.monotonic() - t0 put("asap_query_seconds", {"method": "raw", "backend": "loki"}, latency, end) top = sorted(((r["metric"].get(LOKI_LABEL, ""), float(r["value"][1])) for r in rows), key=lambda kc: -kc[1]) - with lock: - for key in [k for k in samples if k[0] == "asap_topk"]: - del samples[key] - for rank, (key, count) in enumerate(top, 1): - put("asap_topk", {"method": "raw", "attribute_key": ATTRIBUTE, "bound": "estimate", - "rank": str(rank), "key": key}, count, end) + replace("asap_topk", [({"method": "raw", "attribute_key": ATTRIBUTE, "bound": "estimate", + "rank": str(rank), "key": key}, count) for rank, (key, count) in enumerate(top, 1)], end) + sketch_rows = at_window(f'asap_topk{{method="sketch",{TOPK_SELECTOR}}}', end) + replace("asap_compare_topk", + [({"method": m, "rank": r["rank"], "key": r["key"]}, v) + for m, rows_ in (("sketch", sketch_rows), ("exact", exact_rows)) for r, v in rows_] + + [({"method": "raw", "rank": str(rank), "key": key}, count) for rank, (key, count) in enumerate(top, 1)], + None) + put("asap_compare_window_end", {}, end * 1000.0, None) exact = sorted(((m["key"], v) for m, v in exact_rows), key=lambda kc: -kc[1]) acc = heavy_hitter_accuracy(exact_points, exact, top) labels = {"method": "raw", "attribute_key": ATTRIBUTE, "phi": f"{PHI:g}"} @@ -162,7 +235,8 @@ def loop(): if end not in done[method] and end > started - 2 * WINDOW_S: time_prometheus_answer(method, end) done[method].add(end) - if USE_CASE == "topk": + raw_answer = {"topk": raw_topk, "cardinality": raw_distinct}.get(USE_CASE) + if raw_answer: ready = window_ends('asap_answer_delay{method="raw"}') for end in sorted(e for e, n in ready.items() if n >= RAW_SOURCES): # Accuracy is against the exact answer of the same window. @@ -192,7 +266,7 @@ def do_GET(self): with lock: for (name, labels), (value, ts) in sorted(samples.items()): text = ",".join(f'{k}="{escape(v)}"' for k, v in labels) - lines.append(f"{name}{{{text}}} {value} {ts}") + lines.append(f"{name}{{{text}}} {value}" + ("" if ts is None else f" {ts}")) body = ("\n".join(lines) + "\n").encode() self.send_response(200) self.send_header("Content-Type", "text/plain; version=0.0.4")