From e1ce307499f9196e4f365fa3591cf8695847a53f Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 16 Jul 2026 21:47:50 -0600 Subject: [PATCH] =?UTF-8?q?feat(gos):=20split=20Discipline=20B=20=E2=80=94?= =?UTF-8?q?=20retire=20alerting,=20decouple=20sample=5Fp=20from=20countdow?= =?UTF-8?q?n?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Discipline B's global-threshold alerting (Monitor::rebroadcast's CMY slack-countdown -> Action::Alert -> the control-plane violation sink) retires per ASAPCollector/docs/design-gos-unified-edge-telemetry.md §11: "Alerting and any other query-time decision is made entirely at the backend against [the reconstructed sketch state]; the edge no longer makes alerting decisions itself." A collector/edge must never be the thing that fires an alert. The whole-sketch ε-floor sampling law (p_i = 1/(1+ε²·rate_i)) was ALWAYS a pure per-edge function with no cross-edge coordination or global-estimate dependency, so decoupling it from the retired countdown is a genuine simplification, not a workaround: - Monitor drops known_value/departed_mass/alerted/tau/slack/gap entirely — just tracks each edge's last-reported rate. - on_report answers ONLY the reporting edge with a freshly computed grant (no more re-broadcast to every registered edge on every report — that fan-out existed only because slack sizing depended on the full edge set). - The retired "<2 edges => p=1" special case (needed only for the alert-fire safety proof under sampling noise) is gone: a lone high-rate edge is now sampled exactly like it would be in a larger edge set. Removes alert.rs (AlertSink/global_threshold_violation) and the now-dead ViolationKind::GlobalThresholdCrossed. MonitorConfig.tau / Grant.LocalSlack stay on the wire (vestigial) for compatibility with the Go edge side, which still sends them. Companion PR: ASAPCollector split/pr-retire-discipline-b-alerting — the edge side. Engine.Observe now reports its rate on its own periodic cadence (ReportEveryN observations), decoupled from any value/slack threshold. Verified with a real cross-language run (Go e2edriver <-> Rust monitor_coordinator_harness over live gRPC): the coordinator answers the edge's periodic rate report with a genuine computed sample_p grant. 🤖 Generated with Claude Code Co-Authored-By: Claude Sonnet 5 --- control_plane/src/monitor/mod.rs | 6 - .../src/bin/monitor_coordinator_harness.rs | 72 ++- data_plane/src/main.rs | 22 +- data_plane/src/monitor/alert.rs | 45 -- data_plane/src/monitor/coordinator.rs | 590 +++++------------- data_plane/src/monitor/mod.rs | 48 +- data_plane/src/monitor/server.rs | 149 ++--- data_plane/tests/monitor_grpc.rs | 228 +++---- 8 files changed, 365 insertions(+), 795 deletions(-) delete mode 100644 data_plane/src/monitor/alert.rs diff --git a/control_plane/src/monitor/mod.rs b/control_plane/src/monitor/mod.rs index e0cfc5327..3ac7e69ce 100644 --- a/control_plane/src/monitor/mod.rs +++ b/control_plane/src/monitor/mod.rs @@ -27,11 +27,6 @@ pub enum ViolationKind { Bandwidth, Accuracy, Cpu, - /// A continuous-monitoring (CDM) global aggregate crossed its threshold τ. - /// Emitted by the data-plane monitor coordinator, not the metrics scraper: - /// `agent_id` carries the monitor id, `observed` the global estimate at - /// fire time, `threshold` the τ. - GlobalThresholdCrossed, } impl std::fmt::Display for ViolationKind { @@ -40,7 +35,6 @@ impl std::fmt::Display for ViolationKind { ViolationKind::Bandwidth => write!(f, "bandwidth"), ViolationKind::Accuracy => write!(f, "accuracy"), ViolationKind::Cpu => write!(f, "cpu"), - ViolationKind::GlobalThresholdCrossed => write!(f, "global_threshold_crossed"), } } } diff --git a/data_plane/src/bin/monitor_coordinator_harness.rs b/data_plane/src/bin/monitor_coordinator_harness.rs index 679e36763..35e3963a0 100644 --- a/data_plane/src/bin/monitor_coordinator_harness.rs +++ b/data_plane/src/bin/monitor_coordinator_harness.rs @@ -1,20 +1,28 @@ //! Minimal CDM monitor-coordinator harness for the cross-language e2e -//! (deploy/mvp-multinode/scripts/monitor_e2e.sh in ASAPCollector). Runs ONLY -//! the tonic MonitorService with a single Sum monitor and prints a line to -//! stdout when the global threshold fires, then exits 0 — so a driver script -//! can assert the Go edge ⇄ Rust coordinator wire path end-to-end without -//! standing up the full data-plane. +//! (deploy/mvp-multinode/scripts/monitor_e2e.sh in ASAPCollector). Runs the +//! tonic MonitorService with a single monitor and prints a line to stdout once +//! the coordinator has computed a REAL coordinated-sampling grant for the +//! connecting edge (i.e. the edge's periodic rate report reached the +//! coordinator and a non-trivial sample_p came back), then exits 0 — so a +//! driver script can assert the Go edge ⇄ Rust coordinator wire path +//! end-to-end without standing up the full data-plane. +//! +//! Global-threshold alerting is retired (see `data_plane::monitor` module +//! docs) — this harness no longer waits for or reports an alert. //! //! Usage: monitor_coordinator_harness [timeout_secs] [key] //! -//! `key` is the monitor's group key (the edge's canonical `k=v;k2=v2` label -//! encoding). Scalar (Sum) monitors register under their series-group key, so -//! the harness config must carry the same key or registrations are rejected -//! as unconfigured. Empty (default) = ungrouped. +//! `tau` is accepted for CLI/back-compat but unused. `key` is the monitor's +//! group key (the edge's canonical `k=v;k2=v2` label encoding). Scalar (Sum) +//! monitors register under their series-group key, so the harness config must +//! carry the same key or registrations are rejected as unconfigured. Empty +//! (default) = ungrouped. -use std::sync::Arc; +use data_plane::monitor::{MonitorConfig, MonitorCoordinator, MonitorServiceImpl}; -use data_plane::monitor::{AlertSink, MonitorConfig, MonitorCoordinator, MonitorServiceImpl}; +/// The edge_id the Go e2edriver connects as (asap-precompute-go/monitor/ +/// grpcclient/cmd/e2edriver/main.go: `monitor.NewEngine("e2e-edge", ...)`). +const E2E_EDGE_ID: &str = "e2e-edge"; #[tokio::main] async fn main() -> Result<(), Box> { @@ -26,40 +34,46 @@ async fn main() -> Result<(), Box> { let timeout_secs: u64 = args.next().and_then(|s| s.parse().ok()).unwrap_or(30); let key: Vec = args.next().map(|s| s.into_bytes()).unwrap_or_default(); - // Exit-on-alert: the sink prints a machine-greppable line and signals the - // run to finish. - let (done_tx, mut done_rx) = tokio::sync::mpsc::channel::<()>(1); - let sink: AlertSink = Arc::new(move |v| { - println!( - "MONITOR_ALERT monitor={} observed={} threshold={}", - v.agent_id, v.observed, v.threshold - ); - let _ = done_tx.try_send(()); - }); - let cfg = MonitorConfig { agg_id, - key, + key: key.clone(), tau, epsilon: 0.05, window_ms, ..Default::default() }; - let coord = MonitorCoordinator::new(vec![cfg], sink); - let svc = MonitorServiceImpl::new(coord).into_server(); + let coord = MonitorCoordinator::new(vec![cfg]); + let svc = MonitorServiceImpl::new(coord.clone()).into_server(); let addr = format!("0.0.0.0:{port}").parse()?; - eprintln!("harness: serving MonitorService on {addr} (agg_id={agg_id}, tau={tau}, window_ms={window_ms})"); + eprintln!("harness: serving MonitorService on {addr} (agg_id={agg_id}, window_ms={window_ms})"); println!("HARNESS_READY {port}"); + // Poll for a real coordinated-sampling grant reaching this edge — the + // proof that the edge's periodic rate report made the live gRPC round + // trip and the coordinator answered with a computed sample_p. + let poll = { + let coord = coord.clone(); + let key = key.clone(); + async move { + loop { + if let Some(p) = coord.granted_sample_p(agg_id, &key, E2E_EDGE_ID).await { + println!("MONITOR_GRANT edge={E2E_EDGE_ID} agg_id={agg_id} sample_p={p}"); + return; + } + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + } + } + }; + tokio::select! { r = tonic::transport::Server::builder().add_service(svc).serve(addr) => { r?; } - _ = done_rx.recv() => { - eprintln!("harness: alert fired — exiting"); + _ = poll => { + eprintln!("harness: coordinated-sampling grant observed — exiting"); } _ = tokio::time::sleep(std::time::Duration::from_secs(timeout_secs)) => { - eprintln!("harness: timeout after {timeout_secs}s with no alert"); + eprintln!("harness: timeout after {timeout_secs}s with no grant"); std::process::exit(2); } } diff --git a/data_plane/src/main.rs b/data_plane/src/main.rs index 45f113b0f..9be3dc2be 100644 --- a/data_plane/src/main.rs +++ b/data_plane/src/main.rs @@ -594,14 +594,12 @@ async fn main() -> Result<()> { None }; - // CDM monitor coordinator: bidi MonitorService gRPC that runs the - // slack-countdown protocol over the streaming-config `monitors:` specs and - // fires a global-threshold alert through the violation sink (logged here; - // the control-plane replanner can subscribe via the same Violation shape). + // Coordinated-sampling monitor coordinator: bidi MonitorService gRPC that + // answers each edge's periodic rate report with its whole-sketch ε-floor + // sample_p grant. Global-threshold alerting is retired (see + // data_plane::monitor module docs) — this coordinator never fires one. let monitor_handle = if args.enable_monitor_coordinator { - use data_plane::monitor::{ - AlertSink, Functional, MonitorConfig, MonitorCoordinator, MonitorServiceImpl, - }; + use data_plane::monitor::{Functional, MonitorConfig, MonitorCoordinator, MonitorServiceImpl}; let specs: Vec = streaming_config .monitors() .iter() @@ -617,15 +615,7 @@ async fn main() -> Result<()> { if specs.is_empty() { warn!("--enable-monitor-coordinator set but streaming-config has no `monitors:` yet — the coordinator will pick them up live when the control plane pushes a config (hot-reload)"); } - let sink: AlertSink = Arc::new(|v| { - warn!( - monitor = %v.agent_id, - observed = v.observed, - threshold = v.threshold, - "CDM global threshold crossed" - ); - }); - let coord = MonitorCoordinator::new(specs, sink); + let coord = MonitorCoordinator::new(specs); // Hot-reload watcher: the coordinator reads `monitors:` once at boot, but // the control plane pushes the real config slightly AFTER boot via the diff --git a/data_plane/src/monitor/alert.rs b/data_plane/src/monitor/alert.rs deleted file mode 100644 index 46fd26d31..000000000 --- a/data_plane/src/monitor/alert.rs +++ /dev/null @@ -1,45 +0,0 @@ -//! Global-threshold alert emission. Reuses the control-plane violation sink -//! (`OnViolationFn`) rather than inventing a new egress, so a crossed CDM -//! threshold flows through the same path operators already watch for -//! bandwidth/accuracy/cpu violations. - -use std::sync::Arc; - -use control_plane::monitor::{Violation, ViolationKind}; - -/// A sink for global-threshold-crossed alerts. The default wiring is the same -/// `OnViolationFn` the control-plane scraper uses. -pub type AlertSink = Arc; - -/// Build a `Violation` for a monitor whose global aggregate crossed τ. -pub fn global_threshold_violation( - monitor_id: impl Into, - global_estimate: f64, - tau: f64, -) -> Violation { - Violation { - agent_id: monitor_id.into(), - kind: ViolationKind::GlobalThresholdCrossed, - observed: global_estimate, - threshold: tau, - } -} - -#[cfg(test)] -mod tests { - use super::*; - use std::sync::Mutex; - - #[test] - fn routes_through_violation_sink() { - let seen: Arc>> = Arc::new(Mutex::new(Vec::new())); - let seen2 = seen.clone(); - let sink: AlertSink = Arc::new(move |v| seen2.lock().unwrap().push(v)); - sink(global_threshold_violation("agg:1/sum", 98.0, 100.0)); - let got = seen.lock().unwrap(); - assert_eq!(got.len(), 1); - assert_eq!(got[0].kind, ViolationKind::GlobalThresholdCrossed); - assert_eq!(got[0].observed, 98.0); - assert_eq!(got[0].threshold, 100.0); - } -} diff --git a/data_plane/src/monitor/coordinator.rs b/data_plane/src/monitor/coordinator.rs index 5e3e92af0..2c17a560b 100644 --- a/data_plane/src/monitor/coordinator.rs +++ b/data_plane/src/monitor/coordinator.rs @@ -1,39 +1,44 @@ -//! The global-threshold coordinator state machine — the Cormode–Muthukrishnan–Yi -//! distributed functional monitoring countdown, realized in the lost-mass-free -//! single-phase form. +//! The per-edge coordinated-sampling state machine. //! -//! Per monitor `(agg_id, key)` the coordinator tracks, for each edge, the LAST -//! VALUE that edge reported (its "known value"). Because the monitored -//! functional is additive and monotone within a tumbling window, an edge that -//! has not reported is guaranteed to be within its granted slack of its known -//! value. So: +//! Historically this module also ran the Cormode–Muthukrishnan–Yi distributed +//! functional-monitoring countdown (global-threshold alerting: register → +//! grant `(slack, sample_p)` → countdown → report → alert). That alerting half +//! is RETIRED as of the 2026-07 insert-time-GOS redesign +//! (`ASAPCollector/docs/design-gos-unified-edge-telemetry.md` §11): "Alerting +//! and any other query-time decision is made entirely at the backend against +//! [the reconstructed sketch state]; the edge no longer makes alerting +//! decisions itself." An edge/collector must never be the thing that fires an +//! alert — that decision now belongs entirely to query-time reads of the +//! backend's synced state, not to this streaming protocol. //! -//! ```text -//! global_estimate = Σ known_value_i + departed_mass -//! true_global ≤ global_estimate + Σ slack_i -//! ``` +//! What's left, and what changed: `obsCount`/rate-tracking (feeding the +//! coordinated-sampling `p_i` grant) is a genuinely separate concern from +//! alerting and is PRESERVED — but it no longer piggybacks on the alerting +//! trigger. The whole-sketch ε-floor `p_i = 1/(1 + ε²·rate_i)` +//! ([`super::sampling_alloc::epsilon_sample_floor`]) is a PURE PER-EDGE +//! function of that edge's own reported rate — it never depended on other +//! edges' state or on a global estimate/τ, so retiring the countdown that used +//! to gate reporting doesn't remove anything the sampling law needed. An edge +//! now reports its rate on its own periodic cadence (see the Go edge's +//! `Engine.Observe` — decoupled from any value/slack threshold), and the +//! coordinator answers that ONE edge with its own fresh grant immediately — +//! no more re-broadcasting to every registered edge on every report (that +//! fan-out existed only because slack sizing depended on the full edge set). //! -//! Allocating each edge `slack = Δ / (2k)` where `Δ = τ − global_estimate` and -//! `k` = #edges keeps the total uncertainty `Σ slack = Δ/2`, so -//! `true_global < (τ + global_estimate)/2 < τ` as long as `global_estimate < τ` -//! — the safety guarantee (no missed crossing). Each report raises -//! `global_estimate`, which shrinks `Δ` and hence the slack; a re-grant to all -//! edges lowers the bar for the next report. When `Δ ≤ ε·τ` the coordinator -//! fires the alert; at that point `true_global ∈ [(1−ε)τ, (1−ε)τ + ετ/2]`, i.e. -//! within `ε·τ` of `τ`. -//! -//! The baseline at each edge tracks its last reported value (NOT reset on -//! grant), which is exactly this coordinator's `known_value` — so no -//! below-slack mass is lost when the slack shrinks, and no separate poll/collect -//! round is needed. This module is pure (no I/O) and fully unit-tested; the -//! tonic server (`server.rs`) translates `Action`s to wire messages. +//! `MonitorConfig.tau` is no longer consulted here (alerting is gone); it +//! stays in the struct only because it still rides the shared streaming-config +//! `monitors:` schema alongside `epsilon`/`window_ms` — trimming it is a +//! config-schema change, not a coordinator behavior change, and out of scope +//! here. use std::collections::HashMap; use super::sampling_alloc::epsilon_sample_floor; -/// Which readout of a series' sketch the monitor thresholds. Scalar functionals -/// (`Sum`/`CmsPoint`/`LinearBuckets`) drive the CMY slack-countdown [`Monitor`]. +/// Which readout of a series' sketch a monitor's rate is scoped to. Retained +/// for identity/config-schema compatibility (a monitor is still keyed by +/// `(agg_id, key)`, and `key` is only meaningful for `CmsPoint`) even though +/// no functional-specific THRESHOLDING happens here anymore. #[derive(Clone, Copy, Debug, PartialEq, Eq, Default)] pub enum Functional { #[default] @@ -55,16 +60,16 @@ impl Functional { } /// Static configuration for one monitor, sourced from the streaming-config -/// `monitors:` section (τ authoritative here, not at the edge). +/// `monitors:` section. #[derive(Clone, Debug, PartialEq)] pub struct MonitorConfig { pub agg_id: u64, pub key: Vec, + /// No longer consulted (alerting retired) — see the module doc comment. pub tau: f64, + /// The ε-floor tolerance feeding [`epsilon_sample_floor`]. pub epsilon: f64, pub window_ms: u64, - /// Which functional this monitor thresholds. Defaults to `Sum` (the scalar - /// countdown) so existing scalar construction sites are unaffected. pub functional: Functional, } @@ -81,13 +86,17 @@ impl Default for MonitorConfig { } } -/// An action the coordinator wants the transport to perform. +/// An action the coordinator wants the transport to perform. `Alert` is gone — +/// this coordinator no longer makes alerting decisions (see module docs). #[derive(Clone, Debug, PartialEq)] pub enum Action { - /// Send a per-round slack grant to one edge. `sample_p` is the distributed- - /// NitroSketch update-sampling probability the coordinator allocates for this - /// edge (1.0 = no sampling); it is an ADDITIONAL field orthogonal to the - /// slack countdown. + /// Send this ONE edge its freshly-computed coordinated-sampling grant, in + /// direct response to its rate report. `sample_p` is the whole-sketch + /// ε-floor for the edge's own just-reported rate — independent of every + /// other edge, so (unlike the retired countdown) this never needs to + /// re-grant the rest of the edge set. `local_slack` is always 0: kept on + /// the wire message for `SlackGrant` compatibility, but carries no + /// meaning post-retirement — the edge no longer gates anything on it. Grant { edge_id: String, round: u64, @@ -95,32 +104,23 @@ pub enum Action { window_start_ms: u64, sample_p: f64, }, - /// The global aggregate crossed τ (within ε): fire exactly once per epoch. - Alert { - global_estimate: f64, - tau: f64, - window_start_ms: u64, - }, } #[derive(Clone, Debug, Default)] struct EdgeView { - known_value: f64, - last_seq: u64, - seen_seq: bool, - /// Last `rate` (items/window) the edge reported; 0 = unknown ⇒ no sampling - /// for this edge. Feeds the coordinated `p_i` allocation. + /// Last `rate` (items/window) the edge reported; 0 = unknown ⇒ no + /// sampling for this edge. The only per-edge state left — there is no + /// more `known_value`/global estimate to track once alerting is gone. rate: f64, } -/// One monitor's coordinator state for the current epoch. +/// One monitor's coordinator state for the current epoch: which edges are +/// registered and each one's last-reported rate. pub struct Monitor { cfg: MonitorConfig, window_start_ms: u64, round: u64, edges: HashMap, - departed_mass: f64, - alerted: bool, } impl Monitor { @@ -130,8 +130,6 @@ impl Monitor { window_start_ms: 0, round: 0, edges: HashMap::new(), - departed_mass: 0.0, - alerted: false, } } @@ -145,29 +143,23 @@ impl Monitor { self.edges.len() } - /// Σ known values + departed mass — a lower bound on the true global. - pub fn global_estimate(&self) -> f64 { - self.departed_mass + self.edges.values().map(|e| e.known_value).sum::() - } - - /// Δ = τ − estimate (clamped at 0). - fn gap(&self) -> f64 { - (self.cfg.tau - self.global_estimate()).max(0.0) - } - - /// Current per-edge slack = Δ / (2k); 0 when there are no edges. - fn slack(&self) -> f64 { - let k = self.edges.len(); - if k == 0 { - return 0.0; + /// The coordinated-sampling grant this monitor would currently compute for + /// `edge_id`, or `None` if the edge hasn't reported a rate yet + /// (test/observability — e.g. the cross-language e2e harness polls this to + /// confirm a real grant reached an edge, without needing a callback hook + /// into `dispatch`). + pub fn sample_p_for_edge(&self, edge_id: &str) -> Option { + let edge = self.edges.get(edge_id)?; + if edge.rate <= 0.0 { + return None; } - self.gap() / (2.0 * k as f64) + Some(epsilon_sample_floor(self.cfg.epsilon, edge.rate)) } /// Ensure the monitor is on the epoch for `window_start`. Advancing resets - /// the epoch (sketches reset at the boundary, so known values restart at 0). - /// Returns false if the timestamp is for a stale (already-closed) epoch, in - /// which case the caller should drop the event. + /// every edge's rate (a new window starts a fresh rate measurement). + /// Returns false if the timestamp is for a stale (already-closed) epoch, + /// in which case the caller should drop the event. fn ensure_epoch(&mut self, window_start: u64) -> bool { if window_start == self.window_start_ms { return true; @@ -175,8 +167,6 @@ impl Monitor { if window_start > self.window_start_ms { self.window_start_ms = window_start; self.round = 0; - self.departed_mass = 0.0; - self.alerted = false; for e in self.edges.values_mut() { *e = EdgeView::default(); } @@ -186,17 +176,18 @@ impl Monitor { } /// Register (or refresh) an edge for the epoch it reports as - /// `window_start_ms` (the edge's own aligned window start, which avoids any - /// coordinator clock skew). Adds the edge to the membership set and re-grants - /// the (now smaller) slack to everyone, since k changed. + /// `window_start_ms`. Adds the edge to the membership set. No grant is + /// returned here — an edge is safely unsampled (p=1) until its first rate + /// report, and unlike the retired countdown, a new registration doesn't + /// need to perturb any other edge's already-granted `p`. pub fn on_register( &mut self, edge_id: &str, epoch_window_ms: u64, window_start_ms: u64, ) -> Vec { - // Alignment guard: a mismatched window size would silently corrupt the - // global estimate. Refuse the registration (no grants) and let the + // Alignment guard: a mismatched window size would silently corrupt + // epoch bookkeeping. Refuse the registration (no grants) and let the // caller log/close the stream. if epoch_window_ms != self.cfg.window_ms { return Vec::new(); @@ -205,20 +196,13 @@ impl Monitor { return Vec::new(); } self.edges.entry(edge_id.to_string()).or_default(); - self.rebroadcast() + Vec::new() } - /// Handle an edge report. Returns grants (re-broadcast with shrunken slack) - /// or an alert. Stale-epoch, unknown-edge, and duplicate-seq reports are - /// dropped. - pub fn on_report( - &mut self, - edge_id: &str, - window_start_ms: u64, - local_value: f64, - seq: u64, - rate: f64, - ) -> Vec { + /// Handle an edge's periodic rate report: record the rate and answer that + /// SAME edge with its freshly-computed coordinated-sampling grant. + /// Stale-epoch and unknown-edge reports are dropped. + pub fn on_report(&mut self, edge_id: &str, window_start_ms: u64, rate: f64) -> Vec { if window_start_ms != self.window_start_ms { // Could be a future epoch we haven't advanced to yet — advance on // strictly-greater, drop on stale. @@ -233,156 +217,29 @@ impl Monitor { let Some(edge) = self.edges.get_mut(edge_id) else { return Vec::new(); // unknown edge — registration precedes reports }; - if edge.seen_seq && seq <= edge.last_seq { - return Vec::new(); // idempotent re-delivery - } - edge.last_seq = seq; - edge.seen_seq = true; - // Monotone within an epoch: never let a known value go backwards. - if local_value > edge.known_value { - edge.known_value = local_value; - } - // Track the edge's latest observed per-window rate (ignore non-positive - // = unknown), which drives the coordinated sampling allocation. + // Ignore non-positive = unknown; keeps the last known-good rate. if rate > 0.0 { edge.rate = rate; } - self.rebroadcast() - } - - /// An edge left (stream closed). Fold its last-known value into departed mass - /// so the estimate keeps it, drop it from membership, and re-grant. - pub fn on_leave(&mut self, edge_id: &str) -> Vec { - if let Some(e) = self.edges.remove(edge_id) { - self.departed_mass += e.known_value; - } - self.rebroadcast() - } - - /// Coordinated update-sampling allocation for the current edge set, keyed by - /// edge id. Returns `p_i ∈ (0,1]` per edge, set by the whole-sketch ε-floor - /// so each edge's sampling noise stays within the CDM band. - /// - /// The allocation is the per-edge CDM-threshold coupling floor - /// `p_i = 1/(1 + ε²·rate_i)` (see `epsilon_sample_floor`), NOT the earlier - /// per-key `√(f_i/rate_i)` KKT water-filling — that allocation has been - /// retired (see the note in the function body for why sketch sampling is - /// governed by the whole-sketch ε-floor rather than a per-key frequency - /// split). `rate_i` is each edge's observed items/window, used directly as - /// the mass whose L2 contribution the floor keeps within ε. With <2 edges or - /// all rates unknown the allocation degenerates to `p=1` everywhere (no - /// sampling). - fn allocate_p(&self) -> HashMap { - // 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 / 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() - } - - /// Sampling margin `m_sa` for the safe-fire rule (derivations §10): - /// `Ĝ + m_sa + m_cdm ≥ (1−ε)τ`. When an edge's readout comes from a sampled - /// sketch, `known_value` is a two-sided Horvitz–Thompson estimate, not an - /// exact lower bound, so the coordinator can under-estimate the global by the - /// sampling noise and fire a real crossing late. `m_sa` is the aggregate - /// 1-σ sampling standard deviation `√(Σ_i kv_i·(1−p_i)/p_i)`, with `p_i` the - /// coordinator's own allocated floor `1/(1+ε²·rate_i)` — the same `p` it - /// grants (self-consistent with its sampling model). It is **0 whenever no - /// edge is sampled** (`rate=0 ⇒ p=1`), so the unsampled path is unchanged. - /// - /// This is the 1-σ heuristic §4.2/§6.2 use, not the high-probability Bernstein - /// margin — a theorem-grade alert would use `z_δ·` this. It is deliberately - /// conservative for a Sum monitor (whose `local_value` is exact and carries no - /// sampling noise even if the coordinator allocated `p<1`): the extra margin - /// only fires earlier, never later, which is the safe direction for a - /// threshold alert (a false-early beat a missed crossing). - fn sampling_margin(&self) -> f64 { - // Mirror the grant policy exactly: `rebroadcast` allocates p<1 only for - // ≥2 edges (a single edge is never sampled — `allocate_p` returns p=1), - // so a lone edge's readout is exact and carries no sampling margin. - if self.edges.len() < 2 { - return 0.0; - } - let var: f64 = self - .edges - .values() - .map(|e| { - if e.rate <= 0.0 { - return 0.0; - } - let p = epsilon_sample_floor(self.cfg.epsilon, e.rate); - if p >= 1.0 { - return 0.0; - } - // HT variance of a 1/p-weighted Bernoulli readout of mass kv_i. - e.known_value.max(0.0) * (1.0 - p) / p - }) - .sum(); - var.sqrt() - } - - /// Recompute Δ; fire the alert once if within ε (plus the sampling margin), - /// otherwise advance the round and emit a fresh slack grant to every edge. - fn rebroadcast(&mut self) -> Vec { - if self.alerted || self.edges.is_empty() { - return Vec::new(); - } - let est = self.global_estimate(); - // Safe-fire rule (§10): fold the sampling margin into the band so a - // sampling-under-estimated global still fires within [(1−ε)τ, τ). m_cdm - // is already realized by the slack countdown (Ĝ = Σ known_value is a - // lower bound each edge maintains within its granted slack), so only the - // sampling term is added here. - if self.gap() <= self.cfg.epsilon * self.cfg.tau + self.sampling_margin() { - self.alerted = true; - return vec![Action::Alert { - global_estimate: est, - tau: self.cfg.tau, - window_start_ms: self.window_start_ms, - }]; - } self.round += 1; - let slack = self.slack(); - let round = self.round; - let ws = self.window_start_ms; - // With a single edge there is no rate vector to coordinate over → p=1. - let p_by_edge = if self.edges.len() >= 2 { - self.allocate_p() + let p = if edge.rate > 0.0 { + epsilon_sample_floor(self.cfg.epsilon, edge.rate) } else { - HashMap::new() + 1.0 }; - self.edges - .keys() - .map(|edge_id| Action::Grant { - edge_id: edge_id.clone(), - round, - local_slack: slack, - window_start_ms: ws, - sample_p: p_by_edge.get(edge_id).copied().unwrap_or(1.0), - }) - .collect() + vec![Action::Grant { + edge_id: edge_id.to_string(), + round: self.round, + local_slack: 0.0, + window_start_ms: self.window_start_ms, + sample_p: p, + }] + } + + /// An edge left (stream closed). Just drops it from membership — there is + /// no mass/estimate to fold anywhere anymore. + pub fn on_leave(&mut self, edge_id: &str) { + self.edges.remove(edge_id); } } @@ -390,12 +247,12 @@ impl Monitor { mod tests { use super::*; - fn cfg(tau: f64) -> MonitorConfig { + fn cfg(epsilon: f64) -> MonitorConfig { MonitorConfig { agg_id: 1, key: Vec::new(), - tau, - epsilon: 0.05, + tau: 0.0, + epsilon, window_ms: 60_000, ..Default::default() } @@ -407,13 +264,6 @@ mod tests { .find(|a| matches!(a, Action::Grant { edge_id, .. } if edge_id == edge)) } - fn slack_in(actions: &[Action], edge: &str) -> f64 { - match grant_for(actions, edge) { - Some(Action::Grant { local_slack, .. }) => *local_slack, - _ => panic!("no grant for {edge}"), - } - } - fn sample_p_in(actions: &[Action], edge: &str) -> f64 { match grant_for(actions, edge) { Some(Action::Grant { sample_p, .. }) => *sample_p, @@ -422,243 +272,117 @@ mod tests { } #[test] - fn registration_grants_initial_slack() { - let mut m = Monitor::new(cfg(100.0)); + fn registration_grants_nothing() { + // Unlike the retired countdown, registering does not perturb any + // edge's grant — an edge stays unsampled (p=1, the edge-side default) + // until it actually reports a rate. + let mut m = Monitor::new(cfg(0.05)); let a = m.on_register("e1", 60_000, 0); - // Δ=100, k=1 → slack=50. - assert_eq!(slack_in(&a, "e1"), 50.0); - assert_eq!(m.round(), 1); + assert!(a.is_empty()); + assert_eq!(m.edge_count(), 1); + assert_eq!(m.round(), 0); } #[test] fn alignment_guard_rejects_mismatched_window() { - let mut m = Monitor::new(cfg(100.0)); + let mut m = Monitor::new(cfg(0.05)); let a = m.on_register("e1", 30_000, 0); // wrong window size assert!(a.is_empty()); assert_eq!(m.edge_count(), 0); } #[test] - fn slack_shrinks_as_estimate_climbs() { - let mut m = Monitor::new(cfg(100.0)); - m.on_register("e1", 60_000, 0); // slack 50 - // e1 reports 50 → estimate=50, Δ=50, slack=25. - let a = m.on_report("e1", 0, 50.0, 1, 0.0); - assert_eq!(slack_in(&a, "e1"), 25.0); - // reports 75 → estimate=75, Δ=25, slack=12.5. - let a = m.on_report("e1", 0, 75.0, 2, 0.0); - assert_eq!(slack_in(&a, "e1"), 12.5); - } - - #[test] - fn fires_alert_within_epsilon() { - let mut m = Monitor::new(cfg(100.0)); // epsilon 0.05 → alert when Δ ≤ 5 - m.on_register("e1", 60_000, 0); - assert!(matches!( - m.on_report("e1", 0, 50.0, 1, 0.0).as_slice(), - [Action::Grant { .. }] - )); - assert!(matches!( - m.on_report("e1", 0, 90.0, 2, 0.0).as_slice(), - [Action::Grant { .. }] - )); - // estimate 96 → Δ=4 ≤ 5 → alert. - let a = m.on_report("e1", 0, 96.0, 3, 0.0); - match a.as_slice() { - [Action::Alert { - global_estimate, - tau, - .. - }] => { - assert_eq!(*global_estimate, 96.0); - assert_eq!(*tau, 100.0); - } - other => panic!("expected alert, got {other:?}"), - } - // Further reports do not re-fire. - assert!(m.on_report("e1", 0, 200.0, 4, 0.0).is_empty()); - } - - #[test] - fn stays_silent_below_tau() { - let mut m = Monitor::new(cfg(100.0)); + fn report_grants_only_the_reporting_edge() { + // Unlike the retired countdown (which re-broadcast to every edge on + // any report, since slack depended on the full edge set), a report + // now answers ONLY the reporting edge — the sampling law is per-edge. + let mut m = Monitor::new(cfg(0.05)); m.on_register("e1", 60_000, 0); m.on_register("e2", 60_000, 0); - // Both report modestly; estimate stays well below τ → only grants, never alert. - for seq in 1..=5 { - let a = m.on_report("e1", 0, seq as f64 * 2.0, seq, 0.0); - assert!(a.iter().all(|x| matches!(x, Action::Grant { .. }))); - let b = m.on_report("e2", 0, seq as f64 * 2.0, seq, 0.0); - assert!(b.iter().all(|x| matches!(x, Action::Grant { .. }))); - } - assert!(m.global_estimate() < 100.0); + let a = m.on_report("e1", 0, 100_000.0); + assert_eq!(a.len(), 1, "report should grant only the reporting edge"); + assert!(grant_for(&a, "e1").is_some()); + assert!(grant_for(&a, "e2").is_none()); } #[test] - fn multi_edge_grants_all_and_uncertainty_bounded() { - let mut m = Monitor::new(cfg(120.0)); + fn sample_p_matches_epsilon_floor_of_reported_rate() { + let mut m = Monitor::new(cfg(0.05)); m.on_register("e1", 60_000, 0); - let a = m.on_register("e2", 60_000, 0); - m.on_register("e3", 60_000, 0); - // After 3 edges, a register re-grants ALL three; Δ=120, k=3 → slack=20. - let a3 = m.on_report("e1", 0, 0.0, 1, 0.0); - assert_eq!(a3.len(), 3, "re-grant should reach all edges"); - let _ = a; - // Safety invariant: Σ slack = k·slack = Δ/2 at all times. - let est = m.global_estimate(); - let delta = 120.0 - est; - let total_slack = 3.0 * slack_in(&a3, "e1"); - assert!((total_slack - delta / 2.0).abs() < 1e-9); + let a = m.on_report("e1", 0, 100_000.0); + let want = epsilon_sample_floor(0.05, 100_000.0); + assert!((sample_p_in(&a, "e1") - want).abs() < 1e-12); } #[test] - fn sampling_margin_fires_alert_earlier() { - // τ=100, ε=0.05 → unsampled band fires at Δ≤5 (estimate ≥ 95). Sampling is - // granted only with ≥2 edges, so this uses two. Each sampled edge has - // rate=400 → p = 1/(1+0.05²·400) = 0.5; at kv=45 each, the aggregate - // margin m_sa = √(45·(1−0.5)/0.5 + 45·(1−0.5)/0.5) = √90 ≈ 9.49, loosening - // the fire threshold to Δ ≤ 5 + 9.49 → estimate ≥ ~85.5. - - // Unsampled control (rate=0): estimate 90 is below the 95 fire point → grant. - let mut m = Monitor::new(cfg(100.0)); + fn single_edge_is_sampled_like_any_other() { + // The retired countdown special-cased "<2 edges ⇒ p=1" purely to keep + // its alert-fire safety proof correct under sampling noise. With + // alerting gone, a lone high-rate edge is sampled exactly like it + // would be in a larger edge set — the ε-floor law never depended on + // edge count. + let mut m = Monitor::new(cfg(0.05)); m.on_register("e1", 60_000, 0); - m.on_register("e2", 60_000, 0); - m.on_report("e1", 0, 45.0, 1, 0.0); - assert!( - matches!( - m.on_report("e2", 0, 45.0, 1, 0.0).as_slice(), - [Action::Grant { .. }, ..] - ), - "unsampled: estimate 90 must not alert (m_sa=0)" - ); - - // Sampled (rate=400): same estimate 90, but the margin fires the alert. - let mut ms = Monitor::new(cfg(100.0)); - ms.on_register("e1", 60_000, 0); - ms.on_register("e2", 60_000, 0); - ms.on_report("e1", 0, 45.0, 1, 400.0); - let a = ms.on_report("e2", 0, 45.0, 1, 400.0); - assert!( - matches!(a.as_slice(), [Action::Alert { .. }]), - "sampled: m_sa≈9.49 must fire the alert at estimate 90, got {a:?}" - ); + let a = m.on_report("e1", 0, 100_000.0); + let p = sample_p_in(&a, "e1"); + assert!(p < 1.0, "a lone high-rate edge should still be sampled, got p={p}"); } #[test] - fn sampling_margin_zero_without_sampling() { - // rate=0 (nothing to sample) ⇒ 0 margin, even with ≥2 edges. - let mut m = Monitor::new(cfg(100.0)); + fn unknown_rate_yields_unsampled_grant() { + let mut m = Monitor::new(cfg(0.05)); m.on_register("e1", 60_000, 0); - m.on_register("e2", 60_000, 0); - m.on_report("e1", 0, 40.0, 1, 0.0); - m.on_report("e2", 0, 40.0, 1, 0.0); - assert_eq!(m.sampling_margin(), 0.0, "rate=0 edges add no margin"); - - // A lone edge is never sampled (grant policy) ⇒ 0 margin even at high rate. - let mut solo = Monitor::new(cfg(100.0)); - solo.on_register("e1", 60_000, 0); - solo.on_report("e1", 0, 40.0, 1, 10_000.0); - assert_eq!( - solo.sampling_margin(), - 0.0, - "single edge is unsampled → no margin" - ); + let a = m.on_report("e1", 0, 0.0); // rate unknown/non-positive + assert_eq!(sample_p_in(&a, "e1"), 1.0); } #[test] - fn seq_dedup_ignores_retransmits() { - let mut m = Monitor::new(cfg(100.0)); - m.on_register("e1", 60_000, 0); - m.on_report("e1", 0, 40.0, 1, 0.0); - let before = m.global_estimate(); - // Same seq re-delivered → ignored. - let a = m.on_report("e1", 0, 40.0, 1, 0.0); - assert!(a.is_empty()); - assert_eq!(m.global_estimate(), before); + fn report_before_register_is_dropped() { + let mut m = Monitor::new(cfg(0.05)); + let a = m.on_report("e1", 0, 100_000.0); + assert!(a.is_empty(), "unregistered edge's report must be dropped"); } #[test] fn stale_epoch_report_dropped() { - let mut m = Monitor::new(cfg(100.0)); + let mut m = Monitor::new(cfg(0.05)); m.on_register("e1", 60_000, 120_000); // epoch starts at 120_000 assert_eq!(m.window_start_ms(), 120_000); - // A report tagged with the previous epoch is dropped. - let a = m.on_report("e1", 60_000, 99.0, 1, 0.0); + let a = m.on_report("e1", 60_000, 100_000.0); // previous epoch assert!(a.is_empty()); - assert_eq!(m.global_estimate(), 0.0); - } - - #[test] - fn join_and_leave_mass_accounting() { - let mut m = Monitor::new(cfg(100.0)); - m.on_register("e1", 60_000, 0); - m.on_register("e2", 60_000, 0); - m.on_report("e1", 0, 30.0, 1, 0.0); - m.on_report("e2", 0, 20.0, 1, 0.0); - assert_eq!(m.global_estimate(), 50.0); - // e2 leaves: its 20 stays in the estimate via departed_mass. - m.on_leave("e2"); - assert_eq!(m.global_estimate(), 50.0); - assert_eq!(m.edge_count(), 1); } #[test] - fn single_edge_grants_no_sampling() { - // With one edge there is no rate vector to coordinate over → p=1. - let mut m = Monitor::new(cfg(1000.0)); + fn epoch_advance_resets_rate() { + let mut m = Monitor::new(cfg(0.05)); m.on_register("e1", 60_000, 0); - let a = m.on_report("e1", 0, 10.0, 1, 50_000.0); + m.on_report("e1", 0, 100_000.0); + // A report for the next epoch advances and resets the rate: an + // unknown (0) rate on the fresh epoch yields an unsampled grant. + let a = m.on_report("e1", 60_000, 0.0); + assert_eq!(m.window_start_ms(), 60_000); assert_eq!(sample_p_in(&a, "e1"), 1.0); } #[test] - 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 + fn leave_drops_membership() { + let mut m = Monitor::new(cfg(0.05)); m.on_register("e1", 60_000, 0); m.on_register("e2", 60_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"); - let eps = m.cfg.epsilon; - 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 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); - 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, same rate - let p_e1 = sample_p_in(&a, "e1"); - let p_e2 = sample_p_in(&a, "e2"); - 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); + assert_eq!(m.edge_count(), 2); + m.on_leave("e2"); + assert_eq!(m.edge_count(), 1); } #[test] - fn epoch_advance_resets_estimate() { - let mut m = Monitor::new(cfg(100.0)); + fn rate_stays_last_known_good_on_non_positive_report() { + let mut m = Monitor::new(cfg(0.05)); m.on_register("e1", 60_000, 0); - m.on_report("e1", 0, 80.0, 1, 0.0); - assert_eq!(m.global_estimate(), 80.0); - // A report for the next epoch advances and resets. - m.on_report("e1", 60_000, 5.0, 2, 0.0); - assert_eq!(m.window_start_ms(), 60_000); - assert_eq!(m.global_estimate(), 5.0); + m.on_report("e1", 0, 100_000.0); + // A subsequent non-positive rate report doesn't clobber the last + // known-good rate. + let a = m.on_report("e1", 0, 0.0); + let want = epsilon_sample_floor(0.05, 100_000.0); + assert!((sample_p_in(&a, "e1") - want).abs() < 1e-12); } } diff --git a/data_plane/src/monitor/mod.rs b/data_plane/src/monitor/mod.rs index c3c5b6b64..6a15d6dd4 100644 --- a/data_plane/src/monitor/mod.rs +++ b/data_plane/src/monitor/mod.rs @@ -1,43 +1,39 @@ -//! Continuous distributed monitoring (CDM) coordinator — the backend side of -//! Discipline B from ASAPCollector +//! Coordinated update-sampling — the backend side of what used to be called +//! "Discipline B" in ASAPCollector //! docs/continuous-monitoring-tumbling-cost-analysis.md. //! -//! Edges run the slack-countdown protocol locally and stream reports over the -//! bidirectional `MonitorService` gRPC (see `monitor.proto`); this module hosts -//! that service, maintains the per-monitor global state machine, and fires an -//! alert through the control-plane violation sink when a global aggregate -//! crosses its threshold τ. +//! Global-threshold ALERTING (the CMY slack-countdown: register → grant +//! `(slack, sample_p)` → countdown → report → alert) is RETIRED as of the +//! 2026-07 insert-time-GOS redesign — see the `coordinator` module doc +//! comment for the full rationale. An edge/collector must never be the thing +//! that fires an alert; that decision belongs entirely to query-time reads of +//! the backend's synced sketch state. +//! +//! What's left: edges report their per-window observation RATE on their own +//! periodic cadence (decoupled from any value/slack threshold), and this +//! module answers each one with its coordinated-sampling grant — the +//! whole-sketch ε-floor `p_i = 1/(1+ε²·rate_i)`, a pure per-edge function with +//! no cross-edge coordination or global state involved. //! //! Layout: -//! - [`coordinator`] — the pure, unit-tested CMY round/slack state machine. +//! - [`coordinator`] — the pure, unit-tested per-edge sampling state machine. //! - [`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). //! -//! # Two orthogonal axes — DON'T conflate them -//! -//! 1. **What to monitor / alert on** (the `functional`): `sum` or `cms_point` -//! (a declared point `f(x)`). This sets the THRESHOLD `g vs τ` and what -//! `known_value` means. Identity is `(agg_id,key)` — `cms_point` carries a -//! key; `sum` is 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). +//! **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 any monitored functional/key — because +//! the sampling protects a SKETCH, whose accuracy is bounded by the stream +//! norm (rate/L2), not by any single key. Applies to additive sketches — +//! Count-Min, Count-Sketch, DDSketch, KLL, Sum — but NOT HLL (sampling biases +//! cardinality). -pub mod alert; pub mod coordinator; pub mod epoch; pub mod sampling_alloc; pub mod server; -pub use alert::{global_threshold_violation, AlertSink}; pub use coordinator::{Action, Functional, Monitor, MonitorConfig}; pub use sampling_alloc::epsilon_sample_floor; pub use server::{MonitorCoordinator, MonitorServiceImpl}; diff --git a/data_plane/src/monitor/server.rs b/data_plane/src/monitor/server.rs index 6fd403666..1fd5a6823 100644 --- a/data_plane/src/monitor/server.rs +++ b/data_plane/src/monitor/server.rs @@ -2,8 +2,9 @@ //! `Monitor` stream and multiplexes all of its monitors over it. The server //! reads `EdgeToCoord` (register / report) messages, drives the per-monitor //! [`Monitor`] state machine, and pushes the resulting `CoordToEdge` -//! (slack-grant) messages back to the addressed edge's stream. Global-threshold -//! alerts go out-of-band through the control-plane violation sink. +//! (coordinated-sampling grant) back to the addressed edge's stream. +//! Global-threshold alerting is retired (see `coordinator` module docs) — this +//! server never fires one. use std::collections::HashMap; use std::pin::Pin; @@ -21,7 +22,6 @@ use asap_otel_proto::monitor::v1::{ CoordToEdge, EdgeToCoord, SlackGrant, }; -use super::alert::{global_threshold_violation, AlertSink}; use super::coordinator::{Action, Monitor, MonitorConfig}; type MonKey = (u64, Vec); @@ -31,20 +31,18 @@ type EdgeTx = mpsc::Sender>; pub struct MonitorCoordinator { /// Live per-monitor scalar state machines, created lazily from `cfgs`. monitors: Mutex>, - /// Monitor specs (τ, ε, window) from the streaming-config `monitors:` - /// section — the authoritative source of τ. Behind an `RwLock` so the - /// control plane's hot-reload (`reconfigure`) can add/update/remove monitors - /// on a live coordinator without a process restart. Held only for brief, - /// non-`await` critical sections, so a `std` lock is safe in async code. + /// Monitor specs (ε, window) from the streaming-config `monitors:` + /// section. Behind an `RwLock` so the control plane's hot-reload + /// (`reconfigure`) can add/update/remove monitors on a live coordinator + /// without a process restart. Held only for brief, non-`await` critical + /// sections, so a `std` lock is safe in async code. cfgs: std::sync::RwLock>, /// Outbound stream sender per connected edge. edges: Mutex>, - /// Alert egress (control-plane violation sink). - alert_sink: AlertSink, } impl MonitorCoordinator { - pub fn new(cfgs: Vec, alert_sink: AlertSink) -> Arc { + pub fn new(cfgs: Vec) -> Arc { let cfgs = cfgs .into_iter() .map(|c| ((c.agg_id, c.key.clone()), c)) @@ -53,7 +51,6 @@ impl MonitorCoordinator { monitors: Mutex::new(HashMap::new()), cfgs: std::sync::RwLock::new(cfgs), edges: Mutex::new(HashMap::new()), - alert_sink, }) } @@ -62,6 +59,17 @@ impl MonitorCoordinator { self.cfgs.read().unwrap().len() } + /// The coordinated-sampling grant currently computed for `edge_id` under + /// monitor `(agg_id, key)`, or `None` if the monitor doesn't exist yet or + /// that edge hasn't reported a rate (test/observability — see + /// `Monitor::sample_p_for_edge`). + pub async fn granted_sample_p(&self, agg_id: u64, key: &[u8], edge_id: &str) -> Option { + let monitors = self.monitors.lock().await; + monitors + .get(&(agg_id, key.to_vec()))? + .sample_p_for_edge(edge_id) + } + /// Hot-reload the monitor set from a freshly-pushed streaming-config. /// /// The coordinator originally read `monitors:` only at boot; a monitor the @@ -76,7 +84,7 @@ impl MonitorCoordinator { /// spec — the edge re-registers every epoch, so this self-heals within /// one window; /// * **removed** specs drop both the spec and any live state. - /// Unchanged monitors keep their in-flight slack-countdown state untouched. + /// Unchanged monitors keep their in-flight per-edge rate state untouched. /// /// Returns `(added, changed, removed)` counts for observability. Idempotent: /// re-applying the same specs is a no-op that returns `(0, 0, 0)`. @@ -129,14 +137,6 @@ impl MonitorCoordinator { (added, changed, removed) } - fn monitor_id(agg_id: u64, key: &[u8]) -> String { - if key.is_empty() { - format!("agg:{agg_id}/sum") - } else { - format!("agg:{agg_id}/{}", String::from_utf8_lossy(key)) - } - } - /// Apply a register and return the resulting actions, or None if no monitor /// is configured for (agg_id, key). async fn apply_register( @@ -175,57 +175,43 @@ impl MonitorCoordinator { key: Vec, edge_id: &str, window_start_ms: u64, - local_value: f64, - seq: u64, rate: f64, ) -> Option> { let mk = (agg_id, key); let mut monitors = self.monitors.lock().await; let mon = monitors.get_mut(&mk)?; - Some(mon.on_report(edge_id, window_start_ms, local_value, seq, rate)) + Some(mon.on_report(edge_id, window_start_ms, rate)) } - /// Dispatch coordinator actions: grants to the addressed edge's stream, - /// alerts to the violation sink. + /// Dispatch coordinator actions: grants to the addressed edge's stream. + /// No alert path — this coordinator no longer makes alerting decisions + /// (see `coordinator` module docs). async fn dispatch(&self, agg_id: u64, key: &[u8], actions: Vec) { for action in actions { - match action { - Action::Grant { - edge_id, + let Action::Grant { + edge_id, + round, + local_slack, + window_start_ms, + sample_p, + } = action; + let msg = CoordToEdge { + msg: Some(coord_to_edge::Msg::Grant(SlackGrant { + agg_id, + key: key.to_vec(), round, local_slack, window_start_ms, sample_p, - } => { - let msg = CoordToEdge { - msg: Some(coord_to_edge::Msg::Grant(SlackGrant { - agg_id, - key: key.to_vec(), - round, - local_slack, - window_start_ms, - sample_p, - })), - }; - let tx = { - let edges = self.edges.lock().await; - edges.get(&edge_id).cloned() - }; - if let Some(tx) = tx { - if tx.send(Ok(msg)).await.is_err() { - debug!(edge = %edge_id, "grant send failed (edge gone)"); - } - } - } - Action::Alert { - global_estimate, - tau, - window_start_ms, - } => { - let id = Self::monitor_id(agg_id, key); - info!(monitor = %id, global_estimate, tau, window_start_ms, - "CDM global threshold crossed — firing alert"); - (self.alert_sink)(global_threshold_violation(id, global_estimate, tau)); + })), + }; + let tx = { + let edges = self.edges.lock().await; + edges.get(&edge_id).cloned() + }; + if let Some(tx) = tx { + if tx.send(Ok(msg)).await.is_err() { + debug!(edge = %edge_id, "grant send failed (edge gone)"); } } } @@ -266,17 +252,12 @@ impl MonitorCoordinator { } } Some(edge_to_coord::Msg::Report(rep)) => { - // Every report drives the scalar CMY slack-countdown Monitor. + // A rate report: reply to this edge alone with its freshly + // computed coordinated-sampling grant (rep.local_value / seq + // are no longer consulted — no countdown left to feed them + // into; they still ride the wire message for compatibility). if let Some(actions) = self - .apply_report( - rep.agg_id, - rep.key.clone(), - &rep.edge_id, - rep.window_start_ms, - rep.local_value, - rep.seq, - rep.rate, - ) + .apply_report(rep.agg_id, rep.key.clone(), &rep.edge_id, rep.window_start_ms, rep.rate) .await { self.dispatch(rep.agg_id, &rep.key, actions).await; @@ -286,22 +267,14 @@ impl MonitorCoordinator { } } - /// An edge stream closed: drop its sender and fold its last-known value into - /// every monitor's departed mass, then route any re-grants. Actions are - /// collected under the monitors lock and dispatched after releasing it, so - /// no send is awaited while the lock is held. + /// An edge stream closed: drop its sender and its membership in every + /// monitor it was part of. No re-grant fan-out needed — each edge's grant + /// is independent, so one edge leaving never perturbs another's. async fn handle_disconnect(&self, edge_id: &str) { self.edges.lock().await.remove(edge_id); - let pending: Vec<(u64, Vec, Vec)> = { - let mut monitors = self.monitors.lock().await; - monitors - .iter_mut() - .map(|(mk, mon)| (mk.0, mk.1.clone(), mon.on_leave(edge_id))) - .filter(|(_, _, actions)| !actions.is_empty()) - .collect() - }; - for (agg_id, key, actions) in pending { - self.dispatch(agg_id, &key, actions).await; + let mut monitors = self.monitors.lock().await; + for mon in monitors.values_mut() { + mon.on_leave(edge_id); } } } @@ -358,13 +331,7 @@ impl MonitorService for MonitorServiceImpl { #[cfg(test)] mod reconfigure_tests { use super::MonitorCoordinator; - use crate::monitor::alert::AlertSink; use crate::monitor::coordinator::MonitorConfig; - use std::sync::Arc; - - fn sink() -> AlertSink { - Arc::new(|_v| {}) - } fn cfg(agg_id: u64, key: &str, tau: f64) -> MonitorConfig { MonitorConfig { @@ -380,7 +347,7 @@ mod reconfigure_tests { #[tokio::test] async fn reconfigure_adds_changes_and_removes_specs() { // Boot with one sum monitor. - let coord = MonitorCoordinator::new(vec![cfg(1, "", 100.0)], sink()); + let coord = MonitorCoordinator::new(vec![cfg(1, "", 100.0)]); assert_eq!(coord.monitor_count(), 1); // Add a cms_point monitor (agg 2, key s0) and change agg 1's τ; agg 1's @@ -408,7 +375,7 @@ mod reconfigure_tests { #[tokio::test] async fn reconfigure_evicts_live_state_for_changed_monitor() { - let coord = MonitorCoordinator::new(vec![cfg(7, "s0", 5000.0)], sink()); + let coord = MonitorCoordinator::new(vec![cfg(7, "s0", 5000.0)]); // A register materializes live Monitor state for (7, s0). let actions = coord .apply_register(7, b"s0".to_vec(), "edge-a", 15_000, 15_000) diff --git a/data_plane/tests/monitor_grpc.rs b/data_plane/tests/monitor_grpc.rs index 03c92467a..91eabc2cc 100644 --- a/data_plane/tests/monitor_grpc.rs +++ b/data_plane/tests/monitor_grpc.rs @@ -1,14 +1,14 @@ -//! End-to-end transport test for the CDM coordinator: a real tonic -//! `MonitorService` server on a loopback port, driven by a real +//! End-to-end transport test for the coordinated-sampling coordinator: a real +//! tonic `MonitorService` server on a loopback port, driven by a real //! `MonitorServiceClient` over a bidi stream. Validates that registration + -//! reports flow over gRPC, grants come back, and the global-threshold alert -//! fires exactly once within ε when the scripted reports climb past τ — and -//! that staying below τ produces only grants, never an alert. +//! rate reports flow over gRPC and the coordinator ships back a computed +//! `SlackGrant.sample_p`. //! -//! The scripted "edge" here sends a fixed report sequence rather than -//! re-implementing the Go slack-countdown engine (which is unit-tested on its -//! own); the point of this test is the wire path: server ⇄ coordinator ⇄ -//! dispatch ⇄ alert sink. +//! Global-threshold alerting is retired (see `data_plane::monitor` module +//! docs) — there is no more alert path to test here. The scripted "edge" here +//! sends a fixed report sequence rather than re-implementing the Go engine's +//! reportEveryN cadence (which is unit-tested on its own); the point of this +//! test is the wire path: server ⇄ coordinator ⇄ dispatch ⇄ grant. use std::sync::{Arc, Mutex}; use std::time::Duration; @@ -22,15 +22,10 @@ use asap_otel_proto::monitor::v1::{ edge_to_coord, monitor_service_client::MonitorServiceClient, EdgeToCoord, MonitorRegister, MonitorReport, }; -use control_plane::monitor::{Violation, ViolationKind}; -use data_plane::monitor::{AlertSink, MonitorConfig, MonitorCoordinator, MonitorServiceImpl}; +use data_plane::monitor::{MonitorConfig, MonitorCoordinator, MonitorServiceImpl}; -async fn start_server(cfgs: Vec) -> (String, Arc>>) { - let alerts = Arc::new(Mutex::new(Vec::::new())); - let captured = alerts.clone(); - let sink: AlertSink = Arc::new(move |v| captured.lock().unwrap().push(v)); - - let coord = MonitorCoordinator::new(cfgs, sink); +async fn start_server(cfgs: Vec) -> String { + let coord = MonitorCoordinator::new(cfgs); let svc = MonitorServiceImpl::new(coord).into_server(); let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); @@ -43,179 +38,94 @@ async fn start_server(cfgs: Vec) -> (String, Arc EdgeToCoord { +fn register_edge(edge_id: &str) -> EdgeToCoord { EdgeToCoord { - msg: Some(edge_to_coord::Msg::Report(MonitorReport { - edge_id: "e1".into(), + msg: Some(edge_to_coord::Msg::Reg(MonitorRegister { + edge_id: edge_id.into(), agg_id: 1, key: vec![], + epoch_window_ms: 60_000, window_start_ms: 0, - local_value: value, - round: 0, - seq, - rate: 0.0, - sketch: Vec::new(), })), } } -fn register() -> EdgeToCoord { +fn report_edge(edge_id: &str, seq: u64, rate: f64) -> EdgeToCoord { EdgeToCoord { - msg: Some(edge_to_coord::Msg::Reg(MonitorRegister { - edge_id: "e1".into(), + msg: Some(edge_to_coord::Msg::Report(MonitorReport { + edge_id: edge_id.into(), agg_id: 1, key: vec![], - epoch_window_ms: 60_000, window_start_ms: 0, + local_value: 0.0, // vestigial (alerting retired) — unused by the coordinator + round: 0, + seq, + rate, })), } } -fn sum_cfg(tau: f64) -> MonitorConfig { +fn sum_cfg(epsilon: f64) -> MonitorConfig { MonitorConfig { agg_id: 1, key: vec![], - tau, - epsilon: 0.05, + tau: 0.0, // vestigial (alerting retired) + epsilon, window_ms: 60_000, ..Default::default() } } -#[tokio::test] -async fn fires_alert_once_past_tau() { - let (url, alerts) = start_server(vec![sum_cfg(100.0)]).await; - let mut client = MonitorServiceClient::connect(url).await.unwrap(); - - let (tx, rx) = mpsc::channel(16); - let mut inbound = client - .monitor(ReceiverStream::new(rx)) - .await - .unwrap() - .into_inner(); - // Drain grants in the background so the stream stays healthy. - tokio::spawn(async move { while let Some(Ok(_grant)) = inbound.next().await {} }); - - tx.send(register()).await.unwrap(); - // Climb past τ=100. Alert fires when estimate ≥ (1-ε)τ = 95. - for (i, v) in [20.0, 50.0, 80.0, 96.0].into_iter().enumerate() { - tx.send(report(v, i as u64 + 1)).await.unwrap(); - tokio::time::sleep(Duration::from_millis(20)).await; - } - tokio::time::sleep(Duration::from_millis(150)).await; - - let got = alerts.lock().unwrap(); - assert_eq!( - got.len(), - 1, - "expected exactly one alert, got {}", - got.len() - ); - assert_eq!(got[0].kind, ViolationKind::GlobalThresholdCrossed); - assert_eq!(got[0].threshold, 100.0); - assert!( - got[0].observed >= 95.0, - "alert fired at estimate {} which is below (1-ε)τ=95", - got[0].observed - ); -} - -#[tokio::test] -async fn stays_quiet_below_tau() { - let (url, alerts) = start_server(vec![sum_cfg(100.0)]).await; - let mut client = MonitorServiceClient::connect(url).await.unwrap(); +async fn connect_edge(url: &str) -> (mpsc::Sender, Arc>>) { + use asap_otel_proto::monitor::v1::coord_to_edge; - let (tx, rx) = mpsc::channel(16); + let mut client = MonitorServiceClient::connect(url.to_string()).await.unwrap(); + let (tx, rx) = mpsc::channel(32); let mut inbound = client .monitor(ReceiverStream::new(rx)) .await .unwrap() .into_inner(); - let grant_count = Arc::new(Mutex::new(0usize)); - let gc = grant_count.clone(); + let ps = Arc::new(Mutex::new(Vec::::new())); + let psc = ps.clone(); tokio::spawn(async move { - while let Some(Ok(_grant)) = inbound.next().await { - *gc.lock().unwrap() += 1; + while let Some(Ok(env)) = inbound.next().await { + if let Some(coord_to_edge::Msg::Grant(g)) = env.msg { + psc.lock().unwrap().push(g.sample_p); + } } }); - - tx.send(register()).await.unwrap(); - for (i, v) in [10.0, 20.0, 30.0].into_iter().enumerate() { - tx.send(report(v, i as u64 + 1)).await.unwrap(); - tokio::time::sleep(Duration::from_millis(20)).await; - } - tokio::time::sleep(Duration::from_millis(150)).await; - - assert!( - alerts.lock().unwrap().is_empty(), - "must not alert while below τ" - ); - assert!( - *grant_count.lock().unwrap() >= 1, - "edge should have received at least the initial grant" - ); + (tx, ps) } -fn register_edge(edge_id: &str) -> EdgeToCoord { - EdgeToCoord { - msg: Some(edge_to_coord::Msg::Reg(MonitorRegister { - edge_id: edge_id.into(), - agg_id: 1, - key: vec![], - epoch_window_ms: 60_000, - window_start_ms: 0, - })), - } -} +/// A single edge reporting a real rate gets a real (< 1.0) sample_p grant — +/// the retired countdown special-cased "<2 edges ⇒ p=1" purely for its +/// alert-fire safety proof; the ε-floor sampling law never depended on edge +/// count, so a lone edge is sampled exactly like any other now. +#[tokio::test] +async fn single_edge_gets_sampled_over_grpc() { + let url = start_server(vec![sum_cfg(0.05)]).await; + let (tx, p) = connect_edge(&url).await; -fn report_edge(edge_id: &str, value: f64, seq: u64, rate: f64) -> EdgeToCoord { - EdgeToCoord { - msg: Some(edge_to_coord::Msg::Report(MonitorReport { - edge_id: edge_id.into(), - agg_id: 1, - key: vec![], - window_start_ms: 0, - local_value: value, - round: 0, - seq, - rate, - sketch: Vec::new(), - })), - } + tx.send(register_edge("e1")).await.unwrap(); + tokio::time::sleep(Duration::from_millis(40)).await; + tx.send(report_edge("e1", 1, 100_000.0)).await.unwrap(); + tokio::time::sleep(Duration::from_millis(150)).await; + + let got = *p.lock().unwrap().last().expect("edge received a grant"); + assert!(got > 0.0 && got < 1.0, "expected a real sampling grant, got {got}"); } /// Live multi-edge coupling: two edges open real gRPC streams, report SKEWED /// rates, and the coordinator ships back differentiated `SlackGrant.sample_p` /// over the wire — the hot edge sampled harder (smaller p) than the quiet one, -/// both at/above the ε-derived coupling floor. τ is large so only grants flow. +/// both at/above the ε-derived coupling floor. #[tokio::test] async fn two_edges_get_differentiated_sample_p_over_grpc() { - use asap_otel_proto::monitor::v1::coord_to_edge; - - async fn connect_edge(url: &str) -> (mpsc::Sender, Arc>>) { - let mut client = MonitorServiceClient::connect(url.to_string()).await.unwrap(); - let (tx, rx) = mpsc::channel(32); - let mut inbound = client - .monitor(ReceiverStream::new(rx)) - .await - .unwrap() - .into_inner(); - let ps = Arc::new(Mutex::new(Vec::::new())); - let psc = ps.clone(); - tokio::spawn(async move { - while let Some(Ok(env)) = inbound.next().await { - if let Some(coord_to_edge::Msg::Grant(g)) = env.msg { - psc.lock().unwrap().push(g.sample_p); - } - } - }); - (tx, ps) - } - - let (url, _alerts) = start_server(vec![sum_cfg(1_000_000.0)]).await; + let url = start_server(vec![sum_cfg(0.05)]).await; let (tx_hot, p_hot) = connect_edge(&url).await; let (tx_quiet, p_quiet) = connect_edge(&url).await; @@ -223,11 +133,10 @@ async fn two_edges_get_differentiated_sample_p_over_grpc() { tx_quiet.send(register_edge("e2")).await.unwrap(); tokio::time::sleep(Duration::from_millis(60)).await; - // e1 hot (100k items/win), e2 quiet (1k/win); small local values stay far - // below τ so only grants (carrying sample_p) flow, never an alert. + // e1 hot (100k items/win), e2 quiet (1k/win). for seq in 1..=4 { - tx_hot.send(report_edge("e1", 1.0, seq, 100_000.0)).await.unwrap(); - tx_quiet.send(report_edge("e2", 1.0, seq, 1_000.0)).await.unwrap(); + tx_hot.send(report_edge("e1", seq, 100_000.0)).await.unwrap(); + tx_quiet.send(report_edge("e2", seq, 1_000.0)).await.unwrap(); tokio::time::sleep(Duration::from_millis(40)).await; } tokio::time::sleep(Duration::from_millis(250)).await; @@ -243,3 +152,24 @@ async fn two_edges_get_differentiated_sample_p_over_grpc() { assert!(hot > 0.0 && hot <= 1.0, "hot p out of range: {hot}"); assert!(quiet > 0.0 && quiet <= 1.0, "quiet p out of range: {quiet}"); } + +/// A report answers ONLY the reporting edge — unlike the retired countdown +/// (which re-broadcast to every edge on any report, since slack sizing +/// depended on the full edge set), a quiet edge that never reports should +/// never receive a grant it didn't ask for. +#[tokio::test] +async fn report_grants_only_the_reporting_edge_over_grpc() { + let url = start_server(vec![sum_cfg(0.05)]).await; + let (tx_a, p_a) = connect_edge(&url).await; + let (tx_b, p_b) = connect_edge(&url).await; + + tx_a.send(register_edge("e1")).await.unwrap(); + tx_b.send(register_edge("e2")).await.unwrap(); + tokio::time::sleep(Duration::from_millis(60)).await; + + tx_a.send(report_edge("e1", 1, 50_000.0)).await.unwrap(); + tokio::time::sleep(Duration::from_millis(150)).await; + + assert!(!p_a.lock().unwrap().is_empty(), "e1 should have received its own grant"); + assert!(p_b.lock().unwrap().is_empty(), "e2 must not receive a grant it never reported for"); +}