diff --git a/asap-precompute-rs/src/answers.rs b/asap-precompute-rs/src/answers.rs index 91f7fcf..e8a6300 100644 --- a/asap-precompute-rs/src/answers.rs +++ b/asap-precompute-rs/src/answers.rs @@ -20,7 +20,7 @@ //! q ± error. //! * Count-Min: counts never underestimate; with width w and N points the //! overestimate is at most e/w * N with probability 1 - e^-depth. -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use crate::envelope::{SketchEnvelope, SketchType}; use crate::observation::KeyValue; @@ -374,10 +374,121 @@ impl DriftTracker { } } +/// The `(key, count)` of every `asap.topk` estimate in `answers`, by rank. +pub fn topk_of(answers: &[SketchEnvelope]) -> Vec<(String, f64)> { + let label = |e: &SketchEnvelope, k: &str| { + e.labels + .iter() + .find(|kv| kv.key == k) + .map(|kv| kv.value.clone()) + }; + let mut top: Vec<(u32, String, f64)> = answers + .iter() + .filter(|e| { + e.metric_name == "asap.topk" && label(e, "bound").as_deref() == Some("estimate") + }) + .filter_map(|e| Some((label(e, "rank")?.parse().ok()?, label(e, "key")?, e.value))) + .collect(); + top.sort_by_key(|t| t.0); + top.into_iter().map(|(_, k, v)| (k, v)).collect() +} + +/// Flags heavy hitters that were not among the top keys of the previous +/// windows. Sketch and exact use the same rule. +#[derive(Default)] +pub struct NoveltyTracker { + /// Windows whose top keys form the baseline. + pub baseline_windows: usize, + history: HashMap>>, +} + +impl NoveltyTracker { + /// Tracker remembering `baseline_windows` windows. + pub fn new(baseline_windows: usize) -> Self { + Self { + baseline_windows, + history: HashMap::new(), + } + } + + /// New heavy hitters in this window's `top` (key, count), as gauges and events. + pub fn observe( + &mut self, + template: &SketchEnvelope, + method: &str, + top: &[(String, f64)], + ) -> (Vec, Vec) { + let key = format!("{}|{:?}", template.metric_name, template.labels); + let window = self.history.entry(key).or_default(); + let (mut out, mut events) = (Vec::new(), Vec::new()); + if window.len() >= 3 { + let attribute = template + .labels + .iter() + .find(|kv| kv.key == crate::matchers::ATTRIBUTE_KEY_LABEL) + .map(|kv| kv.value.clone()) + .unwrap_or_default(); + for (rank, (k, count)) in top.iter().enumerate() { + if window.iter().any(|seen| seen.contains(k)) { + continue; + } + out.push(gauge( + template, + "asap.topk.new", + method, + &[("key", k.clone())], + *count, + )); + events.push(Event { + window_end_ms: template.window_end_ms, + method: method.into(), + kind: "novelty".into(), + attributes: vec![ + KeyValue::new("attribute.key", attribute.clone()), + KeyValue::new("key", k.clone()), + KeyValue::new("rank", (rank + 1).to_string()), + KeyValue::new("count", format!("{count:.0}")), + ], + body: format!( + "new heavy hitter {attribute}={k}: rank {} with {count:.0} points, absent from the top {} of the previous {} windows", + rank + 1, + top.len(), + window.len() + ), + }); + } + } + window.push_back(top.iter().map(|(k, _)| k.clone()).collect()); + if window.len() > self.baseline_windows { + window.pop_front(); + } + (out, events) + } +} + #[cfg(test)] mod classifier_tests { use super::*; + #[test] + fn novelty_after_three_windows() { + let mut t = NoveltyTracker::new(5); + let s = SketchEnvelope { + metric_name: "m".into(), + ..Default::default() + }; + let usual = vec![("a".to_string(), 10.0), ("b".to_string(), 5.0)]; + for _ in 0..3 { + assert!(t.observe(&s, "sketch", &usual).1.is_empty()); + } + let now = vec![("new".to_string(), 20.0), ("a".to_string(), 10.0)]; + let (out, events) = t.observe(&s, "sketch", &now); + assert_eq!(out.len(), 1); + assert!(events[0].body.contains("new heavy hitter =new: rank 1")); + // The next window, the key is part of the baseline. + assert!(t.observe(&s, "sketch", &now).1.is_empty()); + } + #[test] fn ks_and_drift_events() { let uniform = |x: f64| x.clamp(0.0, 1.0); diff --git a/asap-precompute-rs/src/otap/answer_nodes.rs b/asap-precompute-rs/src/otap/answer_nodes.rs index 741e969..5056c6b 100644 --- a/asap-precompute-rs/src/otap/answer_nodes.rs +++ b/asap-precompute-rs/src/otap/answer_nodes.rs @@ -44,8 +44,8 @@ use otel_arrow_dfe_pdata::OtapPayload; use super::codec::decode_pdata_to_observations; use crate::answers::{ - ks_distance, sketch_answers, CardinalityClassifier, CardinalityRules, DriftRules, DriftTracker, - Event, + ks_distance, sketch_answers, topk_of, CardinalityClassifier, CardinalityRules, DriftRules, + DriftTracker, Event, NoveltyTracker, }; use crate::envelope::SketchEnvelope; use crate::envelope::SketchType; @@ -86,6 +86,13 @@ pub struct DetectorConfig { /// Thresholds for distribution drift of quantile series. #[serde(default)] pub drift: DriftRules, + /// Windows whose top keys form the novelty baseline. + #[serde(default = "default_novelty_windows")] + pub novelty_windows: usize, +} + +pub(super) fn default_novelty_windows() -> usize { + 10 } pub(super) fn now_ms() -> u64 { @@ -246,6 +253,7 @@ pub(super) async fn observations( struct SketchDetector { config: DetectorConfig, cardinality: CardinalityClassifier, + novelty: NoveltyTracker, drift: DriftTracker, /// Recent KLL windows per series, the drift baseline. baselines: HashMap>, @@ -333,6 +341,12 @@ impl local::Processor for SketchDetector { out.extend(gauges); events.extend(event); } + let top = topk_of(&answers); + if !top.is_empty() { + let (gauges, new) = self.novelty.observe(&env, "sketch", &top); + out.extend(gauges); + events.extend(new); + } if let Some(d) = estimate_of(&answers, "asap.distinct") { let (gauges, event) = self.cardinality.observe(&env, "sketch", d, env.count); @@ -368,6 +382,7 @@ static SKETCH_DETECTOR: ProcessorFactory = ProcessorFactory { Ok(ProcessorWrapper::local( SketchDetector { cardinality: CardinalityClassifier::new(config.cardinality.clone()), + novelty: NoveltyTracker::new(config.novelty_windows), drift: DriftTracker::new(config.drift.clone()), baselines: HashMap::new(), config, diff --git a/asap-precompute-rs/src/otap/baseline_nodes.rs b/asap-precompute-rs/src/otap/baseline_nodes.rs index 3b8525a..3013f92 100644 --- a/asap-precompute-rs/src/otap/baseline_nodes.rs +++ b/asap-precompute-rs/src/otap/baseline_nodes.rs @@ -65,13 +65,13 @@ use otel_arrow_dfe_pdata::OtapPayload; use serde::Deserialize; use super::answer_nodes::{ - default_grace, default_k, default_quantiles, default_window, emit, gauges_pdata, invalid, - now_ms, observations, series_key, + default_grace, default_k, default_novelty_windows, default_quantiles, default_window, emit, + gauges_pdata, invalid, now_ms, observations, series_key, }; use super::codec::OtapSketchEncoder; use crate::answers::{ - exact_answers, gauge, ks_distance, CardinalityClassifier, CardinalityRules, DriftRules, - DriftTracker, Event, ExactWindow, + exact_answers, gauge, ks_distance, topk_of, CardinalityClassifier, CardinalityRules, + DriftRules, DriftTracker, Event, ExactWindow, NoveltyTracker, }; use crate::envelope::{SketchEnvelope, SketchType}; use crate::matchers::ATTRIBUTE_KEY_LABEL; @@ -135,6 +135,9 @@ pub struct ExactDetectorConfig { /// Thresholds for distribution drift of quantile series. #[serde(default)] pub drift: DriftRules, + /// Windows whose top keys form the novelty baseline. + #[serde(default = "default_novelty_windows")] + pub novelty_windows: usize, } /// `raw_sink` configuration. @@ -362,6 +365,7 @@ impl local::Processor for ExactMerge { struct ExactDetector { config: ExactDetectorConfig, cardinality: CardinalityClassifier, + novelty: NoveltyTracker, drift: DriftTracker, /// Recent windows' sorted values per quantile series. baselines: HashMap>>, @@ -385,12 +389,14 @@ impl ExactDetector { out.extend(gauges); events.extend(event); } - out.extend(exact_answers( - &template, - &mut w, - &self.config.quantiles, - self.config.k, - )); + let answers = exact_answers(&template, &mut w, &self.config.quantiles, self.config.k); + let top = topk_of(&answers); + if !top.is_empty() { + let (gauges, new) = self.novelty.observe(&template, "exact", &top); + out.extend(gauges); + events.extend(new); + } + out.extend(answers); if !w.values.is_empty() { // exact_answers left the values sorted. let (gauges, event) = self.drift(&template, std::mem::take(&mut w.values)); @@ -760,6 +766,7 @@ static EXACT_DETECTOR: ProcessorFactory = ProcessorFactory { Ok(ProcessorWrapper::local( ExactDetector { cardinality: CardinalityClassifier::new(config.cardinality.clone()), + novelty: NoveltyTracker::new(config.novelty_windows), drift: DriftTracker::new(config.drift.clone()), baselines: HashMap::new(), config, @@ -894,6 +901,7 @@ mod tests { serde_json::from_value(serde_json::json!({"k": 2})).unwrap(); let mut detector = ExactDetector { cardinality: CardinalityClassifier::new(config.cardinality.clone()), + novelty: NoveltyTracker::new(config.novelty_windows), drift: DriftTracker::new(config.drift.clone()), baselines: HashMap::new(), config, diff --git a/deploy/telemetry-functions/grafana/dashboards/topk.json b/deploy/telemetry-functions/grafana/dashboards/topk.json new file mode 100644 index 0000000..0bef626 --- /dev/null +++ b/deploy/telemetry-functions/grafana/dashboards/topk.json @@ -0,0 +1,556 @@ +{ + "uid": "asap-topk", + "title": "Heavy hitters and new keys: sketch vs exact (real collector containers)", + "schemaVersion": 39, + "version": 1, + "refresh": "10s", + "time": { + "from": "now-30m", + "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.", + "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-10 heap per 1-min window
→
merge
Stage 2: merges both sources' sketches per window
→
detector
top-10 with bounds; new heavy hitters
→
Prometheus + Loki
method="sketch"
exact-generator-a/b
same points
→
exact-branch-a/b
Stage 1 per source: route → count map per 1-min window
→
exact-merge
Stage 2: adds both sources' exact windows
→
exact-detector
exact top-10; new heavy hitters
→
Prometheus + Loki
method="exact" (ground truth)
raw-generator-a/b
same points
→
raw-branch-a/b
Stage 1 per source: forwards every raw point
→
raw-merge
Stage 2: forwards both sources' raw points
→
raw-sink
writes every raw point; measures when the window's last point is stored
→
loki-raw
one log record per raw point
" + }, + "id": 1 + }, + { + "type": "bargauge", + "title": "Top-10 routes, latest window: CMS + heap (lower bound to estimate)", + "datasource": { + "type": "prometheus", + "uid": "asap-prometheus" + }, + "gridPos": { + "h": 10, + "w": 12, + "x": 0, + "y": 10 + }, + "targets": [ + { + "refId": "A", + "expr": "sort_desc(asap_topk{method=\"sketch\",bound=\"estimate\"})", + "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", + "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\"})", + "legendFormat": "#{{rank}} {{key}}", + "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": [ + { + "refId": "A", + "expr": "asap_topk{key=\"/new/hot-route\",method=\"sketch\",bound=\"estimate\"}", + "legendFormat": "sketch" + }, + { + "refId": "B", + "expr": "asap_topk{key=\"/new/hot-route\",method=\"sketch\",bound=\"lower\"}", + "legendFormat": "sketch lower" + }, + { + "refId": "C", + "expr": "asap_topk{key=\"/new/hot-route\",method=\"exact\"}", + "legendFormat": "exact" + } + ], + "fieldConfig": { + "defaults": { + "custom": { + "spanNulls": true + } + }, + "overrides": [ + { + "matcher": { + "id": "byName", + "options": "sketch" + }, + "properties": [ + { + "id": "custom.fillBelowTo", + "value": "sketch lower" + }, + { + "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": "exact" + }, + "properties": [ + { + "id": "color", + "value": { + "mode": "fixed", + "fixedColor": "orange" + } + }, + { + "id": "custom.lineStyle", + "value": { + "fill": "dash", + "dash": [ + 6, + 4 + ] + } + } + ] + } + ] + }, + "id": 4 + }, + { + "type": "logs", + "title": "New heavy hitters (Loki)", + "datasource": { + "type": "loki", + "uid": "asap-loki" + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 20 + }, + "targets": [ + { + "refId": "A", + "expr": "{service_name=\"asap-detector\"} | event_kind=\"novelty\"" + } + ], + "options": { + "showTime": true, + "wrapLogMessage": true, + "sortOrder": "Descending" + }, + "id": 5 + }, + { + "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": 28 + }, + "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": 28 + }, + "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": 36 + }, + "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": 36 + }, + "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/topk/branch.yaml b/deploy/telemetry-functions/topk/branch.yaml new file mode 100644 index 0000000..c7301e0 --- /dev/null +++ b/deploy/telemetry-functions/topk/branch.yaml @@ -0,0 +1,40 @@ +# Sketch pipeline, stage 1 at each source: one Count-Min sketch (4 x 512) + +# top-10 heap of routes per 1-minute window goes to merge. +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: countminsketch_heap + window_size: 60s + output_metric_name: http.server.request.duration + per_attribute: [http.route] + omit_resource_attrs: true + sketch_params: {rows: 4, columns: 512, k: 10} + 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/topk/detector.yaml b/deploy/telemetry-functions/topk/detector.yaml new file mode 100644 index 0000000..6465f03 --- /dev/null +++ b/deploy/telemetry-functions/topk/detector.yaml @@ -0,0 +1,36 @@ +# Top-10 routes with Count-Min bounds to Prometheus; routes new to the top 10 +# (vs the last 10 windows) 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: {k: 10, novelty_windows: 10} + 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/topk/exact-branch.yaml b/deploy/telemetry-functions/topk/exact-branch.yaml new file mode 100644 index 0000000..33579bb --- /dev/null +++ b/deploy/telemetry-functions/topk/exact-branch.yaml @@ -0,0 +1,34 @@ +# Exact pipeline, stage 1 at each source: a route -> count map 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, per_attribute: [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/topk/exact-detector.yaml b/deploy/telemetry-functions/topk/exact-detector.yaml new file mode 100644 index 0000000..61e560a --- /dev/null +++ b/deploy/telemetry-functions/topk/exact-detector.yaml @@ -0,0 +1,36 @@ +# 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: {k: 10, novelty_windows: 10} + 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/topk/generator-a.yaml b/deploy/telemetry-functions/topk/generator-a.yaml new file mode 100644 index 0000000..4e50035 --- /dev/null +++ b/deploy/telemetry-functions/topk/generator-a.yaml @@ -0,0 +1,36 @@ +# Source a: 100,000 routes (Zipf 1.1); a new route takes 15% of traffic 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: 100000, zipf: 1.1} + value: {lognormal: {median: 50.0, sigma: 0.6}} + changes: + - {after_seconds: 600, attribute: http.route, inject: /new/hot-route, share: 0.15} + 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/topk/generator-b.yaml b/deploy/telemetry-functions/topk/generator-b.yaml new file mode 100644 index 0000000..4804d35 --- /dev/null +++ b/deploy/telemetry-functions/topk/generator-b.yaml @@ -0,0 +1,36 @@ +# Source b: 100,000 routes (Zipf 1.1); a new route takes 15% of traffic 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: 100000, zipf: 1.1} + value: {lognormal: {median: 50.0, sigma: 0.6}} + changes: + - {after_seconds: 600, attribute: http.route, inject: /new/hot-route, share: 0.15} + 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/topk/merge.yaml b/deploy/telemetry-functions/topk/merge.yaml new file mode 100644 index 0000000..cb392f2 --- /dev/null +++ b/deploy/telemetry-functions/topk/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/docs/images/topk-novelty-grafana.png b/docs/images/topk-novelty-grafana.png new file mode 100644 index 0000000..8d8d134 Binary files /dev/null and b/docs/images/topk-novelty-grafana.png differ