From c5385c049bbefb2c8d7e2c07706eef4d717b0e0d Mon Sep 17 00:00:00 2001 From: Zeying Zhu <50204836+zzylol@users.noreply.github.com> Date: Mon, 19 Oct 2026 15:30:53 +0000 Subject: [PATCH 1/2] Report heavy-hitter precision, recall, F1 and ARE in the top-k demo A new observer stage, hh-accuracy (urn:asap:processor:heavy_hitter_accuracy), sits outside the three pipelines. In the top-k use case both detectors send it a copy of their answers through the official fanout. Per window and top-k series it takes the heavy hitters as the keys with at least phi (1%) of the window's points by the exact counts, and the sketch's reported keys as those whose estimate reaches the same threshold. It then emits asap.hh.precision, asap.hh.recall, asap.hh.f1, asap.hh.are (mean relative count error over the true positives), asap.hh.truth and asap.hh.reported. About 10 routes reach 1% in this scenario, so the sketch heap and both detectors now report the top 20, and the top-10 panels and checks filter on rank. A window whose exact top 20 is all above the threshold could hide heavy hitters and only gets asap.hh.incomplete. The dashboard gains precision/recall/F1 and ARE panels; acceptance checks precision and recall >= 0.9, and the accuracy-size experiment records them. --- .github/workflows/ci.yml | 1 + asap-precompute-rs/src/answers.rs | 94 ++++++ asap-precompute-rs/src/otap/hh_accuracy.rs | 296 ++++++++++++++++++ asap-precompute-rs/src/otap/mod.rs | 2 + deploy/telemetry-functions/README.md | 18 +- deploy/telemetry-functions/acceptance.py | 10 +- .../common/hh-accuracy.yaml | 36 +++ deploy/telemetry-functions/compose.yaml | 13 +- deploy/telemetry-functions/experiments.py | 9 +- .../grafana/dashboards/topk.json | 159 +++++++++- deploy/telemetry-functions/prometheus.yml | 3 +- deploy/telemetry-functions/topk/branch.yaml | 4 +- deploy/telemetry-functions/topk/detector.yaml | 25 +- .../topk/exact-detector.yaml | 22 +- 14 files changed, 666 insertions(+), 26 deletions(-) create mode 100644 asap-precompute-rs/src/otap/hh_accuracy.rs create mode 100644 deploy/telemetry-functions/common/hh-accuracy.yaml diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 7b056ae..6a4c16a 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -62,6 +62,7 @@ jobs: CPU_RAW_BRANCH_B: "0-3" CPU_RAW_MERGE: "0-3" CPU_RAW_SINK: "0-3" + CPU_HH_ACCURACY: "0-3" steps: - uses: actions/checkout@v4 - uses: actions/setup-python@v5 diff --git a/asap-precompute-rs/src/answers.rs b/asap-precompute-rs/src/answers.rs index e8a6300..823a0df 100644 --- a/asap-precompute-rs/src/answers.rs +++ b/asap-precompute-rs/src/answers.rs @@ -393,6 +393,72 @@ pub fn topk_of(answers: &[SketchEnvelope]) -> Vec<(String, f64)> { top.into_iter().map(|(_, k, v)| (k, v)).collect() } +/// How well a sketch's top-k finds and counts one window's heavy hitters: the +/// keys with at least `phi` of the window's points. +#[derive(Clone, Debug, PartialEq)] +pub struct HeavyHitterAccuracy { + /// Heavy hitters in the exact counts. + pub truth: usize, + /// Keys whose sketch estimate reaches the threshold. + pub reported: usize, + /// Reported keys that are heavy hitters. + pub true_positives: usize, + /// True positives / reported (1 when nothing is reported). + pub precision: f64, + /// True positives / heavy hitters (1 when there are none). + pub recall: f64, + /// Harmonic mean of precision and recall. + pub f1: f64, + /// Mean of |estimate - exact| / exact over the true positives (`None` + /// without any): the sketch has no count for a heavy hitter it missed. + pub average_relative_error: Option, +} + +/// Compares a window's sketch top-k with its exact top-k at threshold +/// `phi * points`. Both lists are `(key, count)` as reported, at most `k` +/// long. Returns `None` when the exact list is full and its smallest count +/// still reaches the threshold: the window may have more heavy hitters than +/// the list shows. +pub fn heavy_hitter_accuracy( + phi: f64, + points: f64, + k: usize, + exact: &[(String, f64)], + sketch: &[(String, f64)], +) -> Option { + let threshold = phi * points; + let smallest = exact.iter().map(|e| e.1).fold(f64::INFINITY, f64::min); + if exact.len() >= k && smallest >= threshold { + return None; + } + let truth: HashMap<&str, f64> = exact + .iter() + .filter(|e| e.1 >= threshold) + .map(|(key, c)| (key.as_str(), *c)) + .collect(); + let reported: Vec<&(String, f64)> = sketch.iter().filter(|e| e.1 >= threshold).collect(); + let errors: Vec = reported + .iter() + .filter_map(|(key, estimate)| truth.get(key.as_str()).map(|c| (estimate - c).abs() / c)) + .collect(); + let tp = errors.len(); + let ratio = |n: usize, d: usize| if d == 0 { 1.0 } else { n as f64 / d as f64 }; + let (precision, recall) = (ratio(tp, reported.len()), ratio(tp, truth.len())); + Some(HeavyHitterAccuracy { + truth: truth.len(), + reported: reported.len(), + true_positives: tp, + precision, + recall, + f1: if precision + recall == 0.0 { + 0.0 + } else { + 2.0 * precision * recall / (precision + recall) + }, + average_relative_error: (tp > 0).then(|| errors.iter().sum::() / tp as f64), + }) +} + /// Flags heavy hitters that were not among the top keys of the previous /// windows. Sketch and exact use the same rule. #[derive(Default)] @@ -821,3 +887,31 @@ impl CardinalityClassifier { (out, event) } } + +#[cfg(test)] +mod heavy_hitter_tests { + use super::*; + + #[test] + fn heavy_hitter_accuracy_counts_false_positives_and_misses() { + let list = |v: &[(&str, f64)]| { + v.iter() + .map(|(k, c)| (k.to_string(), *c)) + .collect::>() + }; + // 1,000 points, phi 10%: a, b, c are heavy; the sketch overestimates + // d past the threshold and misses c. + let exact = list(&[("a", 400.0), ("b", 200.0), ("c", 100.0), ("d", 90.0)]); + let sketch = list(&[("a", 410.0), ("b", 220.0), ("d", 105.0)]); + let acc = heavy_hitter_accuracy(0.1, 1000.0, 20, &exact, &sketch).unwrap(); + assert_eq!((acc.truth, acc.reported, acc.true_positives), (3, 3, 2)); + assert!((acc.precision - 2.0 / 3.0).abs() < 1e-12); + assert!((acc.recall - 2.0 / 3.0).abs() < 1e-12); + assert!((acc.f1 - 2.0 / 3.0).abs() < 1e-12); + // (10/400 + 20/200) / 2 + assert!((acc.average_relative_error.unwrap() - 0.0625).abs() < 1e-12); + // A full exact list whose last count is still heavy may be cut short. + assert!(heavy_hitter_accuracy(0.05, 1000.0, 4, &exact, &sketch).is_none()); + assert!(heavy_hitter_accuracy(0.1, 1000.0, 4, &exact, &sketch).is_some()); + } +} diff --git a/asap-precompute-rs/src/otap/hh_accuracy.rs b/asap-precompute-rs/src/otap/hh_accuracy.rs new file mode 100644 index 0000000..b7ea789 --- /dev/null +++ b/asap-precompute-rs/src/otap/hh_accuracy.rs @@ -0,0 +1,296 @@ +//! `urn:asap:processor:heavy_hitter_accuracy` watches the answers of the +//! sketch and exact pipelines (their detectors each send it a copy) and, per +//! window and top-k series, reports how well the sketch found and counted the +//! heavy hitters: keys with at least `phi` of the window's points. It sits +//! outside both pipelines and only reads their output. +//! +//! Gauges, labelled with the series labels and `phi`: `asap.hh.precision`, +//! `asap.hh.recall`, `asap.hh.f1`, `asap.hh.are` (average relative count error +//! over the true positives), `asap.hh.truth` and `asap.hh.reported`. A window +//! whose exact top-k may hide more heavy hitters (`k` entries, all above the +//! threshold) gets only `asap.hh.incomplete`. +//! +//! ```yaml +//! accuracy: +//! type: urn:asap:processor:heavy_hitter_accuracy +//! config: {phi: 0.01, k: 20} # k: the detectors' top-k length +//! ``` +use std::collections::BTreeMap; +use std::sync::Arc; + +use async_trait::async_trait; +use linkme::distributed_slice; +use otel_arrow_dfe_config::node::NodeUserConfig; +use otel_arrow_dfe_engine::config::ProcessorConfig; +use otel_arrow_dfe_engine::context::PipelineContext; +use otel_arrow_dfe_engine::control::AckMsg; +use otel_arrow_dfe_engine::error::Error; +use otel_arrow_dfe_engine::local::processor as local; +use otel_arrow_dfe_engine::message::Message; +use otel_arrow_dfe_engine::node::NodeId; +use otel_arrow_dfe_engine::processor::ProcessorWrapper; +use otel_arrow_dfe_engine::ConsumerEffectHandlerExtension; +use otel_arrow_dfe_engine::{MessageSourceLocalEffectHandlerExtension, ProcessorFactory}; +use otel_arrow_dfe_otap::pdata::OtapPdata; +use otel_arrow_dfe_otap::OTAP_PROCESSOR_FACTORIES; +use serde::Deserialize; + +use super::answer_nodes::{default_k, gauges_pdata, invalid, observations}; +use crate::answers::heavy_hitter_accuracy; +use crate::envelope::SketchEnvelope; +use crate::observation::{KeyValue, Observation}; + +/// Registered URN of the heavy-hitter accuracy observer. +pub const HH_ACCURACY_URN: &str = "urn:asap:processor:heavy_hitter_accuracy"; + +/// Windows kept waiting for the other method's answers. +const MAX_OPEN_WINDOWS: usize = 16; + +fn default_phi() -> f64 { + 0.01 +} + +/// `heavy_hitter_accuracy` configuration. +#[derive(Clone, Debug, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct HhAccuracyConfig { + /// A heavy hitter has at least this fraction of the window's points. + #[serde(default = "default_phi")] + pub phi: f64, + /// Length of the detectors' top-k lists. + #[serde(default = "default_k")] + pub k: usize, +} + +/// One method's answers for one window and series. +#[derive(Default)] +struct Answers { + points: Option, + /// `(rank, key, count)`. + top: Vec<(u32, String, f64)>, +} + +/// Window end, then series labels (without `method`, `key`, `rank`, `bound`). +type Key = (u64, Vec<(String, String)>); + +struct HhAccuracy { + config: HhAccuracyConfig, + open: BTreeMap, +} + +fn label<'a>(obs: &'a Observation, key: &str) -> Option<&'a str> { + obs.labels + .iter() + .find(|kv| kv.key == key) + .map(|kv| kv.value.as_str()) +} + +impl HhAccuracy { + /// Records one detector answer: a top-k estimate or a window's points. + fn observe(&mut self, obs: &Observation) { + let topk = obs.metric == "asap.topk" && label(obs, "bound") == Some("estimate"); + if !topk && obs.metric != "asap.points" { + return; + } + let method = match label(obs, "method") { + Some("sketch") => 0, + Some("exact") => 1, + _ => return, + }; + let mut series: Vec<(String, String)> = obs + .labels + .iter() + .filter(|kv| !["method", "key", "rank", "bound"].contains(&kv.key.as_str())) + .map(|kv| (kv.key.clone(), kv.value.clone())) + .collect(); + series.sort(); + let entry = self.open.entry((obs.timestamp_ms, series)).or_default(); + let answers = if method == 0 { + &mut entry.0 + } else { + &mut entry.1 + }; + if topk { + let rank = label(obs, "rank").and_then(|r| r.parse().ok()); + if let (Some(rank), Some(key)) = (rank, label(obs, "key")) { + answers.top.push((rank, key.to_string(), obs.value.float)); + } + } else { + answers.points = Some(obs.value.float); + } + } + + /// Accuracy gauges of the windows that have both methods' answers. + fn ready(&mut self) -> Vec { + let done: Vec = self + .open + .iter() + .filter(|(_, (s, e))| e.points.is_some() && !s.top.is_empty() && !e.top.is_empty()) + .map(|(k, _)| k.clone()) + .collect(); + let mut out = Vec::new(); + for key in done { + let (sketch, exact) = self.open.remove(&key).expect("listed above"); + out.extend(self.gauges(&key, sketch, exact)); + } + // A window one method never answered (e.g. an idle series). + while self.open.len() > MAX_OPEN_WINDOWS { + self.open.pop_first(); + } + out + } + + fn gauges(&self, (end, series): &Key, sketch: Answers, exact: Answers) -> Vec { + let list = |mut a: Answers| { + a.top.sort_by_key(|t| t.0); + a.top + .into_iter() + .map(|(_, k, c)| (k, c)) + .collect::>() + }; + let points = exact.points.unwrap_or(0.0); + let mut labels: Vec = series.iter().map(|(k, v)| KeyValue::new(k, v)).collect(); + labels.push(KeyValue::new("phi", self.config.phi.to_string())); + let gauge = |metric: &str, value: f64| SketchEnvelope { + labels: labels.clone(), + metric_name: metric.into(), + window_end_ms: *end, + window_start_ms: *end, + value, + ..Default::default() + }; + let (k, phi) = (self.config.k, self.config.phi); + let Some(acc) = heavy_hitter_accuracy(phi, points, k, &list(exact), &list(sketch)) else { + return vec![gauge("asap.hh.incomplete", 1.0)]; + }; + let mut out = vec![ + gauge("asap.hh.precision", acc.precision), + gauge("asap.hh.recall", acc.recall), + gauge("asap.hh.f1", acc.f1), + gauge("asap.hh.truth", acc.truth as f64), + gauge("asap.hh.reported", acc.reported as f64), + ]; + out.extend( + acc.average_relative_error + .map(|are| gauge("asap.hh.are", are)), + ); + out + } +} + +#[async_trait(?Send)] +impl local::Processor for HhAccuracy { + async fn process( + &mut self, + msg: Message, + effects: &mut local::EffectHandler, + ) -> Result<(), Error> { + let Message::PData(pdata) = msg else { + return Ok(()); + }; + effects.notify_ack(AckMsg::new(pdata.clone())).await?; + for obs in observations(pdata, effects).await { + self.observe(&obs); + } + let out = self.ready(); + if out.is_empty() { + return Ok(()); + } + match gauges_pdata(&out) { + Ok(pdata) => effects.send_message_with_source_node(pdata).await?, + Err(e) => { + effects + .info(&format!("hh_accuracy: encode failed: {e}")) + .await + } + } + Ok(()) + } +} + +#[allow(unsafe_code)] +#[distributed_slice(OTAP_PROCESSOR_FACTORIES)] +static HH_ACCURACY: ProcessorFactory = ProcessorFactory { + name: HH_ACCURACY_URN, + create: |_: PipelineContext, + node: NodeId, + node_config: Arc, + processor_config: &ProcessorConfig, + _: &otel_arrow_dfe_engine::capability::registry::Capabilities| { + let config: HhAccuracyConfig = + serde_json::from_value(node_config.config.clone()).map_err(invalid)?; + Ok(ProcessorWrapper::local( + HhAccuracy { + config, + open: BTreeMap::new(), + }, + node, + node_config, + processor_config, + )) + }, + wiring_contract: otel_arrow_dfe_engine::wiring_contract::WiringContract::UNRESTRICTED, + validate_config: otel_arrow_dfe_config::validation::validate_typed_config::, +}; + +#[cfg(test)] +mod tests { + use super::*; + use crate::observation::ObservationValue; + + fn answer(method: &str, metric: &str, extra: &[(&str, &str)], value: f64) -> Observation { + let mut labels = vec![ + KeyValue::new("attribute.key", "http.route"), + KeyValue::new("method", method), + ]; + labels.extend(extra.iter().map(|(k, v)| KeyValue::new(*k, *v))); + Observation::new( + 60_000, + metric, + vec![], + labels, + ObservationValue::float(value), + ) + } + + #[test] + fn reports_a_window_once_both_methods_answered() { + let mut node = HhAccuracy { + config: HhAccuracyConfig { phi: 0.1, k: 20 }, + open: BTreeMap::new(), + }; + let top = |method: &str, rank: &str, key: &str, v: f64| { + answer( + method, + "asap.topk", + &[("rank", rank), ("key", key), ("bound", "estimate")], + v, + ) + }; + node.observe(&answer("sketch", "asap.points", &[], 1000.0)); + node.observe(&top("sketch", "1", "a", 410.0)); + node.observe(&top("sketch", "2", "d", 105.0)); + // The sketch's lower bounds are not estimates. + node.observe(&answer( + "sketch", + "asap.topk", + &[("rank", "1"), ("key", "a"), ("bound", "lower")], + 300.0, + )); + assert!(node.ready().is_empty(), "waits for exact"); + node.observe(&answer("exact", "asap.points", &[], 1000.0)); + node.observe(&top("exact", "1", "a", 400.0)); + node.observe(&top("exact", "2", "c", 100.0)); + node.observe(&top("exact", "3", "d", 90.0)); + let out = node.ready(); + let get = |m: &str| out.iter().find(|e| e.metric_name == m).map(|e| e.value); + assert_eq!(get("asap.hh.precision"), Some(0.5)); + assert_eq!(get("asap.hh.recall"), Some(0.5)); + assert_eq!(get("asap.hh.are"), Some(0.025)); + assert!(out.iter().all(|e| e.window_end_ms == 60_000 + && e.labels + .contains(&KeyValue::new("attribute.key", "http.route")) + && e.labels.contains(&KeyValue::new("phi", "0.1")))); + assert!(node.open.is_empty()); + } +} diff --git a/asap-precompute-rs/src/otap/mod.rs b/asap-precompute-rs/src/otap/mod.rs index 838e0ba..b8bf06a 100644 --- a/asap-precompute-rs/src/otap/mod.rs +++ b/asap-precompute-rs/src/otap/mod.rs @@ -17,6 +17,8 @@ pub mod baseline_nodes; #[cfg(feature = "otap-engine")] pub mod codec; #[cfg(feature = "otap-engine")] +pub mod hh_accuracy; +#[cfg(feature = "otap-engine")] pub mod processor; #[cfg(feature = "otap-engine")] pub mod scenario_generator; diff --git a/deploy/telemetry-functions/README.md b/deploy/telemetry-functions/README.md index 8c35d32..0c5c226 100644 --- a/deploy/telemetry-functions/README.md +++ b/deploy/telemetry-functions/README.md @@ -27,6 +27,18 @@ of scenario time, so all three pipelines receive identical points. | exact-detector | `exact_detector`: the same answers and events from the exact windows (`method="exact"`): the ground truth | | raw-branch-a/b, raw-merge | forward raw points (OTLP receiver -> exporter) | | raw-sink | `raw_sink`: writes every raw point as one log record to `loki-raw` and reports when `loki-raw` acknowledged each window's last point | +| 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 +least 1% of the window's points by the exact counts; the sketch reports a +route when its estimate reaches 1%. Per window, `hh-accuracy` emits +`asap_hh_precision` (reported routes that are heavy hitters / reported), +`asap_hh_recall` (heavy hitters reported / heavy hitters), `asap_hh_f1`, +`asap_hh_are` (mean |estimate - exact| / exact over the heavy hitters found), +`asap_hh_truth` and `asap_hh_reported`. Both detectors report their top 20 so +that every heavy hitter is in the exact list (about 10 routes reach 1% here); +a window whose exact top 20 is all above the threshold only gets +`asap_hh_incomplete`. Every pipeline reports its window completion delay as `asap_answer_delay` (seconds from a window's end until its result is complete at the end of the @@ -45,7 +57,7 @@ reports each point's time from its event time to that acknowledgement | `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 | | `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-10 heap per attribute key | 100,000 Zipf routes; a new route takes 15% of traffic at minute 10 | `/d/asap-topk`: top-10 with bound vs exact, new heavy hitters | +| `topk` | Count-Min (4 x 512) + top-20 heap per attribute key | 100,000 Zipf routes; a new route takes 15% of traffic at minute 10 | `/d/asap-topk`: top-10 with bound vs exact, new heavy hitters, heavy-hitter precision/recall/F1/ARE | Answers reach Prometheus over OTLP as `asap_quantile`, `asap_distinct`, `asap_topk`, `asap_points`, `asap_cardinality_state`, `asap_drift_ks`, @@ -59,7 +71,7 @@ and payload bytes per receiver, labelled with the stage. ## Run -Linux, Docker with Compose, 18 free CPUs (0-17 by default; `CPU_BRANCH_A=20` +Linux, Docker with Compose, 19 free CPUs (0-18 by default; `CPU_BRANCH_A=20` etc. move a stage) and memory for `loki-raw` (capped at `LOKI_RAW_MEMORY`, default 16g). From the repository root: @@ -89,7 +101,7 @@ fails. The checks: | 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 | | 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; novelty 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; novelty event in Loki | CI runs it on every PR (`.github/workflows/ci.yml`) with the stages sharing the runner's CPUs: a correctness check, not a capacity measurement. diff --git a/deploy/telemetry-functions/acceptance.py b/deploy/telemetry-functions/acceptance.py index db61e40..f01398e 100755 --- a/deploy/telemetry-functions/acceptance.py +++ b/deploy/telemetry-functions/acceptance.py @@ -39,7 +39,8 @@ USE_CASES = ("latency", "cardinality", "drift", "topk") STAGES = tuple(f"{p}{s}" for p in ("", "exact-", "raw-") for s in ("generator-a", "generator-b")) STAGES += ("branch-a", "branch-b", "merge", "detector", "exact-branch-a", "exact-branch-b", - "exact-merge", "exact-detector", "raw-branch-a", "raw-branch-b", "raw-merge", "raw-sink") + "exact-merge", "exact-detector", "raw-branch-a", "raw-branch-b", "raw-merge", "raw-sink", + "hh-accuracy") METHODS = ("sketch", "exact", "raw") @@ -200,10 +201,15 @@ def topk_checks(stack, t): new = stack.vector('max_over_time(asap_topk{key="/new/hot-route",bound=~"estimate|"}[10m])') yield check("new route in both top-10s", {m["method"] for m, _ in new} == {"sketch", "exact"}, str(sorted(m["method"] for m, _ in new))) - top = {m: {r["key"] for r, _ in stack.vector(f'asap_topk{{method="{m}",bound=~"estimate|"}} @ {t}')} + top10 = 'rank=~"[1-9]|10"' + top = {m: {r["key"] for r, _ in stack.vector(f'asap_topk{{method="{m}",bound=~"estimate|",{top10}}} @ {t}')} for m in ("sketch", "exact")} recall = len(top["sketch"] & top["exact"]) / max(1, len(top["exact"])) yield check("sketch top-10 recall vs exact >= 0.9", recall >= 0.9, f"recall {recall:.2f}") + hh = {m["__name__"]: v for m, v in stack.vector(f'{{__name__=~"asap_hh_(precision|recall|are)"}} @ {t}')} + yield check("heavy hitters (>= 1% of points): precision and recall >= 0.9", + hh.get("asap_hh_precision", 0) >= 0.9 and hh.get("asap_hh_recall", 0) >= 0.9, + ", ".join(f"{k.removeprefix('asap_hh_')} {v:.3f}" for k, v in sorted(hh.items())) or "no asap_hh gauges") events = stack.events('| event_kind="novelty" | method="sketch"') flagged = stack.vector('max_over_time(asap_topk_new{key="/new/hot-route"}[10m])') yield check("novelty event in Loki", any("/new/hot-route" in e for e in events), diff --git a/deploy/telemetry-functions/common/hh-accuracy.yaml b/deploy/telemetry-functions/common/hh-accuracy.yaml new file mode 100644 index 0000000..beda466 --- /dev/null +++ b/deploy/telemetry-functions/common/hh-accuracy.yaml @@ -0,0 +1,36 @@ +# Outside the pipelines: compares the sketch and exact detectors' top-k per +# window and reports heavy-hitter precision, recall, F1 and average relative +# error (keys with at least 1% of the window's points) to Prometheus. +version: otel_dataflow/v1 +engine: + http_admin: + bind_address: 0.0.0.0:8085 +policies: + resources: + core_allocation: + type: core_count + count: 1 +groups: + default: + pipelines: + main: + nodes: + receiver: + type: receiver:otlp + config: + protocols: + http: {listening_addr: "0.0.0.0:4318", wait_for_result: false} + accuracy: + type: urn:asap:processor:heavy_hitter_accuracy + config: {phi: 0.01, k: 20} + to_prometheus: + type: exporter:otlp_http + config: + endpoint: http://prometheus:9090 + metrics_endpoint: http://prometheus:9090/api/v1/otlp/v1/metrics + client_pool_size: 1 + max_in_flight: 1 + http: {timeout: 30s, compression: none} + connections: + - {from: receiver, to: accuracy} + - {from: accuracy, to: to_prometheus} diff --git a/deploy/telemetry-functions/compose.yaml b/deploy/telemetry-functions/compose.yaml index 033fc0c..9fd0627 100644 --- a/deploy/telemetry-functions/compose.yaml +++ b/deploy/telemetry-functions/compose.yaml @@ -4,6 +4,8 @@ # exact-merge -> exact-detector # raw: raw-generator-a/b -> raw-branch-a/b -> raw-merge -> raw-sink # (every raw point into loki-raw) +# hh-accuracy sits outside the pipelines: in the top-k use case both detectors +# send it a copy of their answers, and it reports heavy-hitter accuracy. # Each stage is the official OTAP engine (asap-collector) in its own container, # pinned to one CPU, talking OTLP/HTTP over the compose network. The generators # of a source share one config (`downstream` is their pipeline's first stage) @@ -58,7 +60,7 @@ services: command: ["--config", "/pipelines/detector.yaml"] volumes: ["${PIPELINES:-./latency}:/pipelines:ro", "${COMMON:-./common}:/common:ro"] cpuset: "${CPU_DETECTOR:-5}" - depends_on: [prometheus] + depends_on: [prometheus, hh-accuracy] # exact pipeline exact-generator-a: <<: *generator @@ -97,7 +99,7 @@ services: command: ["--config", "/pipelines/exact-detector.yaml"] volumes: ["${PIPELINES:-./latency}:/pipelines:ro", "${COMMON:-./common}:/common:ro"] cpuset: "${CPU_EXACT_DETECTOR:-11}" - depends_on: [prometheus] + depends_on: [prometheus, hh-accuracy] # raw pipeline raw-generator-a: <<: *generator @@ -137,6 +139,13 @@ services: volumes: ["${PIPELINES:-./latency}:/pipelines:ro", "${COMMON:-./common}:/common:ro"] cpuset: "${CPU_RAW_SINK:-17}" depends_on: [prometheus, loki-raw] + # Outside the pipelines: compares the sketch and exact top-k answers. + hh-accuracy: + <<: *collector + command: ["--config", "/common/hh-accuracy.yaml"] + volumes: ["${PIPELINES:-./latency}:/pipelines:ro", "${COMMON:-./common}:/common:ro"] + cpuset: "${CPU_HH_ACCURACY:-18}" + depends_on: [prometheus] prometheus: image: prom/prometheus:v3.5.0 diff --git a/deploy/telemetry-functions/experiments.py b/deploy/telemetry-functions/experiments.py index c702eb9..ed7a12f 100755 --- a/deploy/telemetry-functions/experiments.py +++ b/deploy/telemetry-functions/experiments.py @@ -113,11 +113,16 @@ def latency_error(stack): def topk_error(stack): t = acc.latest_window(stack, 'attribute_key="http.route"') - get = lambda m: {r["key"]: v for r, v in stack.vector(f'asap_topk{{method="{m}",bound=~"estimate|"}} @ {t}')} + top10 = 'rank=~"[1-9]|10"' + get = lambda m: {r["key"]: v for r, v in + stack.vector(f'asap_topk{{method="{m}",bound=~"estimate|",{top10}}} @ {t}')} sketch, exact = get("sketch"), get("exact") errs = [abs(sketch[k] - exact[k]) / exact[k] for k in exact if k in sketch] + hh = {m["__name__"].removeprefix("asap_hh_"): v + for m, v in stack.vector(f'{{__name__=~"asap_hh_(precision|recall|f1|are)"}} @ {t}')} return {"recall_at_10": len(set(sketch) & set(exact)) / max(1, len(exact)), - "count_rel_error_mean": statistics.mean(errs) if errs else float("nan"), **merge_bytes(stack)} + "count_rel_error_mean": statistics.mean(errs) if errs else float("nan"), + **{f"hh_{k}": v for k, v in sorted(hh.items())}, **merge_bytes(stack)} def pipelines(args): diff --git a/deploy/telemetry-functions/grafana/dashboards/topk.json b/deploy/telemetry-functions/grafana/dashboards/topk.json index 0bef626..3626c58 100644 --- a/deploy/telemetry-functions/grafana/dashboards/topk.json +++ b/deploy/telemetry-functions/grafana/dashboards/topk.json @@ -21,7 +21,7 @@ }, "options": { "mode": "html", - "content": "
generator-a/b
sources A and B
→
branch-a/b
Stage 1 per source: Count-Min 4×512 + top-10 heap per 1-min window
→
merge
Stage 2: merges both sources' sketches per window
→
detector
top-10 with bounds; new heavy hitters
→
Prometheus + Loki
method="sketch"
exact-generator-a/b
same points
→
exact-branch-a/b
Stage 1 per source: route → count map per 1-min window
→
exact-merge
Stage 2: adds both sources' exact windows
→
exact-detector
exact top-10; new heavy hitters
→
Prometheus + Loki
method="exact" (ground truth)
raw-generator-a/b
same points
→
raw-branch-a/b
Stage 1 per source: forwards every raw point
→
raw-merge
Stage 2: forwards both sources' raw points
→
raw-sink
writes every raw point; measures when the window's last point is stored
→
loki-raw
one log record per raw point
" + "content": "
generator-a/b
sources A and B
→
branch-a/b
Stage 1 per source: Count-Min 4×512 + top-20 heap per 1-min window
→
merge
Stage 2: merges both sources' sketches per window
→
detector
top-20 with bounds; new heavy hitters
→
Prometheus + Loki
method="sketch"
exact-generator-a/b
same points
→
exact-branch-a/b
Stage 1 per source: route → count map per 1-min window
→
exact-merge
Stage 2: adds both sources' exact windows
→
exact-detector
exact top-20; new heavy hitters
→
Prometheus + Loki
method="exact" (ground truth)
raw-generator-a/b
same points
→
raw-branch-a/b
Stage 1 per source: forwards every raw point
→
raw-merge
Stage 2: forwards both sources' raw points
→
raw-sink
writes every raw point; measures when the window's last point is stored
→
loki-raw
one log record per raw point

hh-accuracy (outside the pipelines) receives a copy of both detectors' answers and, per window, compares the heavy hitters (routes with at least 1% of the window's points): precision, recall, F1 and average relative count error.

" }, "id": 1 }, @@ -41,7 +41,7 @@ "targets": [ { "refId": "A", - "expr": "sort_desc(asap_topk{method=\"sketch\",bound=\"estimate\"})", + "expr": "sort_desc(asap_topk{method=\"sketch\",bound=\"estimate\",rank=~\"[1-9]|10\"})", "legendFormat": "#{{rank}} {{key}}", "instant": true } @@ -81,7 +81,7 @@ "targets": [ { "refId": "A", - "expr": "sort_desc(asap_topk{method=\"exact\",bound=\"estimate\"})", + "expr": "sort_desc(asap_topk{method=\"exact\",bound=\"estimate\",rank=~\"[1-9]|10\"})", "legendFormat": "#{{rank}} {{key}}", "instant": true } @@ -241,7 +241,7 @@ }, { "type": "timeseries", - "title": "Window completion delay: seconds from a window's end until its result is complete", + "title": "Heavy hitters (routes with >= 1% of the window's points): sketch vs exact, per window", "datasource": { "type": "prometheus", "uid": "asap-prometheus" @@ -252,6 +252,151 @@ "x": 0, "y": 28 }, + "targets": [ + { + "refId": "A", + "expr": "asap_hh_precision", + "legendFormat": "precision" + }, + { + "refId": "B", + "expr": "asap_hh_recall", + "legendFormat": "recall" + }, + { + "refId": "C", + "expr": "asap_hh_f1", + "legendFormat": "F1" + } + ], + "fieldConfig": { + "defaults": { + "min": 0, + "max": 1.05, + "decimals": 2, + "custom": { + "spanNulls": true, + "drawStyle": "line", + "pointSize": 6, + "showPoints": "always" + } + }, + "overrides": [] + }, + "options": { + "legend": { + "displayMode": "list", + "placement": "bottom" + } + }, + "id": 10, + "description": "Precision = reported heavy hitters that are real / all reported (low: Count-Min overestimates push small routes over the threshold). Recall = real heavy hitters the sketch reported / all real ones (low: the heap dropped one). A heavy hitter has at least 1% of the window's points by the exact counts; the sketch reports a route when its estimate reaches 1%." + }, + { + "type": "timeseries", + "title": "Average relative count error of the heavy hitters found (ARE)", + "datasource": { + "type": "prometheus", + "uid": "asap-prometheus" + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 28 + }, + "targets": [ + { + "refId": "A", + "expr": "asap_hh_are", + "legendFormat": "ARE" + }, + { + "refId": "B", + "expr": "asap_hh_truth", + "legendFormat": "heavy hitters (exact)" + }, + { + "refId": "C", + "expr": "asap_hh_reported", + "legendFormat": "reported (sketch)" + } + ], + "fieldConfig": { + "defaults": { + "custom": { + "spanNulls": true, + "showPoints": "always", + "pointSize": 6 + } + }, + "overrides": [ + { + "matcher": { + "id": "byName", + "options": "ARE" + }, + "properties": [ + { + "id": "unit", + "value": "percentunit" + }, + { + "id": "decimals", + "value": 2 + } + ] + }, + { + "matcher": { + "id": "byRegexp", + "options": "heavy hitters.*|reported.*" + }, + "properties": [ + { + "id": "custom.axisPlacement", + "value": "right" + }, + { + "id": "decimals", + "value": 0 + }, + { + "id": "custom.lineStyle", + "value": { + "fill": "dash", + "dash": [ + 6, + 4 + ] + } + } + ] + } + ] + }, + "options": { + "legend": { + "displayMode": "list", + "placement": "bottom" + } + }, + "id": 11, + "description": "Mean of |sketch estimate - exact count| / exact count over the heavy hitters the sketch found. Right axis: how many heavy hitters exact has and how many the sketch reported." + }, + { + "type": "timeseries", + "title": "Window completion delay: seconds from a window's end until its result is complete", + "datasource": { + "type": "prometheus", + "uid": "asap-prometheus" + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 36 + }, "targets": [ { "refId": "A", @@ -328,7 +473,7 @@ "h": 8, "w": 12, "x": 12, - "y": 28 + "y": 36 }, "targets": [ { @@ -405,7 +550,7 @@ "h": 8, "w": 12, "x": 0, - "y": 36 + "y": 44 }, "targets": [ { @@ -486,7 +631,7 @@ "h": 8, "w": 12, "x": 12, - "y": 36 + "y": 44 }, "targets": [ { diff --git a/deploy/telemetry-functions/prometheus.yml b/deploy/telemetry-functions/prometheus.yml index b2eff75..122b3b2 100644 --- a/deploy/telemetry-functions/prometheus.yml +++ b/deploy/telemetry-functions/prometheus.yml @@ -11,7 +11,8 @@ scrape_configs: detector:8085, exact-generator-a:8085, exact-generator-b:8085, exact-branch-a:8085, exact-branch-b:8085, exact-merge:8085, exact-detector:8085, raw-generator-a:8085, raw-generator-b:8085, - raw-branch-a:8085, raw-branch-b:8085, raw-merge:8085, raw-sink:8085] + raw-branch-a:8085, raw-branch-b:8085, raw-merge:8085, raw-sink:8085, + hh-accuracy:8085] relabel_configs: - source_labels: [__address__] regex: (.+):8085 diff --git a/deploy/telemetry-functions/topk/branch.yaml b/deploy/telemetry-functions/topk/branch.yaml index c7301e0..39b691e 100644 --- a/deploy/telemetry-functions/topk/branch.yaml +++ b/deploy/telemetry-functions/topk/branch.yaml @@ -1,5 +1,5 @@ # Sketch pipeline, stage 1 at each source: one Count-Min sketch (4 x 512) + -# top-10 heap of routes per 1-minute window goes to merge. +# top-20 heap of routes per 1-minute window goes to merge. version: otel_dataflow/v1 engine: http_admin: @@ -27,7 +27,7 @@ groups: output_metric_name: http.server.request.duration per_attribute: [http.route] omit_resource_attrs: true - sketch_params: {rows: 4, columns: 512, k: 10} + sketch_params: {rows: 4, columns: 512, k: 20} to_merge: type: exporter:otlp_http config: diff --git a/deploy/telemetry-functions/topk/detector.yaml b/deploy/telemetry-functions/topk/detector.yaml index 6465f03..8a70f18 100644 --- a/deploy/telemetry-functions/topk/detector.yaml +++ b/deploy/telemetry-functions/topk/detector.yaml @@ -1,5 +1,6 @@ -# Top-10 routes with Count-Min bounds to Prometheus; routes new to the top 10 -# (vs the last 10 windows) as events to Loki. +# Top-20 routes with Count-Min bounds to Prometheus; routes new to the top 20 +# (vs the last 10 windows) as events to Loki. A copy of the answers goes to +# hh-accuracy, which compares them with the exact pipeline's. version: otel_dataflow/v1 engine: http_admin: @@ -21,7 +22,14 @@ groups: http: {listening_addr: "0.0.0.0:4318", wait_for_result: false} detector: type: urn:asap:processor:sketch_detector - config: {k: 10, novelty_windows: 10} + config: {k: 20, novelty_windows: 10} + fanout: + type: processor:fanout + outputs: [answers, accuracy] + config: + destinations: + - {port: answers} + - {port: accuracy} to_prometheus: type: exporter:otlp_http config: @@ -31,6 +39,15 @@ groups: client_pool_size: 1 max_in_flight: 1 http: {timeout: 30s, compression: none} + to_accuracy: + type: exporter:otlp_http + config: + endpoint: http://hh-accuracy:4318 + client_pool_size: 1 + max_in_flight: 1 + http: {timeout: 30s, compression: none} connections: - {from: receiver, to: detector} - - {from: detector, to: to_prometheus} + - {from: detector, to: fanout} + - {from: 'fanout["answers"]', to: to_prometheus} + - {from: 'fanout["accuracy"]', to: to_accuracy} diff --git a/deploy/telemetry-functions/topk/exact-detector.yaml b/deploy/telemetry-functions/topk/exact-detector.yaml index 61e560a..463b4e4 100644 --- a/deploy/telemetry-functions/topk/exact-detector.yaml +++ b/deploy/telemetry-functions/topk/exact-detector.yaml @@ -1,5 +1,5 @@ # Exact pipeline, end: exact answers and events of each merged window, the -# ground truth (method="exact"). +# ground truth (method="exact"). A copy of the answers goes to hh-accuracy. version: otel_dataflow/v1 engine: http_admin: @@ -21,7 +21,14 @@ groups: http: {listening_addr: "0.0.0.0:4318", wait_for_result: false, max_request_body_size: 1GiB} detector: type: urn:asap:processor:exact_detector - config: {k: 10, novelty_windows: 10} + config: {k: 20, novelty_windows: 10} + fanout: + type: processor:fanout + outputs: [answers, accuracy] + config: + destinations: + - {port: answers} + - {port: accuracy} to_prometheus: type: exporter:otlp_http config: @@ -31,6 +38,15 @@ groups: client_pool_size: 1 max_in_flight: 1 http: {timeout: 30s, compression: none} + to_accuracy: + type: exporter:otlp_http + config: + endpoint: http://hh-accuracy:4318 + client_pool_size: 1 + max_in_flight: 1 + http: {timeout: 30s, compression: none} connections: - {from: receiver, to: detector} - - {from: detector, to: to_prometheus} + - {from: detector, to: fanout} + - {from: 'fanout["answers"]', to: to_prometheus} + - {from: 'fanout["accuracy"]', to: to_accuracy} From 8383a6e71a81a61dbb02edd30d9b8f41c91d7459 Mon Sep 17 00:00:00 2001 From: Zeying Zhu <50204836+zzylol@users.noreply.github.com> Date: Tue, 20 Oct 2026 23:20:27 +0000 Subject: [PATCH 2/2] Treat a full sketch top-k as incomplete too When the sketch's top-k is full and its smallest estimate still reaches the threshold, more keys may be reported than the list shows, so precision would come out too high; such a window is reported as incomplete, like a full exact list. --- asap-precompute-rs/src/answers.rs | 14 +++++++++----- 1 file changed, 9 insertions(+), 5 deletions(-) diff --git a/asap-precompute-rs/src/answers.rs b/asap-precompute-rs/src/answers.rs index 823a0df..ade18fd 100644 --- a/asap-precompute-rs/src/answers.rs +++ b/asap-precompute-rs/src/answers.rs @@ -416,9 +416,9 @@ pub struct HeavyHitterAccuracy { /// Compares a window's sketch top-k with its exact top-k at threshold /// `phi * points`. Both lists are `(key, count)` as reported, at most `k` -/// long. Returns `None` when the exact list is full and its smallest count -/// still reaches the threshold: the window may have more heavy hitters than -/// the list shows. +/// long. Returns `None` when either list is full and its smallest count still +/// reaches the threshold: the window may have more heavy hitters (exact) or +/// more reported keys (sketch) than the list shows. pub fn heavy_hitter_accuracy( phi: f64, points: f64, @@ -427,8 +427,10 @@ pub fn heavy_hitter_accuracy( sketch: &[(String, f64)], ) -> Option { let threshold = phi * points; - let smallest = exact.iter().map(|e| e.1).fold(f64::INFINITY, f64::min); - if exact.len() >= k && smallest >= threshold { + let cut_short = |list: &[(String, f64)]| { + list.len() >= k && list.iter().map(|e| e.1).fold(f64::INFINITY, f64::min) >= threshold + }; + if cut_short(exact) || cut_short(sketch) { return None; } let truth: HashMap<&str, f64> = exact @@ -913,5 +915,7 @@ mod heavy_hitter_tests { // A full exact list whose last count is still heavy may be cut short. assert!(heavy_hitter_accuracy(0.05, 1000.0, 4, &exact, &sketch).is_none()); assert!(heavy_hitter_accuracy(0.1, 1000.0, 4, &exact, &sketch).is_some()); + // So may a full sketch list whose last estimate is still reported. + assert!(heavy_hitter_accuracy(0.1, 1000.0, 3, &exact[3..], &sketch).is_none()); } }