Flag new heavy hitters: keys that enter the top-k

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<String, std::collections::VecDeque<HashSet<String>>>,
+}
+
+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<SketchEnvelope>, Vec<Event>) {
+        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);

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<String, VecDeque<KLLWrapper>>,
@@ -333,6 +341,12 @@ impl local::Processor<OtapPdata> 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<OtapPdata> = 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,

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<OtapPdata> for ExactMerge {
 struct ExactDetector {
     config: ExactDetectorConfig,
     cardinality: CardinalityClassifier,
+    novelty: NoveltyTracker,
     drift: DriftTracker,
     /// Recent windows' sorted values per quantile series.
     baselines: HashMap<String, VecDeque<Vec<f64>>>,
@@ -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<OtapPdata> = 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,