diff --git a/asap-precompute-rs/src/answers.rs b/asap-precompute-rs/src/answers.rs index a423908..7d70f13 100644 --- a/asap-precompute-rs/src/answers.rs +++ b/asap-precompute-rs/src/answers.rs @@ -257,6 +257,47 @@ pub fn exact_answers( out } +#[cfg(test)] +mod classifier_tests { + use super::*; + + fn attr(key: &str) -> SketchEnvelope { + SketchEnvelope { + metric_name: "m".into(), + labels: vec![KeyValue::new("attribute.key", key)], + ..Default::default() + } + } + + fn state(out: &[SketchEnvelope]) -> String { + out.iter() + .find(|e| e.metric_name == "asap.cardinality.state" && e.value == 1.0) + .and_then(|e| e.labels.iter().find(|kv| kv.key == "state")) + .map(|kv| kv.value.clone()) + .unwrap() + } + + #[test] + fn identifier_growing_and_stable() { + let mut c = CardinalityClassifier::default(); + let (out, event) = c.observe(&attr("user.id"), "sketch", 980.0, 1000); + assert_eq!(state(&out), "identifier-like"); + assert!(event.unwrap().body.contains("identifier-like")); + for _ in 0..4 { + let (out, _) = c.observe(&attr("pod"), "sketch", 50.0, 1000); + assert_eq!(state(&out), "stable"); + } + let (out, event) = c.observe(&attr("pod"), "sketch", 70.0, 1000); + assert_eq!(state(&out), "growing"); + assert_eq!(event.unwrap().attributes[1].value, "growing"); + // No event while the state does not change. + let (_, event) = c.observe(&attr("pod"), "sketch", 90.0, 1000); + assert!(event.is_none()); + // Too few points: no answer. + assert!(c.observe(&attr("rare"), "sketch", 5.0, 5).0.is_empty()); + } +} + #[cfg(test)] mod tests { use super::*; @@ -355,3 +396,176 @@ mod tests { .any(|kv| kv.key == "method" && kv.value == "sketch"))); } } + +/// A detection result worth telling a person about, emitted as a log record. +#[derive(Clone, Debug, PartialEq)] +pub struct Event { + /// Window the event refers to. + pub window_end_ms: u64, + /// `sketch` or `exact`. + pub method: String, + /// Event kind, e.g. `cardinality`. + pub kind: String, + /// Series labels plus event fields. + pub attributes: Vec, + /// One readable sentence. + pub body: String, +} + +/// Cardinality class of an attribute key in one window. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum CardinalityState { + /// Distinct values close to the baseline. + Stable, + /// Distinct values well above the baseline mean. + Growing, + /// Almost every point carries a new value, like an identifier. + IdentifierLike, +} + +impl CardinalityState { + /// Name used in labels and events. + pub fn name(self) -> &'static str { + match self { + Self::Stable => "stable", + Self::Growing => "growing", + Self::IdentifierLike => "identifier-like", + } + } +} + +/// Thresholds for [`CardinalityClassifier`]. +#[derive(Clone, Debug, serde::Deserialize)] +#[serde(deny_unknown_fields)] +pub struct CardinalityRules { + /// Windows kept as the baseline. + #[serde(default = "default_baseline")] + pub baseline_windows: usize, + /// distinct / points at or above which a key is identifier-like. + #[serde(default = "default_identifier_ratio")] + pub identifier_ratio: f64, + /// Fraction above the baseline mean at which a key is growing. + #[serde(default = "default_growth")] + pub growth: f64, + /// Points needed before classifying. + #[serde(default = "default_min_points")] + pub min_points: u64, +} + +fn default_baseline() -> usize { + 10 +} +fn default_identifier_ratio() -> f64 { + 0.9 +} +fn default_growth() -> f64 { + 0.2 +} +fn default_min_points() -> u64 { + 100 +} + +impl Default for CardinalityRules { + fn default() -> Self { + Self { + baseline_windows: default_baseline(), + identifier_ratio: default_identifier_ratio(), + growth: default_growth(), + min_points: default_min_points(), + } + } +} + +/// Classifies each attribute key against its own recent windows. The same +/// rules run on sketch and exact answers so their labels can be compared. +#[derive(Default)] +pub struct CardinalityClassifier { + rules: CardinalityRules, + history: HashMap, Option)>, +} + +impl CardinalityClassifier { + /// Classifier with the given thresholds. + pub fn new(rules: CardinalityRules) -> Self { + Self { + rules, + history: HashMap::new(), + } + } + + /// Classifies one window of `series` (its distinct count and points), and + /// returns its state gauge plus an event when the state changed. + pub fn observe( + &mut self, + template: &SketchEnvelope, + method: &str, + distinct: f64, + points: u64, + ) -> (Vec, Option) { + if points < self.rules.min_points { + return (Vec::new(), None); + } + let key = format!("{}|{:?}", template.metric_name, template.labels); + let (window, previous) = self.history.entry(key).or_default(); + let baseline = + (window.len() >= 3).then(|| window.iter().sum::() / window.len() as f64); + let ratio = distinct / points as f64; + let state = if ratio >= self.rules.identifier_ratio { + CardinalityState::IdentifierLike + } else if baseline.is_some_and(|b| distinct > b * (1.0 + self.rules.growth)) { + CardinalityState::Growing + } else { + CardinalityState::Stable + }; + window.push_back(distinct); + if window.len() > self.rules.baseline_windows { + window.pop_front(); + } + let mut out = vec![gauge(template, "asap.distinct.ratio", method, &[], ratio)]; + out.extend( + [ + CardinalityState::Stable, + CardinalityState::Growing, + CardinalityState::IdentifierLike, + ] + .into_iter() + .map(|s| { + let v = if s == state { 1.0 } else { 0.0 }; + gauge( + template, + "asap.cardinality.state", + method, + &[("state", s.name().into())], + v, + ) + }), + ); + let changed = *previous != Some(state); + *previous = Some(state); + let attribute = template + .labels + .iter() + .find(|kv| kv.key == crate::matchers::ATTRIBUTE_KEY_LABEL) + .map(|kv| kv.value.clone()) + .unwrap_or_default(); + let event = changed.then(|| Event { + window_end_ms: template.window_end_ms, + method: method.into(), + kind: "cardinality".into(), + attributes: vec![ + KeyValue::new("attribute.key", attribute.clone()), + KeyValue::new("state", state.name()), + KeyValue::new("distinct", format!("{distinct:.0}")), + KeyValue::new("points", points.to_string()), + ], + body: format!( + "attribute {attribute} of {} is {}: {distinct:.0} distinct values in {points} points ({:.0}%){}", + template.metric_name, + state.name(), + ratio * 100.0, + baseline.map(|b| format!(", baseline {b:.0}")).unwrap_or_default() + ), + }); + (out, event) + } +} diff --git a/asap-precompute-rs/src/matchers.rs b/asap-precompute-rs/src/matchers.rs index 4f2c99c..bc04473 100644 --- a/asap-precompute-rs/src/matchers.rs +++ b/asap-precompute-rs/src/matchers.rs @@ -205,6 +205,29 @@ pub fn series_attrs(labels: &[KeyValue], aggregate_by: &[String]) -> Vec`, value = +/// the attribute's value as bytes. A cardinality or frequency sketch over +/// these series answers "how many distinct values" and "which values +/// dominate" per attribute key. +pub fn per_attribute(obs: &Observation, keys: &[String]) -> Vec { + let all = keys.iter().any(|k| k == "*"); + obs.labels + .iter() + .filter(|kv| all || keys.contains(&kv.key)) + .map(|kv| Observation { + timestamp_ms: obs.timestamp_ms, + metric: obs.metric.clone(), + resource_labels: obs.resource_labels.clone(), + labels: vec![KeyValue::new(ATTRIBUTE_KEY_LABEL, kv.key.clone())], + value: crate::observation::ObservationValue::bytes(kv.value.as_bytes().to_vec()), + }) + .collect() +} + #[cfg(test)] mod tests { use super::*; diff --git a/asap-precompute-rs/src/otap/answer_nodes.rs b/asap-precompute-rs/src/otap/answer_nodes.rs index 5cb7ace..dde6f56 100644 --- a/asap-precompute-rs/src/otap/answer_nodes.rs +++ b/asap-precompute-rs/src/otap/answer_nodes.rs @@ -1,6 +1,7 @@ //! `urn:asap:processor:sketch_detector` reads merged sketch windows (the -//! output of `window_merge`) and emits each window's answers with bounds. The -//! exact and raw baselines are in [`super::baseline_nodes`]. +//! output of `window_merge`) and emits each window's answers with bounds and +//! its detection events. The exact and raw baselines are in +//! [`super::baseline_nodes`]. //! //! ```yaml //! detector: @@ -16,11 +17,13 @@ use linkme::distributed_slice; use otel_arrow_dfe_config::node::NodeUserConfig; use otel_arrow_dfe_engine::config::ProcessorConfig; use otel_arrow_dfe_engine::context::PipelineContext; +use otel_arrow_dfe_engine::control::AckMsg; use otel_arrow_dfe_engine::error::Error; use otel_arrow_dfe_engine::local::processor as local; use otel_arrow_dfe_engine::message::Message; use otel_arrow_dfe_engine::node::NodeId; use otel_arrow_dfe_engine::processor::ProcessorWrapper; +use otel_arrow_dfe_engine::ConsumerEffectHandlerExtension; use otel_arrow_dfe_engine::{MessageSourceLocalEffectHandlerExtension, ProcessorFactory}; use otel_arrow_dfe_otap::pdata::OtapPdata; use otel_arrow_dfe_otap::OTAP_PROCESSOR_FACTORIES; @@ -29,6 +32,9 @@ use serde::Deserialize; use otel_arrow_dfe_pdata::proto::opentelemetry::common::v1::{ AnyValue, InstrumentationScope, KeyValue as OtlpKeyValue, }; +use otel_arrow_dfe_pdata::proto::opentelemetry::logs::v1::{ + LogRecord, LogsData, ResourceLogs, ScopeLogs, SeverityNumber, +}; use otel_arrow_dfe_pdata::proto::opentelemetry::metrics::v1::{ Gauge, Metric, MetricsData, NumberDataPoint, ResourceMetrics, ScopeMetrics, }; @@ -37,7 +43,7 @@ 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; +use crate::answers::{sketch_answers, CardinalityClassifier, CardinalityRules, Event}; use crate::envelope::SketchEnvelope; use crate::observation::{KeyValue, Observation}; @@ -67,6 +73,9 @@ pub struct DetectorConfig { /// Heavy hitters answered from CMS + heap windows. #[serde(default = "default_k")] pub k: usize, + /// Thresholds for classifying per-attribute cardinality. + #[serde(default)] + pub cardinality: CardinalityRules, } pub(super) fn now_ms() -> u64 { @@ -144,17 +153,67 @@ pub(super) fn gauges_pdata(envs: &[SketchEnvelope]) -> Result Ok(OtapPdata::new_todo_context(payload)) } +/// Detection events as OTLP log records (for Loki or any log backend). +fn events_pdata(events: &[Event]) -> Result { + let records = events + .iter() + .map(|e| { + let mut attributes = vec![ + OtlpKeyValue::new("event.kind", AnyValue::new_string(e.kind.clone())), + OtlpKeyValue::new("method", AnyValue::new_string(e.method.clone())), + ]; + attributes.extend(e.attributes.iter().map(|kv| { + OtlpKeyValue::new(kv.key.clone(), AnyValue::new_string(kv.value.clone())) + })); + LogRecord::build() + .time_unix_nano(e.window_end_ms * 1_000_000) + .observed_time_unix_nano(e.window_end_ms * 1_000_000) + .severity_number(SeverityNumber::Info) + .severity_text("INFO") + .body(AnyValue::new_string(e.body.clone())) + .attributes(attributes) + .finish() + }) + .collect::>(); + let data = LogsData::new(vec![ResourceLogs::new( + Resource::build() + .attributes(vec![OtlpKeyValue::new( + "service.name", + AnyValue::new_string("asap-detector"), + )]) + .finish(), + vec![ScopeLogs::new( + InstrumentationScope::build().name("asap.events").finish(), + records, + )], + )]); + let payload: OtapPayload = OtlpProtoMessage::Logs(data) + .try_into() + .map_err(|e| format!("{e:?}"))?; + Ok(OtapPdata::new_todo_context(payload)) +} + pub(super) async fn emit( envs: &[SketchEnvelope], + events: &[Event], effects: &mut local::EffectHandler, ) -> Result<(), Error> { - if envs.is_empty() { - return Ok(()); + if !envs.is_empty() { + let envs = with_delay(envs, now_ms()); + match gauges_pdata(&envs) { + Ok(pdata) => effects.send_message_with_source_node(pdata).await?, + Err(e) => effects.info(&format!("answers: encode failed: {e}")).await, + } } - let envs = with_delay(envs, now_ms()); - match gauges_pdata(&envs) { - Ok(pdata) => effects.send_message_with_source_node(pdata).await?, - Err(e) => effects.info(&format!("answers: encode failed: {e}")).await, + if !events.is_empty() { + match events_pdata(events) { + Ok(pdata) => effects.send_message_with_source_node(pdata).await?, + Err(e) => { + effects + .info(&format!("answers: event encode failed: {e}")) + .await + } + } } Ok(()) } @@ -181,6 +240,20 @@ pub(super) async fn observations( struct SketchDetector { config: DetectorConfig, + cardinality: CardinalityClassifier, +} + +/// The `bound="estimate"` value of `metric` among `answers`. +fn estimate_of(answers: &[SketchEnvelope], metric: &str) -> Option { + answers + .iter() + .find(|e| { + e.metric_name == metric + && e.labels + .iter() + .any(|kv| kv.key == "bound" && kv.value == "estimate") + }) + .map(|e| e.value) } #[async_trait(?Send)] @@ -193,17 +266,27 @@ impl local::Processor for SketchDetector { let Message::PData(pdata) = msg else { return Ok(()); }; - let mut out = Vec::new(); + // Consumed into state here: ack so upstream (e.g. fanout) can move on. + effects.notify_ack(AckMsg::new(pdata.clone())).await?; + let (mut out, mut events) = (Vec::new(), Vec::new()); for obs in observations(pdata, effects).await { let Some(env) = obs.value.envelope.filter(|e| !e.payload.is_empty()) else { continue; }; match sketch_answers(&env, &self.config.quantiles, self.config.k) { - Ok(answers) => out.extend(answers), + Ok(answers) => { + if let Some(d) = estimate_of(&answers, "asap.distinct") { + let (gauges, event) = + self.cardinality.observe(&env, "sketch", d, env.count); + out.extend(gauges); + events.extend(event); + } + out.extend(answers); + } Err(e) => effects.info(&format!("sketch_detector: {e}")).await, } } - emit(&out, effects).await + emit(&out, &events, effects).await } } @@ -225,7 +308,10 @@ static SKETCH_DETECTOR: ProcessorFactory = ProcessorFactory { let config: DetectorConfig = serde_json::from_value(node_config.config.clone()).map_err(invalid)?; Ok(ProcessorWrapper::local( - SketchDetector { config }, + SketchDetector { + cardinality: CardinalityClassifier::new(config.cardinality.clone()), + config, + }, node, node_config, processor_config, diff --git a/asap-precompute-rs/src/otap/baseline_nodes.rs b/asap-precompute-rs/src/otap/baseline_nodes.rs index a38d151..625a399 100644 --- a/asap-precompute-rs/src/otap/baseline_nodes.rs +++ b/asap-precompute-rs/src/otap/baseline_nodes.rs @@ -69,9 +69,11 @@ use super::answer_nodes::{ now_ms, observations, series_key, }; use super::codec::OtapSketchEncoder; -use crate::answers::{exact_answers, gauge, ExactWindow}; +use crate::answers::{ + exact_answers, gauge, CardinalityClassifier, CardinalityRules, Event, ExactWindow, +}; use crate::envelope::{SketchEnvelope, SketchType}; - +use crate::matchers::ATTRIBUTE_KEY_LABEL; use crate::observation::{KeyValue, Observation, ObservationValueKind}; /// Registered URN of the exact branch. @@ -126,6 +128,9 @@ pub struct ExactDetectorConfig { /// Heavy hitters per attribute key. #[serde(default = "default_k")] pub k: usize, + /// Thresholds for classifying per-attribute cardinality. + #[serde(default)] + pub cardinality: CardinalityRules, } /// `raw_sink` configuration. @@ -227,7 +232,7 @@ impl ExactBranch { continue; } let (_, w) = state.entry(format!("a|{}", kv.key)).or_insert_with(|| { - let labels = vec![KeyValue::new("attribute.key", kv.key.clone())]; + let labels = vec![KeyValue::new(ATTRIBUTE_KEY_LABEL, kv.key.clone())]; (template(labels), ExactWindow::default()) }); *w.counts.entry(kv.value.clone()).or_default() += 1; @@ -352,25 +357,34 @@ impl local::Processor for ExactMerge { struct ExactDetector { config: ExactDetectorConfig, + cardinality: CardinalityClassifier, } impl ExactDetector { - /// Answers of one merged window. - fn answer(&mut self, mut template: SketchEnvelope) -> Vec { - let mut out = Vec::new(); + /// Answers and events of one merged window. + fn answer(&mut self, mut template: SketchEnvelope) -> (Vec, Vec) { + let (mut out, mut events) = (Vec::new(), Vec::new()); let Ok(mut w) = ExactWindow::decode(&template.payload) else { - return out; + return (out, events); }; template.payload = Vec::new(); template.sketch_type = SketchType::Unspecified; template.count = w.points; + if !w.counts.is_empty() { + let distinct = w.counts.len() as f64; + let (gauges, event) = self + .cardinality + .observe(&template, "exact", distinct, w.points); + out.extend(gauges); + events.extend(event); + } out.extend(exact_answers( &template, &mut w, &self.config.quantiles, self.config.k, )); - out + (out, events) } } @@ -385,14 +399,16 @@ impl local::Processor for ExactDetector { return Ok(()); }; effects.notify_ack(AckMsg::new(pdata.clone())).await?; - let mut out = Vec::new(); + let (mut out, mut events) = (Vec::new(), Vec::new()); for obs in observations(pdata, effects).await { let Some(env) = obs.value.envelope.filter(|e| !e.payload.is_empty()) else { continue; }; - out.extend(self.answer(*env)); + let (gauges, new) = self.answer(*env); + out.extend(gauges); + events.extend(new); } - emit(&out, effects).await + emit(&out, &events, effects).await } } @@ -696,7 +712,10 @@ static EXACT_DETECTOR: ProcessorFactory = ProcessorFactory { _: &otel_arrow_dfe_engine::capability::registry::Capabilities| { let config: ExactDetectorConfig = config(&node_config)?; Ok(ProcessorWrapper::local( - ExactDetector { config }, + ExactDetector { + cardinality: CardinalityClassifier::new(config.cardinality.clone()), + config, + }, node, node_config, processor_config, @@ -825,10 +844,13 @@ mod tests { assert!(merged.iter().all(|e| e.count == 100)); let config: ExactDetectorConfig = serde_json::from_value(serde_json::json!({"k": 2})).unwrap(); - let mut detector = ExactDetector { config }; + let mut detector = ExactDetector { + cardinality: CardinalityClassifier::new(config.cardinality.clone()), + config, + }; let out: Vec<_> = merged .into_iter() - .flat_map(|e| detector.answer(e)) + .flat_map(|e| detector.answer(e).0) .collect(); assert_eq!(get(&out, "asap.quantile", ("quantile", "0.5")), Some(50.0)); assert_eq!( diff --git a/asap-precompute-rs/src/otap/config.rs b/asap-precompute-rs/src/otap/config.rs index d268d2e..3ad6245 100644 --- a/asap-precompute-rs/src/otap/config.rs +++ b/asap-precompute-rs/src/otap/config.rs @@ -116,6 +116,8 @@ pub struct PluginConfig { pub quantiles: Vec, /// Labels that identify a series; empty keys series by every label. pub aggregate_by: Vec, + /// Attribute keys sketched as their own series; `"*"` = every attribute. + pub per_attribute: Vec, } impl Default for PluginConfig { @@ -134,6 +136,7 @@ impl Default for PluginConfig { transmit_sketch: true, quantiles: Vec::new(), aggregate_by: Vec::new(), + per_attribute: Vec::new(), } } } diff --git a/asap-precompute-rs/src/otap/processor.rs b/asap-precompute-rs/src/otap/processor.rs index 52a107f..0adaf84 100644 --- a/asap-precompute-rs/src/otap/processor.rs +++ b/asap-precompute-rs/src/otap/processor.rs @@ -74,6 +74,8 @@ use std::time::Duration; use async_trait::async_trait; use linkme::distributed_slice; +use otel_arrow_dfe_engine::control::AckMsg; +use otel_arrow_dfe_engine::ConsumerEffectHandlerExtension; use serde::{Deserialize, Serialize}; use otel_arrow_dfe_config::error::Error as OtapConfigError; @@ -93,6 +95,7 @@ use otel_arrow_dfe_otap::OTAP_PROCESSOR_FACTORIES; use crate::config::PrecomputeConfigSet; use crate::envelope::SketchEnvelope; +use crate::matchers::per_attribute; use crate::observation::ObservationValueKind; use crate::otap::config::resolve as resolve_plugin_config; use crate::otap::PluginConfig; @@ -155,6 +158,10 @@ pub struct AsapSketchesUserConfig { /// Labels that identify a series; empty keys series by every label. #[serde(default)] pub aggregate_by: Vec, + /// Sketch each listed attribute's values as its own series + /// (`attribute.key=`); `"*"` = every attribute. + #[serde(default)] + pub per_attribute: Vec, } impl AsapSketchesUserConfig { @@ -193,6 +200,7 @@ impl AsapSketchesUserConfig { transmit_sketch: self.transmit_sketch, quantiles: self.quantiles, aggregate_by: self.aggregate_by, + per_attribute: self.per_attribute, }; // Bounce the config through the resolver so config mistakes // (unsupported sketch_type, zero window) surface here rather @@ -411,6 +419,10 @@ impl local::Processor for AsapSketchesProcessor { match msg { Message::PData(pdata) => { + // Consumed into state here: ack so upstream (e.g. fanout) can move on. + effect_handler + .notify_ack(AckMsg::new(pdata.clone())) + .await?; let outcome = match decode_pdata_to_observations(pdata) { Ok(outcome) => outcome, Err(error) => { @@ -442,7 +454,16 @@ impl local::Processor for AsapSketchesProcessor { self.emit_envelopes(&envs, effect_handler).await?; } } - for obs in &outcome.observations { + let observations = if self.plugin_config.per_attribute.is_empty() { + outcome.observations + } else { + outcome + .observations + .iter() + .flat_map(|o| per_attribute(o, &self.plugin_config.per_attribute)) + .collect() + }; + for obs in &observations { // Note: `observe` here handles both "genuine OTLP // metric" and "sketch shipped as binary inside an // OTAP metric" (an upstream asap_sketches node's diff --git a/asap-precompute-rs/src/otap/window_merge.rs b/asap-precompute-rs/src/otap/window_merge.rs index 6a5534b..e61db4a 100644 --- a/asap-precompute-rs/src/otap/window_merge.rs +++ b/asap-precompute-rs/src/otap/window_merge.rs @@ -20,12 +20,14 @@ use linkme::distributed_slice; use otel_arrow_dfe_config::node::NodeUserConfig; use otel_arrow_dfe_engine::config::ProcessorConfig; use otel_arrow_dfe_engine::context::PipelineContext; +use otel_arrow_dfe_engine::control::AckMsg; use otel_arrow_dfe_engine::control::NodeControlMsg; use otel_arrow_dfe_engine::error::Error; use otel_arrow_dfe_engine::local::processor as local; use otel_arrow_dfe_engine::message::Message; use otel_arrow_dfe_engine::node::NodeId; use otel_arrow_dfe_engine::processor::ProcessorWrapper; +use otel_arrow_dfe_engine::ConsumerEffectHandlerExtension; use otel_arrow_dfe_engine::{MessageSourceLocalEffectHandlerExtension, ProcessorFactory}; use otel_arrow_dfe_otap::pdata::OtapPdata; use otel_arrow_dfe_otap::OTAP_PROCESSOR_FACTORIES; @@ -181,6 +183,8 @@ impl local::Processor for WindowMergeProcessor { } match msg { Message::PData(pdata) => { + // Consumed into state here: ack so upstream (e.g. fanout) can move on. + effects.notify_ack(AckMsg::new(pdata.clone())).await?; let observations = match decode_pdata_to_observations(pdata) { Ok(outcome) => outcome.observations, Err(e) => { diff --git a/deploy/telemetry-functions/README.md b/deploy/telemetry-functions/README.md index 7c84c72..f573002 100644 --- a/deploy/telemetry-functions/README.md +++ b/deploy/telemetry-functions/README.md @@ -18,13 +18,13 @@ of scenario time, so all three pipelines receive identical points. | Stage | Config | What it runs | | --- | --- | --- | -| generator-a/b, exact-generator-a/b, raw-generator-a/b | `pipelines/generator-*.yaml` | `scenario_generator`: seeded latency points over 20 routes (temporary stand-in for the official traffic generator) | -| branch-a/b | `pipelines/branch.yaml` | `asap_sketches`: one KLL per route per 1-minute window | -| merge | `pipelines/merge.yaml` | `window_merge`: merges both sources' sketches per window, 5 s after the window ends | -| detector | `pipelines/detector.yaml` | `sketch_detector`: p50/p99 per route with lower/upper bounds (`method="sketch"`) | -| exact-branch-a/b | `pipelines/exact-branch.yaml` | `exact_branch`: every value per route per 1-minute window | +| generator-a/b, exact-generator-a/b, raw-generator-a/b | `latency/generator-*.yaml` | `scenario_generator`: seeded latency points over 20 routes (temporary stand-in for the official traffic generator) | +| branch-a/b | `latency/branch.yaml` | `asap_sketches`: one KLL per route per 1-minute window | +| merge | `latency/merge.yaml` | `window_merge`: merges both sources' sketches per window, 5 s after the window ends | +| detector | `latency/detector.yaml` | `sketch_detector`: p50/p99 per route with lower/upper bounds (`method="sketch"`) | +| exact-branch-a/b | `latency/exact-branch.yaml` | `exact_branch`: every value per route per 1-minute window | | exact-merge | `common/exact-merge.yaml` | `exact_merge`: adds both sources' exact windows, 5 s after the window ends | -| exact-detector | `pipelines/exact-detector.yaml` | `exact_detector`: exact p50/p99 per route (`method="exact"`): the ground truth | +| exact-detector | `latency/exact-detector.yaml` | `exact_detector`: exact p50/p99 per route (`method="exact"`): the ground truth | | raw-branch-a/b, raw-merge | `common/raw-*.yaml` | forward raw points (OTLP receiver -> exporter) | | raw-sink | `common/raw-sink.yaml` | `raw_sink`: writes every raw point as one log record to `loki-raw` and reports when `loki-raw` acknowledged each window's last point | diff --git a/deploy/telemetry-functions/cardinality/branch.yaml b/deploy/telemetry-functions/cardinality/branch.yaml new file mode 100644 index 0000000..707698c --- /dev/null +++ b/deploy/telemetry-functions/cardinality/branch.yaml @@ -0,0 +1,39 @@ +# Sketch pipeline, stage 1 at each source: one HLL per attribute key 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: hll + window_size: 60s + output_metric_name: http.server.request.duration + per_attribute: ["*"] + omit_resource_attrs: true + 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/cardinality/detector.yaml b/deploy/telemetry-functions/cardinality/detector.yaml new file mode 100644 index 0000000..e57e3cb --- /dev/null +++ b/deploy/telemetry-functions/cardinality/detector.yaml @@ -0,0 +1,37 @@ +# Distinct values per attribute key (with bounds) and its class (stable, +# growing, identifier-like) to Prometheus; class changes 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: + cardinality: {baseline_windows: 10, identifier_ratio: 0.9, growth: 0.2} + 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/cardinality/exact-branch.yaml b/deploy/telemetry-functions/cardinality/exact-branch.yaml new file mode 100644 index 0000000..28509f6 --- /dev/null +++ b/deploy/telemetry-functions/cardinality/exact-branch.yaml @@ -0,0 +1,34 @@ +# Exact pipeline, stage 1 at each source: a value -> count map per attribute key 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: ["*"]} + 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/cardinality/exact-detector.yaml b/deploy/telemetry-functions/cardinality/exact-detector.yaml new file mode 100644 index 0000000..d08a777 --- /dev/null +++ b/deploy/telemetry-functions/cardinality/exact-detector.yaml @@ -0,0 +1,37 @@ +# 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: + cardinality: {baseline_windows: 10, identifier_ratio: 0.9, growth: 0.2} + 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/cardinality/generator-a.yaml b/deploy/telemetry-functions/cardinality/generator-a.yaml new file mode 100644 index 0000000..165b1d8 --- /dev/null +++ b/deploy/telemetry-functions/cardinality/generator-a.yaml @@ -0,0 +1,37 @@ +# Source a: four attributes with known cardinality behaviour. +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} # stable + - {key: http.method, cardinality: 5} # stable + - {key: k8s.pod.name, cardinality: 50, growth_per_minute: 0.05} # growing + - {key: user.id, unique: true} # identifier-like + value: {lognormal: {median: 50.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/cardinality/generator-b.yaml b/deploy/telemetry-functions/cardinality/generator-b.yaml new file mode 100644 index 0000000..1ff2ac7 --- /dev/null +++ b/deploy/telemetry-functions/cardinality/generator-b.yaml @@ -0,0 +1,37 @@ +# Source b: four attributes with known cardinality behaviour. +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} # stable + - {key: http.method, cardinality: 5} # stable + - {key: k8s.pod.name, cardinality: 50, growth_per_minute: 0.05} # growing + - {key: user.id, unique: true} # identifier-like + value: {lognormal: {median: 50.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/pipelines/merge.yaml b/deploy/telemetry-functions/cardinality/merge.yaml similarity index 100% rename from deploy/telemetry-functions/pipelines/merge.yaml rename to deploy/telemetry-functions/cardinality/merge.yaml diff --git a/deploy/telemetry-functions/compose.yaml b/deploy/telemetry-functions/compose.yaml index 8931593..9b279f8 100644 --- a/deploy/telemetry-functions/compose.yaml +++ b/deploy/telemetry-functions/compose.yaml @@ -8,9 +8,9 @@ # 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) # and SCENARIO_START, so all three pipelines get identical points. -# ./pipelines holds the use case's stage configs, ./common the stages that do -# not depend on it; CPU_* variables move a stage to another core. Prometheus, -# Grafana and Loki are the observers. +# USE_CASE picks the stage-config directory (latency, cardinality, ...), and +# ./common holds the stages every use case shares; CPU_* variables move a stage +# to another core. Prometheus, Grafana and Loki are the observers. x-collector: &collector image: asap-collector:local restart: "no" @@ -24,117 +24,117 @@ services: generator-a: <<: *generator command: ["--config", "/pipelines/generator-a.yaml"] - volumes: ["./pipelines:/pipelines:ro", "./common:/common:ro"] + volumes: ["./${USE_CASE:-latency}:/pipelines:ro", "./common:/common:ro"] cpuset: "${CPU_GENERATOR_A:-0}" links: ["branch-a:downstream"] depends_on: [branch-a] generator-b: <<: *generator command: ["--config", "/pipelines/generator-b.yaml"] - volumes: ["./pipelines:/pipelines:ro", "./common:/common:ro"] + volumes: ["./${USE_CASE:-latency}:/pipelines:ro", "./common:/common:ro"] cpuset: "${CPU_GENERATOR_B:-1}" links: ["branch-b:downstream"] depends_on: [branch-b] branch-a: <<: *collector command: ["--config", "/pipelines/branch.yaml"] - volumes: ["./pipelines:/pipelines:ro", "./common:/common:ro"] + volumes: ["./${USE_CASE:-latency}:/pipelines:ro", "./common:/common:ro"] cpuset: "${CPU_BRANCH_A:-2}" depends_on: [merge] branch-b: <<: *collector command: ["--config", "/pipelines/branch.yaml"] - volumes: ["./pipelines:/pipelines:ro", "./common:/common:ro"] + volumes: ["./${USE_CASE:-latency}:/pipelines:ro", "./common:/common:ro"] cpuset: "${CPU_BRANCH_B:-3}" depends_on: [merge] merge: <<: *collector command: ["--config", "/pipelines/merge.yaml"] - volumes: ["./pipelines:/pipelines:ro", "./common:/common:ro"] + volumes: ["./${USE_CASE:-latency}:/pipelines:ro", "./common:/common:ro"] cpuset: "${CPU_MERGE:-4}" depends_on: [detector] detector: <<: *collector command: ["--config", "/pipelines/detector.yaml"] - volumes: ["./pipelines:/pipelines:ro", "./common:/common:ro"] + volumes: ["./${USE_CASE:-latency}:/pipelines:ro", "./common:/common:ro"] cpuset: "${CPU_DETECTOR:-5}" depends_on: [prometheus] # exact pipeline exact-generator-a: <<: *generator command: ["--config", "/pipelines/generator-a.yaml"] - volumes: ["./pipelines:/pipelines:ro", "./common:/common:ro"] + volumes: ["./${USE_CASE:-latency}:/pipelines:ro", "./common:/common:ro"] cpuset: "${CPU_EXACT_GENERATOR_A:-6}" links: ["exact-branch-a:downstream"] depends_on: [exact-branch-a] exact-generator-b: <<: *generator command: ["--config", "/pipelines/generator-b.yaml"] - volumes: ["./pipelines:/pipelines:ro", "./common:/common:ro"] + volumes: ["./${USE_CASE:-latency}:/pipelines:ro", "./common:/common:ro"] cpuset: "${CPU_EXACT_GENERATOR_B:-7}" links: ["exact-branch-b:downstream"] depends_on: [exact-branch-b] exact-branch-a: <<: *collector command: ["--config", "/pipelines/exact-branch.yaml"] - volumes: ["./pipelines:/pipelines:ro", "./common:/common:ro"] + volumes: ["./${USE_CASE:-latency}:/pipelines:ro", "./common:/common:ro"] cpuset: "${CPU_EXACT_BRANCH_A:-8}" depends_on: [exact-merge] exact-branch-b: <<: *collector command: ["--config", "/pipelines/exact-branch.yaml"] - volumes: ["./pipelines:/pipelines:ro", "./common:/common:ro"] + volumes: ["./${USE_CASE:-latency}:/pipelines:ro", "./common:/common:ro"] cpuset: "${CPU_EXACT_BRANCH_B:-9}" depends_on: [exact-merge] exact-merge: <<: *collector command: ["--config", "/common/exact-merge.yaml"] - volumes: ["./pipelines:/pipelines:ro", "./common:/common:ro"] + volumes: ["./${USE_CASE:-latency}:/pipelines:ro", "./common:/common:ro"] cpuset: "${CPU_EXACT_MERGE:-10}" depends_on: [exact-detector] exact-detector: <<: *collector command: ["--config", "/pipelines/exact-detector.yaml"] - volumes: ["./pipelines:/pipelines:ro", "./common:/common:ro"] + volumes: ["./${USE_CASE:-latency}:/pipelines:ro", "./common:/common:ro"] cpuset: "${CPU_EXACT_DETECTOR:-11}" depends_on: [prometheus] # raw pipeline raw-generator-a: <<: *generator command: ["--config", "/pipelines/generator-a.yaml"] - volumes: ["./pipelines:/pipelines:ro", "./common:/common:ro"] + volumes: ["./${USE_CASE:-latency}:/pipelines:ro", "./common:/common:ro"] cpuset: "${CPU_RAW_GENERATOR_A:-12}" links: ["raw-branch-a:downstream"] depends_on: [raw-branch-a] raw-generator-b: <<: *generator command: ["--config", "/pipelines/generator-b.yaml"] - volumes: ["./pipelines:/pipelines:ro", "./common:/common:ro"] + volumes: ["./${USE_CASE:-latency}:/pipelines:ro", "./common:/common:ro"] cpuset: "${CPU_RAW_GENERATOR_B:-13}" links: ["raw-branch-b:downstream"] depends_on: [raw-branch-b] raw-branch-a: <<: *collector command: ["--config", "/common/raw-branch.yaml"] - volumes: ["./pipelines:/pipelines:ro", "./common:/common:ro"] + volumes: ["./${USE_CASE:-latency}:/pipelines:ro", "./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:/pipelines:ro", "./common:/common:ro"] + volumes: ["./${USE_CASE:-latency}:/pipelines:ro", "./common:/common:ro"] cpuset: "${CPU_RAW_BRANCH_B:-15}" depends_on: [raw-merge] raw-merge: <<: *collector command: ["--config", "/common/raw-merge.yaml"] - volumes: ["./pipelines:/pipelines:ro", "./common:/common:ro"] + volumes: ["./${USE_CASE:-latency}:/pipelines:ro", "./common:/common:ro"] cpuset: "${CPU_RAW_MERGE:-16}" depends_on: [raw-sink] raw-sink: <<: *collector command: ["--config", "/common/raw-sink.yaml"] - volumes: ["./pipelines:/pipelines:ro", "./common:/common:ro"] + volumes: ["./${USE_CASE:-latency}:/pipelines:ro", "./common:/common:ro"] cpuset: "${CPU_RAW_SINK:-17}" depends_on: [prometheus, loki-raw] diff --git a/deploy/telemetry-functions/grafana/dashboards/cardinality.json b/deploy/telemetry-functions/grafana/dashboards/cardinality.json new file mode 100644 index 0000000..fc60a0e --- /dev/null +++ b/deploy/telemetry-functions/grafana/dashboards/cardinality.json @@ -0,0 +1,608 @@ +{ + "uid": "asap-cardinality", + "title": "Attribute cardinality: sketch vs exact (real collector containers)", + "schemaVersion": 39, + "version": 1, + "refresh": "10s", + "time": { + "from": "now-30m", + "to": "now" + }, + "templating": { + "list": [ + { + "name": "attribute", + "type": "query", + "datasource": { + "type": "prometheus", + "uid": "asap-prometheus" + }, + "query": { + "query": "label_values(asap_distinct, attribute_key)", + "refId": "attribute" + }, + "refresh": 2, + "current": { + "text": "k8s.pod.name", + "value": "k8s.pod.name" + } + } + ] + }, + "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: 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
" + }, + "id": 1 + }, + { + "type": "timeseries", + "title": "Distinct values of $attribute per minute: sketch with 95% bound vs exact", + "datasource": { + "type": "prometheus", + "uid": "asap-prometheus" + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 10 + }, + "targets": [ + { + "refId": "A", + "expr": "asap_distinct{attribute_key=\"$attribute\",method=\"sketch\",bound=\"estimate\"}", + "legendFormat": "sketch" + }, + { + "refId": "B", + "expr": "asap_distinct{attribute_key=\"$attribute\",method=\"sketch\",bound=\"lower\"}", + "legendFormat": "sketch lower" + }, + { + "refId": "C", + "expr": "asap_distinct{attribute_key=\"$attribute\",method=\"sketch\",bound=\"upper\"}", + "legendFormat": "sketch upper" + }, + { + "refId": "D", + "expr": "asap_distinct{attribute_key=\"$attribute\",method=\"exact\"}", + "legendFormat": "exact" + } + ], + "fieldConfig": { + "defaults": { + "custom": { + "spanNulls": true + } + }, + "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": 2 + }, + { + "type": "table", + "title": "Class of every attribute key (latest window)", + "datasource": { + "type": "prometheus", + "uid": "asap-prometheus" + }, + "gridPos": { + "h": 10, + "w": 12, + "x": 12, + "y": 10 + }, + "targets": [ + { + "refId": "A", + "expr": "asap_cardinality_state == 1", + "format": "table", + "instant": true + } + ], + "transformations": [ + { + "id": "organize", + "options": { + "excludeByName": { + "Time": true, + "Value": true, + "__name__": true, + "job": true, + "instance": true, + "metric": true + }, + "renameByName": { + "attribute_key": "attribute", + "method": "method", + "state": "class" + } + } + }, + { + "id": "sortBy", + "options": { + "sort": [ + { + "field": "class", + "desc": false + }, + { + "field": "attribute" + } + ] + } + } + ], + "id": 3 + }, + { + "type": "timeseries", + "title": "Distinct / points per attribute key (1.0 = every point new)", + "datasource": { + "type": "prometheus", + "uid": "asap-prometheus" + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 20 + }, + "targets": [ + { + "refId": "A", + "expr": "asap_distinct_ratio", + "legendFormat": "{{attribute_key}} {{method}}" + } + ], + "fieldConfig": { + "defaults": { + "unit": "percentunit", + "max": 1, + "min": 0 + }, + "overrides": [] + }, + "id": 4 + }, + { + "type": "logs", + "title": "Cardinality events (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=\"cardinality\"" + } + ], + "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/pipelines/branch.yaml b/deploy/telemetry-functions/latency/branch.yaml similarity index 100% rename from deploy/telemetry-functions/pipelines/branch.yaml rename to deploy/telemetry-functions/latency/branch.yaml diff --git a/deploy/telemetry-functions/pipelines/detector.yaml b/deploy/telemetry-functions/latency/detector.yaml similarity index 100% rename from deploy/telemetry-functions/pipelines/detector.yaml rename to deploy/telemetry-functions/latency/detector.yaml diff --git a/deploy/telemetry-functions/pipelines/exact-branch.yaml b/deploy/telemetry-functions/latency/exact-branch.yaml similarity index 100% rename from deploy/telemetry-functions/pipelines/exact-branch.yaml rename to deploy/telemetry-functions/latency/exact-branch.yaml diff --git a/deploy/telemetry-functions/pipelines/exact-detector.yaml b/deploy/telemetry-functions/latency/exact-detector.yaml similarity index 88% rename from deploy/telemetry-functions/pipelines/exact-detector.yaml rename to deploy/telemetry-functions/latency/exact-detector.yaml index 740c6b2..0902c9b 100644 --- a/deploy/telemetry-functions/pipelines/exact-detector.yaml +++ b/deploy/telemetry-functions/latency/exact-detector.yaml @@ -1,4 +1,4 @@ -# Exact pipeline, end: exact answers of each merged window, the +# Exact pipeline, end: exact answers and events of each merged window, the # ground truth (method="exact"). version: otel_dataflow/v1 engine: @@ -27,6 +27,7 @@ groups: 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} diff --git a/deploy/telemetry-functions/pipelines/generator-a.yaml b/deploy/telemetry-functions/latency/generator-a.yaml similarity index 100% rename from deploy/telemetry-functions/pipelines/generator-a.yaml rename to deploy/telemetry-functions/latency/generator-a.yaml diff --git a/deploy/telemetry-functions/pipelines/generator-b.yaml b/deploy/telemetry-functions/latency/generator-b.yaml similarity index 100% rename from deploy/telemetry-functions/pipelines/generator-b.yaml rename to deploy/telemetry-functions/latency/generator-b.yaml diff --git a/deploy/telemetry-functions/latency/merge.yaml b/deploy/telemetry-functions/latency/merge.yaml new file mode 100644 index 0000000..cb392f2 --- /dev/null +++ b/deploy/telemetry-functions/latency/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/run.sh b/deploy/telemetry-functions/run.sh index 5f27c09..1323597 100755 --- a/deploy/telemetry-functions/run.sh +++ b/deploy/telemetry-functions/run.sh @@ -1,5 +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. # DOCKER_COMMAND="sudo -n docker" when Docker needs sudo. Stop with: # $DOCKER_COMMAND compose -f deploy/telemetry-functions/compose.yaml down set -euo pipefail @@ -13,8 +14,9 @@ trap 'rm -rf "$context"' EXIT cp "$target/release/asap-collector" "$context/" strip --strip-debug "$context/asap-collector" "${docker[@]}" build -q -f deploy/collector.Dockerfile -t asap-collector:local "$context" +export USE_CASE=${USE_CASE:-latency} # Every generator counts scenario time from this second, so the three # pipelines get identical points. export SCENARIO_START=${SCENARIO_START:-$(($(date +%s) + 10))} "${docker[@]}" compose -f deploy/telemetry-functions/compose.yaml up -d -echo "Grafana: http://localhost:${GRAFANA_PORT:-3000}/d/asap-latency" +echo "Grafana: http://localhost:${GRAFANA_PORT:-3000}/d/asap-$USE_CASE" diff --git a/docs/images/cardinality-grafana.png b/docs/images/cardinality-grafana.png new file mode 100644 index 0000000..48412be Binary files /dev/null and b/docs/images/cardinality-grafana.png differ