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..ade18fd 100644 --- a/asap-precompute-rs/src/answers.rs +++ b/asap-precompute-rs/src/answers.rs @@ -393,6 +393,74 @@ 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 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, + k: usize, + exact: &[(String, f64)], + sketch: &[(String, f64)], +) -> Option { + let threshold = phi * points; + 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 + .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 +889,33 @@ 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()); + // 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()); + } +} 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}