diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 6a4c16a..d947a92 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -58,10 +58,9 @@ jobs: CPU_EXACT_DETECTOR: "0-3" CPU_RAW_GENERATOR_A: "0-3" CPU_RAW_GENERATOR_B: "0-3" - CPU_RAW_BRANCH_A: "0-3" - CPU_RAW_BRANCH_B: "0-3" - CPU_RAW_MERGE: "0-3" - CPU_RAW_SINK: "0-3" + CPU_RAW_SINK_A: "0-3" + CPU_RAW_SINK_B: "0-3" + CPU_QUERY_JOB: "0-3" CPU_HH_ACCURACY: "0-3" steps: - uses: actions/checkout@v4 diff --git a/asap-precompute-rs/src/otap/answer_nodes.rs b/asap-precompute-rs/src/otap/answer_nodes.rs index 2000ad5..6ca99e7 100644 --- a/asap-precompute-rs/src/otap/answer_nodes.rs +++ b/asap-precompute-rs/src/otap/answer_nodes.rs @@ -129,6 +129,84 @@ fn with_delay(envs: &[SketchEnvelope], now_ms: u64) -> Vec { out } +/// When a stage got the last input of a window and when it emitted the +/// window, in seconds after the window's end: `asap.stage.received` and +/// `asap.stage.emitted`, labelled `method` and `stage`. Together they give the +/// latency breakdown of each pipeline (the gap between one stage's emission +/// and the next one's receipt is transmission). +pub(super) fn stage_timing( + window_start_ms: u64, + window_end_ms: u64, + method: &str, + stage: &str, + received_ms: u64, + emitted_ms: u64, +) -> Vec { + [ + ("asap.stage.received", received_ms), + ("asap.stage.emitted", emitted_ms), + ] + .into_iter() + .map(|(metric, at)| SketchEnvelope { + labels: vec![ + KeyValue::new("method", method), + KeyValue::new("stage", stage), + ], + metric_name: metric.into(), + window_start_ms, + window_end_ms, + value: at.saturating_sub(window_end_ms) as f64 / 1000.0, + ..Default::default() + }) + .collect() +} + +/// Stage timings an upstream stage (the merge) sent along with its windows, +/// for a detector to pass on to the answer backend. +pub(super) fn upstream_timing(observations: &[Observation]) -> Vec { + observations + .iter() + .filter(|o| o.metric.starts_with("asap.stage.") && o.value.envelope.is_none()) + .map(|o| SketchEnvelope { + labels: o.labels.clone(), + metric_name: o.metric.clone(), + window_start_ms: o.timestamp_ms, + window_end_ms: o.timestamp_ms, + value: o.value.float, + ..Default::default() + }) + .collect() +} + +/// Sends gauges as they are, without `asap.answer.delay`. +pub(super) async fn send_gauges( + envs: &[SketchEnvelope], + effects: &mut local::EffectHandler, +) -> Result<(), Error> { + if envs.is_empty() { + return Ok(()); + } + match gauges_pdata(envs) { + Ok(pdata) => effects.send_message_with_source_node(pdata).await?, + Err(e) => effects.info(&format!("answers: encode failed: {e}")).await, + } + Ok(()) +} + +/// The detector's own timing of each window in `windows`, received at +/// `received_ms` and emitted now. +pub(super) fn detector_timing( + windows: &std::collections::BTreeSet<(u64, u64)>, + method: &str, + received_ms: u64, +) -> Vec { + let now = now_ms(); + windows + .iter() + .flat_map(|(start, end)| stage_timing(*start, *end, method, "detector", received_ms, now)) + .collect() +} + /// Plain OTLP gauges with every label on the data point, so backends such /// as Prometheus keep them as series labels. pub(super) fn gauges_pdata(envs: &[SketchEnvelope]) -> Result { @@ -332,11 +410,16 @@ impl local::Processor for SketchDetector { }; // Consumed into state here: ack so upstream (e.g. fanout) can move on. effects.notify_ack(AckMsg::new(pdata.clone())).await?; + let received = now_ms(); let (mut out, mut events) = (Vec::new(), Vec::new()); - for obs in observations(pdata, effects).await { + let observations = observations(pdata, effects).await; + send_gauges(&upstream_timing(&observations), effects).await?; + let mut windows = std::collections::BTreeSet::new(); + for obs in observations { let Some(env) = obs.value.envelope.filter(|e| !e.payload.is_empty()) else { continue; }; + let _ = windows.insert((env.window_start_ms, env.window_end_ms)); match sketch_answers(&env, &self.config.quantiles, self.config.k) { Ok(answers) => { if env.sketch_type == SketchType::KLLSketch { @@ -361,6 +444,7 @@ impl local::Processor for SketchDetector { Err(e) => effects.info(&format!("sketch_detector: {e}")).await, } } + out.extend(detector_timing(&windows, "sketch", received)); emit(&out, &events, effects).await } } diff --git a/asap-precompute-rs/src/otap/baseline_nodes.rs b/asap-precompute-rs/src/otap/baseline_nodes.rs index a6f1352..7318f51 100644 --- a/asap-precompute-rs/src/otap/baseline_nodes.rs +++ b/asap-precompute-rs/src/otap/baseline_nodes.rs @@ -5,14 +5,19 @@ //! of each event-time window as an [`ExactWindow`]: the values of each //! quantile series and a value -> count map per attribute key. It emits a //! window when data crosses its end, or 5 s after its end if the stream idles. -//! * `urn:asap:processor:exact_merge` adds up the sources' windows, `grace` -//! after each window's end. +//! * `urn:asap:processor:exact_merge` adds up the sources' windows and emits +//! each as soon as all `sources` have sent it (each sends a window in one +//! message, in window order), or `grace` after its end if one is missing. +//! Like `window_merge` it counts messages, not distinct sources; windows +//! arriving after they were emitted are logged and dropped. //! * `urn:asap:processor:exact_detector` computes each merged window's exact //! answers and events, with `method="exact"`: the ground truth. //! * `urn:asap:processor:raw_sink` ends the raw pipeline, which only forwards //! raw points: it writes every point as one log record to a store through -//! its `logs` output and watches the store's acknowledgements. For each -//! event-time window it reports on its `latency` output when the store +//! its `logs` output and watches the store's acknowledgements. Once its +//! source has sent a point past a window's end (or `grace` after the end if +//! the source idles) and the store answered every message of the window, +//! it reports on its `latency` output when the store //! acknowledged the window's last point (`asap.answer.delay`, //! `method="raw"`), the per-point time from event to acknowledgement //! (`asap.raw.transit`) and rejected points (`asap.raw.failed`). @@ -23,7 +28,7 @@ //! config: {window: 60s, group_by: [http.route], per_attribute: [user.id]} //! exact_merge: //! type: urn:asap:processor:exact_merge -//! config: {grace: 5s} +//! config: {sources: 2, grace: 5s} //! exact_detector: //! type: urn:asap:processor:exact_detector //! config: {quantiles: [0.5, 0.99], k: 10} @@ -65,8 +70,9 @@ use otel_arrow_dfe_pdata::OtapPayload; use serde::Deserialize; use super::answer_nodes::{ - default_grace, default_k, default_novelty_windows, default_quantiles, default_window, emit, - gauges_pdata, invalid, now_ms, observations, series_key, + default_grace, default_k, default_novelty_windows, default_quantiles, default_window, + detector_timing, emit, gauges_pdata, invalid, now_ms, observations, send_gauges, series_key, + stage_timing, upstream_timing, }; use super::codec::{decode_each, OtapSketchEncoder}; use crate::answers::{ @@ -114,9 +120,12 @@ pub struct ExactBranchConfig { #[derive(Clone, Debug, Deserialize)] #[serde(deny_unknown_fields)] pub struct ExactMergeConfig { - /// How long after a window's end to wait for the last source. + /// How long after a window's end to wait for a missing source. #[serde(default = "default_grace", with = "humantime_serde")] pub grace: Duration, + /// Sources sending each window; 0 waits the full grace. + #[serde(default)] + pub sources: usize, } /// `exact_detector` configuration. @@ -147,9 +156,14 @@ pub struct RawSinkConfig { /// Event-time window size. #[serde(default = "default_window", with = "humantime_serde")] pub window: Duration, - /// Wait after a window's end before reporting it. + /// Wait after a window's end before reporting it when the source sends + /// nothing later. #[serde(default = "default_grace", with = "humantime_serde")] pub grace: Duration, + /// Labels added to every gauge, e.g. `{source: a}` when each source has + /// its own sink. + #[serde(default)] + pub labels: BTreeMap, } /// Open windows by window end, then series. @@ -301,18 +315,24 @@ impl local::Processor for ExactBranch { struct ExactMerge { config: ExactMergeConfig, windows: Windows, + /// When the last source's window arrived, by window end (ms). + arrived: BTreeMap, + /// Messages (one per source) that carried each open window. + messages: BTreeMap, emitted_until: u64, encoder: OtapSketchEncoder, timer_started: bool, } impl ExactMerge { - /// Adds one source's window; windows already emitted drop it as late. - fn add(&mut self, env: SketchEnvelope) -> Result<(), crate::precompute::PrecomputeError> { + /// Adds one source's window; returns false when the window was already + /// emitted and the input is dropped as late. + fn add(&mut self, env: SketchEnvelope) -> Result { if env.window_end_ms <= self.emitted_until { - return Ok(()); + return Ok(false); } let w = ExactWindow::decode(&env.payload)?; + let _ = self.arrived.insert(env.window_end_ms, now_ms()); let key = format!("{}|{}", series_key(&env), env.window_start_ms); let window = self.windows.entry(env.window_end_ms).or_default(); match window.get_mut(&key) { @@ -321,11 +341,24 @@ impl ExactMerge { window.insert(key, (env, w)); } } - Ok(()) + Ok(true) } + /// Every window all sources sent or that ended `grace` before `now`; + /// sources send windows in order, so earlier windows are complete too. fn due(&mut self, now: u64) -> Vec { - let cutoff = now.saturating_sub(self.config.grace.as_millis() as u64); + let sources = self.config.sources; + let complete = self + .messages + .iter() + .filter(|(_, n)| sources > 0 && **n >= sources) + .map(|(end, _)| *end) + .max() + .unwrap_or(0); + let cutoff = now + .saturating_sub(self.config.grace.as_millis() as u64) + .max(complete); + self.messages = self.messages.split_off(&(cutoff.saturating_add(1))); let out = close(&mut self.windows, cutoff); if let Some(end) = out.iter().map(|e| e.window_end_ms).max() { self.emitted_until = self.emitted_until.max(end); @@ -350,21 +383,49 @@ impl local::Processor for ExactMerge { let out = match msg { Message::PData(pdata) => { effects.notify_ack(AckMsg::new(pdata.clone())).await?; + let mut carried = std::collections::BTreeSet::new(); + let mut late = 0; for obs in observations(pdata, effects).await { let Some(env) = obs.value.envelope else { continue; }; - if let Err(e) = self.add(*env) { - effects.info(&format!("exact_merge: {e}")).await; + let _ = carried.insert(env.window_end_ms); + match self.add(*env) { + Ok(true) => {} + Ok(false) => late += 1, + Err(e) => effects.info(&format!("exact_merge: {e}")).await, } } - Vec::new() + if late > 0 { + effects + .info(&format!( + "exact_merge: {late} windows arrived after they were emitted" + )) + .await; + } + for end in carried.into_iter().filter(|e| *e > self.emitted_until) { + *self.messages.entry(end).or_default() += 1; + } + self.due(now_ms()) } Message::Control(NodeControlMsg::TimerTick { .. }) => self.due(now_ms()), Message::Control(NodeControlMsg::Shutdown { .. }) => self.due(u64::MAX), _ => Vec::new(), }; - send(&mut self.encoder, &out, effects).await + send(&mut self.encoder, &out, effects).await?; + let now = now_ms(); + let windows: BTreeMap = out + .iter() + .map(|e| (e.window_end_ms, e.window_start_ms)) + .collect(); + let timing: Vec = windows + .into_iter() + .flat_map(|(end, start)| { + let arrived = self.arrived.remove(&end).unwrap_or(now); + stage_timing(start, end, "exact", "merge", arrived, now) + }) + .collect(); + send_gauges(&timing, effects).await } } @@ -457,15 +518,21 @@ impl local::Processor for ExactDetector { return Ok(()); }; effects.notify_ack(AckMsg::new(pdata.clone())).await?; + let received = now_ms(); let (mut out, mut events) = (Vec::new(), Vec::new()); - for obs in observations(pdata, effects).await { + let observations = observations(pdata, effects).await; + send_gauges(&upstream_timing(&observations), effects).await?; + let mut windows = std::collections::BTreeSet::new(); + for obs in observations { let Some(env) = obs.value.envelope.filter(|e| !e.payload.is_empty()) else { continue; }; + let _ = windows.insert((env.window_start_ms, env.window_end_ms)); let (gauges, new) = self.answer(*env); out.extend(gauges); events.extend(new); } + out.extend(detector_timing(&windows, "exact", received)); emit(&out, &events, effects).await } } @@ -491,6 +558,8 @@ struct RawSink { in_flight: HashMap)>>, next_id: u64, emitted_until: u64, + /// Latest event time (ms) the source has sent. + sent_until_ms: u64, timer_started: bool, } @@ -545,6 +614,7 @@ impl RawSink { let size = self.config.window.as_millis().max(1) as u64; let mut by_window: BTreeMap> = BTreeMap::new(); for p in points { + self.sent_until_ms = self.sent_until_ms.max(p.timestamp_ms); let end = p.timestamp_ms / size * size + size; if end > self.emitted_until { by_window.entry(end).or_default().push(p.timestamp_ms); @@ -576,8 +646,11 @@ impl RawSink { } } - /// Gauges of every window that ended `grace` or more before `now` and - /// has no message waiting for the store (or ended 10 graces ago). + /// Gauges of every window the source has moved past (a later point, or + /// `grace` after its end) with no message waiting for the store, or that + /// ended 10 graces ago: then `asap.raw.unacknowledged` counts the points + /// still waiting and the delay is the time of the report. The source sends + /// points in time order. fn due(&mut self, now: u64) -> Vec { let grace = self.config.grace.as_millis() as u64; let size = self.config.window.as_millis() as u64; @@ -585,7 +658,8 @@ impl RawSink { .windows .iter() .filter(|(end, w)| { - **end + grace <= now && (w.in_flight == 0 || **end + 10 * grace <= now) + (**end <= self.sent_until_ms || **end + grace <= now) + && (w.in_flight == 0 || **end + 10 * grace <= now) }) .map(|(end, _)| *end) .collect(); @@ -593,6 +667,19 @@ impl RawSink { for end in ready { let mut w = self.windows.remove(&end).unwrap_or_default(); self.emitted_until = self.emitted_until.max(end); + let unacknowledged: usize = self + .in_flight + .values() + .flat_map(|windows| windows.iter()) + .filter(|(e, _)| *e == end) + .map(|(_, times)| times.len()) + .sum(); + // On shutdown `now` is u64::MAX: report the real time instead. + let complete_ms = if unacknowledged > 0 { + now.min(now_ms()) + } else { + w.last_ack_ms + }; let template = SketchEnvelope { window_start_ms: end - size, window_end_ms: end, @@ -603,6 +690,12 @@ impl RawSink { let mut e = gauge(&template, name, "raw", extra, v); // Window-level gauges: no source metric label. e.labels.retain(|kv| kv.key != "metric"); + e.labels.extend( + self.config + .labels + .iter() + .map(|(k, v)| KeyValue::new(k.clone(), v.clone())), + ); e }; // Complete once the last point is stored, and not before the @@ -610,7 +703,12 @@ impl RawSink { out.push(metric( "asap.answer.delay", &[], - w.last_ack_ms.saturating_sub(end) as f64 / 1000.0, + complete_ms.saturating_sub(end) as f64 / 1000.0, + )); + out.push(metric( + "asap.raw.unacknowledged", + &[], + unacknowledged as f64, )); out.push(metric("asap.points", &[], w.transit_s.len() as f64)); out.push(metric("asap.raw.failed", &[], w.failed as f64)); @@ -667,19 +765,20 @@ impl local::Processor for RawSink { } Err(e) => effects.info(&format!("raw_sink: encode failed: {e}")).await, } - return Ok(()); + // A later point completes the windows already stored. + self.due(now_ms()) } Message::Control(NodeControlMsg::Ack(ack)) => { if let Some(id) = ack.unwind.route.calldata.first() { self.answered(u64::from(*id), now_ms(), true); } - return Ok(()); + self.due(now_ms()) } Message::Control(NodeControlMsg::Nack(nack)) => { if let Some(id) = nack.unwind.route.calldata.first() { self.answered(u64::from(*id), now_ms(), false); } - return Ok(()); + self.due(now_ms()) } Message::Control(NodeControlMsg::TimerTick { .. }) => self.due(now_ms()), Message::Control(NodeControlMsg::Shutdown { .. }) => self.due(u64::MAX), @@ -746,6 +845,8 @@ static EXACT_MERGE: ProcessorFactory = ProcessorFactory { ExactMerge { config: config(&node_config)?, windows: Windows::new(), + arrived: BTreeMap::new(), + messages: BTreeMap::new(), emitted_until: 0, encoder: OtapSketchEncoder::default(), timer_started: false, @@ -802,6 +903,7 @@ static RAW_SINK: ProcessorFactory = ProcessorFactory { in_flight: HashMap::new(), next_id: 0, emitted_until: 0, + sent_until_ms: 0, timer_started: false, }, node, @@ -892,13 +994,18 @@ mod tests { let mut merge = ExactMerge { config: ExactMergeConfig { grace: Duration::from_secs(5), + sources: 0, }, windows: Windows::new(), + arrived: BTreeMap::new(), + messages: BTreeMap::new(), emitted_until: 0, encoder: OtapSketchEncoder::default(), timer_started: false, }; - sources.into_iter().for_each(|e| merge.add(e).unwrap()); + for e in sources { + assert!(merge.add(e).unwrap()); + } assert!(merge.due(64_000).is_empty()); let merged = merge.due(65_000); assert_eq!(merged.len(), 2); @@ -924,10 +1031,95 @@ mod tests { assert_eq!(get(&out, "asap.topk", ("rank", "1")), Some(26.0)); assert!(out.iter().all(|e| e.window_end_ms == 60_000)); // A late window is dropped. - merge.add(branch_window(2_000)).unwrap(); + assert!(!merge.add(branch_window(2_000)).unwrap()); assert!(merge.due(200_000).is_empty()); } + #[test] + fn exact_merge_emits_once_every_source_sent_the_window() { + let mut merge = ExactMerge { + config: ExactMergeConfig { + grace: Duration::from_secs(5), + sources: 2, + }, + windows: Windows::new(), + arrived: BTreeMap::new(), + messages: BTreeMap::new(), + emitted_until: 0, + encoder: OtapSketchEncoder::default(), + timer_started: false, + }; + merge.add(branch_window(1_000)).unwrap(); + *merge.messages.entry(60_000).or_default() += 1; + assert!(merge.due(60_010).is_empty(), "one of two sources"); + merge.add(branch_window(2_000)).unwrap(); + *merge.messages.entry(60_000).or_default() += 1; + let out = merge.due(60_020); + assert_eq!(out.len(), 1); + assert_eq!(out[0].count, 2); + // A missing source: emitted at the grace. + merge.add(branch_window(61_000)).unwrap(); + *merge.messages.entry(120_000).or_default() += 1; + assert!(merge.due(124_000).is_empty()); + assert_eq!(merge.due(125_000).len(), 1); + } + + #[test] + fn raw_sink_reports_unacknowledged_points_after_ten_graces() { + let mut raw = RawSink { + config: serde_json::from_value(serde_json::json!({})).unwrap(), + windows: BTreeMap::new(), + in_flight: HashMap::new(), + next_id: 0, + emitted_until: 0, + sent_until_ms: 0, + timer_started: false, + }; + let stored = raw.send(&[point(10_000, "/a", "u0", 1.0)]); + let _waiting = raw.send(&[ + point(59_000, "/a", "u1", 1.0), + point(59_500, "/b", "u2", 1.0), + ]); + raw.answered(stored, 10_100, true); + assert!(raw.due(109_000).is_empty(), "still waiting for the store"); + let out = raw.due(110_000); + assert_eq!( + get(&out, "asap.raw.unacknowledged", ("method", "raw")), + Some(2.0) + ); + assert_eq!( + get(&out, "asap.answer.delay", ("method", "raw")), + Some(50.0) + ); + assert_eq!(get(&out, "asap.points", ("method", "raw")), Some(1.0)); + } + + #[test] + fn raw_sink_reports_once_the_source_moved_past_the_window() { + let mut raw = RawSink { + config: serde_json::from_value(serde_json::json!({})).unwrap(), + windows: BTreeMap::new(), + in_flight: HashMap::new(), + next_id: 0, + emitted_until: 0, + sent_until_ms: 0, + timer_started: false, + }; + let last = raw.send(&[point(59_900, "/a", "u1", 1.0)]); + raw.answered(last, 60_050, true); + assert!(raw.due(60_060).is_empty(), "the source may still send"); + let next = raw.send(&[point(60_100, "/a", "u1", 1.0)]); + let out = raw.due(60_110); + assert_eq!( + get(&out, "asap.answer.delay", ("method", "raw")), + Some(0.05) + ); + assert_eq!(get(&out, "asap.points", ("method", "raw")), Some(1.0)); + // The next window waits for its own message. + raw.answered(next, 60_200, true); + assert!(raw.due(60_210).is_empty()); + } + fn branch_window(ts: u64) -> SketchEnvelope { let mut b = branch(); b.observe(&point(ts, "/a", "u0", 1.0)); @@ -942,6 +1134,7 @@ mod tests { in_flight: HashMap::new(), next_id: 0, emitted_until: 0, + sent_until_ms: 0, timer_started: false, }; let first = raw.send(&[point(10_000, "/a", "u0", 1.0)]); diff --git a/asap-precompute-rs/src/otap/hh_accuracy.rs b/asap-precompute-rs/src/otap/hh_accuracy.rs index b7ea789..89ca887 100644 --- a/asap-precompute-rs/src/otap/hh_accuracy.rs +++ b/asap-precompute-rs/src/otap/hh_accuracy.rs @@ -4,7 +4,7 @@ //! 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`, +//! Gauges, labelled with the series labels, `phi` and `method="sketch"`: `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 @@ -151,6 +151,8 @@ impl HhAccuracy { 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())); + // The sketch's accuracy; the store-and-query job reports `method="raw"`. + labels.push(KeyValue::new("method", "sketch")); let gauge = |metric: &str, value: f64| SketchEnvelope { labels: labels.clone(), metric_name: metric.into(), diff --git a/asap-precompute-rs/src/otap/scenario_generator.rs b/asap-precompute-rs/src/otap/scenario_generator.rs index a696cbd..f6e26b4 100644 --- a/asap-precompute-rs/src/otap/scenario_generator.rs +++ b/asap-precompute-rs/src/otap/scenario_generator.rs @@ -24,6 +24,8 @@ //! - {after_seconds: 1800, attribute: http.route, equals: http.route-3, //! value: {lognormal: {median: 100.0, sigma: 0.6}}} //! - {after_seconds: 2700, attribute: http.route, inject: new-hot, share: 0.2} +//! - {after_seconds: 600, signals_per_second: 100000} +//! - {after_seconds: 900, attribute: http.route, cardinality: 1000} //! ``` //! //! Attribute values are `-` (`--` when unique); with @@ -135,14 +137,23 @@ pub enum Distribution { }, } -/// A change starting `after_seconds` into the run. +/// A change starting `after_seconds` into the run: a new rate, or a change +/// to one attribute (its values' distribution, an injected value, or its +/// number of distinct values). #[derive(Clone, Debug, Deserialize)] #[serde(deny_unknown_fields)] pub struct Change { /// Seconds after start. pub after_seconds: u64, - /// Attribute the change refers to. + /// New data points per second from this time on. + #[serde(default)] + pub signals_per_second: Option, + /// Attribute the change refers to; empty for a rate change. + #[serde(default)] pub attribute: String, + /// New number of distinct values of `attribute` from this time on. + #[serde(default)] + pub cardinality: Option, /// With `value`: only points whose attribute equals this value change. #[serde(default)] pub equals: Option, @@ -198,8 +209,8 @@ impl Rng { pub struct Scenario { cfg: ScenarioConfig, rng: Rng, - /// Zipf CDF per attribute index, when `zipf` is set. - cdfs: HashMap>, + /// Zipf CDF per (attribute index, cardinality), when `zipf` is set. + cdfs: HashMap<(usize, u64), Vec>, unique_counter: u64, } @@ -214,6 +225,18 @@ impl Scenario { if cfg.signals_per_second == 0 || cfg.max_batch_size == 0 { return Err("signals_per_second and max_batch_size must be positive".into()); } + for c in &cfg.changes { + let rate_only = c.signals_per_second.is_some() && c.attribute.is_empty(); + if c.signals_per_second == Some(0) + || c.cardinality == Some(0) + || (!rate_only && !cfg.attributes.iter().any(|a| a.key == c.attribute)) + { + return Err(format!( + "change at {} s: a positive rate, or an existing attribute", + c.after_seconds + )); + } + } let mut cdfs = HashMap::new(); for (i, a) in cfg.attributes.iter().enumerate() { if a.cardinality == 0 || (a.zipf.is_some() && a.growth_per_minute != 0.0) { @@ -222,16 +245,24 @@ impl Scenario { a.key )); } - if let Some(s) = a.zipf { - let mut total = 0.0; - let mut cdf: Vec = (1..=a.cardinality) - .map(|rank| { - total += (rank as f64).powf(-s); - total - }) - .collect(); - cdf.iter_mut().for_each(|c| *c /= total); - cdfs.insert(i, cdf); + let Some(s) = a.zipf else { continue }; + let changed = cfg + .changes + .iter() + .filter(|c| c.attribute == a.key) + .filter_map(|c| c.cardinality); + for n in std::iter::once(a.cardinality).chain(changed) { + cdfs.entry((i, n)).or_insert_with(|| { + let mut total = 0.0; + let mut cdf: Vec = (1..=n) + .map(|rank| { + total += (rank as f64).powf(-s); + total + }) + .collect(); + cdf.iter_mut().for_each(|c| *c /= total); + cdf + }); } } Ok(Self { @@ -242,41 +273,85 @@ impl Scenario { }) } - fn attribute_value(&mut self, i: usize, minutes: f64) -> String { + /// Data points per second at scenario second `second`. + pub fn rate_at(&self, second: u64) -> usize { + self.cfg + .changes + .iter() + .filter(|c| second >= c.after_seconds) + .filter_map(|c| c.signals_per_second.map(|r| (c.after_seconds, r))) + .max_by_key(|(after, _)| *after) + .map_or(self.cfg.signals_per_second, |(_, r)| r) + } + + /// The highest rate of the scenario: it sets the batches per second. + fn max_rate(&self) -> usize { + self.cfg + .changes + .iter() + .filter_map(|c| c.signals_per_second) + .fold(self.cfg.signals_per_second, usize::max) + } + + /// Batches per second: enough for the highest rate in `max_batch_size`s. + pub fn batches_per_second(&self) -> usize { + self.max_rate().div_ceil(self.cfg.max_batch_size) + } + + /// Distinct values of attribute `i` at scenario second `second`. + fn cardinality_at(&self, i: usize, second: u64) -> u64 { + let a = &self.cfg.attributes[i]; + self.cfg + .changes + .iter() + .filter(|c| second >= c.after_seconds && c.attribute == a.key) + .filter_map(|c| c.cardinality.map(|n| (c.after_seconds, n))) + .max_by_key(|(after, _)| *after) + .map_or(a.cardinality, |(_, n)| n) + } + + fn attribute_value(&mut self, i: usize, second: u64) -> String { + let minutes = second as f64 / 60.0; + let cardinality = self.cardinality_at(i, second); let a = &self.cfg.attributes[i]; if a.unique { // The seed keeps sources with different seeds from reusing values. self.unique_counter += 1; return format!("{}-{}-{}", a.key, self.cfg.seed, self.unique_counter); } - let n = if let Some(cdf) = self.cdfs.get(&i) { + let n = if let Some(cdf) = self.cdfs.get(&(i, cardinality)) { let u = self.rng.unit(); cdf.partition_point(|&c| c < u).min(cdf.len() - 1) as u64 } else { - let live = (a.cardinality as f64 * (1.0 + a.growth_per_minute).powf(minutes)).max(1.0); + let live = (cardinality as f64 * (1.0 + a.growth_per_minute).powf(minutes)).max(1.0); (self.rng.unit() * live) as u64 }; format!("{}-{n}", a.key) } - /// Batch `index` (of `max_batch_size` points, fewer at the end) of scenario - /// second `second`, stamped `time_ns`. The same arguments always give the - /// same points. + /// First point and number of points of batch `index` of second `second`. + pub fn batch_range(&self, second: u64, index: usize) -> (usize, usize) { + let (rate, slots) = (self.rate_at(second), self.batches_per_second()); + let first = index * rate / slots; + (first, (index + 1) * rate / slots - first) + } + + /// Batch `index` of the [`Self::batches_per_second`] batches of scenario + /// second `second`, stamped `time_ns`: that second's points split evenly + /// over its batches, so at most `max_batch_size`. The same arguments always + /// give the same points. pub fn batch(&mut self, second: u64, index: usize, time_ns: u64) -> MetricsData { - let (rate, size) = (self.cfg.signals_per_second, self.cfg.max_batch_size); - let first = index * size; - let count = rate.saturating_sub(first).min(size); + let (first, count) = self.batch_range(second, index); self.rng = Rng(mix(self.cfg.seed ^ mix(second ^ mix(index as u64)))); - self.unique_counter = second * rate as u64 + first as u64; + self.unique_counter = second * self.max_rate() as u64 + first as u64; let secs = second; - let minutes = second as f64 / 60.0; let points = (0..count) .map(|_| { let mut values: Vec<(String, String)> = (0..self.cfg.attributes.len()) .map(|i| { ( self.cfg.attributes[i].key.clone(), - self.attribute_value(i, minutes), + self.attribute_value(i, second), ) }) .collect(); @@ -388,11 +463,7 @@ impl local::Receiver for ScenarioReceiver { // Batch g (second g / n, index g % n) is sent at start + g / n seconds; // a generator that falls behind skips batches rather than shifting // later ones. - let (rate, size) = ( - self.scenario.cfg.signals_per_second, - self.scenario.cfg.max_batch_size, - ); - let per_second = rate.div_ceil(size) as u64; + let per_second = self.scenario.batches_per_second() as u64; let period_ns = 1_000_000_000 / per_second; let start = scenario_start_ns(); let now = now_ns(); @@ -418,7 +489,8 @@ impl local::Receiver for ScenarioReceiver { let batch = first_batch + (elapsed + period_ns / 2) / period_ns; let (second, index) = (batch / per_second, batch % per_second); let time_ns = start + second * 1_000_000_000 + index * 1_000_000_000 / per_second; - { + // Below the highest rate some batches are empty. + if self.scenario.batch_range(second, index as usize).1 > 0 { let data = self.scenario.batch(second, index as usize, time_ns); let payload: OtapPayload = OtlpProtoMessage::Metrics(data) .try_into() @@ -523,6 +595,43 @@ mod tests { assert!((800..1200).contains(&hot), "{hot}"); } + #[test] + fn rate_and_cardinality_change_after_their_time() { + let cfg = r#"{"signals_per_second":1000,"max_batch_size":500,"metric_name":"m", + "value":{"uniform":{"min":0,"max":1}}, + "attributes":[{"key":"route","cardinality":10,"zipf":1.1}], + "changes":[{"after_seconds":60,"signals_per_second":4000}, + {"after_seconds":60,"attribute":"route","cardinality":1000}]}"#; + let s = Scenario::new(config(cfg)).unwrap(); + assert_eq!(s.batches_per_second(), 8); + let sizes = |at| (0..8).map(|i| s.batch_range(at, i).1).collect::>(); + assert_eq!(sizes(0).iter().sum::(), 1000); + assert_eq!(sizes(60), vec![500; 8]); + let mut s = Scenario::new(config(cfg)).unwrap(); + let routes = |s: &mut Scenario, at| { + (0..8) + .flat_map(|i| { + let data = s.batch(at, i, 1); + match &data.resource_metrics[0].scope_metrics[0].metrics[0].data { + Some(otel_arrow_dfe_pdata::proto::opentelemetry::metrics::v1::metric::Data::Gauge(g)) => { + g.data_points.iter().map(|p| attr(p, "route")).collect::>() + } + _ => panic!("expected a gauge"), + } + }) + .collect::>() + }; + assert!(routes(&mut s, 0).len() <= 10); + assert!(routes(&mut s, 60).len() > 100); + } + + #[test] + fn changes_need_a_rate_or_an_existing_attribute() { + let bad = r#"{"signals_per_second":1,"metric_name":"m","value":{"uniform":{"min":0,"max":1}}, + "changes":[{"after_seconds":1,"attribute":"nope","cardinality":5}]}"#; + assert!(Scenario::new(config(bad)).is_err()); + } + #[test] fn same_seed_same_points() { let cfg = r#"{"seed":7,"signals_per_second":1,"metric_name":"m","value":{"lognormal":{"median":10,"sigma":1}}, diff --git a/asap-precompute-rs/src/otap/window_merge.rs b/asap-precompute-rs/src/otap/window_merge.rs index e61db4a..731ed34 100644 --- a/asap-precompute-rs/src/otap/window_merge.rs +++ b/asap-precompute-rs/src/otap/window_merge.rs @@ -3,13 +3,20 @@ //! //! Each input envelope names its series (labels, sketch type, agg id) and its //! window. Envelopes with the same series and window are merged. A window is -//! emitted once, `grace` after its end on this host's clock; envelopes for a -//! window already emitted are counted as late and dropped. Scalar (non-sketch) -//! inputs pass through unchanged. +//! emitted once: as soon as each of `sources` sources has sent it, or `grace` +//! after its end on this host's clock if a source is missing. Each source +//! (`asap_sketches`) sends all of a window's sketches in one message, in window +//! order, so a window is complete when `sources` messages carried it. +//! Envelopes for a window already emitted are counted as late, logged and +//! dropped. The count is of messages, not of distinct sources: a source that +//! sends a window twice (a retried request, or a branch restarted or +//! reconfigured mid-window) completes it early, and the other source's +//! sketch then arrives late and is logged as such. +//! Scalar (non-sketch) inputs pass through unchanged. //! //! ```yaml //! type: urn:asap:processor:window_merge -//! config: {grace: 5s} +//! config: {sources: 2, grace: 5s} //! ``` use std::collections::BTreeMap; use std::sync::Arc; @@ -33,6 +40,7 @@ use otel_arrow_dfe_otap::pdata::OtapPdata; use otel_arrow_dfe_otap::OTAP_PROCESSOR_FACTORIES; use serde::Deserialize; +use super::answer_nodes::{send_gauges, stage_timing}; use super::codec::{decode_pdata_to_observations, OtapSketchEncoder}; use crate::envelope::SketchEnvelope; use crate::observation::ObservationValueKind; @@ -46,9 +54,12 @@ pub const WINDOW_MERGE_URN: &str = "urn:asap:processor:window_merge"; #[derive(Clone, Debug, Deserialize)] #[serde(deny_unknown_fields)] pub struct WindowMergeConfig { - /// How long after a window's end to wait for the last source. + /// How long after a window's end to wait for a missing source. #[serde(default = "default_grace", with = "humantime_serde")] pub grace: Duration, + /// Sources sending each window; 0 waits the full grace. + #[serde(default)] + pub sources: usize, } fn default_grace() -> Duration { @@ -60,7 +71,7 @@ struct Pending { sketch: Box, } -/// Windows waiting for their grace period, keyed by window end then series. +/// Windows waiting for their sources, keyed by window end then series. #[derive(Default)] pub struct WindowMerger { windows: BTreeMap>, @@ -68,6 +79,10 @@ pub struct WindowMerger { emitted_until: u64, /// Envelopes that arrived after their window was emitted. pub late: u64, + /// When the last source's sketch of each open window arrived (ms). + arrived: BTreeMap, + /// Messages (one per source) that carried each open window. + messages: BTreeMap, } fn series_key(e: &SketchEnvelope) -> String { @@ -97,6 +112,7 @@ impl WindowMerger { self.late += 1; return Ok(()); } + let _ = self.arrived.insert(envelope.window_end_ms, now_ms()); let window = self.windows.entry(envelope.window_end_ms).or_default(); match window.entry(series_key(&envelope)) { std::collections::btree_map::Entry::Occupied(mut slot) => { @@ -112,10 +128,33 @@ impl WindowMerger { Ok(()) } - /// Merged envelopes of every window that ended `grace` or more before `now_ms`. - pub fn due(&mut self, now_ms: u64, grace: Duration) -> Vec { - let cutoff = now_ms.saturating_sub(grace.as_millis() as u64); - let open = self.windows.split_off(&(cutoff + 1)); + /// Counts one source's message carrying window `end`. + pub fn message_carried(&mut self, end: u64) { + if end > self.emitted_until { + *self.messages.entry(end).or_default() += 1; + } + } + + /// Removes and returns when the last sketch of window `end` arrived. + pub fn take_arrival(&mut self, end: u64) -> Option { + self.arrived.remove(&end) + } + + /// Merged envelopes of every window that `sources` messages carried or + /// that ended `grace` or more before `now_ms`. Sources send windows in + /// order, so every earlier window is complete too. + pub fn due(&mut self, now_ms: u64, grace: Duration, sources: usize) -> Vec { + let timed_out = now_ms.saturating_sub(grace.as_millis() as u64); + let complete = self + .messages + .iter() + .filter(|(_, n)| sources > 0 && **n >= sources) + .map(|(end, _)| *end) + .max() + .unwrap_or(0); + let cutoff = timed_out.max(complete); + self.messages = self.messages.split_off(&(cutoff.saturating_add(1))); + let open = self.windows.split_off(&(cutoff.saturating_add(1))); let ready = std::mem::replace(&mut self.windows, open); let mut out = Vec::new(); for (end, series) in ready { @@ -164,7 +203,22 @@ impl WindowMergeProcessor { .await } } - Ok(()) + // Merged windows (not passthrough) carry this stage's timing along to + // the detector. + let now = now_ms(); + let windows: BTreeMap = envs + .iter() + .filter(|e| !e.payload.is_empty()) + .map(|e| (e.window_end_ms, e.window_start_ms)) + .collect(); + let timing: Vec = windows + .into_iter() + .flat_map(|(end, start)| { + let arrived = self.merger.take_arrival(end).unwrap_or(now); + stage_timing(start, end, "sketch", "merge", arrived, now) + }) + .collect(); + send_gauges(&timing, effects).await } } @@ -195,9 +249,12 @@ impl local::Processor for WindowMergeProcessor { } }; let mut passthrough = Vec::new(); + let mut carried = std::collections::BTreeSet::new(); + let late_before = self.merger.late; for obs in observations { match (obs.value.kind, obs.value.envelope) { (ObservationValueKind::Envelope, Some(env)) if !env.payload.is_empty() => { + let _ = carried.insert(env.window_end_ms); if let Err(e) = self.merger.add(*env) { effects.info(&format!("window_merge: {e}")).await; } @@ -206,14 +263,31 @@ impl local::Processor for WindowMergeProcessor { _ => {} } } - self.emit(passthrough, effects).await + if self.merger.late > late_before { + effects + .info(&format!( + "window_merge: {} sketches arrived after their window was emitted", + self.merger.late - late_before + )) + .await; + } + for end in carried { + self.merger.message_carried(end); + } + self.emit(passthrough, effects).await?; + let due = self + .merger + .due(now_ms(), self.config.grace, self.config.sources); + self.emit(due, effects).await } Message::Control(NodeControlMsg::TimerTick { .. }) => { - let due = self.merger.due(now_ms(), self.config.grace); + let due = self + .merger + .due(now_ms(), self.config.grace, self.config.sources); self.emit(due, effects).await } Message::Control(NodeControlMsg::Shutdown { .. }) => { - let all = self.merger.due(u64::MAX, Duration::ZERO); + let all = self.merger.due(u64::MAX, Duration::ZERO, 0); self.emit(all, effects).await } _ => Ok(()), @@ -280,8 +354,8 @@ mod tests { m.add(hll(500..1500, 0)).unwrap(); m.add(hll(0..10, 60_000)).unwrap(); let grace = Duration::from_secs(5); - assert!(m.due(64_000, grace).is_empty()); - let out = m.due(65_000, grace); + assert!(m.due(64_000, grace, 0).is_empty()); + let out = m.due(65_000, grace, 0); assert_eq!(out.len(), 1); assert_eq!(out[0].count, 2000); let merged = HLLWrapper::from_snapshot(&out[0].payload).unwrap(); @@ -290,6 +364,48 @@ mod tests { // A source arriving after its window was emitted is late. m.add(hll(0..5, 0)).unwrap(); assert_eq!(m.late, 1); - assert_eq!(m.due(125_000, grace).len(), 1); + assert_eq!(m.due(125_000, grace, 0).len(), 1); + } + + #[test] + fn emits_a_window_once_every_source_sent_it() { + let mut m = WindowMerger::default(); + let grace = Duration::from_secs(5); + m.add(hll(0..1000, 0)).unwrap(); + m.message_carried(60_000); + // One of two sources: wait. + assert!(m.due(60_010, grace, 2).is_empty()); + m.add(hll(500..1500, 0)).unwrap(); + m.message_carried(60_000); + let out = m.due(60_020, grace, 2); + assert_eq!(out.len(), 1); + assert_eq!(out[0].count, 2000); + // A missing source: emitted at the grace. + m.add(hll(0..10, 60_000)).unwrap(); + m.message_carried(120_000); + assert!(m.due(124_000, grace, 2).is_empty()); + assert_eq!(m.due(125_000, grace, 2).len(), 1); + } + + #[test] + fn remembers_when_a_window_last_got_a_sketch() { + let mut m = WindowMerger::default(); + let before = now_ms(); + m.add(hll(0..10, 0)).unwrap(); + let arrived = m.take_arrival(60_000).unwrap(); + assert!(arrived >= before && arrived <= now_ms()); + assert_eq!(m.take_arrival(60_000), None); + } + + #[test] + fn stage_timing_is_seconds_after_the_window_end() { + let gauges = + super::super::answer_nodes::stage_timing(0, 60_000, "sketch", "merge", 61_500, 65_250); + let get = |name: &str| gauges.iter().find(|g| g.metric_name == name).unwrap(); + assert_eq!(get("asap.stage.received").value, 1.5); + assert_eq!(get("asap.stage.emitted").value, 5.25); + assert!(get("asap.stage.emitted") + .labels + .contains(&KeyValue::new("stage", "merge"))); } } diff --git a/deploy/telemetry-functions/README.md b/deploy/telemetry-functions/README.md index 1437e55..edf098b 100644 --- a/deploy/telemetry-functions/README.md +++ b/deploy/telemetry-functions/README.md @@ -4,14 +4,18 @@ Every stage is the official OTAP dataflow engine (`asap-collector`: the upstream controller with the ASAP nodes linked in), one container per stage, each pinned to its own CPU, connected by OTLP/HTTP over the compose network. -Three pipelines run side by side over the same input and share no processor: +Three approaches run side by side over the same input and share no processor: ```text -sketch: generator-a/b -> branch-a/b (sketch per window) -> merge -> detector -> Prometheus, Loki -exact: exact-generator-a/b -> exact-branch-a/b (exact window) -> exact-merge -> exact-detector -> Prometheus, Loki -raw: raw-generator-a/b -> raw-branch-a/b (forward) -> raw-merge (forward) -> raw-sink -> loki-raw +sketch: generator-a/b -> branch-a/b (sketch per window) -> merge -> detector -> Prometheus, Loki +exact: exact-generator-a/b -> exact-branch-a/b (exact window) -> exact-merge -> exact-detector -> Prometheus, Loki +store and query: raw-generator-a/b -> raw-sink-a/b -> loki-raw; query-job runs the answer query over loki-raw ``` +The store-and-query pipeline (`method="raw"`) keeps every raw point and +computes the answer later, the way a log store would be used without inline +functions. + The two generators of each pipeline (sources a and b) run the same configs with the same scenario start (`SCENARIO_START`), and every point is a function of scenario time, so all three pipelines receive identical points. @@ -20,13 +24,13 @@ of scenario time, so all three pipelines receive identical points. | --- | --- | | generator-a/b, exact-generator-a/b, raw-generator-a/b | `scenario_generator`: seeded traffic for the use case (temporary stand-in for the official traffic generator) | | branch-a/b | `asap_sketches`: one sketch per series per 1-minute window | -| merge | `window_merge`: merges both sources' sketches per window, 5 s after the window ends | +| merge | `window_merge`: merges both sources' sketches per window and emits it as soon as both have sent it (5 s timeout for a missing source) | | detector | `sketch_detector`: answers with bounds, baseline comparisons, events (`method="sketch"`) | | exact-branch-a/b | `exact_branch`: every value per quantile series and a value -> count map per attribute key, per 1-minute window | -| exact-merge | `exact_merge`: adds both sources' exact windows, 5 s after the window ends | +| 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-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 | +| 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"`) | | 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 @@ -42,8 +46,34 @@ 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, against the offered -2 x 20,000 points/s. +`loki-raw` acknowledged) divided by the window length. + +**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 +end (`asap_stage_received` and `asap_stage_emitted`, labelled `method` and +`stage`); detectors pass the merge's timing on to Prometheus. With the query +job's backend query time, the top-k dashboard stacks each approach's freshness: +the window wait (up to 60 s for a window's first sample), branch and transfer to +the merge, the merge's wait for both sources, transfer to the detector, the detector, and the +query. For store and query: the window wait, ingestion into `loki-raw` (last +point acknowledged) and the LogQL query. + +**Whole-system cost.** Prometheus also scrapes `loki-raw`, itself and the +query job, so the top-k dashboard shows CPU and memory per component and per +approach: sketch and exact count their four stages, store and query its sinks, +`loki-raw` and the query job; Prometheus, which holds every approach's answers +and the monitoring data, is shown on its own. + +The merges count a window's messages, not distinct sources: a source that sends +a window twice (a retried request, or a branch restarted or reconfigured +mid-window) completes it early, and the other source's late input is logged and +dropped. If raw-sink is still waiting for `loki-raw` 50 s after a window's end, +it reports the window anyway with `asap_raw_unacknowledged` (points still +waiting) and that time as its delay; the query job flags such windows with +`asap_raw_incomplete`. The generators size their batches for the scenario's +peak rate, so at the starting rate a batch holds about 60 points instead of +1,024. A cardinality change on a growing attribute applies the growth from the +start of the scenario, not from the change. 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 @@ -62,7 +92,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-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 | +| `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 | Answers reach Prometheus over OTLP as `asap_quantile`, `asap_distinct`, `asap_topk`, `asap_points`, `asap_cardinality_state`, `asap_drift_ks`, @@ -76,7 +106,7 @@ and payload bytes per receiver, labelled with the stage. ## Run -Linux, Docker with Compose, 19 free CPUs (0-18 by default; `CPU_BRANCH_A=20` +Linux, Docker with Compose, 18 free CPUs (0-17 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: @@ -96,17 +126,18 @@ deploy/telemetry-functions/acceptance.py --report acceptance.json ``` Runs all four use cases on the containers with time scaled down 6x (10-second -windows, changes at second 100) and checks the sketch answers against the +windows, changes at second 100, at each scenario's starting rate without its +rate and cardinality ramps) and checks the sketch answers against the exact ground truth through Prometheus and Loki. It exits non-zero if any check fails. The checks: | Use case | 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) | +| 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 | | 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; 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; store and query answers from `loki-raw` with heavy-hitter 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. @@ -115,8 +146,8 @@ the runner's CPUs: a correctness check, not a capacity measurement. `experiments.py pipelines` runs the three pipelines side by side over offered rates and records each stage's CPU, memory, delivered rate and window -completion delay (the sketch branch's CPU minus the raw branch's is the cost of -the function); `experiments.py accuracy-size` measures error +completion delay (the raw pipeline's per-source stage is raw-sink-a/b); +`experiments.py accuracy-size` measures error against exact as KLL `k` and Count-Min width change. Results and the merge/serialization benchmark (`asap-merge-bench`) are in [docs/results](../../docs/results/). diff --git a/deploy/telemetry-functions/acceptance.py b/deploy/telemetry-functions/acceptance.py index f01398e..678e013 100755 --- a/deploy/telemetry-functions/acceptance.py +++ b/deploy/telemetry-functions/acceptance.py @@ -9,9 +9,10 @@ fails. Time is scaled down 6x so a use case takes a few minutes: 10-second windows, a -5-second merge grace, changes after 1/6 of their configured time and growth 6x +5-second timeout for a missing source, changes after 1/6 of their configured time and growth 6x faster per minute. Rules counted in windows (baselines, thresholds) are as -deployed. +deployed. Rate and cardinality ramps are left out: acceptance checks the +answers at the starting rate; how each approach scales is what the demo shows. deploy/telemetry-functions/acceptance.py --docker-command "sudo docker" deploy/telemetry-functions/acceptance.py --use-cases drift topk @@ -34,16 +35,20 @@ HERE = Path(__file__).resolve().parent REPO = HERE.parents[1] SCALE = 6 -RATE = 20000 # signals/s per source, as in the generator configs WINDOW_S = 60 // SCALE 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", - "hh-accuracy") + "exact-merge", "exact-detector", "raw-sink-a", "raw-sink-b", "hh-accuracy") METHODS = ("sketch", "exact", "raw") +def rate(directory): + """Data points per second per source at the start of the scenario.""" + text = (HERE / directory / "generator-a.yaml").read_text() + return int(re.search(r"signals_per_second: (\d+)", text)[1]) + + def scaled_copy(directory, into): """A stage-config directory with time scaled down SCALE times.""" into.mkdir(exist_ok=True) @@ -51,8 +56,9 @@ def scaled_copy(directory, into): text = src.read_text() text = text.replace("window_size: 60s", f"window_size: {WINDOW_S}s") text = text.replace("window: 60s", f"window: {WINDOW_S}s") - # The merge grace is not scaled: shared CI runners need the full 5 s - # for every source's sketch to arrive. + # 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) 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) @@ -60,8 +66,10 @@ def scaled_copy(directory, into): class Stack: - def __init__(self, docker, pipelines, common, prometheus, loki): - self.docker, self.pipelines, self.common = docker, pipelines, common + def __init__(self, docker, use_case, pipelines, common, prometheus, loki): + self.docker, self.use_case = docker, use_case + self.pipelines, self.common = pipelines, common + self.rate = rate(use_case) self.prometheus, self.loki = prometheus, loki # The generators of all three pipelines count from the same second. self.start = int(time.time()) + 10 @@ -69,7 +77,8 @@ def __init__(self, docker, pipelines, common, prometheus, loki): def compose(self, *args): # A CI runner has about 16 GB; the raw pipeline's store gets 4. env = dict(os.environ, PIPELINES=str(self.pipelines), COMMON=str(self.common), - SCENARIO_START=str(self.start), LOKI_RAW_MEMORY="4g") + SCENARIO_START=str(self.start), LOKI_RAW_MEMORY="4g", + USE_CASE=self.use_case, WINDOW_SECONDS=str(WINDOW_S)) cmd = self.docker + ["compose", "-f", str(HERE / "compose.yaml"), *args] return subprocess.run(cmd, env=env, capture_output=True, text=True, check=True) @@ -123,7 +132,7 @@ def common_checks(stack, flt): f"{sum(v == 1 for _, v in up)}/{len(STAGES)} collectors up") w = windows(stack, flt) shared = sorted(set(w["sketch"]) & set(w["exact"]))[1:-1] # skip partial edges - expected = 2 * RATE * WINDOW_S + expected = 2 * stack.rate * WINDOW_S full = [w["exact"][t] for t in shared] yield check("full rate reaches the end (no backpressure)", full and all(abs(v - expected) <= 0.01 * expected for v in full), @@ -139,6 +148,14 @@ def common_checks(stack, flt): set(delay) >= set(METHODS), ", ".join(f"{m} {delay[m]:.1f} s" for m in METHODS if m in delay) + f"; raw points/window {sorted(set(raw))} vs {expected}") + stages = {(m["method"], m["stage"]) for m, _ in stack.vector( + "max by (method, stage) (max_over_time(asap_stage_emitted[5m]))")} + want = {(m, s) for m in ("sketch", "exact") for s in ("merge", "detector")} + yield check("merge and detector report their timing (latency breakdown)", stages >= want, + f"{sorted(stages)}") + queried = {m["method"] for m, _ in stack.vector("max_over_time(asap_query_seconds[5m])")} + yield check("the query job times the backend query of sketch and exact", + queried >= {"sketch", "exact"}, f"{sorted(queried)}") def latency_checks(stack, t): @@ -199,17 +216,25 @@ def drift_checks(stack, t): 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"}, + yield check("new route in the sketch and exact top-10s", {m["method"] for m, _ in new} >= {"sketch", "exact"}, str(sorted(m["method"] for m, _ in new))) 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}')} + hh = {m["__name__"]: v for m, v in stack.vector( + f'{{__name__=~"asap_hh_(precision|recall|are)",method="sketch"}} @ {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") + raw = {m["__name__"]: v for m, v in stack.vector( + '{__name__=~"asap_hh_(precision|recall)",method="raw"}')} + loki = stack.vector('max_over_time(asap_query_seconds{method="raw",backend="loki"}[5m])') + yield check("store and query: top-k from loki-raw, heavy-hitter recall >= 0.9", + loki and raw.get("asap_hh_recall", 0) >= 0.9, + ", ".join(f"{k.removeprefix('asap_hh_')} {v:.3f}" for k, v in sorted(raw.items())) + + (f"; Loki query {loki[0][1]:.2f} s" if loki else "; no Loki query timed")) 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), @@ -235,7 +260,8 @@ def run_case(args, use_case): scaled_copy("common", common) for f in [pipelines, common, *pipelines.iterdir(), *common.iterdir()]: f.chmod(0o755 if f.is_dir() else 0o644) - stack = Stack(shlex.split(args.docker_command), pipelines, common, args.prometheus, args.loki) + stack = Stack(shlex.split(args.docker_command), use_case, pipelines, common, + args.prometheus, args.loki) stack.compose("down", "--remove-orphans") stack.compose("up", "-d") try: diff --git a/deploy/telemetry-functions/cardinality/merge.yaml b/deploy/telemetry-functions/cardinality/merge.yaml index cb392f2..7a27162 100644 --- a/deploy/telemetry-functions/cardinality/merge.yaml +++ b/deploy/telemetry-functions/cardinality/merge.yaml @@ -1,4 +1,4 @@ -# Stage 2: merge every source's sketch for a window, 5 s after it ends. +# Stage 2: merge both sources' sketches for a window as soon as both sent it (5 s timeout). version: otel_dataflow/v1 engine: http_admin: @@ -20,7 +20,7 @@ groups: http: {listening_addr: "0.0.0.0:4318", wait_for_result: false} merge: type: urn:asap:processor:window_merge - config: {grace: 5s} + config: {sources: 2, grace: 5s} to_detector: type: exporter:otlp_http config: diff --git a/deploy/telemetry-functions/common/exact-merge.yaml b/deploy/telemetry-functions/common/exact-merge.yaml index e8f44de..e6bfdf6 100644 --- a/deploy/telemetry-functions/common/exact-merge.yaml +++ b/deploy/telemetry-functions/common/exact-merge.yaml @@ -1,5 +1,5 @@ -# Exact pipeline, stage 2: add up both sources' exact windows, 5 s after -# each window ends. +# Exact pipeline, stage 2: add up both sources' exact windows as soon as both +# sent a window (5 s timeout). version: otel_dataflow/v1 engine: http_admin: @@ -21,7 +21,7 @@ groups: http: {listening_addr: "0.0.0.0:4318", wait_for_result: false, max_request_body_size: 1GiB} merge: type: urn:asap:processor:exact_merge - config: {grace: 5s} + config: {sources: 2, grace: 5s} to_detector: type: exporter:otlp_http config: diff --git a/deploy/telemetry-functions/common/raw-branch.yaml b/deploy/telemetry-functions/common/raw-branch.yaml deleted file mode 100644 index ea1b866..0000000 --- a/deploy/telemetry-functions/common/raw-branch.yaml +++ /dev/null @@ -1,29 +0,0 @@ -# Raw pipeline, stage 1 at each source: forward every raw point. -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} - to_merge: - type: exporter:otlp_http - config: - endpoint: http://raw-merge:4318 - client_pool_size: 1 - max_in_flight: 1 - http: {timeout: 30s, compression: none} - connections: - - {from: receiver, to: to_merge} diff --git a/deploy/telemetry-functions/common/raw-merge.yaml b/deploy/telemetry-functions/common/raw-merge.yaml deleted file mode 100644 index e9d3664..0000000 --- a/deploy/telemetry-functions/common/raw-merge.yaml +++ /dev/null @@ -1,29 +0,0 @@ -# Raw pipeline, stage 2: forward both sources' raw points to one place. -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} - to_sink: - type: exporter:otlp_http - config: - endpoint: http://raw-sink:4318 - client_pool_size: 1 - max_in_flight: 1 - http: {timeout: 30s, compression: none} - connections: - - {from: receiver, to: to_sink} diff --git a/deploy/telemetry-functions/common/raw-sink.yaml b/deploy/telemetry-functions/common/raw-sink-a.yaml similarity index 73% rename from deploy/telemetry-functions/common/raw-sink.yaml rename to deploy/telemetry-functions/common/raw-sink-a.yaml index fa34ce2..bf569e4 100644 --- a/deploy/telemetry-functions/common/raw-sink.yaml +++ b/deploy/telemetry-functions/common/raw-sink-a.yaml @@ -1,6 +1,8 @@ -# Raw pipeline, end: write every raw point as a log record to loki-raw, and -# report per window when loki-raw acknowledged its last point -# (asap.answer.delay, method="raw") and each point's time to acknowledgement. +# Store-and-query pipeline, source a: the only stage. Writes every raw point as +# a log record to loki-raw, and reports per window when loki-raw acknowledged +# its last point (asap.answer.delay, method="raw", source="a") and each +# point's time to acknowledgement. The query job computes the answer from +# loki-raw. version: otel_dataflow/v1 engine: http_admin: @@ -23,14 +25,14 @@ groups: sink: type: urn:asap:processor:raw_sink outputs: [logs, latency] - config: {window: 60s, grace: 5s} + config: {window: 60s, grace: 5s, labels: {source: "a"}} to_loki_raw: type: exporter:otlp_http config: endpoint: http://loki-raw:3100 logs_endpoint: http://loki-raw:3100/otlp/v1/logs - client_pool_size: 1 - max_in_flight: 1 + client_pool_size: 4 + max_in_flight: 4 http: {timeout: 30s, compression: none} to_prometheus: type: exporter:otlp_http diff --git a/deploy/telemetry-functions/common/raw-sink-b.yaml b/deploy/telemetry-functions/common/raw-sink-b.yaml new file mode 100644 index 0000000..7624027 --- /dev/null +++ b/deploy/telemetry-functions/common/raw-sink-b.yaml @@ -0,0 +1,48 @@ +# Store-and-query pipeline, source b: the only stage. Writes every raw point as +# a log record to loki-raw, and reports per window when loki-raw acknowledged +# its last point (asap.answer.delay, method="raw", source="b") and each +# point's time to acknowledgement. The query job computes the answer from +# loki-raw. +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} + sink: + type: urn:asap:processor:raw_sink + outputs: [logs, latency] + config: {window: 60s, grace: 5s, labels: {source: "b"}} + to_loki_raw: + type: exporter:otlp_http + config: + endpoint: http://loki-raw:3100 + logs_endpoint: http://loki-raw:3100/otlp/v1/logs + client_pool_size: 4 + max_in_flight: 4 + http: {timeout: 30s, compression: none} + 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: sink} + - {from: 'sink["logs"]', to: to_loki_raw} + - {from: 'sink["latency"]', to: to_prometheus} diff --git a/deploy/telemetry-functions/compose.yaml b/deploy/telemetry-functions/compose.yaml index 9fd0627..5046d44 100644 --- a/deploy/telemetry-functions/compose.yaml +++ b/deploy/telemetry-functions/compose.yaml @@ -2,10 +2,11 @@ # sketch: generator-a/b -> branch-a/b (sketch) -> merge -> detector # exact: exact-generator-a/b -> exact-branch-a/b (exact window) -> # exact-merge -> exact-detector -# raw: raw-generator-a/b -> raw-branch-a/b -> raw-merge -> raw-sink -# (every raw point into loki-raw) +# store and query (raw): raw-generator-a/b -> raw-sink-a/b (every raw point +# into loki-raw); query-job computes the answer from 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. +# query-job also times each approach's backend query (Prometheus or Loki). # 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) @@ -100,51 +101,48 @@ services: volumes: ["${PIPELINES:-./latency}:/pipelines:ro", "${COMMON:-./common}:/common:ro"] cpuset: "${CPU_EXACT_DETECTOR:-11}" depends_on: [prometheus, hh-accuracy] - # raw pipeline + # store-and-query pipeline: one stage per source, then the query job raw-generator-a: <<: *generator command: ["--config", "/pipelines/generator-a.yaml"] volumes: ["${PIPELINES:-./latency}:/pipelines:ro", "${COMMON:-./common}:/common:ro"] cpuset: "${CPU_RAW_GENERATOR_A:-12}" - links: ["raw-branch-a:downstream"] - depends_on: [raw-branch-a] + links: ["raw-sink-a:downstream"] + depends_on: [raw-sink-a] raw-generator-b: <<: *generator command: ["--config", "/pipelines/generator-b.yaml"] volumes: ["${PIPELINES:-./latency}:/pipelines:ro", "${COMMON:-./common}:/common:ro"] cpuset: "${CPU_RAW_GENERATOR_B:-13}" - links: ["raw-branch-b:downstream"] - depends_on: [raw-branch-b] - raw-branch-a: + links: ["raw-sink-b:downstream"] + depends_on: [raw-sink-b] + raw-sink-a: <<: *collector - command: ["--config", "/common/raw-branch.yaml"] + command: ["--config", "/common/raw-sink-a.yaml"] volumes: ["${PIPELINES:-./latency}:/pipelines:ro", "${COMMON:-./common}:/common:ro"] - cpuset: "${CPU_RAW_BRANCH_A:-14}" - depends_on: [raw-merge] - raw-branch-b: - <<: *collector - command: ["--config", "/common/raw-branch.yaml"] - volumes: ["${PIPELINES:-./latency}:/pipelines:ro", "${COMMON:-./common}:/common:ro"] - cpuset: "${CPU_RAW_BRANCH_B:-15}" - depends_on: [raw-merge] - raw-merge: - <<: *collector - command: ["--config", "/common/raw-merge.yaml"] - volumes: ["${PIPELINES:-./latency}:/pipelines:ro", "${COMMON:-./common}:/common:ro"] - cpuset: "${CPU_RAW_MERGE:-16}" - depends_on: [raw-sink] - raw-sink: + cpuset: "${CPU_RAW_SINK_A:-14}" + depends_on: [prometheus, loki-raw] + raw-sink-b: <<: *collector - command: ["--config", "/common/raw-sink.yaml"] + command: ["--config", "/common/raw-sink-b.yaml"] volumes: ["${PIPELINES:-./latency}:/pipelines:ro", "${COMMON:-./common}:/common:ro"] - cpuset: "${CPU_RAW_SINK:-17}" + cpuset: "${CPU_RAW_SINK_B:-15}" + depends_on: [prometheus, loki-raw] + query-job: + image: python:3.12-slim + command: ["python3", "/app/query_job.py"] + volumes: ["./query_job.py:/app/query_job.py:ro"] + environment: + USE_CASE: "${USE_CASE:-latency}" + WINDOW_SECONDS: "${WINDOW_SECONDS:-60}" + cpuset: "${CPU_QUERY_JOB:-16}" 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}" + cpuset: "${CPU_HH_ACCURACY:-17}" depends_on: [prometheus] prometheus: diff --git a/deploy/telemetry-functions/drift/merge.yaml b/deploy/telemetry-functions/drift/merge.yaml index cb392f2..7a27162 100644 --- a/deploy/telemetry-functions/drift/merge.yaml +++ b/deploy/telemetry-functions/drift/merge.yaml @@ -1,4 +1,4 @@ -# Stage 2: merge every source's sketch for a window, 5 s after it ends. +# Stage 2: merge both sources' sketches for a window as soon as both sent it (5 s timeout). version: otel_dataflow/v1 engine: http_admin: @@ -20,7 +20,7 @@ groups: http: {listening_addr: "0.0.0.0:4318", wait_for_result: false} merge: type: urn:asap:processor:window_merge - config: {grace: 5s} + config: {sources: 2, grace: 5s} to_detector: type: exporter:otlp_http config: diff --git a/deploy/telemetry-functions/experiments.py b/deploy/telemetry-functions/experiments.py index ed7a12f..0abead1 100755 --- a/deploy/telemetry-functions/experiments.py +++ b/deploy/telemetry-functions/experiments.py @@ -4,8 +4,8 @@ pipelines The three pipelines (sketch, exact, raw) side by side over offered rates, fed identical points: CPU of each stage, memory, points delivered per second and window completion delay. The raw - pipeline's branch only forwards points, so the sketch branch's - CPU minus the raw branch's is the cost of the function. + pipeline has one stage per source (raw-sink-a/b, writing to + loki-raw), reported in the branch columns. accuracy-size Error against exact versus sketch size: KLL k (latency p99 error) and Count-Min width (top-10 recall and count error), with the bytes merge receives per window. @@ -26,11 +26,11 @@ import acceptance as acc -# Stages of each pipeline: per-source branches, merge, end. +# Stages of each pipeline: per-source stage, merge, end (None: no such stage). PIPELINES = { "sketch": ("branch", "merge", "detector"), "exact": ("exact-branch", "exact-merge", "exact-detector"), - "raw": ("raw-branch", "raw-merge", "raw-sink"), + "raw": ("raw-sink", None, None), } @@ -60,7 +60,7 @@ def measure(stack, use_case, seconds): flt = acc.CASES[use_case][0] row = {} for name, (branch, merge, end) in PIPELINES.items(): - stages = [f"{branch}-a", f"{branch}-b", merge, end] + stages = [f"{branch}-a", f"{branch}-b"] + [s for s in (merge, end) if s] # Total points over the range divided by the windows it covers. sel = f'asap_points{{method="{name}"' + ("" if name == "raw" else f",{flt}") + "}" seen = (stack.vector(f"max(count_over_time({sel}[{rng}]))") or [({}, float("nan"))])[0][1] @@ -81,7 +81,8 @@ def measure(stack, use_case, seconds): def run(args, use_case, rate, edits=(), extra=None): with tempfile.TemporaryDirectory(prefix="asap-exp-") as tmp: pipelines, common = prepare(use_case, Path(tmp), rate, edits) - stack = acc.Stack(shlex.split(args.docker_command), pipelines, common, args.prometheus, args.loki) + stack = acc.Stack(shlex.split(args.docker_command), use_case, pipelines, common, + args.prometheus, args.loki) stack.compose("down", "--remove-orphans") stack.compose("up", "-d") try: @@ -119,7 +120,7 @@ def topk_error(stack): 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}')} + for m, v in stack.vector(f'{{__name__=~"asap_hh_(precision|recall|f1|are)",method="sketch"}} @ {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"), **{f"hh_{k}": v for k, v in sorted(hh.items())}, **merge_bytes(stack)} diff --git a/deploy/telemetry-functions/grafana/dashboards/cardinality.json b/deploy/telemetry-functions/grafana/dashboards/cardinality.json index 7d1191e..c06d9f0 100644 --- a/deploy/telemetry-functions/grafana/dashboards/cardinality.json +++ b/deploy/telemetry-functions/grafana/dashboards/cardinality.json @@ -42,7 +42,7 @@ }, "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-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: 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 }, diff --git a/deploy/telemetry-functions/grafana/dashboards/drift.json b/deploy/telemetry-functions/grafana/dashboards/drift.json index c11162a..81c8075 100644 --- a/deploy/telemetry-functions/grafana/dashboards/drift.json +++ b/deploy/telemetry-functions/grafana/dashboards/drift.json @@ -42,7 +42,7 @@ }, "options": { "mode": "html", - "content": "
generator-a/b
sources A and B
→
branch-a/b
Stage 1 per source: KLL (k = 200) per route per 1-min window
→
merge
Stage 2: merges both sources' sketches per window
→
detector
KS distance vs last 5 windows; drift events
→
Prometheus + Loki
method="sketch"
exact-generator-a/b
same points
→
exact-branch-a/b
Stage 1 per source: every value per route per 1-min window
→
exact-merge
Stage 2: adds both sources' exact windows
→
exact-detector
exact KS distance; drift events
→
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: KLL (k = 200) per route per 1-min window
→
merge
Stage 2: merges both sources' sketches per window
→
detector
KS distance vs last 5 windows; drift events
→
Prometheus + Loki
method="sketch"
exact-generator-a/b
same points
→
exact-branch-a/b
Stage 1 per source: every value per route per 1-min window
→
exact-merge
Stage 2: adds both sources' exact windows
→
exact-detector
exact KS distance; drift events
→
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 }, diff --git a/deploy/telemetry-functions/grafana/dashboards/latency.json b/deploy/telemetry-functions/grafana/dashboards/latency.json index 5995081..1f608e5 100644 --- a/deploy/telemetry-functions/grafana/dashboards/latency.json +++ b/deploy/telemetry-functions/grafana/dashboards/latency.json @@ -42,7 +42,7 @@ }, "options": { "mode": "html", - "content": "
generator-a/b
sources A and B
→
branch-a/b
Stage 1 per source: KLL (k = 200) per route per 1-min window
→
merge
Stage 2: merges both sources' sketches per window
→
detector
p50/p99 with bounds
→
Prometheus + Loki
method="sketch"
exact-generator-a/b
same points
→
exact-branch-a/b
Stage 1 per source: every value per route per 1-min window
→
exact-merge
Stage 2: adds both sources' exact windows
→
exact-detector
exact p50/p99
→
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: KLL (k = 200) per route per 1-min window
→
merge
Stage 2: merges both sources' sketches per window
→
detector
p50/p99 with bounds
→
Prometheus + Loki
method="sketch"
exact-generator-a/b
same points
→
exact-branch-a/b
Stage 1 per source: every value per route per 1-min window
→
exact-merge
Stage 2: adds both sources' exact windows
→
exact-detector
exact p50/p99
→
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 }, diff --git a/deploy/telemetry-functions/grafana/dashboards/topk.json b/deploy/telemetry-functions/grafana/dashboards/topk.json index 11e4ac2..83432cc 100644 --- a/deploy/telemetry-functions/grafana/dashboards/topk.json +++ b/deploy/telemetry-functions/grafana/dashboards/topk.json @@ -1,166 +1,179 @@ { "uid": "asap-topk", - "title": "Heavy hitters and new keys: sketch vs exact (real collector containers)", + "title": "Heavy hitters: 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" }, "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: which routes are heavy hitters, and which are new?", + "options": { + "mode": "html", + "content": "
generator-a/b
sources A and B
→
branch-a/b
Count-Min 4×512 + top-20 heap per 1-min window
→
merge
both sources' sketches per window
→
detector
top-20 with bounds; new heavy hitters
→
Prometheus
query: method="sketch"
exact-generator-a/b
same points
→
exact-branch-a/b
route → count map per 1-min window
→
exact-merge
adds both sources' maps: every key
→
exact-detector
exact top-20; new heavy hitters
→
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 top-20 once per window
→
Prometheus
answer: method="raw"

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.

" + }, "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: 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 }, { - "type": "bargauge", - "title": "Top-10 routes, latest window: CMS + heap (lower bound to estimate)", - "datasource": { - "type": "prometheus", - "uid": "asap-prometheus" - }, + "type": "row", + "title": "Answers", + "collapsed": false, + "panels": [], "gridPos": { - "h": 10, - "w": 12, + "h": 1, + "w": 24, "x": 0, "y": 10 }, - "targets": [ - { - "refId": "A", - "expr": "sort_desc(asap_topk{method=\"sketch\",bound=\"estimate\",rank=~\"[1-9]|10\"})", - "legendFormat": "#{{rank}} {{key}}", - "instant": true - } - ], - "options": { - "orientation": "horizontal", - "displayMode": "basic", - "reduceOptions": { - "calcs": [ - "lastNotNull" - ], - "fields": "", - "values": false - } - }, - "fieldConfig": { - "defaults": { - "decimals": 0 - }, - "overrides": [] - }, "id": 2 }, { - "type": "bargauge", - "title": "Top-10 routes, latest window: exact counts", + "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.", "datasource": { "type": "prometheus", "uid": "asap-prometheus" }, - "gridPos": { - "h": 10, - "w": 12, - "x": 12, - "y": 10 - }, "targets": [ { "refId": "A", - "expr": "sort_desc(asap_topk{method=\"exact\",bound=\"estimate\",rank=~\"[1-9]|10\"})", - "legendFormat": "#{{rank}} {{key}}", + "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\", \"(.*)\"))", + "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\", \"(.*)\"))", + "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\", \"(.*)\"))", + "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\", \"(.*)\")))", + "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\", \"(.*)\")))", + "format": "table", "instant": true } ], - "options": { - "orientation": "horizontal", - "displayMode": "basic", - "reduceOptions": { - "calcs": [ - "lastNotNull" - ], - "fields": "", - "values": false - } - }, - "fieldConfig": { - "defaults": { - "decimals": 0 - }, - "overrides": [] - }, - "id": 3 - }, - { - "type": "timeseries", - "title": "Count of /new/hot-route per minute: sketch with Count-Min bound vs exact", - "datasource": { - "type": "prometheus", - "uid": "asap-prometheus" - }, - "gridPos": { - "h": 8, - "w": 12, - "x": 0, - "y": 20 - }, - "targets": [ + "transformations": [ { - "refId": "A", - "expr": "asap_topk{key=\"/new/hot-route\",method=\"sketch\",bound=\"estimate\"}", - "legendFormat": "sketch" + "id": "joinByField", + "options": { + "byField": "key", + "mode": "outer" + } }, { - "refId": "B", - "expr": "asap_topk{key=\"/new/hot-route\",method=\"sketch\",bound=\"lower\"}", - "legendFormat": "sketch lower" + "id": "organize", + "options": { + "excludeByName": { + "Time": true, + "Time 1": true, + "Time 2": true, + "Time 3": true, + "Time 4": true, + "Time 5": true + }, + "indexByName": { + "key": 0, + "exact_rank": 1, + "Value #A": 2, + "sketch_rank": 3, + "Value #B": 4, + "Value #D": 5, + "raw_rank": 6, + "Value #C": 7, + "Value #E": 8 + }, + "renameByName": { + "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" + } + } }, { - "refId": "C", - "expr": "asap_topk{key=\"/new/hot-route\",method=\"exact\"}", - "legendFormat": "exact" + "id": "convertFieldType", + "options": { + "conversions": [ + { + "targetField": "Exact rank", + "destinationType": "number" + }, + { + "targetField": "Sketch rank", + "destinationType": "number" + }, + { + "targetField": "Store+query rank", + "destinationType": "number" + } + ] + } } ], + "options": { + "showHeader": true, + "sortBy": [ + { + "displayName": "Exact count", + "desc": true + } + ] + }, "fieldConfig": { "defaults": { + "decimals": 0, "custom": { - "spanNulls": true + "align": "auto" } }, "overrides": [ { "matcher": { "id": "byName", - "options": "sketch" + "options": "Sketch count" }, "properties": [ - { - "id": "custom.fillBelowTo", - "value": "sketch lower" - }, - { - "id": "custom.fillOpacity", - "value": 20 - }, { "id": "color", "value": { "mode": "fixed", - "fixedColor": "blue" + "fixedColor": "#5794f2" + } + }, + { + "id": "custom.cellOptions", + "value": { + "type": "color-text" } } ] @@ -168,18 +181,20 @@ { "matcher": { "id": "byName", - "options": "sketch lower" + "options": "Exact count" }, "properties": [ - { - "id": "custom.lineWidth", - "value": 0 - }, { "id": "color", "value": { "mode": "fixed", - "fixedColor": "blue" + "fixedColor": "#ff9830" + } + }, + { + "id": "custom.cellOptions", + "value": { + "type": "color-text" } } ] @@ -187,273 +202,372 @@ { "matcher": { "id": "byName", - "options": "exact" + "options": "Store+query count" }, "properties": [ { "id": "color", "value": { "mode": "fixed", - "fixedColor": "orange" + "fixedColor": "#8e8e8e" } }, { - "id": "custom.lineStyle", + "id": "custom.cellOptions", "value": { - "fill": "dash", - "dash": [ - 6, - 4 - ] + "type": "color-text" } } ] + }, + { + "matcher": { + "id": "byName", + "options": "Route" + }, + "properties": [ + { + "id": "custom.width", + "value": 150 + } + ] + }, + { + "matcher": { + "id": "byName", + "options": "Sketch error" + }, + "properties": [ + { + "id": "unit", + "value": "percent" + }, + { + "id": "decimals", + "value": 1 + } + ] + }, + { + "matcher": { + "id": "byName", + "options": "Store+query error" + }, + "properties": [ + { + "id": "unit", + "value": "percent" + }, + { + "id": "decimals", + "value": 1 + } + ] } ] }, - "id": 4 + "gridPos": { + "h": 12, + "w": 24, + "x": 0, + "y": 11 + }, + "id": 3 }, { - "type": "logs", - "title": "New heavy hitters (Loki)", + "type": "bargauge", + "title": "Sketch (CMS + heap): top-10 routes, latest window", "datasource": { - "type": "loki", - "uid": "asap-loki" - }, - "gridPos": { - "h": 8, - "w": 12, - "x": 12, - "y": 20 + "type": "prometheus", + "uid": "asap-prometheus" }, "targets": [ { "refId": "A", - "expr": "{service_name=\"asap-detector\"} | event_kind=\"novelty\"" + "expr": "sort_desc(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\"})))))", + "legendFormat": "#{{rank}} {{key}}", + "instant": true } ], "options": { - "showTime": true, - "wrapLogMessage": true, - "sortOrder": "Descending" + "orientation": "horizontal", + "displayMode": "basic", + "reduceOptions": { + "calcs": [ + "lastNotNull" + ], + "fields": "", + "values": false + } }, - "id": 5 + "fieldConfig": { + "defaults": { + "decimals": 0, + "color": { + "mode": "fixed", + "fixedColor": "#5794f2" + } + }, + "overrides": [] + }, + "gridPos": { + "h": 10, + "w": 8, + "x": 0, + "y": 23 + }, + "id": 4 }, { - "type": "timeseries", - "title": "Heavy hitters (routes with >= 1% of the window's points): sketch vs exact, per window", + "type": "bargauge", + "title": "Exact: top-10 routes, latest window", "datasource": { "type": "prometheus", "uid": "asap-prometheus" }, - "gridPos": { - "h": 8, - "w": 12, - "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" + "expr": "sort_desc(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\"})))))", + "legendFormat": "#{{rank}} {{key}}", + "instant": true } ], + "options": { + "orientation": "horizontal", + "displayMode": "basic", + "reduceOptions": { + "calcs": [ + "lastNotNull" + ], + "fields": "", + "values": false + } + }, "fieldConfig": { "defaults": { - "min": 0, - "max": 1.05, - "decimals": 2, - "custom": { - "spanNulls": true, - "drawStyle": "line", - "pointSize": 6, - "showPoints": "always" + "decimals": 0, + "color": { + "mode": "fixed", + "fixedColor": "#ff9830" } }, "overrides": [] }, - "options": { - "legend": { - "displayMode": "list", - "placement": "bottom" - } + "gridPos": { + "h": 10, + "w": 8, + "x": 8, + "y": 23 }, - "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%." + "id": 5 }, { - "type": "timeseries", - "title": "Average relative count error of the heavy hitters found (ARE)", + "type": "bargauge", + "title": "Store and query (LogQL over loki-raw): top-10 routes, latest window", "datasource": { "type": "prometheus", "uid": "asap-prometheus" }, - "gridPos": { - "h": 8, - "w": 12, - "x": 12, - "y": 28 + "targets": [ + { + "refId": "A", + "expr": "sort_desc(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\"})))))", + "legendFormat": "#{{rank}} {{key}}", + "instant": true + } + ], + "options": { + "orientation": "horizontal", + "displayMode": "basic", + "reduceOptions": { + "calcs": [ + "lastNotNull" + ], + "fields": "", + "values": false + } + }, + "fieldConfig": { + "defaults": { + "decimals": 0, + "color": { + "mode": "fixed", + "fixedColor": "#8e8e8e" + } + }, + "overrides": [] + }, + "gridPos": { + "h": 10, + "w": 8, + "x": 16, + "y": 23 + }, + "id": 6 + }, + { + "type": "logs", + "title": "New heavy hitters (Loki events from both detectors)", + "datasource": { + "type": "loki", + "uid": "asap-loki" + }, + "targets": [ + { + "refId": "A", + "expr": "{service_name=\"asap-detector\"} | event_kind=\"novelty\"" + } + ], + "options": { + "showTime": true, + "wrapLogMessage": true, + "sortOrder": "Descending" + }, + "gridPos": { + "h": 8, + "w": 24, + "x": 0, + "y": 33 + }, + "id": 7 + }, + { + "type": "row", + "title": "Accuracy against exact", + "collapsed": false, + "panels": [], + "gridPos": { + "h": 1, + "w": 24, + "x": 0, + "y": 41 + }, + "id": 8 + }, + { + "type": "timeseries", + "title": "Heavy hitters (routes with >= 1% of the window's points): precision, recall, F1", + "datasource": { + "type": "prometheus", + "uid": "asap-prometheus" }, + "description": "Truth: exact counts >= 1% of the window's points. Store and query loses accuracy only when points never reached the log store.", "targets": [ { "refId": "A", - "expr": "asap_hh_are", - "legendFormat": "ARE" + "expr": "asap_hh_precision{method=\"sketch\"}", + "legendFormat": "sketch precision" }, { "refId": "B", - "expr": "asap_hh_truth", - "legendFormat": "heavy hitters (exact)" + "expr": "asap_hh_recall{method=\"sketch\"}", + "legendFormat": "sketch recall" }, { "refId": "C", - "expr": "asap_hh_reported", - "legendFormat": "reported (sketch)" + "expr": "asap_hh_f1{method=\"sketch\"}", + "legendFormat": "sketch F1" + }, + { + "refId": "D", + "expr": "asap_hh_precision{method=\"raw\"}", + "legendFormat": "store and query precision" + }, + { + "refId": "E", + "expr": "asap_hh_recall{method=\"raw\"}", + "legendFormat": "store and query recall" + }, + { + "refId": "F", + "expr": "asap_hh_f1{method=\"raw\"}", + "legendFormat": "store and query F1" } ], "fieldConfig": { "defaults": { "custom": { - "spanNulls": true, - "showPoints": "always", - "pointSize": 6 - } - }, - "overrides": [ - { - "matcher": { - "id": "byName", - "options": "ARE" - }, - "properties": [ - { - "id": "unit", - "value": "percentunit" - }, - { - "id": "decimals", - "value": 2 - } - ] + "spanNulls": true }, - { - "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 - ] - } - } - ] - } - ] + "unit": "percentunit" + }, + "overrides": [] }, "options": { "legend": { "displayMode": "list", "placement": "bottom" + }, + "tooltip": { + "mode": "multi", + "sort": "none" } }, - "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." + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 42 + }, + "id": 9 }, { "type": "timeseries", - "id": 12, - "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": "Average relative count error of the heavy hitters found (ARE)", "datasource": { "type": "prometheus", "uid": "asap-prometheus" }, - "gridPos": { - "h": 8, - "w": 24, - "x": 0, - "y": 36 - }, + "description": "", "targets": [ { "refId": "A", - "expr": "sum by (method) (asap_points{method=~\"sketch|exact\",attribute_key=\"http.route\"}) / 60", - "legendFormat": "{{method}}" + "expr": "asap_hh_are{method=\"sketch\"}", + "legendFormat": "sketch" }, { "refId": "B", - "expr": "sum by (method) (asap_points{method=\"raw\"}) / 60", - "legendFormat": "{{method}}" - }, - { - "refId": "C", - "expr": "vector(40000)", - "legendFormat": "offered" + "expr": "asap_hh_are{method=\"raw\"}", + "legendFormat": "store and query" } ], "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" + "options": "sketch" }, "properties": [ { - "id": "custom.lineStyle", + "id": "color", "value": { - "fill": "dash", - "dash": [ - 6, - 4 - ] + "mode": "fixed", + "fixedColor": "#5794f2" } - }, + } + ] + }, + { + "matcher": { + "id": "byName", + "options": "store and query" + }, + "properties": [ { "id": "color", "value": { "mode": "fixed", "fixedColor": "#8e8e8e" } - }, - { - "id": "custom.showPoints", - "value": "never" } ] } @@ -463,27 +577,51 @@ "legend": { "displayMode": "list", "placement": "bottom" + }, + "tooltip": { + "mode": "multi", + "sort": "none" } - } + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 42 + }, + "id": 10 + }, + { + "type": "row", + "title": "Traffic over time: volume and shape", + "collapsed": false, + "panels": [], + "gridPos": { + "h": 1, + "w": 24, + "x": 0, + "y": 50 + }, + "id": 11 }, { "type": "timeseries", - "title": "Window completion delay: seconds from a window's end until its result is complete", + "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": 0, - "y": 44 - }, + "description": "Both sources together. The generators skip batches a pipeline cannot take, so a pipeline that falls behind delivers less.", "targets": [ { "refId": "A", - "expr": "avg by (method) (asap_answer_delay)", + "expr": "sum by (method) (asap_points{method=~\"sketch|exact\",attribute_key=\"http.route\"}) / 60", "legendFormat": "{{method}}" + }, + { + "refId": "B", + "expr": "sum(asap_points{method=\"raw\"}) / 60", + "legendFormat": "store and query" } ], "fieldConfig": { @@ -491,7 +629,8 @@ "custom": { "spanNulls": true }, - "unit": "s" + "unit": "short", + "min": 0 }, "overrides": [ { @@ -527,7 +666,7 @@ { "matcher": { "id": "byName", - "options": "raw" + "options": "store and query" }, "properties": [ { @@ -541,27 +680,103 @@ } ] }, - "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": 0, + "y": 51 + }, + "id": 12 }, { "type": "timeseries", - "title": "CPU cores per pipeline (all stages but the generators)", + "title": "Data shape: distinct routes per window (exact)", "datasource": { "type": "prometheus", "uid": "asap-prometheus" }, + "description": "", + "targets": [ + { + "refId": "A", + "expr": "asap_distinct{method=\"exact\",attribute_key=\"http.route\"}", + "legendFormat": "distinct routes" + } + ], + "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": 12, - "y": 44 + "y": 51 + }, + "id": 13 + }, + { + "type": "row", + "title": "Freshness: from a sample's generation to the queryable answer", + "collapsed": false, + "panels": [], + "gridPos": { + "h": 1, + "w": 24, + "x": 0, + "y": 59 + }, + "id": 14 + }, + { + "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(cpu_time_total{job=\"collectors\",stage!~\".*generator.*\",otel_scope_pipeline_id=\"main\"}[1m]), \"pipeline\", \"sketch\", \"stage\", \".*\"), \"pipeline\", \"$1\", \"stage\", \"(exact|raw)-.*\"))", - "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": { @@ -569,7 +784,8 @@ "custom": { "spanNulls": true }, - "unit": "short" + "unit": "s", + "min": 0 }, "overrides": [ { @@ -605,7 +821,7 @@ { "matcher": { "id": "byName", - "options": "raw" + "options": "store and query" }, "properties": [ { @@ -619,26 +835,563 @@ } ] }, - "id": 7 + "options": { + "legend": { + "displayMode": "list", + "placement": "bottom" + }, + "tooltip": { + "mode": "multi", + "sort": "none" + } + }, + "gridPos": { + "h": 7, + "w": 24, + "x": 0, + "y": 60 + }, + "id": 15 }, { "type": "timeseries", - "title": "Bytes received per pipeline (all stages but the generators), log scale", + "title": "Sketch: latency breakdown per window", "datasource": { "type": "prometheus", "uid": "asap-prometheus" }, - "gridPos": { - "h": 8, - "w": 12, - "x": 0, - "y": 52 - }, + "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(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": "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": 8, + "x": 0, + "y": 67 + }, + "id": 16 + }, + { + "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": "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, + "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": 8, + "x": 8, + "y": 67 + }, + "id": 17 + }, + { + "type": "timeseries", + "title": "Store and query: latency breakdown per window", + "datasource": { + "type": "prometheus", + "uid": "asap-prometheus" + }, + "description": "", + "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 top-k 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": 8, + "x": 16, + "y": 67 + }, + "id": 18 + }, + { + "type": "row", + "title": "Resources per approach (generators and the hh-accuracy observer excluded)", + "collapsed": false, + "panels": [], + "gridPos": { + "h": 1, + "w": 24, + "x": 0, + "y": 75 + }, + "id": 19 + }, + { + "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 + }, + "unit": "short", + "min": 0 + }, + "overrides": [] + }, + "options": { + "legend": { + "displayMode": "table", + "placement": "bottom", + "calcs": [ + "lastNotNull" + ] + }, + "tooltip": { + "mode": "multi", + "sort": "none" + } + }, + "gridPos": { + "h": 12, + "w": 8, + "x": 0, + "y": 76 + }, + "id": 20 + }, + { + "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": 76 + }, + "id": 21 + }, + { + "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": 76 + }, + "id": 22 + }, + { + "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": 88 + }, + "id": 23 + }, + { + "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": 88 + }, + "id": 24 + }, + { + "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": 88 + }, + "id": 25 + }, + { + "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": { @@ -652,74 +1405,183 @@ }, "unit": "bps" }, - "overrides": [ - { - "matcher": { - "id": "byName", - "options": "sketch" - }, - "properties": [ - { - "id": "color", - "value": { - "mode": "fixed", - "fixedColor": "#5794f2" - } - } - ] + "overrides": [] + }, + "options": { + "legend": { + "displayMode": "table", + "placement": "bottom", + "calcs": [ + "lastNotNull" + ] + }, + "tooltip": { + "mode": "multi", + "sort": "none" + } + }, + "gridPos": { + "h": 12, + "w": 8, + "x": 0, + "y": 100 + }, + "id": 26 + }, + { + "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 + } }, - { - "matcher": { - "id": "byName", - "options": "exact" - }, - "properties": [ - { - "id": "color", - "value": { - "mode": "fixed", - "fixedColor": "#ff9830" - } - } - ] + "unit": "bps" + }, + "overrides": [] + }, + "options": { + "legend": { + "displayMode": "table", + "placement": "bottom", + "calcs": [ + "lastNotNull" + ] + }, + "tooltip": { + "mode": "multi", + "sort": "none" + } + }, + "gridPos": { + "h": 12, + "w": 8, + "x": 8, + "y": 100 + }, + "id": 27 + }, + { + "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 + } }, - { - "matcher": { - "id": "byName", - "options": "raw" - }, - "properties": [ - { - "id": "color", - "value": { - "mode": "fixed", - "fixedColor": "#8e8e8e" - } - } - ] - } - ] + "unit": "bps" + }, + "overrides": [] }, - "id": 8 + "options": { + "legend": { + "displayMode": "table", + "placement": "bottom", + "calcs": [ + "lastNotNull" + ] + }, + "tooltip": { + "mode": "multi", + "sort": "none" + } + }, + "gridPos": { + "h": 12, + "w": 8, + "x": 16, + "y": 100 + }, + "id": 28 }, { "type": "timeseries", - "title": "Resident memory per pipeline (all stages but the generators)", + "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": 12, - "y": 52 + "x": 0, + "y": 112 + }, + "id": 29 + }, + { + "type": "timeseries", + "title": "Shared by all three: Prometheus resident memory", + "datasource": { + "type": "prometheus", + "uid": "asap-prometheus" }, + "description": "", "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": "max(process_resident_memory_bytes{job=\"prometheus\"})", + "legendFormat": "Prometheus" } ], "fieldConfig": { @@ -727,57 +1589,28 @@ "custom": { "spanNulls": true }, - "unit": "bytes" + "unit": "bytes", + "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": [] }, - "id": 9 + "options": { + "legend": { + "displayMode": "list", + "placement": "bottom" + }, + "tooltip": { + "mode": "multi", + "sort": "none" + } + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 112 + }, + "id": 30 } ] } diff --git a/deploy/telemetry-functions/latency/merge.yaml b/deploy/telemetry-functions/latency/merge.yaml index cb392f2..7a27162 100644 --- a/deploy/telemetry-functions/latency/merge.yaml +++ b/deploy/telemetry-functions/latency/merge.yaml @@ -1,4 +1,4 @@ -# Stage 2: merge every source's sketch for a window, 5 s after it ends. +# Stage 2: merge both sources' sketches for a window as soon as both sent it (5 s timeout). version: otel_dataflow/v1 engine: http_admin: @@ -20,7 +20,7 @@ groups: http: {listening_addr: "0.0.0.0:4318", wait_for_result: false} merge: type: urn:asap:processor:window_merge - config: {grace: 5s} + config: {sources: 2, grace: 5s} to_detector: type: exporter:otlp_http config: diff --git a/deploy/telemetry-functions/loki-raw.yaml b/deploy/telemetry-functions/loki-raw.yaml index 04ff483..efc3c82 100644 --- a/deploy/telemetry-functions/loki-raw.yaml +++ b/deploy/telemetry-functions/loki-raw.yaml @@ -4,6 +4,8 @@ auth_enabled: false server: http_listen_port: 3100 grpc_server_max_recv_msg_size: 1073741824 + http_server_read_timeout: 5m + http_server_write_timeout: 5m common: instance_addr: 127.0.0.1 path_prefix: /loki @@ -31,5 +33,17 @@ 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 + 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). + retention_period: 2h +compactor: + working_directory: /loki/compactor + compaction_interval: 10m + retention_enabled: true + retention_delete_delay: 10m + delete_request_store: filesystem analytics: reporting_enabled: false diff --git a/deploy/telemetry-functions/prometheus.yml b/deploy/telemetry-functions/prometheus.yml index 122b3b2..1b1157b 100644 --- a/deploy/telemetry-functions/prometheus.yml +++ b/deploy/telemetry-functions/prometheus.yml @@ -11,11 +11,19 @@ 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, - hh-accuracy:8085] + raw-sink-a:8085, raw-sink-b:8085, hh-accuracy:8085] relabel_configs: - source_labels: [__address__] regex: (.+):8085 target_label: stage # Answers (sketch, exact) and the raw pipeline's latency arrive through the # OTLP receiver. + # The backends and the query job: their own CPU time and memory, so each + # approach's whole-system cost includes them. The query job's gauges carry + # the window end as their timestamp. + - job_name: prometheus + static_configs: [{targets: ["localhost:9090"], labels: {stage: prometheus}}] + - job_name: loki-raw + static_configs: [{targets: ["loki-raw:3100"], labels: {stage: loki-raw}}] + - job_name: query-job + static_configs: [{targets: ["query-job:9100"], labels: {stage: query-job}}] diff --git a/deploy/telemetry-functions/query_job.py b/deploy/telemetry-functions/query_job.py new file mode 100644 index 0000000..d453c44 --- /dev/null +++ b/deploy/telemetry-functions/query_job.py @@ -0,0 +1,208 @@ +#!/usr/bin/env python3 +"""Query job: the answer step of the store-and-query baseline, and the backend +query latency of every approach. + +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 + 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). + +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 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. + +Configuration (environment): USE_CASE, WINDOW_SECONDS (60), K (20), PHI (0.01), +RAW_SOURCES (2), PROMETHEUS, LOKI_RAW, PORT (9100). +""" +import json +import os +import resource +import threading +import time +import urllib.parse +import urllib.request +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +PROMETHEUS = os.environ.get("PROMETHEUS", "http://prometheus:9090") +LOKI_RAW = os.environ.get("LOKI_RAW", "http://loki-raw:3100") +USE_CASE = os.environ.get("USE_CASE", "topk") +WINDOW_S = int(os.environ.get("WINDOW_SECONDS", "60")) +K = int(os.environ.get("K", "20")) +PHI = float(os.environ.get("PHI", "0.01")) +RAW_SOURCES = int(os.environ.get("RAW_SOURCES", "2")) +PORT = int(os.environ.get("PORT", "9100")) +ATTRIBUTE = "http.route" +# Loki stores OTLP attributes as structured metadata with dots replaced. +LOKI_LABEL = ATTRIBUTE.replace(".", "_") +LOKI_STREAM = '{service_name="checkout"}' +TOPK_SELECTOR = f'attribute_key="{ATTRIBUTE}",bound="estimate"' + +# Latest sample per series: (name, sorted labels) -> (value, timestamp ms). +samples = {} +lock = threading.Lock() + + +def get(base, path, params, timeout=300): + url = base + path + "?" + urllib.parse.urlencode(params) + with urllib.request.urlopen(url, timeout=timeout) as response: + return json.load(response)["data"]["result"] + + +def prom(expr, at=None): + params = {"query": expr} + if at is not None: + params["time"] = f"{at:.3f}" + return get(PROMETHEUS, "/api/v1/query", params, timeout=30) + + +def window_ends(expr): + """Window ends (s) with a sample of `expr` in the last 15 minutes, with + the number of series that have one.""" + ends = {} + for row in prom(f"{expr}[15m]"): + for t, _ in row["values"]: + ends[int(float(t))] = ends.get(int(float(t)), 0) + 1 + return ends + + +def at_window(expr, end): + """Samples of `expr` stamped exactly at window end `end` (s): an instant + query would also return series last seen in earlier windows.""" + rows = prom(f"{expr}[{WINDOW_S}s]", at=end) + return [(r["metric"], float(v)) for r in rows for t, v in r["values"] if int(float(t)) == end] + + +def put(name, labels, value, end_s): + with lock: + samples[(name, tuple(sorted(labels.items())))] = (value, end_s * 1000) + + +def heavy_hitter_accuracy(points, exact, reported_list): + """As `answers::heavy_hitter_accuracy`; None when the exact list may hide + more heavy hitters.""" + threshold = PHI * points + if len(exact) >= K and min((c for _, c in exact), default=float("inf")) >= threshold: + return None + truth = {k: c for k, c in exact if c >= threshold} + reported = [(k, c) for k, c in reported_list if c >= threshold] + errors = [abs(c - truth[k]) / truth[k] for k, c in reported if k in truth] + tp = len(errors) + precision = tp / len(reported) if reported else 1.0 + recall = tp / len(truth) if truth else 1.0 + f1 = 0.0 if precision + recall == 0 else 2 * precision * recall / (precision + recall) + return {"precision": precision, "recall": recall, "f1": f1, "truth": len(truth), + "reported": len(reported), "are": sum(errors) / tp if tp else None} + + +def time_prometheus_answer(method, end): + t0 = time.monotonic() + at_window(f'asap_topk{{method="{method}",{TOPK_SELECTOR}}}', end) + put("asap_query_seconds", {"method": method, "backend": "prometheus"}, time.monotonic() - t0, end) + + +def flag_incomplete(end): + unacknowledged = sum(v for _, v in at_window('asap_raw_unacknowledged{method="raw"}', 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.""" + 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)) + if not exact_rows or not exact_points: + return False + 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)}) + 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) + 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}"} + if acc is None: + put("asap_hh_incomplete", labels, 1.0, end) + else: + for name, value in acc.items(): + if value is not None: + put(f"asap_hh_{name}", labels, float(value), end) + return True + + +def loop(): + done = {"sketch": set(), "exact": set(), "raw": set()} + started = time.time() + while True: + try: + for method in ("sketch", "exact"): + for end in sorted(window_ends(f'asap_answer_delay{{method="{method}"}}')): + 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": + 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. + if (end not in done["raw"] and end in done["exact"] + and end > started - 2 * WINDOW_S and raw_answer(end)): + done["raw"].add(end) + except Exception as e: # keep serving; the next round retries + print(f"query_job: {e}", flush=True) + time.sleep(1) + + +def escape(value): + return value.replace("\\", "\\\\").replace('"', '\\"').replace("\n", "\\n") + + +def own_metrics(): + usage = resource.getrusage(resource.RUSAGE_SELF) + with open("/proc/self/statm") as f: + rss = int(f.read().split()[1]) * os.sysconf("SC_PAGE_SIZE") + return [f"process_cpu_seconds_total {usage.ru_utime + usage.ru_stime}", + f"process_resident_memory_bytes {rss}"] + + +class Metrics(BaseHTTPRequestHandler): + def do_GET(self): + lines = own_metrics() + 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}") + body = ("\n".join(lines) + "\n").encode() + self.send_response(200) + self.send_header("Content-Type", "text/plain; version=0.0.4") + self.end_headers() + self.wfile.write(body) + + def log_message(self, *args): + pass + + +if __name__ == "__main__": + threading.Thread(target=loop, daemon=True).start() + ThreadingHTTPServer(("0.0.0.0", PORT), Metrics).serve_forever() diff --git a/deploy/telemetry-functions/topk/generator-a.yaml b/deploy/telemetry-functions/topk/generator-a.yaml index 4e50035..d669799 100644 --- a/deploy/telemetry-functions/topk/generator-a.yaml +++ b/deploy/telemetry-functions/topk/generator-a.yaml @@ -1,4 +1,7 @@ -# Source a: 100,000 routes (Zipf 1.1); a new route takes 15% of traffic after 10 minutes. +# Source a: routes drawn Zipf(1.1). Traffic grows over 30 minutes so the +# dashboard shows how each approach scales: the rate doubles every 5 minutes +# (5k to 80k points/s), a new route takes 15% of traffic at minute 10, and the +# number of routes grows from 100,000 to 1,000,000 at minute 15. version: otel_dataflow/v1 engine: http_admin: @@ -17,7 +20,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: @@ -25,6 +28,11 @@ groups: value: {lognormal: {median: 50.0, sigma: 0.6}} changes: - {after_seconds: 600, attribute: http.route, inject: /new/hot-route, share: 0.15} + - {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: 900, attribute: http.route, cardinality: 1000000} export: type: exporter:otlp_http config: diff --git a/deploy/telemetry-functions/topk/generator-b.yaml b/deploy/telemetry-functions/topk/generator-b.yaml index 4804d35..529c64c 100644 --- a/deploy/telemetry-functions/topk/generator-b.yaml +++ b/deploy/telemetry-functions/topk/generator-b.yaml @@ -1,4 +1,7 @@ -# Source b: 100,000 routes (Zipf 1.1); a new route takes 15% of traffic after 10 minutes. +# Source b: routes drawn Zipf(1.1). Traffic grows over 30 minutes so the +# dashboard shows how each approach scales: the rate doubles every 5 minutes +# (5k to 80k points/s), a new route takes 15% of traffic at minute 10, and the +# number of routes grows from 100,000 to 1,000,000 at minute 15. version: otel_dataflow/v1 engine: http_admin: @@ -17,7 +20,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: @@ -25,6 +28,11 @@ groups: value: {lognormal: {median: 50.0, sigma: 0.6}} changes: - {after_seconds: 600, attribute: http.route, inject: /new/hot-route, share: 0.15} + - {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: 900, attribute: http.route, cardinality: 1000000} export: type: exporter:otlp_http config: diff --git a/deploy/telemetry-functions/topk/merge.yaml b/deploy/telemetry-functions/topk/merge.yaml index cb392f2..7a27162 100644 --- a/deploy/telemetry-functions/topk/merge.yaml +++ b/deploy/telemetry-functions/topk/merge.yaml @@ -1,4 +1,4 @@ -# Stage 2: merge every source's sketch for a window, 5 s after it ends. +# Stage 2: merge both sources' sketches for a window as soon as both sent it (5 s timeout). version: otel_dataflow/v1 engine: http_admin: @@ -20,7 +20,7 @@ groups: http: {listening_addr: "0.0.0.0:4318", wait_for_result: false} merge: type: urn:asap:processor:window_merge - config: {grace: 5s} + config: {sources: 2, grace: 5s} to_detector: type: exporter:otlp_http config: