feat(monitor): hot-reload the CDM coordinator's monitor set (no restart)

data_plane/src/main.rs

@@ -612,7 +612,7 @@ async fn main() -> Result<()> {
             })
             .collect();
         if specs.is_empty() {
-            warn!("--enable-monitor-coordinator set but streaming-config has no `monitors:` — coordinator will accept streams but serve nothing");
+            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!(
@@ -623,6 +623,52 @@ async fn main() -> Result<()> {
             );
         });
         let coord = MonitorCoordinator::new(specs, sink);
+
+        // Hot-reload watcher: the coordinator reads `monitors:` once at boot, but
+        // the control plane pushes the real config slightly AFTER boot via the
+        // `/api/v1/streaming-config` POST (an ArcSwap in `hot_reload_config`).
+        // Without this, a monitor that arrives post-boot never reaches the
+        // coordinator and every edge registering for it is rejected as
+        // "unconfigured". Watch the ArcSwap and re-apply its `monitors:` to the
+        // live coordinator on each swap (cheap: an atomic load + pointer compare
+        // every 2s; `reconfigure` is a no-op unless the spec set actually
+        // changed). The same path covers controller-driven monitor add/remove.
+        {
+            let coord = coord.clone();
+            let hot = hot_reload_config.clone();
+            tokio::spawn(async move {
+                let mut last = hot.snapshot();
+                loop {
+                    tokio::time::sleep(std::time::Duration::from_secs(2)).await;
+                    let cur = hot.snapshot();
+                    if Arc::ptr_eq(&last, &cur) {
+                        continue;
+                    }
+                    last = cur.clone();
+                    let specs: Vec<MonitorConfig> = cur
+                        .monitors()
+                        .iter()
+                        .map(|m| MonitorConfig {
+                            agg_id: m.agg_id,
+                            key: m.key.clone().into_bytes(),
+                            tau: m.tau,
+                            epsilon: m.epsilon,
+                            window_ms: m.window_ms,
+                        })
+                        .collect();
+                    let (added, changed, removed) = coord.reconfigure(specs).await;
+                    if added + changed + removed > 0 {
+                        info!(
+                            added,
+                            changed,
+                            removed,
+                            "CDM monitor coordinator hot-reloaded monitors from pushed streaming-config"
+                        );
+                    }
+                }
+            });
+        }
+
         let svc = MonitorServiceImpl::new(coord).into_server();
         let port = args.monitor_grpc_port;
         info!("Starting CDM monitor coordinator gRPC on 0.0.0.0:{port}");

data_plane/src/monitor/coordinator.rs

@@ -34,7 +34,7 @@ use super::sampling_alloc::{allocate_sample_rates, epsilon_sample_floor};
 
 /// Static configuration for one monitor, sourced from the streaming-config
 /// `monitors:` section (τ authoritative here, not at the edge).
-#[derive(Clone, Debug)]
+#[derive(Clone, Debug, PartialEq)]
 pub struct MonitorConfig {
     pub agg_id: u64,
     pub key: Vec<u8>,

data_plane/src/monitor/server.rs

@@ -31,9 +31,12 @@ type EdgeTx = mpsc::Sender<Result<CoordToEdge, Status>>;
 pub struct MonitorCoordinator {
     /// Live per-monitor state machines, created lazily from `cfgs`.
     monitors: Mutex<HashMap<MonKey, Monitor>>,
-    /// Static monitor specs (τ, ε, window) from the streaming-config
-    /// `monitors:` section — the authoritative source of τ.
-    cfgs: HashMap<MonKey, MonitorConfig>,
+    /// 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.
+    cfgs: std::sync::RwLock<HashMap<MonKey, MonitorConfig>>,
     /// Outbound stream sender per connected edge.
     edges: Mutex<HashMap<String, EdgeTx>>,
     /// Alert egress (control-plane violation sink).
@@ -48,15 +51,82 @@ impl MonitorCoordinator {
             .collect();
         Arc::new(Self {
             monitors: Mutex::new(HashMap::new()),
-            cfgs,
+            cfgs: std::sync::RwLock::new(cfgs),
             edges: Mutex::new(HashMap::new()),
             alert_sink,
         })
     }
 
     /// Number of configured monitors (test/observability).
     pub fn monitor_count(&self) -> usize {
-        self.cfgs.len()
+        self.cfgs.read().unwrap().len()
+    }
+
+    /// Hot-reload the monitor set from a freshly-pushed streaming-config.
+    ///
+    /// The coordinator originally read `monitors:` only at boot; a monitor the
+    /// control plane published *after* start (via the `/api/v1/streaming-config`
+    /// hot-reload POST) never reached it, so every edge registering for that
+    /// monitor was rejected as "unconfigured". This applies the new spec list to
+    /// the live coordinator:
+    ///   * **added** specs become matchable immediately (the next register
+    ///     lazily builds the `Monitor`);
+    ///   * **changed** specs (different τ/ε/window for an existing key) drop the
+    ///     stale live `Monitor` so the next register rebuilds it under the new
+    ///     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.
+    ///
+    /// Returns `(added, changed, removed)` counts for observability. Idempotent:
+    /// re-applying the same specs is a no-op that returns `(0, 0, 0)`.
+    pub async fn reconfigure(&self, specs: Vec<MonitorConfig>) -> (usize, usize, usize) {
+        let new_cfgs: HashMap<MonKey, MonitorConfig> = specs
+            .into_iter()
+            .map(|c| ((c.agg_id, c.key.clone()), c))
+            .collect();
+
+        // Diff against the current specs to find keys to evict from live state
+        // (removed, or changed so the live Monitor's τ is stale).
+        let (added, changed, removed, evict): (usize, usize, usize, Vec<MonKey>) = {
+            let old = self.cfgs.read().unwrap();
+            let mut added = 0;
+            let mut changed = 0;
+            let mut evict = Vec::new();
+            for (k, c) in new_cfgs.iter() {
+                match old.get(k) {
+                    None => added += 1,
+                    Some(prev) if prev != c => {
+                        changed += 1;
+                        evict.push(k.clone());
+                    }
+                    Some(_) => {}
+                }
+            }
+            let mut removed = 0;
+            for k in old.keys() {
+                if !new_cfgs.contains_key(k) {
+                    removed += 1;
+                    evict.push(k.clone());
+                }
+            }
+            (added, changed, removed, evict)
+        };
+
+        if added == 0 && changed == 0 && removed == 0 {
+            return (0, 0, 0);
+        }
+
+        // Swap the spec map first so any register racing the eviction below sees
+        // the new spec (and rebuilds correctly), then drop stale live state.
+        *self.cfgs.write().unwrap() = new_cfgs;
+        if !evict.is_empty() {
+            let mut monitors = self.monitors.lock().await;
+            for k in &evict {
+                monitors.remove(k);
+            }
+        }
+        (added, changed, removed)
     }
 
     fn monitor_id(agg_id: u64, key: &[u8]) -> String {
@@ -78,7 +148,7 @@ impl MonitorCoordinator {
         window_start_ms: u64,
     ) -> Option<Vec<Action>> {
         let mk = (agg_id, key.clone());
-        let cfg = self.cfgs.get(&mk)?.clone();
+        let cfg = self.cfgs.read().unwrap().get(&mk)?.clone();
         let mut monitors = self.monitors.lock().await;
         let mon = monitors.entry(mk).or_insert_with(|| Monitor::new(cfg));
         Some(mon.on_register(edge_id, epoch_window_ms, window_start_ms))
@@ -268,3 +338,81 @@ impl MonitorService for MonitorServiceImpl {
         Ok(Response::new(Box::pin(ReceiverStream::new(rx))))
     }
 }
+
+#[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 {
+            agg_id,
+            key: key.as_bytes().to_vec(),
+            tau,
+            epsilon: 0.2,
+            window_ms: 15_000,
+        }
+    }
+
+    #[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());
+        assert_eq!(coord.monitor_count(), 1);
+
+        // Add a cms_point monitor (agg 2, key s0) and change agg 1's τ; agg 1's
+        // key "" stays but τ differs -> counts as a change. Nothing removed.
+        let (added, changed, removed) = coord
+            .reconfigure(vec![cfg(1, "", 200.0), cfg(2, "s0", 5000.0)])
+            .await;
+        assert_eq!((added, changed, removed), (1, 1, 0));
+        assert_eq!(coord.monitor_count(), 2);
+
+        // Idempotent: same specs -> no-op.
+        assert_eq!(
+            coord
+                .reconfigure(vec![cfg(1, "", 200.0), cfg(2, "s0", 5000.0)])
+                .await,
+            (0, 0, 0)
+        );
+
+        // Drop agg 1, keep agg 2 unchanged.
+        let (added, changed, removed) =
+            coord.reconfigure(vec![cfg(2, "s0", 5000.0)]).await;
+        assert_eq!((added, changed, removed), (0, 0, 1));
+        assert_eq!(coord.monitor_count(), 1);
+    }
+
+    #[tokio::test]
+    async fn reconfigure_evicts_live_state_for_changed_monitor() {
+        let coord = MonitorCoordinator::new(vec![cfg(7, "s0", 5000.0)], sink());
+        // 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)
+            .await;
+        assert!(actions.is_some(), "register for a configured monitor succeeds");
+        assert_eq!(coord.monitors.lock().await.len(), 1);
+
+        // Changing the spec (new τ) must drop the stale live state so the next
+        // register rebuilds the Monitor under the new τ.
+        let (_, changed, _) = coord.reconfigure(vec![cfg(7, "s0", 9000.0)]).await;
+        assert_eq!(changed, 1);
+        assert_eq!(
+            coord.monitors.lock().await.len(),
+            0,
+            "changed monitor's live state evicted"
+        );
+
+        // An unconfigured key is still rejected after reconfigure.
+        assert!(coord
+            .apply_register(7, b"other".to_vec(), "edge-a", 15_000, 15_000)
+            .await
+            .is_none());
+    }
+}