Skip to content

feat(monitor): whole-sketch F2 functional + unify sampling to the ε-floor #381

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
zzylol merged 2 commits into from
Jun 17, 2026
Merged

feat(monitor): whole-sketch F2 functional + unify sampling to the ε-floor #381

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
5b71b6f
feat(monitor): publish + round-trip monitor functional (incl. f2/whol…
Jun 17, 2026
794359d
refactor(monitor): unify coordinated sampling to the whole-sketch ε-f…
Jun 17, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions control_plane/src/emit/monitor.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,11 @@ pub enum Functional {
Sum,
CmsPoint,
LinearBuckets,
/// Whole-sketch second frequency moment F2 = ‖f‖₂². Monitors the entire
/// sketch's L2 mass (no per-point key) so any future point query stays within
/// ε — see `data_plane::monitor` module docs for the whole-sketch-vs-point
/// decision rule. The edge reports its local F2 = Σ_x f_i(x)² as the value.
F2,
}

impl Functional {
Expand All @@ -33,6 +38,7 @@ impl Functional {
Functional::Sum => "sum",
Functional::CmsPoint => "cms_point",
Functional::LinearBuckets => "linear_buckets",
Functional::F2 => "f2",
}
}

Expand All @@ -41,6 +47,7 @@ impl Functional {
match s {
"cms_point" => Functional::CmsPoint,
"linear_buckets" => Functional::LinearBuckets,
"f2" | "l2" => Functional::F2,
_ => Functional::Sum,
}
}
Expand Down Expand Up @@ -118,6 +125,7 @@ pub fn edge_threshold_block(intent: &MonitorIntent) -> Value {
pub fn streaming_config_monitor_entry(intent: &MonitorIntent) -> serde_json::Value {
serde_json::json!({
"agg_id": agg_id_for_metric(&intent.metric),
"functional": intent.functional.as_str(),
"key": intent.key,
"tau": intent.tau,
"epsilon": intent.epsilon,
Expand Down
8 changes: 7 additions & 1 deletion crates/asap_types/src/streaming_config.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,13 @@ use crate::query_requirements::QueryRequirements;
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MonitorSpec {
pub agg_id: u64,
/// CMS point-frequency key x; empty (default) for Sum / whole-stream.
/// Additive readout the edge reports: "sum" (default), "cms_point", "f2".
/// Pass-through metadata so the edge can auto-learn its reporting mode from
/// the pushed config; the coordinator allocation is value-driven and does not
/// branch on it (p_i ∝ √(value/rate) is the F2 allocation when value=‖f‖²).
#[serde(default)]
pub functional: String,
/// CMS point-frequency key x; empty (default) for Sum / whole-stream / F2.
#[serde(default)]
pub key: String,
/// Threshold τ (authoritative here, not at the edge).
Expand Down
120 changes: 45 additions & 75 deletions data_plane/src/monitor/coordinator.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@

use std::collections::HashMap;

use super::sampling_alloc::{allocate_sample_rates, epsilon_sample_floor};
use super::sampling_alloc::epsilon_sample_floor;

/// Static configuration for one monitor, sourced from the streaming-config
/// `monitors:` section (τ authoritative here, not at the edge).
Expand Down Expand Up @@ -235,42 +235,31 @@ impl Monitor {
/// 1/√rate_i`). With <2 edges or all rates unknown the allocation degenerates
/// to `p=1` everywhere (no sampling).
fn allocate_p(&self) -> HashMap<String, f64> {
// Stable edge order so the rate / freq vectors align with the p vector.
let ids: Vec<&String> = self.edges.keys().collect();
let rates: Vec<f64> = ids.iter().map(|id| self.edges[*id].rate).collect();
// Per-key frequency `f_i` = the monitored functional's value this edge
// knows this epoch (`known_value`: the cms_point key frequency, or the
// edge's contribution to the monitored sum), which is DISTINCT from the
// edge's total update `rate`. Feeding `f_i` (not `rates`) into the KKT
// allocation is what makes `p_i ∝ √(f_i/rate_i)` differentiate by the
// monitored key's *share* of each edge's stream. Previously this passed
// `rates` as both vectors → `√(rate_i/rate_i)=1` for every edge → a
// uniform allocation, with ALL differentiation left to the ε-floor;
// now the allocation itself is freq-driven (an edge that carries more of
// the monitored key at equal total rate keeps a higher p). An edge with
// no monitored-key mass yet (`known_value=0`) falls back to `p_i=1`
// (allocate_sample_rates: `freqs_i ≤ 0 ⇒ p_i=1`), i.e. nothing to sample.
let freqs: Vec<f64> = ids.iter().map(|id| self.edges[*id].known_value).collect();
let var_budget = {
let band = self.cfg.epsilon * self.cfg.tau;
band * band
};
let mut p_vec = allocate_sample_rates(&rates, &freqs, var_budget);
// Enforce the CDM-threshold coupling floor per edge: never sample so hard
// that ε_s = √((1−p)/(p·rate)) exceeds the agg's ε.
for (i, &rate) in rates.iter().enumerate() {
let floor = epsilon_sample_floor(self.cfg.epsilon, rate);
if p_vec[i] < floor {
p_vec[i] = floor;
}
}
ids.into_iter()
.cloned()
.zip(p_vec)
.map(|(id, p)| {
// Single edge or unknown rate ⇒ no sampling (p=1).
let p = if p > 0.0 && p <= 1.0 { p } else { 1.0 };
(id, p)
// Coordinated update-sampling for a SKETCH is governed by the whole-sketch
// ε-floor — a single law, NOT a per-key `√(f_i/rate_i)` KKT allocation.
//
// Why: the sampling protects the warm SKETCH's accuracy. A sketch point/L2
// estimate's error is bounded by the sketch NORM (‖f‖, rate-spread over
// buckets — NitroSketch), never by a single key's `f(x)`. So the only
// accuracy a per-edge `p_i` can buy is "keep this edge's L2 contribution
// within ε", i.e. ε_s = √((1−p)/(p·rate)) ≤ ε, which gives
// p_i = 1/(1 + ε²·rate_i) (the CDM-threshold coupling floor).
// The `√(freq/rate)` allocation only holds when a key is EXACT-counted
// OUTSIDE the sketch (then sampling that one counter is pointless anyway),
// so it is not a valid sketch-sampling regime — it has been retired. The
// monitored functional (cms_point / f2 / sum) still drives the THRESHOLD
// (`known_value` → `global_estimate`/alert); it no longer drives sampling.
// See `monitor` module docs for the derivation.
self.edges
.iter()
.map(|(id, e)| {
// Unknown rate ⇒ no sampling (p=1).
let p = if e.rate > 0.0 {
epsilon_sample_floor(self.cfg.epsilon, e.rate)
} else {
1.0
};
(id.clone(), p)
})
.collect()
}
Expand Down Expand Up @@ -484,61 +473,42 @@ mod tests {
}

#[test]
fn skewed_rates_yield_differentiated_sample_p_above_floor() {
// Two edges, skewed reported rates: the hot edge should be sampled
// harder (smaller p) than the quiet edge, and BOTH must sit at/above the
// ε-derived coupling floor. τ is large so the monitor stays in the grant
// (not alert) regime while we exercise the allocation.
fn skewed_rates_yield_sample_p_at_the_epsilon_floor() {
// The whole-sketch sampling law: p_i = ε-floor(rate_i). The hot (high-rate)
// edge is sampled harder (smaller p) than the quiet edge, and each p sits
// EXACTLY at its rate's ε-floor. τ is large so no alert fires.
let mut m = Monitor::new(cfg(1_000_000.0)); // epsilon 0.05
m.on_register("e1", 60_000, 0);
m.on_register("e2", 60_000, 0);
// e1 hot (100k items/win), e2 quiet (1k/win). Small local values keep us
// far from τ so no alert fires.
m.on_report("e1", 0, 1.0, 1, 100_000.0);
let a = m.on_report("e2", 0, 1.0, 1, 1_000.0);
m.on_report("e1", 0, 1.0, 1, 100_000.0); // hot: 100k items/win
let a = m.on_report("e2", 0, 1.0, 1, 1_000.0); // quiet: 1k/win
let p_hot = sample_p_in(&a, "e1");
let p_quiet = sample_p_in(&a, "e2");
assert!(
p_hot < p_quiet,
"hot edge p {p_hot} should be < quiet edge p {p_quiet}"
);
// Both respect the ε-derived floor for their rate.
let eps = m.cfg.epsilon;
assert!(p_hot >= epsilon_sample_floor(eps, 100_000.0) - 1e-9);
assert!(p_quiet >= epsilon_sample_floor(eps, 1_000.0) - 1e-9);
assert!(p_hot > 0.0 && p_hot <= 1.0);
assert!(p_quiet > 0.0 && p_quiet <= 1.0);
assert!(p_hot < p_quiet, "hot p {p_hot} should be < quiet p {p_quiet}");
assert!((p_hot - epsilon_sample_floor(eps, 100_000.0)).abs() < 1e-12);
assert!((p_quiet - epsilon_sample_floor(eps, 1_000.0)).abs() < 1e-12);
// Slack countdown unaffected: both grants carry the same (positive) slack.
assert!(slack_in(&a, "e1") > 0.0);
assert_eq!(slack_in(&a, "e1"), slack_in(&a, "e2"));
}

#[test]
fn per_key_freq_differentiates_at_equal_rate() {
// EQUAL total rate, but different monitored-key frequency (`known_value`):
// the edge carrying more of the monitored key keeps a HIGHER p (sampled
// less), since p_i ∝ √(f_i/rate_i) and rate cancels. Under the old
// freqs==rates this returned a UNIFORM p (same rate ⇒ same floor ⇒ tie),
// so this test fails on the pre-fix code — it pins the freq-driven branch.
// High rate ⇒ low ε-floor, modest τ ⇒ tight var_budget, so the allocation
// (not the floor) is the operative differentiator.
let mut m = Monitor::new(cfg(2_000.0)); // epsilon 0.05 ⇒ var_budget=(100)^2
fn sampling_is_rate_floor_independent_of_monitored_freq() {
// Unified law: p_i = ε-floor(rate_i), the whole-sketch sampling law — it
// depends ONLY on rate, never on the monitored-key frequency
// (`known_value`). EQUAL rate but very different known_value ⇒ EQUAL p.
// (The retired `√(f_i/rate_i)` allocation would have differentiated here;
// it does not, because sketch accuracy is rate/L2-bounded, not per-key.)
let mut m = Monitor::new(cfg(2_000.0));
m.on_register("e1", 60_000, 0);
m.on_register("e2", 60_000, 0);
// same rate (100k/win); e1 sees the monitored key 10× more than e2.
m.on_report("e1", 0, 900.0, 1, 100_000.0); // known_value=900
let a = m.on_report("e2", 0, 90.0, 1, 100_000.0); // known_value=90
let a = m.on_report("e2", 0, 90.0, 1, 100_000.0); // known_value=90, same rate
let p_e1 = sample_p_in(&a, "e1");
let p_e2 = sample_p_in(&a, "e2");
let floor = epsilon_sample_floor(m.cfg.epsilon, 100_000.0);
assert!(
p_e1 > p_e2,
"high-freq edge p {p_e1} should exceed low-freq p {p_e2} (freq-driven, not floor)"
);
// Both sit ABOVE the (equal) rate-floor, proving the allocation — not the
// floor — produced the spread.
assert!(p_e1 > floor + 1e-6 && p_e2 > floor + 1e-6,
"both p ({p_e1}, {p_e2}) should exceed the equal rate-floor {floor}");
assert_eq!(p_e1, p_e2, "equal rate ⇒ equal p regardless of monitored freq");
assert!((p_e1 - epsilon_sample_floor(m.cfg.epsilon, 100_000.0)).abs() < 1e-12);
}

#[test]
Expand Down
91 changes: 14 additions & 77 deletions data_plane/src/monitor/f2.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,6 @@

use std::collections::HashMap;

use super::sampling_alloc::{allocate_sample_rates, epsilon_sample_floor};

/// murmur3 64-bit finalizer (`fmix64`).
#[inline]
Expand Down Expand Up @@ -183,12 +182,12 @@ pub enum F2Action {
#[derive(Clone, Debug)]
struct EdgeF2 {
sketch: CountSketchF2,
rate: f64,
}

/// Distributed F2 threshold monitor + coordinated-sampling allocator. Holds each
/// edge's latest Count-Sketch + rate, merges on report, estimates the global
/// `F2̂`, fires at `F2̂ ≥ (1−ε)τ`, and allocates `p_i ∝ √(F2̂_i / rate_i)`.
/// Distributed F2 **threshold** monitor. Holds each edge's latest Count-Sketch,
/// linearly merges on report, estimates the global `F2̂`, and fires at
/// `F2̂ ≥ (1−ε)τ`. F2 is a MONITORED quantity only — per-edge update-sampling is
/// the single whole-sketch ε-floor law (`sampling_alloc`), not driven by F2.
pub struct DistributedF2Monitor {
tau: f64,
epsilon: f64,
Expand Down Expand Up @@ -237,23 +236,24 @@ impl DistributedF2Monitor {
false
}

/// Ingest one edge's `(Count-Sketch, rate)` for `window_start_ms`. Replaces
/// that edge's prior report, re-estimates the merged global F2, and returns
/// an Alert iff it crossed `(1−ε)τ`.
/// Ingest one edge's Count-Sketch for `window_start_ms`. Replaces that edge's
/// prior report, re-estimates the merged global F2, and returns an Alert iff
/// it crossed `(1−ε)τ`. (`rate` is part of the edge wire report but the F2
/// threshold needs only the sketch; sampling consumes rate elsewhere.)
pub fn on_report(
&mut self,
edge_id: &str,
window_start_ms: u64,
sketch: CountSketchF2,
rate: f64,
_rate: f64,
) -> Option<F2Action> {
if sketch.dims() != (self.d, self.w, self.seed) {
return None; // mis-dimensioned — drop
}
if !self.ensure_epoch(window_start_ms) {
return None; // stale epoch
}
self.edges.insert(edge_id.to_string(), EdgeF2 { sketch, rate });
self.edges.insert(edge_id.to_string(), EdgeF2 { sketch });
let f2 = self.global_f2();
if !self.alerted && f2 >= (1.0 - self.epsilon) * self.tau {
self.alerted = true;
Expand All @@ -275,37 +275,10 @@ impl DistributedF2Monitor {
merged.estimate_f2()
}

/// L2 sampling variance budget `V = (ε·‖f‖₂)² = ε²·F2̂` (vs L1's `(ε·τ)²`).
pub fn sampling_var_budget(&self) -> f64 {
self.epsilon * self.epsilon * self.global_f2()
}

/// Coordinated per-edge sampling probability `p_i ∝ √(F2̂_i / rate_i)`: the
/// freq weight is each edge's LOCAL F2 (from its own sketch), the rate is its
/// reported update rate, the budget is `ε²·F2̂_global`. An edge contributing
/// more L2 mass is sampled less (higher p) to preserve the F2 estimate.
/// Mirrors the L1 `coordinator::allocate_p` but with L2 freq/budget.
pub fn allocate_p(&self) -> HashMap<String, f64> {
let ids: Vec<&String> = self.edges.keys().collect();
let rates: Vec<f64> = ids.iter().map(|id| self.edges[*id].rate).collect();
let freqs: Vec<f64> = ids
.iter()
.map(|id| self.edges[*id].sketch.estimate_f2()) // local F2̂_i
.collect();
let var_budget = self.sampling_var_budget();
let mut p = allocate_sample_rates(&rates, &freqs, var_budget);
for (i, &rate) in rates.iter().enumerate() {
let floor = epsilon_sample_floor(self.epsilon, rate);
if p[i] < floor {
p[i] = floor;
}
}
ids.into_iter()
.cloned()
.zip(p)
.map(|(id, pi)| (id, if pi > 0.0 && pi <= 1.0 { pi } else { 1.0 }))
.collect()
}
// NOTE: F2 here is a MONITORED quantity (threshold/alert), NOT a sampling
// driver. Per-edge update-sampling is the single whole-sketch ε-floor law
// (`sampling_alloc::epsilon_sample_floor`); the old `√(F2/rate)` allocation
// was retired (it is not a valid sketch-sampling regime — see `monitor` docs).
}

/// Geometric (Sharfman–Schuster–Keren) safe-zone layer for **communication-efficient**
Expand Down Expand Up @@ -492,42 +465,6 @@ mod tests {
assert!(matches!(r_lo, F2Action::Alert { .. }), "above tau ⇒ Alert, got {r_lo:?}");
}

#[test]
fn allocate_p_higher_local_f2_keeps_higher_p_at_equal_rate() {
// Equal rate; edge A carries far more L2 mass than edge B (disjoint keys).
// p_i ∝ √(F2_i/rate) ⇒ A (high F2) keeps a HIGHER p (sampled less) to
// preserve the F2 estimate. High rate ⇒ low floor so the allocation, not
// the floor, drives the spread.
let seed = 0x1357_2468;
let mut mon = DistributedF2Monitor::new(1.0e12, 0.1, 9, 4096, seed);
let mut a = mon.new_edge_sketch();
let mut b = mon.new_edge_sketch();
fill(&mut a, &[(1u64, 100i64)]); // local F2 ≈ 10000
fill(&mut b, &[(2u64, 1i64)]); // local F2 ≈ 1
mon.on_report("a", 0, a, 1_000_000.0);
mon.on_report("b", 0, b, 1_000_000.0);
let p = mon.allocate_p();
let pa = p["a"];
let pb = p["b"];
let floor = epsilon_sample_floor(0.1, 1_000_000.0);
assert!(pa > pb, "high-F2 edge p {pa} should exceed low-F2 p {pb}");
assert!(pa > floor + 1e-9 && pb > floor + 1e-9, "p ({pa},{pb}) above floor {floor}");
}

#[test]
fn sampling_var_budget_is_eps2_times_f2() {
let seed = 0x9999_1111;
let eps = 0.2;
let mut mon = DistributedF2Monitor::new(1.0e12, eps, 9, 4096, seed);
let mut e = mon.new_edge_sketch();
fill(&mut e, &(1..=100u64).map(|k| (k, k as i64)).collect::<Vec<_>>());
mon.on_report("e", 0, e, 1000.0);
let f2 = mon.global_f2();
let v = mon.sampling_var_budget();
assert!((v - eps * eps * f2).abs() < 1e-6);
assert!(v > 0.0);
}

#[test]
fn epoch_advance_clears_edges() {
let mut mon = DistributedF2Monitor::new(1_000.0, 0.1, 5, 1024, 7);
Expand Down
22 changes: 21 additions & 1 deletion data_plane/src/monitor/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,26 @@
//! - [`epoch`] — tumbling-epoch alignment (matches the edge formula).
//! - [`alert`] — alert egress via the control-plane `Violation` sink.
//! - [`server`] — the tonic bidi-streaming `MonitorService` server.
//! - [`sampling_alloc`] — the coordinated update-sampling law (ε-floor).
//! - [`f2`] — whole-sketch L2/F2 *threshold* monitor (Count-Sketch +
//! geometric safe-zone). NOTE: F2 here is a MONITORED quantity (alert), not a
//! sampling driver — see the sampling law below.
//!
//! # Two orthogonal axes — DON'T conflate them
//!
//! 1. **What to monitor / alert on** (the `functional`): `sum`, `cms_point`
//! (a declared point `f(x)`), or `f2` (whole-sketch L2). This sets the
//! THRESHOLD `g vs τ` and what `known_value` means. Identity is `(agg_id,key)`
//! — `cms_point` carries a key; `sum`/`f2` are whole-stream (`key=""`).
//!
//! 2. **How hard to sample** (the per-edge `p_i`): a SINGLE law, the whole-sketch
//! **ε-floor** `p_i = 1/(1+ε²·rate_i)` (see [`sampling_alloc`]). It depends
//! ONLY on each edge's rate, NEVER on the monitored functional — because the
//! sampling protects a SKETCH, whose accuracy is bounded by the stream norm
//! (rate/L2), not by any single key. A per-key `√(f/rate)` allocation is NOT a
//! valid sketch-sampling regime (it needs exact-counting outside the sketch),
//! so it has been retired. Applies to additive sketches — Count-Min, Count-
//! Sketch, DDSketch, KLL, Sum — but NOT HLL (sampling biases cardinality).

pub mod alert;
pub mod coordinator;
Expand All @@ -23,5 +43,5 @@ pub mod server;

pub use alert::{global_threshold_violation, AlertSink};
pub use coordinator::{Action, Monitor, MonitorConfig};
pub use sampling_alloc::{allocate_sample_rates, epsilon_sample_floor, uniform_sample_rate};
pub use sampling_alloc::epsilon_sample_floor;
pub use server::{MonitorCoordinator, MonitorServiceImpl};
Loading