diff --git a/asap-precompute-rs/src/answers.rs b/asap-precompute-rs/src/answers.rs index 7d70f13..91f7fcf 100644 --- a/asap-precompute-rs/src/answers.rs +++ b/asap-precompute-rs/src/answers.rs @@ -257,10 +257,151 @@ pub fn exact_answers( out } +/// Thresholds for [`DriftTracker`]. +#[derive(Clone, Debug, serde::Deserialize)] +#[serde(deny_unknown_fields)] +pub struct DriftRules { + /// Windows merged into the baseline. + #[serde(default = "default_drift_baseline")] + pub baseline_windows: usize, + /// Kolmogorov-Smirnov distance above which a series has drifted. + #[serde(default = "default_ks")] + pub ks_threshold: f64, + /// Points needed in the current window before testing. + #[serde(default = "default_drift_points")] + pub min_points: u64, +} + +fn default_drift_baseline() -> usize { + 5 +} +fn default_ks() -> f64 { + 0.1 +} +fn default_drift_points() -> u64 { + 1000 +} + +impl Default for DriftRules { + fn default() -> Self { + Self { + baseline_windows: default_drift_baseline(), + ks_threshold: default_ks(), + min_points: default_drift_points(), + } + } +} + +/// Largest gap between two CDFs over the points `xs` (Kolmogorov-Smirnov). +pub fn ks_distance(current: impl Fn(f64) -> f64, baseline: impl Fn(f64) -> f64, xs: &[f64]) -> f64 { + xs.iter() + .map(|&x| (current(x) - baseline(x)).abs()) + .fold(0.0, f64::max) +} + +/// Turns each window's KS distance into a drift gauge and an event when a +/// series starts or stops drifting. Sketch and exact use the same rules. +#[derive(Default)] +pub struct DriftTracker { + /// Thresholds. + pub rules: DriftRules, + drifting: HashMap, +} + +impl DriftTracker { + /// Tracker with the given thresholds. + pub fn new(rules: DriftRules) -> Self { + Self { + rules, + drifting: HashMap::new(), + } + } + + /// Records one window's distance from its baseline. + pub fn observe( + &mut self, + template: &SketchEnvelope, + method: &str, + ks: f64, + baseline_p50: f64, + current_p50: f64, + ) -> (Vec, Option) { + let drift = ks > self.rules.ks_threshold; + let out = vec![ + gauge(template, "asap.drift.ks", method, &[], ks), + gauge( + template, + "asap.drift", + method, + &[], + if drift { 1.0 } else { 0.0 }, + ), + ]; + let key = format!("{}|{:?}", template.metric_name, template.labels); + let previous = self.drifting.insert(key, drift); + let series = template + .labels + .iter() + .map(|kv| format!("{}={}", kv.key, kv.value)) + .collect::>() + .join(","); + let event = (previous.unwrap_or(false) != drift).then(|| Event { + window_end_ms: template.window_end_ms, + method: method.into(), + kind: "drift".into(), + attributes: template + .labels + .iter() + .cloned() + .chain([ + KeyValue::new("drift", drift.to_string()), + KeyValue::new("ks", format!("{ks:.3}")), + ]) + .collect(), + body: if drift { + format!( + "{} {{{series}}} drifted from its baseline: KS distance {ks:.3}, median {baseline_p50:.1} -> {current_p50:.1}", + template.metric_name + ) + } else { + format!( + "{} {{{series}}} is back within its baseline: KS distance {ks:.3}", + template.metric_name + ) + }, + }); + (out, event) + } +} + #[cfg(test)] mod classifier_tests { use super::*; + #[test] + fn ks_and_drift_events() { + let uniform = |x: f64| x.clamp(0.0, 1.0); + let shifted = |x: f64| (x - 0.3).clamp(0.0, 1.0); + let xs: Vec = (0..=100).map(|i| i as f64 / 100.0).collect(); + let ks = ks_distance(shifted, uniform, &xs); + assert!((ks - 0.3).abs() < 1e-9, "{ks}"); + let mut t = DriftTracker::default(); + let s = SketchEnvelope { + metric_name: "m".into(), + ..Default::default() + }; + assert!(t.observe(&s, "sketch", 0.02, 1.0, 1.0).1.is_none()); + let (out, event) = t.observe(&s, "sketch", 0.3, 1.0, 2.0); + assert_eq!(out[1].value, 1.0); + assert!(event.unwrap().body.contains("drifted")); + assert!(t + .observe(&s, "sketch", 0.03, 1.0, 1.0) + .1 + .unwrap() + .body + .contains("back")); + } + fn attr(key: &str) -> SketchEnvelope { SketchEnvelope { metric_name: "m".into(), diff --git a/asap-precompute-rs/src/otap/answer_nodes.rs b/asap-precompute-rs/src/otap/answer_nodes.rs index dde6f56..741e969 100644 --- a/asap-precompute-rs/src/otap/answer_nodes.rs +++ b/asap-precompute-rs/src/otap/answer_nodes.rs @@ -43,9 +43,16 @@ use otel_arrow_dfe_pdata::proto::OtlpProtoMessage; use otel_arrow_dfe_pdata::OtapPayload; use super::codec::decode_pdata_to_observations; -use crate::answers::{sketch_answers, CardinalityClassifier, CardinalityRules, Event}; +use crate::answers::{ + ks_distance, sketch_answers, CardinalityClassifier, CardinalityRules, DriftRules, DriftTracker, + Event, +}; use crate::envelope::SketchEnvelope; +use crate::envelope::SketchType; use crate::observation::{KeyValue, Observation}; +use crate::precompute::QuantileSketch; +use crate::sketches::KLLWrapper; +use std::collections::{HashMap, VecDeque}; /// Registered URN of the sketch detector. pub const SKETCH_DETECTOR_URN: &str = "urn:asap:processor:sketch_detector"; @@ -76,6 +83,9 @@ pub struct DetectorConfig { /// Thresholds for classifying per-attribute cardinality. #[serde(default)] pub cardinality: CardinalityRules, + /// Thresholds for distribution drift of quantile series. + #[serde(default)] + pub drift: DriftRules, } pub(super) fn now_ms() -> u64 { @@ -218,11 +228,6 @@ pub(super) async fn emit( Ok(()) } -/// Identity of a series across windows: metric and labels. -pub(super) fn series_key(env: &SketchEnvelope) -> String { - format!("{}|{:?}", env.metric_name, env.labels) -} - pub(super) async fn observations( pdata: OtapPdata, effects: &mut local::EffectHandler, @@ -241,6 +246,54 @@ pub(super) async fn observations( struct SketchDetector { config: DetectorConfig, cardinality: CardinalityClassifier, + drift: DriftTracker, + /// Recent KLL windows per series, the drift baseline. + baselines: HashMap>, +} + +pub(super) fn series_key(env: &SketchEnvelope) -> String { + format!("{}|{:?}", env.metric_name, env.labels) +} + +/// Percentile grid of both distributions: where their CDFs are compared. +fn grid(a: &dyn QuantileSketch, b: &dyn QuantileSketch) -> Vec { + (1..100) + .flat_map(|i| [a.quantile(i as f64 / 100.0), b.quantile(i as f64 / 100.0)]) + .collect() +} + +impl SketchDetector { + /// KS distance of this KLL window from the merge of the previous ones. + fn drift(&mut self, env: &SketchEnvelope) -> (Vec, Option) { + let Ok(current) = KLLWrapper::from_snapshot(&env.payload) else { + return (Vec::new(), None); + }; + let window = self.baselines.entry(series_key(env)).or_default(); + let mut result = (Vec::new(), None); + if window.len() >= 3 && env.count >= self.drift.rules.min_points { + let mut baseline = KLLWrapper::new(current.inner().k() as i32, None); + for w in window.iter() { + let _ = crate::precompute::Sketch::merge(&mut baseline, w); + } + fn cdf(s: &KLLWrapper) -> impl Fn(f64) -> f64 + '_ { + let n = s.inner().count().max(1) as f64; + move |x| s.inner().rank(x) as f64 / n + } + let ks = ks_distance(cdf(¤t), cdf(&baseline), &grid(¤t, &baseline)); + result = self.drift.observe( + env, + "sketch", + ks, + baseline.quantile(0.5), + current.quantile(0.5), + ); + } + window.push_back(current); + if window.len() > self.drift.rules.baseline_windows { + window.pop_front(); + } + result + } } /// The `bound="estimate"` value of `metric` among `answers`. @@ -275,6 +328,11 @@ impl local::Processor for SketchDetector { }; match sketch_answers(&env, &self.config.quantiles, self.config.k) { Ok(answers) => { + if env.sketch_type == SketchType::KLLSketch { + let (gauges, event) = self.drift(&env); + out.extend(gauges); + events.extend(event); + } if let Some(d) = estimate_of(&answers, "asap.distinct") { let (gauges, event) = self.cardinality.observe(&env, "sketch", d, env.count); @@ -310,6 +368,8 @@ static SKETCH_DETECTOR: ProcessorFactory = ProcessorFactory { Ok(ProcessorWrapper::local( SketchDetector { cardinality: CardinalityClassifier::new(config.cardinality.clone()), + drift: DriftTracker::new(config.drift.clone()), + baselines: HashMap::new(), config, }, node, diff --git a/asap-precompute-rs/src/otap/baseline_nodes.rs b/asap-precompute-rs/src/otap/baseline_nodes.rs index 625a399..3b8525a 100644 --- a/asap-precompute-rs/src/otap/baseline_nodes.rs +++ b/asap-precompute-rs/src/otap/baseline_nodes.rs @@ -32,7 +32,7 @@ //! outputs: [logs, latency] //! config: {window: 60s, grace: 5s} //! ``` -use std::collections::{BTreeMap, HashMap}; +use std::collections::{BTreeMap, HashMap, VecDeque}; use std::sync::Arc; use std::time::Duration; @@ -70,7 +70,8 @@ use super::answer_nodes::{ }; use super::codec::OtapSketchEncoder; use crate::answers::{ - exact_answers, gauge, CardinalityClassifier, CardinalityRules, Event, ExactWindow, + exact_answers, gauge, ks_distance, CardinalityClassifier, CardinalityRules, DriftRules, + DriftTracker, Event, ExactWindow, }; use crate::envelope::{SketchEnvelope, SketchType}; use crate::matchers::ATTRIBUTE_KEY_LABEL; @@ -131,6 +132,9 @@ pub struct ExactDetectorConfig { /// Thresholds for classifying per-attribute cardinality. #[serde(default)] pub cardinality: CardinalityRules, + /// Thresholds for distribution drift of quantile series. + #[serde(default)] + pub drift: DriftRules, } /// `raw_sink` configuration. @@ -358,6 +362,9 @@ impl local::Processor for ExactMerge { struct ExactDetector { config: ExactDetectorConfig, cardinality: CardinalityClassifier, + drift: DriftTracker, + /// Recent windows' sorted values per quantile series. + baselines: HashMap>>, } impl ExactDetector { @@ -384,8 +391,47 @@ impl ExactDetector { &self.config.quantiles, self.config.k, )); + if !w.values.is_empty() { + // exact_answers left the values sorted. + let (gauges, event) = self.drift(&template, std::mem::take(&mut w.values)); + out.extend(gauges); + events.extend(event); + } (out, events) } + + /// Exact KS distance of a window's sorted values from the previous windows. + fn drift( + &mut self, + template: &SketchEnvelope, + sorted: Vec, + ) -> (Vec, Option) { + let window = self.baselines.entry(series_key(template)).or_default(); + let mut result = (Vec::new(), None); + if window.len() >= 3 && sorted.len() as u64 >= self.drift.rules.min_points { + let mut baseline: Vec = window.iter().flatten().copied().collect(); + baseline.sort_unstable_by(f64::total_cmp); + let cdf = |v: &[f64], x: f64| v.partition_point(|&y| y <= x) as f64 / v.len() as f64; + let at = |v: &[f64], q: f64| v[((v.len() - 1) as f64 * q).round() as usize]; + let xs: Vec = (1..100) + .flat_map(|i| { + [ + at(&sorted, i as f64 / 100.0), + at(&baseline, i as f64 / 100.0), + ] + }) + .collect(); + let ks = ks_distance(|x| cdf(&sorted, x), |x| cdf(&baseline, x), &xs); + result = + self.drift + .observe(template, "exact", ks, at(&baseline, 0.5), at(&sorted, 0.5)); + } + window.push_back(sorted); + if window.len() > self.drift.rules.baseline_windows { + window.pop_front(); + } + result + } } #[async_trait(?Send)] @@ -714,6 +760,8 @@ static EXACT_DETECTOR: ProcessorFactory = ProcessorFactory { Ok(ProcessorWrapper::local( ExactDetector { cardinality: CardinalityClassifier::new(config.cardinality.clone()), + drift: DriftTracker::new(config.drift.clone()), + baselines: HashMap::new(), config, }, node, @@ -846,6 +894,8 @@ mod tests { serde_json::from_value(serde_json::json!({"k": 2})).unwrap(); let mut detector = ExactDetector { cardinality: CardinalityClassifier::new(config.cardinality.clone()), + drift: DriftTracker::new(config.drift.clone()), + baselines: HashMap::new(), config, }; let out: Vec<_> = merged diff --git a/deploy/telemetry-functions/drift/branch.yaml b/deploy/telemetry-functions/drift/branch.yaml new file mode 100644 index 0000000..8d2d148 --- /dev/null +++ b/deploy/telemetry-functions/drift/branch.yaml @@ -0,0 +1,40 @@ +# Sketch pipeline, stage 1 at each source: one KLL per route per 1-minute +# window goes to merge. +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} + sketch: + type: urn:asap:processor:asap_sketches + config: + sketch_type: kll + window_size: 60s + output_metric_name: http.server.request.duration + aggregate_by: [http.route] + omit_resource_attrs: true + sketch_params: {k: 200} + to_merge: + type: exporter:otlp_http + config: + endpoint: http://merge:4318 + client_pool_size: 1 + max_in_flight: 1 + http: {timeout: 30s, compression: none} + connections: + - {from: receiver, to: sketch} + - {from: sketch, to: to_merge} diff --git a/deploy/telemetry-functions/drift/detector.yaml b/deploy/telemetry-functions/drift/detector.yaml new file mode 100644 index 0000000..0ebd8fe --- /dev/null +++ b/deploy/telemetry-functions/drift/detector.yaml @@ -0,0 +1,38 @@ +# Quantiles with bounds, and each route's KS distance from its last 5 windows, +# to Prometheus; drift start/end as events to Loki. +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} + detector: + type: urn:asap:processor:sketch_detector + config: + quantiles: [0.5, 0.99] + drift: {baseline_windows: 5, ks_threshold: 0.1} + to_prometheus: + type: exporter:otlp_http + config: + endpoint: http://prometheus:9090 + metrics_endpoint: http://prometheus:9090/api/v1/otlp/v1/metrics + logs_endpoint: http://loki:3100/otlp/v1/logs + client_pool_size: 1 + max_in_flight: 1 + http: {timeout: 30s, compression: none} + connections: + - {from: receiver, to: detector} + - {from: detector, to: to_prometheus} diff --git a/deploy/telemetry-functions/drift/exact-branch.yaml b/deploy/telemetry-functions/drift/exact-branch.yaml new file mode 100644 index 0000000..a061d73 --- /dev/null +++ b/deploy/telemetry-functions/drift/exact-branch.yaml @@ -0,0 +1,34 @@ +# Exact pipeline, stage 1 at each source: every value per route per 1-minute window +# goes to exact-merge. +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} + branch: + type: urn:asap:processor:exact_branch + config: {window: 60s, group_by: [http.route]} + to_merge: + type: exporter:otlp_http + config: + endpoint: http://exact-merge:4318 + client_pool_size: 1 + max_in_flight: 1 + http: {timeout: 30s, compression: none} + connections: + - {from: receiver, to: branch} + - {from: branch, to: to_merge} diff --git a/deploy/telemetry-functions/drift/exact-detector.yaml b/deploy/telemetry-functions/drift/exact-detector.yaml new file mode 100644 index 0000000..f66ebcd --- /dev/null +++ b/deploy/telemetry-functions/drift/exact-detector.yaml @@ -0,0 +1,38 @@ +# Exact pipeline, end: exact answers and events of each merged window, the +# ground truth (method="exact"). +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, max_request_body_size: 1GiB} + detector: + type: urn:asap:processor:exact_detector + config: + quantiles: [0.5, 0.99] + drift: {baseline_windows: 5, ks_threshold: 0.1} + to_prometheus: + type: exporter:otlp_http + config: + endpoint: http://prometheus:9090 + metrics_endpoint: http://prometheus:9090/api/v1/otlp/v1/metrics + logs_endpoint: http://loki:3100/otlp/v1/logs + client_pool_size: 1 + max_in_flight: 1 + http: {timeout: 30s, compression: none} + connections: + - {from: receiver, to: detector} + - {from: detector, to: to_prometheus} diff --git a/deploy/telemetry-functions/drift/generator-a.yaml b/deploy/telemetry-functions/drift/generator-a.yaml new file mode 100644 index 0000000..c6895c1 --- /dev/null +++ b/deploy/telemetry-functions/drift/generator-a.yaml @@ -0,0 +1,37 @@ +# Source a: latency over 20 routes; route-3's median triples after 10 minutes. +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: + generator: + type: urn:asap:receiver:scenario_generator + config: + seed: 1 + signals_per_second: 20000 + metric_name: http.server.request.duration + resource_attributes: {service.name: checkout, source: "a"} + attributes: + - {key: http.route, cardinality: 20, zipf: 1.1} + value: {lognormal: {median: 50.0, sigma: 0.6}} + changes: + - {after_seconds: 600, attribute: http.route, equals: http.route-3, + value: {lognormal: {median: 150.0, sigma: 0.6}}} + export: + type: exporter:otlp_http + config: + endpoint: http://downstream:4318 + client_pool_size: 1 + max_in_flight: 1 + http: {timeout: 30s, compression: none} + connections: + - {from: generator, to: export} diff --git a/deploy/telemetry-functions/drift/generator-b.yaml b/deploy/telemetry-functions/drift/generator-b.yaml new file mode 100644 index 0000000..0d50d01 --- /dev/null +++ b/deploy/telemetry-functions/drift/generator-b.yaml @@ -0,0 +1,37 @@ +# Source b: latency over 20 routes; route-3's median triples after 10 minutes. +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: + generator: + type: urn:asap:receiver:scenario_generator + config: + seed: 2 + signals_per_second: 20000 + metric_name: http.server.request.duration + resource_attributes: {service.name: checkout, source: "b"} + attributes: + - {key: http.route, cardinality: 20, zipf: 1.1} + value: {lognormal: {median: 50.0, sigma: 0.6}} + changes: + - {after_seconds: 600, attribute: http.route, equals: http.route-3, + value: {lognormal: {median: 150.0, sigma: 0.6}}} + export: + type: exporter:otlp_http + config: + endpoint: http://downstream:4318 + client_pool_size: 1 + max_in_flight: 1 + http: {timeout: 30s, compression: none} + connections: + - {from: generator, to: export} diff --git a/deploy/telemetry-functions/drift/merge.yaml b/deploy/telemetry-functions/drift/merge.yaml new file mode 100644 index 0000000..cb392f2 --- /dev/null +++ b/deploy/telemetry-functions/drift/merge.yaml @@ -0,0 +1,33 @@ +# Stage 2: merge every source's sketch for a window, 5 s after it ends. +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} + merge: + type: urn:asap:processor:window_merge + config: {grace: 5s} + to_detector: + type: exporter:otlp_http + config: + endpoint: http://detector:4318 + client_pool_size: 1 + max_in_flight: 1 + http: {timeout: 30s, compression: none} + connections: + - {from: receiver, to: merge} + - {from: merge, to: to_detector} diff --git a/deploy/telemetry-functions/grafana/dashboards/drift.json b/deploy/telemetry-functions/grafana/dashboards/drift.json new file mode 100644 index 0000000..d561c44 --- /dev/null +++ b/deploy/telemetry-functions/grafana/dashboards/drift.json @@ -0,0 +1,710 @@ +{ + "uid": "asap-drift", + "title": "Latency drift: sketch vs exact (real collector containers)", + "schemaVersion": 39, + "version": 1, + "refresh": "10s", + "time": { + "from": "now-30m", + "to": "now" + }, + "templating": { + "list": [ + { + "name": "route", + "type": "query", + "datasource": { + "type": "prometheus", + "uid": "asap-prometheus" + }, + "query": { + "query": "label_values(asap_quantile, http_route)", + "refId": "route" + }, + "refresh": 2, + "current": { + "text": "http.route-3", + "value": "http.route-3" + } + } + ] + }, + "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.", + "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: 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
" + }, + "id": 1 + }, + { + "type": "timeseries", + "title": "KS distance from the baseline (last 5 windows), every route: sketch solid, exact dashed", + "datasource": { + "type": "prometheus", + "uid": "asap-prometheus" + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 10 + }, + "targets": [ + { + "refId": "A", + "expr": "asap_drift_ks{method=\"sketch\"}", + "legendFormat": "{{http_route}} sketch" + }, + { + "refId": "B", + "expr": "asap_drift_ks{method=\"exact\"}", + "legendFormat": "{{http_route}} exact" + } + ], + "fieldConfig": { + "defaults": { + "custom": { + "spanNulls": true + }, + "min": 0 + }, + "overrides": [ + { + "matcher": { + "id": "byRegexp", + "options": ".* exact" + }, + "properties": [ + { + "id": "custom.lineStyle", + "value": { + "fill": "dash", + "dash": [ + 6, + 4 + ] + } + } + ] + } + ] + }, + "id": 2 + }, + { + "type": "logs", + "title": "Drift events (Loki)", + "datasource": { + "type": "loki", + "uid": "asap-loki" + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 10 + }, + "targets": [ + { + "refId": "A", + "expr": "{service_name=\"asap-detector\"} | event_kind=\"drift\"" + } + ], + "options": { + "showTime": true, + "wrapLogMessage": true, + "sortOrder": "Descending" + }, + "id": 3 + }, + { + "id": 4, + "type": "timeseries", + "title": "p99 latency for $route: sketch with bound vs exact", + "datasource": { + "type": "prometheus", + "uid": "asap-prometheus" + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 18 + }, + "targets": [ + { + "refId": "A", + "expr": "asap_quantile{quantile=\"0.99\",http_route=\"$route\",method=\"sketch\",bound=\"estimate\"}", + "legendFormat": "sketch" + }, + { + "refId": "B", + "expr": "asap_quantile{quantile=\"0.99\",http_route=\"$route\",method=\"sketch\",bound=\"lower\"}", + "legendFormat": "sketch lower" + }, + { + "refId": "C", + "expr": "asap_quantile{quantile=\"0.99\",http_route=\"$route\",method=\"sketch\",bound=\"upper\"}", + "legendFormat": "sketch upper" + }, + { + "refId": "D", + "expr": "asap_quantile{quantile=\"0.99\",http_route=\"$route\",method=\"exact\"}", + "legendFormat": "exact" + } + ], + "fieldConfig": { + "defaults": { + "custom": { + "spanNulls": true + }, + "unit": "ms" + }, + "overrides": [ + { + "matcher": { + "id": "byName", + "options": "sketch upper" + }, + "properties": [ + { + "id": "custom.fillBelowTo", + "value": "sketch lower" + }, + { + "id": "custom.lineWidth", + "value": 0 + }, + { + "id": "custom.fillOpacity", + "value": 20 + }, + { + "id": "color", + "value": { + "mode": "fixed", + "fixedColor": "blue" + } + } + ] + }, + { + "matcher": { + "id": "byName", + "options": "sketch lower" + }, + "properties": [ + { + "id": "custom.lineWidth", + "value": 0 + }, + { + "id": "color", + "value": { + "mode": "fixed", + "fixedColor": "blue" + } + } + ] + }, + { + "matcher": { + "id": "byName", + "options": "sketch" + }, + "properties": [ + { + "id": "color", + "value": { + "mode": "fixed", + "fixedColor": "blue" + } + } + ] + }, + { + "matcher": { + "id": "byName", + "options": "exact" + }, + "properties": [ + { + "id": "color", + "value": { + "mode": "fixed", + "fixedColor": "orange" + } + }, + { + "id": "custom.lineStyle", + "value": { + "fill": "dash", + "dash": [ + 6, + 4 + ] + } + } + ] + } + ] + } + }, + { + "id": 5, + "type": "timeseries", + "title": "p50 latency for $route: sketch with bound vs exact", + "datasource": { + "type": "prometheus", + "uid": "asap-prometheus" + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 18 + }, + "targets": [ + { + "refId": "A", + "expr": "asap_quantile{quantile=\"0.5\",http_route=\"$route\",method=\"sketch\",bound=\"estimate\"}", + "legendFormat": "sketch" + }, + { + "refId": "B", + "expr": "asap_quantile{quantile=\"0.5\",http_route=\"$route\",method=\"sketch\",bound=\"lower\"}", + "legendFormat": "sketch lower" + }, + { + "refId": "C", + "expr": "asap_quantile{quantile=\"0.5\",http_route=\"$route\",method=\"sketch\",bound=\"upper\"}", + "legendFormat": "sketch upper" + }, + { + "refId": "D", + "expr": "asap_quantile{quantile=\"0.5\",http_route=\"$route\",method=\"exact\"}", + "legendFormat": "exact" + } + ], + "fieldConfig": { + "defaults": { + "custom": { + "spanNulls": true + }, + "unit": "ms" + }, + "overrides": [ + { + "matcher": { + "id": "byName", + "options": "sketch upper" + }, + "properties": [ + { + "id": "custom.fillBelowTo", + "value": "sketch lower" + }, + { + "id": "custom.lineWidth", + "value": 0 + }, + { + "id": "custom.fillOpacity", + "value": 20 + }, + { + "id": "color", + "value": { + "mode": "fixed", + "fixedColor": "blue" + } + } + ] + }, + { + "matcher": { + "id": "byName", + "options": "sketch lower" + }, + "properties": [ + { + "id": "custom.lineWidth", + "value": 0 + }, + { + "id": "color", + "value": { + "mode": "fixed", + "fixedColor": "blue" + } + } + ] + }, + { + "matcher": { + "id": "byName", + "options": "sketch" + }, + "properties": [ + { + "id": "color", + "value": { + "mode": "fixed", + "fixedColor": "blue" + } + } + ] + }, + { + "matcher": { + "id": "byName", + "options": "exact" + }, + "properties": [ + { + "id": "color", + "value": { + "mode": "fixed", + "fixedColor": "orange" + } + }, + { + "id": "custom.lineStyle", + "value": { + "fill": "dash", + "dash": [ + 6, + 4 + ] + } + } + ] + } + ] + } + }, + { + "type": "timeseries", + "title": "Window completion delay: seconds from a window's end until its result is complete", + "datasource": { + "type": "prometheus", + "uid": "asap-prometheus" + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 26 + }, + "targets": [ + { + "refId": "A", + "expr": "avg by (method) (asap_answer_delay)", + "legendFormat": "{{method}}" + } + ], + "fieldConfig": { + "defaults": { + "custom": { + "spanNulls": true + }, + "unit": "s" + }, + "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" + } + } + ] + } + ] + }, + "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 + }, + { + "type": "timeseries", + "title": "CPU cores per pipeline (all stages but the generators)", + "datasource": { + "type": "prometheus", + "uid": "asap-prometheus" + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 26 + }, + "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}}" + } + ], + "fieldConfig": { + "defaults": { + "custom": { + "spanNulls": true + }, + "unit": "short" + }, + "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" + } + } + ] + } + ] + }, + "id": 7 + }, + { + "type": "timeseries", + "title": "Bytes received per pipeline (all stages but the generators), log scale", + "datasource": { + "type": "prometheus", + "uid": "asap-prometheus" + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 34 + }, + "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}}" + } + ], + "fieldConfig": { + "defaults": { + "custom": { + "spanNulls": true, + "scaleDistribution": { + "type": "log", + "log": 10 + } + }, + "unit": "bps" + }, + "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" + } + } + ] + } + ] + }, + "id": 8 + }, + { + "type": "timeseries", + "title": "Resident memory per pipeline (all stages but the generators)", + "datasource": { + "type": "prometheus", + "uid": "asap-prometheus" + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 34 + }, + "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}}" + } + ], + "fieldConfig": { + "defaults": { + "custom": { + "spanNulls": true + }, + "unit": "bytes" + }, + "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" + } + } + ] + } + ] + }, + "id": 9 + } + ] +} diff --git a/deploy/telemetry-functions/run.sh b/deploy/telemetry-functions/run.sh index 1323597..59bd0d4 100755 --- a/deploy/telemetry-functions/run.sh +++ b/deploy/telemetry-functions/run.sh @@ -1,6 +1,6 @@ #!/usr/bin/env bash # Build asap-collector, package it, and start every stage plus the observers. -# USE_CASE=latency (default) or cardinality picks the scenario. +# USE_CASE=latency (default), cardinality, drift or topk picks the scenario. # DOCKER_COMMAND="sudo -n docker" when Docker needs sudo. Stop with: # $DOCKER_COMMAND compose -f deploy/telemetry-functions/compose.yaml down set -euo pipefail diff --git a/docs/images/drift-grafana.png b/docs/images/drift-grafana.png new file mode 100644 index 0000000..d403f14 Binary files /dev/null and b/docs/images/drift-grafana.png differ