From 1fc392e88a4b36cf81078534314049cf9ba3a092 Mon Sep 17 00:00:00 2001 From: GyulyVGC Date: Wed, 21 Oct 2026 15:19:20 +0200 Subject: [PATCH 1/2] feat: attribute ingress congestion per node --- CHANGELOG.md | 1 + members/nullnet-server/src/congestion.rs | 638 ++++++++++++++++-- .../nullnet-server/src/http_server/nodes.rs | 80 ++- .../nullnet-server/src/nullnet_grpc_impl.rs | 140 ++-- members/nullnet-server/ui/src/index.css | 13 +- .../ui/src/lib/pressureChart.ts | 39 ++ members/nullnet-server/ui/src/pages/Nodes.tsx | 141 ++-- members/nullnet-server/ui/src/types.ts | 15 + .../ui/tests/pressureChart.test.mjs | 33 + 9 files changed, 906 insertions(+), 194 deletions(-) create mode 100644 members/nullnet-server/ui/src/lib/pressureChart.ts create mode 100644 members/nullnet-server/ui/tests/pressureChart.test.mjs diff --git a/CHANGELOG.md b/CHANGELOG.md index a1a5532..6575943 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,7 @@ All Nullnet releases with the relative changes are documented in this file. - Per-service egress/ingress traffic filters: arbitrary AND/OR/group combinations of Country, Organization, Src IP (ingress), and Dst IP (egress) conditions, evaluated via `rpn-predicate-interpreter` — replaces the country-only egress/ingress policy ([#171](https://github.com/NullNet-ai/nullnet/pull/171) — fixes [#143](https://github.com/NullNet-ai/nullnet/issues/143)) - Persist ingress and egress sessions to SQLite and show the full history on the Sessions page, filterable by status, service, direction, and policy verdict, with its own retention window ([#170](https://github.com/NullNet-ai/nullnet/pull/170) — fixes [#156](https://github.com/NullNet-ai/nullnet/issues/156)) ### Changed +- Attribute encrypted ingress setup pressure and adaptive multiplexing to each destination node, and show per-node setup metrics with proxy-only routing pressure on the Nodes page ([#207](https://github.com/NullNet-ai/nullnet/pull/207)). - Show bounded live and recent ingress pressure, congestion-control state, and active ingress groups on the Nodes page ([#206](https://github.com/NullNet-ai/nullnet/pull/206)). - Adapt encrypted ingress multiplexing, with or without a configured hard cap, to sustained RPC admission congestion by pausing dedicated setups and conservatively probing verified recovery per host ([#206](https://github.com/NullNet-ai/nullnet/pull/206)). - Multiplex encrypted ingress overflow with client-IP policy isolation, persistent host-pair IPsec transports, and direct same-host forwarding ([#206](https://github.com/NullNet-ai/nullnet/pull/206)). diff --git a/members/nullnet-server/src/congestion.rs b/members/nullnet-server/src/congestion.rs index b3f410d..d6d25bb 100644 --- a/members/nullnet-server/src/congestion.rs +++ b/members/nullnet-server/src/congestion.rs @@ -41,8 +41,8 @@ pub(crate) struct PressureWindowSnapshot { pub(crate) duration: Duration, pub(crate) arrivals: u64, pub(crate) completions: u64, - pub(crate) waiters: u32, - pub(crate) mean_wait_us: u64, + pub(crate) pending: u32, + pub(crate) mean_latency_us: u64, pub(crate) trend: PressureTrend, } @@ -73,8 +73,10 @@ pub(crate) struct HostStatusSnapshot { #[derive(Clone, Debug, Default)] pub(crate) struct NodeCongestionSnapshot { - pub(crate) pressure_age: Option, - pub(crate) windows: Vec, + pub(crate) proxy_pressure_age: Option, + pub(crate) proxy_windows: Vec, + pub(crate) setup_pressure_age: Option, + pub(crate) setup_windows: Vec, pub(crate) control: Option, } @@ -84,8 +86,8 @@ struct PressureWindow { duration: Duration, arrivals: u64, completions: u64, - waiters: u32, - mean_wait_us: u64, + pending: u32, + mean_latency_us: u64, trend: PressureTrend, } @@ -95,13 +97,23 @@ struct PressureView { trend: Trend, } -struct ProxyPressure { +#[derive(Clone, Copy)] +struct LoadSample { + epoch: u64, + arrivals: u64, + completions: u64, + pending: u32, +} + +struct Pressure { window_started: Instant, - start: ProxyLoad, - last: ProxyLoad, - wait_sum: u128, - samples: u64, - previous_wait_mean: Option, + start: LoadSample, + last: LoadSample, + latency_sum: u128, + latency_samples: u64, + previous_latency_mean: Option, + stalled_is_growing: bool, + carry_boundary_latency: bool, congested_windows: u8, sequence: u64, trend: Trend, @@ -109,15 +121,23 @@ struct ProxyPressure { history: VecDeque, } -impl ProxyPressure { - fn new(now: Instant, sample: ProxyLoad) -> Self { +impl Pressure { + fn new( + now: Instant, + sample: LoadSample, + latency_us: Option, + stalled_is_growing: bool, + carry_boundary_latency: bool, + ) -> Self { Self { window_started: now, start: sample, last: sample, - wait_sum: u128::from(sample.wait_us), - samples: 1, - previous_wait_mean: None, + latency_sum: latency_us.map_or(0, u128::from), + latency_samples: u64::from(latency_us.is_some()), + previous_latency_mean: None, + stalled_is_growing, + carry_boundary_latency, congested_windows: 0, sequence: 0, trend: Trend::Neutral, @@ -126,18 +146,26 @@ impl ProxyPressure { } } - fn reset(&mut self, now: Instant, sample: ProxyLoad) { + fn reset(&mut self, now: Instant, sample: LoadSample, latency_us: Option) { let sequence = self.sequence.saturating_add(1); + let stalled_is_growing = self.stalled_is_growing; + let carry_boundary_latency = self.carry_boundary_latency; let history = std::mem::replace(&mut self.history, VecDeque::with_capacity(PRESSURE_HISTORY)); - *self = Self::new(now, sample); + *self = Self::new( + now, + sample, + latency_us, + stalled_is_growing, + carry_boundary_latency, + ); self.sequence = sequence; self.history = history; } - fn observe(&mut self, now: Instant, sample: ProxyLoad) { + fn observe(&mut self, now: Instant, sample: LoadSample, latency_us: Option) { if sample.epoch != self.last.epoch { - self.reset(now, sample); + self.reset(now, sample, latency_us); return; } if sample.arrivals < self.last.arrivals || sample.completions < self.last.completions { @@ -145,25 +173,34 @@ impl ProxyPressure { } self.last = sample; self.last_observed = now; - self.wait_sum += u128::from(sample.wait_us); - self.samples += 1; + if let Some(latency_us) = latency_us { + self.latency_sum += u128::from(latency_us); + self.latency_samples += 1; + } if now.duration_since(self.window_started) < PRESSURE_WINDOW { return; } let arrivals = self.last.arrivals - self.start.arrivals; let completions = self.last.completions - self.start.completions; - let wait_mean = u64::try_from(self.wait_sum / u128::from(self.samples)).unwrap_or(u64::MAX); - let wait_rising = self - .previous_wait_mean - .is_none_or(|previous| wait_mean > previous); - let wait_not_rising = self.previous_wait_mean.is_none_or(|previous| { - wait_mean <= previous.saturating_add(previous / 10).saturating_add(100) + let latency_mean = if self.latency_samples == 0 { + self.previous_latency_mean.unwrap_or_default() + } else { + u64::try_from(self.latency_sum / u128::from(self.latency_samples)).unwrap_or(u64::MAX) + }; + let latency_rising = self + .previous_latency_mean + .is_none_or(|previous| latency_mean > previous); + let latency_not_rising = self.previous_latency_mean.is_none_or(|previous| { + latency_mean <= previous.saturating_add(previous / 10).saturating_add(100) }); - let growing = - arrivals > completions && self.last.waiters > self.start.waiters && wait_rising; - let healthy = - arrivals <= completions && self.last.waiters <= self.start.waiters && wait_not_rising; + let stalled = self.stalled_is_growing && arrivals > completions && completions == 0; + let growing = arrivals > completions + && self.last.pending > self.start.pending + && (latency_rising || stalled); + let healthy = arrivals <= completions + && self.last.pending <= self.start.pending + && latency_not_rising; let display_trend = if growing { PressureTrend::Growing @@ -197,16 +234,20 @@ impl ProxyPressure { duration: now.duration_since(self.window_started), arrivals, completions, - waiters: self.last.waiters, - mean_wait_us: wait_mean, + pending: self.last.pending, + mean_latency_us: latency_mean, trend: display_trend, }); - self.previous_wait_mean = Some(wait_mean); + self.previous_latency_mean = Some(latency_mean); self.window_started = now; self.start = sample; self.last = sample; - self.wait_sum = u128::from(sample.wait_us); - self.samples = 1; + self.latency_sum = if self.carry_boundary_latency { + latency_us.map_or(0, u128::from) + } else { + 0 + }; + self.latency_samples = u64::from(self.carry_boundary_latency && latency_us.is_some()); } fn view(&self) -> PressureView { @@ -215,6 +256,84 @@ impl ProxyPressure { trend: self.trend, } } + + fn current_window(&self, now: Instant) -> Option { + let arrivals = self.last.arrivals - self.start.arrivals; + let completions = self.last.completions - self.start.completions; + if arrivals == 0 && completions == 0 && self.last.pending == self.start.pending { + return None; + } + let elapsed = now.duration_since(self.window_started); + let duration = elapsed.min(PRESSURE_WINDOW); + if duration.is_zero() { + return None; + } + let mean_latency_us = if self.latency_samples == 0 { + self.previous_latency_mean.unwrap_or_default() + } else { + u64::try_from(self.latency_sum / u128::from(self.latency_samples)).unwrap_or(u64::MAX) + }; + Some(PressureWindow { + ended_at: self.window_started + duration, + duration, + arrivals, + completions, + pending: self.last.pending, + mean_latency_us, + trend: PressureTrend::Neutral, + }) + } +} + +struct SetupPressure { + pressure: Pressure, + arrivals: u64, + completions: u64, + pending: u32, +} + +impl SetupPressure { + fn new(now: Instant) -> Self { + Self { + pressure: Pressure::new( + now, + LoadSample { + epoch: 0, + arrivals: 0, + completions: 0, + pending: 0, + }, + None, + true, + false, + ), + arrivals: 0, + completions: 0, + pending: 0, + } + } + + fn sample(&self) -> LoadSample { + LoadSample { + epoch: 0, + arrivals: self.arrivals, + completions: self.completions, + pending: self.pending, + } + } + + fn started(&mut self, now: Instant) { + self.arrivals = self.arrivals.saturating_add(1); + self.pending = self.pending.saturating_add(1); + self.pressure.observe(now, self.sample(), None); + } + + fn completed(&mut self, now: Instant, latency: Duration) { + self.completions = self.completions.saturating_add(1); + self.pending = self.pending.saturating_sub(1); + let latency_us = u64::try_from(latency.as_micros()).unwrap_or(u64::MAX); + self.pressure.observe(now, self.sample(), Some(latency_us)); + } } #[derive(Clone, Copy)] @@ -224,13 +343,19 @@ struct ControlTransition { source_proxy: IpAddr, } +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] +enum PressureSource { + Proxy(IpAddr), + Setup(IpAddr), +} + struct HostRate { rate: Option, tokens: f64, last_refill: Instant, last_denied: Option, completions: VecDeque, - pressure_sequences: HashMap, + pressure_sequences: HashMap, healthy_windows: u8, state_since: Option, source_proxy: Option, @@ -324,24 +449,41 @@ impl HostRate { self.completions.clear(); } - fn apply( + fn apply_all( &mut self, host: IpAddr, proxy: IpAddr, - pressure: PressureView, + pressures: &[(PressureSource, PressureView)], now: Instant, ) -> Option { - if pressure.sequence - <= self - .pressure_sequences - .get(&proxy) - .copied() - .unwrap_or_default() - { + let changed = pressures.iter().any(|(source, pressure)| { + pressure.sequence + > self + .pressure_sequences + .get(source) + .copied() + .unwrap_or_default() + }); + if !changed { return None; } - self.pressure_sequences.insert(proxy, pressure.sequence); - match pressure.trend { + for (source, pressure) in pressures { + self.pressure_sequences.insert(*source, pressure.sequence); + } + let trend = if pressures + .iter() + .any(|(_, pressure)| pressure.trend == Trend::Congested) + { + Trend::Congested + } else if pressures + .iter() + .all(|(_, pressure)| pressure.trend == Trend::Healthy) + { + Trend::Healthy + } else { + Trend::Neutral + }; + match trend { Trend::Congested => { let activating = self.rate != Some(0.0); self.rate = Some(0.0); @@ -388,6 +530,22 @@ impl HostRate { None } + #[cfg(test)] + fn apply( + &mut self, + host: IpAddr, + proxy: IpAddr, + pressure: PressureView, + now: Instant, + ) -> Option { + self.apply_all( + host, + proxy, + &[(PressureSource::Proxy(proxy), pressure)], + now, + ) + } + fn refill(&mut self, now: Instant) { let Some(rate) = self.rate else { self.last_refill = now; @@ -401,7 +559,8 @@ impl HostRate { #[derive(Default)] struct State { - proxies: HashMap, + proxies: HashMap, + setups: HashMap, hosts: HashMap, } @@ -410,6 +569,28 @@ pub(crate) struct CongestionController { state: Mutex, } +pub(crate) struct SetupAttempt { + controller: std::sync::Arc, + host: IpAddr, + started: Instant, +} + +impl Drop for SetupAttempt { + fn drop(&mut self) { + let now = Instant::now(); + let mut state = self + .controller + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + state + .setups + .get_mut(&self.host) + .expect("setup attempt is registered") + .completed(now, now.duration_since(self.started)); + } +} + impl CongestionController { pub(crate) fn observe_proxy(&self, proxy: IpAddr, sample: ProxyLoad) { let mut state = self @@ -420,11 +601,36 @@ impl CongestionController { } fn observe_proxy_locked(state: &mut State, proxy: IpAddr, sample: ProxyLoad, now: Instant) { + let load = LoadSample { + epoch: sample.epoch, + arrivals: sample.arrivals, + completions: sample.completions, + pending: sample.waiters, + }; state .proxies .entry(proxy) - .and_modify(|pressure| pressure.observe(now, sample)) - .or_insert_with(|| ProxyPressure::new(now, sample)); + .and_modify(|pressure| pressure.observe(now, load, Some(sample.wait_us))) + .or_insert_with(|| Pressure::new(now, load, Some(sample.wait_us), false, true)); + } + + pub(crate) fn setup_attempt(self: &std::sync::Arc, host: IpAddr) -> SetupAttempt { + let now = Instant::now(); + let mut state = self + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + state + .setups + .entry(host) + .or_insert_with(|| SetupPressure::new(now)) + .started(now); + drop(state); + SetupAttempt { + controller: self.clone(), + host, + started: now, + } } pub(crate) fn allow_dedicated(&self, proxy: IpAddr, destination: IpAddr) -> AdmissionDecision { @@ -441,7 +647,11 @@ impl CongestionController { destination: IpAddr, now: Instant, ) -> AdmissionDecision { - let pressure = state.proxies.get(&proxy).map(ProxyPressure::view); + let proxy_pressure = state.proxies.get(&proxy).map(Pressure::view); + let setup_pressure = state + .setups + .get(&destination) + .map(|setup| setup.pressure.view()); let hosts = [proxy, destination]; let host_count = if proxy == destination { 1 } else { 2 }; let mut transitions = Vec::new(); @@ -451,8 +661,28 @@ impl CongestionController { .entry(*host) .or_insert_with(|| HostRate::new(now)); rate.refill(now); - if let Some(pressure) = pressure - && let Some(transition) = rate.apply(*host, proxy, pressure, now) + let mut pressures = [( + PressureSource::Proxy(proxy), + PressureView { + sequence: 0, + trend: Trend::Neutral, + }, + ); 2]; + let mut pressure_count = 0; + if *host == proxy + && let Some(pressure) = proxy_pressure + { + pressures[pressure_count] = (PressureSource::Proxy(proxy), pressure); + pressure_count += 1; + } + if *host == destination + && let Some(pressure) = setup_pressure + { + pressures[pressure_count] = (PressureSource::Setup(destination), pressure); + pressure_count += 1; + } + if let Some(transition) = + rate.apply_all(*host, proxy, &pressures[..pressure_count], now) { transitions.push(transition); } @@ -522,17 +752,35 @@ impl CongestionController { nodes .into_iter() .map(|node| { - let pressure = state.proxies.get(&node); - let windows = pressure + let proxy_pressure = state.proxies.get(&node); + let proxy_windows = proxy_pressure .into_iter() .flat_map(|pressure| pressure.history.iter()) + .copied() + .chain(proxy_pressure.and_then(|pressure| pressure.current_window(now))) .map(|window| PressureWindowSnapshot { age: now.duration_since(window.ended_at), duration: window.duration, arrivals: window.arrivals, completions: window.completions, - waiters: window.waiters, - mean_wait_us: window.mean_wait_us, + pending: window.pending, + mean_latency_us: window.mean_latency_us, + trend: window.trend, + }) + .collect(); + let setup_pressure = state.setups.get(&node); + let setup_windows = setup_pressure + .into_iter() + .flat_map(|setup| setup.pressure.history.iter()) + .copied() + .chain(setup_pressure.and_then(|setup| setup.pressure.current_window(now))) + .map(|window| PressureWindowSnapshot { + age: now.duration_since(window.ended_at), + duration: window.duration, + arrivals: window.arrivals, + completions: window.completions, + pending: window.pending, + mean_latency_us: window.mean_latency_us, trend: window.trend, }) .collect(); @@ -564,9 +812,12 @@ impl CongestionController { ( node, NodeCongestionSnapshot { - pressure_age: pressure + proxy_pressure_age: proxy_pressure .map(|pressure| now.duration_since(pressure.last_observed)), - windows, + proxy_windows, + setup_pressure_age: setup_pressure + .map(|setup| now.duration_since(setup.pressure.last_observed)), + setup_windows, control, }, ) @@ -593,6 +844,16 @@ mod tests { } } + fn congest_setup(state: &mut State, host: IpAddr, start: Instant) { + let setup = state + .setups + .entry(host) + .or_insert_with(|| SetupPressure::new(start)); + setup.started(start); + setup.started(start + Duration::from_secs(1)); + setup.started(start + Duration::from_secs(2)); + } + #[test] fn stable_waiters_do_not_activate_control() { let controller = CongestionController::default(); @@ -625,11 +886,30 @@ mod tests { #[test] fn pressure_history_is_bounded_and_survives_proxy_epoch_changes() { let start = Instant::now(); - let mut pressure = ProxyPressure::new(start, sample(0, 0, 0, 0)); + let first = sample(0, 0, 0, 0); + let mut pressure = Pressure::new( + start, + LoadSample { + epoch: first.epoch, + arrivals: first.arrivals, + completions: first.completions, + pending: first.waiters, + }, + Some(first.wait_us), + false, + true, + ); for second in 1..=65 { + let next = sample(second * 10, second * 10, 0, 100); pressure.observe( start + Duration::from_secs(second), - sample(second * 10, second * 10, 0, 100), + LoadSample { + epoch: next.epoch, + arrivals: next.arrivals, + completions: next.completions, + pending: next.waiters, + }, + Some(next.wait_us), ); } assert_eq!(pressure.history.len(), PRESSURE_HISTORY); @@ -637,12 +917,21 @@ mod tests { let mut reset = sample(1, 1, 0, 100); reset.epoch = 2; - pressure.observe(start + Duration::from_secs(66), reset); + pressure.observe( + start + Duration::from_secs(66), + LoadSample { + epoch: reset.epoch, + arrivals: reset.arrivals, + completions: reset.completions, + pending: reset.waiters, + }, + Some(reset.wait_us), + ); assert_eq!(pressure.history.len(), PRESSURE_HISTORY); } #[test] - fn growing_backlog_activates_rate_control() { + fn growing_proxy_backlog_activates_only_proxy_control() { let controller = CongestionController::default(); let start = Instant::now(); let mut state = controller.state.lock().unwrap(); @@ -674,12 +963,9 @@ mod tests { start + Duration::from_secs(2), ); assert!(!first.dedicated); - assert_eq!(first.transitions.len(), 2); - assert!( - first - .transitions - .iter() - .all(|change| matches!(change, RateTransition::Activated { .. })) + assert_eq!( + first.transitions, + vec![RateTransition::Activated { host: ip(1) }] ); assert!( !CongestionController::allow_dedicated_locked( @@ -692,6 +978,191 @@ mod tests { ); } + #[test] + fn proxy_backlog_does_not_activate_an_unmeasured_destination() { + let controller = CongestionController::default(); + let start = Instant::now(); + let mut state = controller.state.lock().unwrap(); + CongestionController::observe_proxy_locked(&mut state, ip(1), sample(0, 0, 0, 0), start); + CongestionController::observe_proxy_locked( + &mut state, + ip(1), + sample(100, 70, 30, 1_000), + start + Duration::from_secs(1), + ); + CongestionController::observe_proxy_locked( + &mut state, + ip(1), + sample(200, 140, 60, 2_000), + start + Duration::from_secs(2), + ); + + let decision = CongestionController::allow_dedicated_locked( + &mut state, + ip(1), + ip(2), + start + Duration::from_secs(2), + ); + + assert_eq!( + decision.transitions, + vec![RateTransition::Activated { host: ip(1) }] + ); + assert_eq!(state.hosts[&ip(2)].mode(), ControlMode::Normal); + } + + #[test] + fn destination_setup_backlog_activates_only_its_node() { + let controller = CongestionController::default(); + let start = Instant::now(); + let mut state = controller.state.lock().unwrap(); + congest_setup(&mut state, ip(2), start); + + let decision = CongestionController::allow_dedicated_locked( + &mut state, + ip(1), + ip(2), + start + Duration::from_secs(2), + ); + + assert!(!decision.dedicated); + assert_eq!( + decision.transitions, + vec![RateTransition::Activated { host: ip(2) }] + ); + assert_eq!(state.hosts[&ip(1)].mode(), ControlMode::Normal); + } + + #[test] + fn destination_pressure_never_spills_into_another_destination() { + let controller = CongestionController::default(); + let start = Instant::now(); + let mut state = controller.state.lock().unwrap(); + congest_setup(&mut state, ip(2), start); + + let decision = CongestionController::allow_dedicated_locked( + &mut state, + ip(1), + ip(3), + start + Duration::from_secs(2), + ); + + assert!(decision.dedicated); + assert!(decision.transitions.is_empty()); + assert_eq!(state.hosts[&ip(3)].mode(), ControlMode::Normal); + } + + #[test] + fn cross_host_setup_requires_both_independent_controllers() { + let controller = CongestionController::default(); + let start = Instant::now(); + let mut state = controller.state.lock().unwrap(); + CongestionController::observe_proxy_locked(&mut state, ip(1), sample(0, 0, 0, 0), start); + CongestionController::observe_proxy_locked( + &mut state, + ip(1), + sample(100, 70, 30, 1_000), + start + Duration::from_secs(1), + ); + CongestionController::observe_proxy_locked( + &mut state, + ip(1), + sample(200, 140, 60, 2_000), + start + Duration::from_secs(2), + ); + congest_setup(&mut state, ip(2), start); + + let decision = CongestionController::allow_dedicated_locked( + &mut state, + ip(1), + ip(2), + start + Duration::from_secs(2), + ); + + assert!(!decision.dedicated); + assert_eq!( + decision.transitions, + vec![ + RateTransition::Activated { host: ip(1) }, + RateTransition::Activated { host: ip(2) }, + ] + ); + } + + #[test] + fn setup_latency_must_worsen_while_completions_continue() { + let start = Instant::now(); + let mut pressure = Pressure::new( + start, + LoadSample { + epoch: 0, + arrivals: 0, + completions: 0, + pending: 0, + }, + None, + true, + false, + ); + let load = |arrivals, completions, pending| LoadSample { + epoch: 0, + arrivals, + completions, + pending, + }; + pressure.observe(start + Duration::from_secs(1), load(10, 5, 5), Some(1_000)); + pressure.observe(start + Duration::from_secs(2), load(20, 10, 10), Some(500)); + assert_eq!(pressure.view().trend, Trend::Neutral); + pressure.observe( + start + Duration::from_secs(3), + load(30, 15, 15), + Some(1_000), + ); + pressure.observe( + start + Duration::from_secs(4), + load(40, 20, 20), + Some(2_000), + ); + assert_eq!(pressure.view().trend, Trend::Congested); + } + + #[test] + fn setup_attempts_complete_independently_per_host() { + let controller = std::sync::Arc::new(CongestionController::default()); + let proxy_attempt = controller.setup_attempt(ip(1)); + let destination_attempt = controller.setup_attempt(ip(2)); + { + let state = controller.state.lock().unwrap(); + assert_eq!(state.setups[&ip(1)].arrivals, 1); + assert_eq!(state.setups[&ip(2)].arrivals, 1); + assert_eq!(state.setups[&ip(1)].pending, 1); + assert_eq!(state.setups[&ip(2)].pending, 1); + } + drop(destination_attempt); + { + let state = controller.state.lock().unwrap(); + assert_eq!(state.setups[&ip(1)].completions, 0); + assert_eq!(state.setups[&ip(2)].completions, 1); + assert_eq!(state.setups[&ip(1)].pending, 1); + assert_eq!(state.setups[&ip(2)].pending, 0); + } + drop(proxy_attempt); + assert_eq!(controller.state.lock().unwrap().setups[&ip(1)].pending, 0); + } + + #[test] + fn separate_same_host_setup_operations_are_counted() { + let controller = std::sync::Arc::new(CongestionController::default()); + let first = controller.setup_attempt(ip(1)); + let second = controller.setup_attempt(ip(1)); + assert_eq!(controller.state.lock().unwrap().setups[&ip(1)].arrivals, 2); + drop((first, second)); + assert_eq!( + controller.state.lock().unwrap().setups[&ip(1)].completions, + 2 + ); + } + #[test] fn new_proxy_epoch_discards_old_pressure_and_can_reactivate() { let controller = CongestionController::default(); @@ -748,7 +1219,7 @@ mod tests { ip(2), start + Duration::from_secs(4), ); - assert_eq!(decision.transitions.len(), 2); + assert_eq!(decision.transitions.len(), 1); } #[test] @@ -781,7 +1252,7 @@ mod tests { ip(2), start + Duration::from_secs(2), ); - assert_eq!(decision.transitions.len(), 2); + assert_eq!(decision.transitions.len(), 1); } #[test] @@ -918,6 +1389,25 @@ mod tests { assert_eq!(rate.rate, None); } + #[test] + fn one_off_setup_is_visible_without_waiting_for_another_setup() { + let start = Instant::now(); + let mut setup = SetupPressure::new(start); + setup.started(start); + setup.completed(start + Duration::from_millis(20), Duration::from_millis(20)); + + let window = setup + .pressure + .current_window(start + Duration::from_secs(5)) + .expect("one-off setup has a display window"); + assert_eq!(window.duration, PRESSURE_WINDOW); + assert_eq!(window.ended_at, start + PRESSURE_WINDOW); + assert_eq!(window.arrivals, 1); + assert_eq!(window.completions, 1); + assert_eq!(window.pending, 0); + assert_eq!(window.mean_latency_us, 20_000); + } + #[test] fn idle_time_does_not_remove_the_probe_limit() { let start = Instant::now(); diff --git a/members/nullnet-server/src/http_server/nodes.rs b/members/nullnet-server/src/http_server/nodes.rs index b461057..8df0f03 100644 --- a/members/nullnet-server/src/http_server/nodes.rs +++ b/members/nullnet-server/src/http_server/nodes.rs @@ -2,6 +2,7 @@ use super::AppState; use super::auth::{AuthContext, require_scope}; use crate::auth::Scope; use crate::congestion::{ControlMode, NodeCongestionSnapshot, PressureTrend}; +use crate::env::PROXY_IP; use crate::services::service_info::ServiceInfo; use axum::extract::{Extension, Path, State}; use axum::response::{IntoResponse, Response}; @@ -19,11 +20,14 @@ struct HostedServiceJson { #[derive(Serialize)] struct NodeJson { ip: String, + is_proxy: bool, hosted_services: Vec, ingress_groups: IngressGroupsJson, ingress_control: IngressControlJson, pressure_age_ms: Option, - pressure_windows: Vec, + pressure_windows: Vec, + setup_pressure_age_ms: Option, + setup_pressure_windows: Vec, } #[derive(Default, Serialize)] @@ -52,7 +56,7 @@ struct IngressStateJson { } #[derive(Serialize)] -struct PressureWindowJson { +struct ProxyPressureWindowJson { age_ms: u64, duration_ms: u64, arrivals: u64, @@ -64,6 +68,19 @@ struct PressureWindowJson { trend: &'static str, } +#[derive(Serialize)] +struct SetupPressureWindowJson { + age_ms: u64, + duration_ms: u64, + arrivals: u64, + completions: u64, + arrivals_per_sec: f64, + completions_per_sec: f64, + in_flight: u32, + mean_setup_us: u64, + trend: &'static str, +} + fn millis(duration: Duration) -> u64 { u64::try_from(duration.as_millis()).unwrap_or(u64::MAX) } @@ -78,7 +95,13 @@ fn control_mode(mode: ControlMode) -> &'static str { fn congestion_json( snapshot: NodeCongestionSnapshot, -) -> (IngressControlJson, Option, Vec) { +) -> ( + IngressControlJson, + Option, + Vec, + Option, + Vec, +) { let control = snapshot.control.map_or_else( || IngressControlJson { state: "normal", @@ -109,20 +132,42 @@ fn congestion_json( .collect(), }, ); - let windows = snapshot - .windows + let proxy_windows = snapshot + .proxy_windows + .into_iter() + .map(|window| { + let seconds = window.duration.as_secs_f64(); + ProxyPressureWindowJson { + age_ms: millis(window.age), + duration_ms: millis(window.duration), + arrivals: window.arrivals, + completions: window.completions, + arrivals_per_sec: window.arrivals as f64 / seconds, + completions_per_sec: window.completions as f64 / seconds, + waiters: window.pending, + mean_wait_us: window.mean_latency_us, + trend: match window.trend { + PressureTrend::Growing => "growing", + PressureTrend::Healthy => "healthy", + PressureTrend::Neutral => "neutral", + }, + } + }) + .collect(); + let setup_windows = snapshot + .setup_windows .into_iter() .map(|window| { let seconds = window.duration.as_secs_f64(); - PressureWindowJson { + SetupPressureWindowJson { age_ms: millis(window.age), duration_ms: millis(window.duration), arrivals: window.arrivals, completions: window.completions, arrivals_per_sec: window.arrivals as f64 / seconds, completions_per_sec: window.completions as f64 / seconds, - waiters: window.waiters, - mean_wait_us: window.mean_wait_us, + in_flight: window.pending, + mean_setup_us: window.mean_latency_us, trend: match window.trend { PressureTrend::Growing => "growing", PressureTrend::Healthy => "healthy", @@ -131,7 +176,13 @@ fn congestion_json( } }) .collect(); - (control, snapshot.pressure_age.map(millis), windows) + ( + control, + snapshot.proxy_pressure_age.map(millis), + proxy_windows, + snapshot.setup_pressure_age.map(millis), + setup_windows, + ) } pub(super) async fn nodes_handler( @@ -194,9 +245,16 @@ pub(super) async fn nodes_handler( let groups = ingress_groups.remove(&ip).unwrap_or_default(); let multiplexed = groups.values().filter(|policy| **policy).count(); let snapshot = congestion.remove(&ip).unwrap_or_default(); - let (ingress_control, pressure_age_ms, pressure_windows) = congestion_json(snapshot); + let ( + ingress_control, + pressure_age_ms, + pressure_windows, + setup_pressure_age_ms, + setup_pressure_windows, + ) = congestion_json(snapshot); NodeJson { ip: ip.to_string(), + is_proxy: Some(ip) == *PROXY_IP, hosted_services: svcs, ingress_groups: IngressGroupsJson { dedicated: groups.len() - multiplexed, @@ -205,6 +263,8 @@ pub(super) async fn nodes_handler( ingress_control, pressure_age_ms, pressure_windows, + setup_pressure_age_ms, + setup_pressure_windows, } }) .collect(); diff --git a/members/nullnet-server/src/nullnet_grpc_impl.rs b/members/nullnet-server/src/nullnet_grpc_impl.rs index 95a2f6d..2959e5b 100644 --- a/members/nullnet-server/src/nullnet_grpc_impl.rs +++ b/members/nullnet-server/src/nullnet_grpc_impl.rs @@ -92,6 +92,12 @@ struct DedicatedAdmission { key: (String, String), } +#[derive(Default)] +struct EntrySetupContext { + admission: Option, + pressure: Option>, +} + struct ProxyReplica { ip: IpAddr, port: u16, @@ -105,6 +111,7 @@ struct ProxyChainRequest<'a> { client_ip: &'a str, replica: &'a ProxyReplica, policy: bool, + measure_congestion: bool, } impl Drop for DedicatedAdmission { @@ -842,6 +849,7 @@ impl NullnetGrpcImpl { client_ip, replica: &replica, policy: policy_overflow, + measure_congestion: policy_eligible, }, dedicated_admission.take(), ) @@ -1032,16 +1040,7 @@ impl NullnetGrpcImpl { request.client_ip.to_string(), request.proxy_ip, ); - let outcome = self - .setup_proxy_chain( - request.stack, - &key, - request.replica.ip, - request.replica.docker.as_deref(), - request.policy, - admission, - ) - .await?; + let outcome = self.setup_proxy_chain(&request, &key, admission).await?; // Suspended replicas are unpaused per-edge inside `net_chain_setup`, so by // the time the chain is built every container in it is already serving. @@ -1130,30 +1129,36 @@ impl NullnetGrpcImpl { async fn setup_proxy_chain( &self, - stack: &str, + request: &ProxyChainRequest<'_>, key: &ProxyKey, - service_ip: IpAddr, - service_docker: Option<&str>, - policy: bool, admission: Option, ) -> Result { let (service_name, client_ip, proxy_ip) = key; let branches = self - .proxy_branches(stack, service_name, service_ip, service_docker) + .proxy_branches( + request.stack, + service_name, + request.replica.ip, + request.replica.docker.as_deref(), + ) .await?; let mut entry = RegisteredEdge::new( *proxy_ip, Client::new(client_ip.clone(), Some(*proxy_ip)), None, - service_ip, + request.replica.ip, Client::new(service_name.to_string(), None), - service_docker.map(String::from), + request.replica.docker.clone(), ); - entry.policy = policy; + entry.policy = request.policy; + let entry_setup = EntrySetupContext { + admission, + pressure: request.measure_congestion.then(|| self.congestion.clone()), + }; match self - .net_chain_setup_with_admission(stack, entry, branches, admission) + .net_chain_setup_with_admission(request.stack, entry, branches, entry_setup) .await? { ChainOutcome::Built(None) => { @@ -1509,8 +1514,14 @@ impl NullnetGrpcImpl { ); // An invalidated owner unwinds its increments instead of retaining a hold. let built = matches!( - self.net_chain_setup_owned(stack, None, vec![branch], owner, None) - .await?, + self.net_chain_setup_owned( + stack, + None, + vec![branch], + owner, + EntrySetupContext::default(), + ) + .await?, ChainOutcome::Built(_) ); println!("[trigger] net_chain_setup completed for '{initiator_name}' port {port}"); @@ -1986,9 +1997,9 @@ impl NullnetGrpcImpl { stack: &str, entry: RegisteredEdge, branches: Vec, - admission: Option, + entry_setup: EntrySetupContext, ) -> Result { - self.net_chain_setup_owned(stack, Some(entry), branches, None, admission) + self.net_chain_setup_owned(stack, Some(entry), branches, None, entry_setup) .await } @@ -1998,7 +2009,7 @@ impl NullnetGrpcImpl { entry: Option, branches: Vec, owner: Option<(BackendKey, uuid::Uuid)>, - admission: Option, + entry_setup: EntrySetupContext, ) -> Result { let services = self.services.clone(); let orchestrator = self.orchestrator.clone(); @@ -2013,7 +2024,7 @@ impl NullnetGrpcImpl { entry, branches, owner, - admission, + entry_setup, ) .await, ); @@ -2128,7 +2139,7 @@ async fn run_net_chain_setup( entry: Option, branches: Vec, owner: Option<(BackendKey, uuid::Uuid)>, - admission: Option, + entry_setup: EntrySetupContext, ) -> Result { let mut join_set_outer = JoinSet::new(); @@ -2151,7 +2162,8 @@ async fn run_net_chain_setup( edge.backend_entry_port, edge.egress, edge.policy, - admission, + entry_setup.admission, + entry_setup.pressure, ) .await; match outcome { @@ -2204,6 +2216,7 @@ async fn run_net_chain_setup( false, false, None, + None, ) .await; match outcome { @@ -2384,6 +2397,7 @@ async fn setup_edge( egress: bool, policy: bool, mut admission: Option, + setup_pressure: Option>, ) -> EdgeOutcome { let init_time = std::time::Instant::now(); // Egress edges steer on the initiator (client) side and intercept on the @@ -2542,37 +2556,53 @@ async fn setup_edge( let orch = orchestrator.clone(); let cd = client_docker.clone(); let sd = server_docker.clone(); - let server_res = orch.send_net_setup( - server_ethernet, - None, - net_id, - client_ethernet, - (cd, sd), - None, - encryption_key, - xfrm_transport_id, - dstport, - encrypted, - server_egress, - server_policy, - ); + let server_attempt = setup_pressure + .as_ref() + .map(|pressure| pressure.setup_attempt(server_ethernet)); + let server_res = async move { + let _attempt = server_attempt; + orch.send_net_setup( + server_ethernet, + None, + net_id, + client_ethernet, + (cd, sd), + None, + encryption_key, + xfrm_transport_id, + dstport, + encrypted, + server_egress, + server_policy, + ) + .await + }; let orch2 = orchestrator.clone(); let cd = client_docker.clone(); let sd = server_docker.clone(); - let client_res = orch2.send_net_setup( - client_ethernet, - Some(server_name.clone()), - net_id, - server_ethernet, - (cd, sd), - backend_entry_port, - encryption_key, - xfrm_transport_id, - dstport, - encrypted, - client_egress, - client_policy, - ); + let client_attempt = setup_pressure + .as_ref() + .map(|pressure| pressure.setup_attempt(client_ethernet)); + let client_server_name = server_name.clone(); + let client_res = async move { + let _attempt = client_attempt; + orch2 + .send_net_setup( + client_ethernet, + Some(client_server_name), + net_id, + server_ethernet, + (cd, sd), + backend_entry_port, + encryption_key, + xfrm_transport_id, + dstport, + encrypted, + client_egress, + client_policy, + ) + .await + }; let (server_ok, client_ok) = tokio::join!(server_res, client_res); diff --git a/members/nullnet-server/ui/src/index.css b/members/nullnet-server/ui/src/index.css index e7bbf22..916b2e6 100644 --- a/members/nullnet-server/ui/src/index.css +++ b/members/nullnet-server/ui/src/index.css @@ -237,12 +237,13 @@ body { /* ── Nodes ── */ .workspace { display: flex; flex: 1; overflow: hidden; } -.nodes-content { padding: 28px; flex: 1; overflow-y: auto; } -.nodes-grid { display: grid; grid-template-columns: 1fr 1fr; gap: 12px; } +.nodes-content { padding: 28px; flex: 1; min-width: 0; overflow-y: auto; } +.nodes-grid { display: grid; grid-template-columns: repeat(2, minmax(0, 1fr)); gap: 12px; } .node-card { background: var(--g1); backdrop-filter: blur(20px); -webkit-backdrop-filter: blur(20px); border: 1px solid var(--gb); box-shadow: inset 0 1px 0 rgba(255,255,255,.08), 0 4px 20px rgba(0,0,0,.3); border-radius: 14px; cursor: pointer; transition: all .18s; overflow: hidden; } .node-card:hover { background: var(--g2); border-color: rgba(255,255,255,.12); } .node-card.sel { border-color: rgba(91,156,246,.4); box-shadow: inset 0 1px 0 rgba(255,255,255,.1), 0 0 0 1px rgba(91,156,246,.15), 0 8px 32px rgba(0,0,0,.4); } .nc-head { padding: 16px 18px; display: flex; align-items: flex-start; justify-content: space-between; } +.nc-badges { display: flex; align-items: center; justify-content: flex-end; gap: 5px; } .nc-name { font-weight: 700; font-size: 14px; display: flex; align-items: center; gap: 8px; } .nc-ip { font-family: 'JetBrains Mono', monospace; font-size: 11px; color: var(--cyan); margin-top: 3px; } .nc-observed { font-size: 10px; color: var(--t2); margin: 3px 0 0 16px; } @@ -261,6 +262,7 @@ body { .drag-handle { position: absolute; left: 0; top: 0; bottom: 0; width: 4px; cursor: col-resize; z-index: 1; transition: background .15s; } .drag-handle:hover { background: rgba(255,255,255,.09); } .dp-head { padding: 16px 18px; border-bottom: 1px solid var(--t3); display: flex; align-items: center; justify-content: space-between; } +.dp-title-group { display: flex; align-items: center; min-width: 0; gap: 7px; } .dp-title { font-size: 13px; font-weight: 700; } .dp-close { background: none; border: none; color: var(--t2); cursor: pointer; font-size: 18px; padding: 0; transition: color .1s; } .dp-close:hover { color: var(--t0); } @@ -284,7 +286,9 @@ body { .legend-line { width: 11px; height: 2px; border-radius: 2px; display: inline-block; } .legend-line.arrivals { background: var(--cyan); } .legend-line.completions { background: var(--green); } -.pressure-chart { display: block; width: 100%; height: 82px; margin-top: 5px; overflow: visible; } +.pressure-plot-row { display: grid; grid-template-columns: 34px minmax(0, 1fr); gap: 6px; margin-top: 5px; } +.pressure-chart { display: block; width: 100%; height: 78px; overflow: visible; } +.chart-y-scale, .queue-y-scale { color: var(--t2); font-family: 'JetBrains Mono', monospace; font-size: 8px; display: flex; flex-direction: column; justify-content: space-between; text-align: right; } .chart-line { fill: none; stroke-width: 2; vector-effect: non-scaling-stroke; stroke-linejoin: round; stroke-linecap: round; } .chart-line.arrivals { stroke: var(--cyan); } .chart-line.completions { stroke: var(--green); } @@ -296,7 +300,8 @@ body { .queue-chart-row { height: 46px; display: grid; grid-template-columns: 64px minmax(0, 1fr) 58px; gap: 8px; align-items: center; padding: 5px 9px; } .queue-chart-row + .queue-chart-row { border-top: 1px solid var(--t3); } .queue-chart-row > span { color: var(--t2); font-size: 9px; } -.queue-chart-row svg { width: 100%; height: 30px; overflow: visible; } +.queue-plot { height: 31px; display: grid; grid-template-columns: 34px minmax(0, 1fr); gap: 5px; min-width: 0; } +.queue-chart-row svg { width: 100%; height: 31px; overflow: visible; } .queue-chart-row strong { color: var(--t1); font-family: 'JetBrains Mono', monospace; font-size: 9px; font-weight: 500; text-align: right; } .pressure-empty { min-height: 92px; border: 1px dashed var(--t3); border-radius: 9px; display: flex; align-items: center; justify-content: center; color: var(--t2); font-size: 10px; } .pressure-empty.small { min-height: 54px; } diff --git a/members/nullnet-server/ui/src/lib/pressureChart.ts b/members/nullnet-server/ui/src/lib/pressureChart.ts new file mode 100644 index 0000000..d5fb84c --- /dev/null +++ b/members/nullnet-server/ui/src/lib/pressureChart.ts @@ -0,0 +1,39 @@ +export const PRESSURE_RANGE_MS = 60_000; + +interface TimedWindow { + age_ms: number; + duration_ms: number; +} + +export function recentPressureWindows(windows: T[]) { + return windows + .filter(window => window.age_ms <= PRESSURE_RANGE_MS) + .toSorted((left, right) => right.age_ms - left.age_ms); +} + +function coordinate(value: number) { + return Number(value.toFixed(3)); +} + +export function timedLinePoints( + windows: T[], + value: (window: T) => number, + width: number, + height: number, + max: number, +) { + const points = [`0,${height}`]; + let cursor = 0; + for (const window of windows) { + const endAge = Math.min(PRESSURE_RANGE_MS, Math.max(0, window.age_ms)); + const startAge = Math.min(PRESSURE_RANGE_MS, endAge + window.duration_ms); + const endX = coordinate((1 - endAge / PRESSURE_RANGE_MS) * width); + const startX = coordinate(Math.max(cursor, (1 - startAge / PRESSURE_RANGE_MS) * width)); + const y = coordinate(height - (value(window) / max) * height); + if (startX > cursor) points.push(`${startX},${height}`); + points.push(`${endX},${y}`); + cursor = Math.max(cursor, endX); + } + if (cursor < width) points.push(`${width},${height}`); + return points.join(' '); +} diff --git a/members/nullnet-server/ui/src/pages/Nodes.tsx b/members/nullnet-server/ui/src/pages/Nodes.tsx index 8fa134b..77d5353 100644 --- a/members/nullnet-server/ui/src/pages/Nodes.tsx +++ b/members/nullnet-server/ui/src/pages/Nodes.tsx @@ -3,8 +3,9 @@ import { useState } from 'react'; import Layout from '../components/Layout'; import { useApi } from '../hooks/useApi'; import { useStack } from '../StackContext'; -import type { IngressControlState, NodeJson, PressureWindowJson } from '../types'; +import type { IngressControlState, NodeJson, PressureWindowJson, SetupPressureWindowJson } from '../types'; import { useDragResize } from '../hooks/useDragResize'; +import { recentPressureWindows, timedLinePoints } from '../lib/pressureChart'; function stateLabel(state: IngressControlState) { if (state === 'multiplexing') return 'Multiplexing'; @@ -40,20 +41,9 @@ function formatRate(value: number | null | undefined) { return value >= 100 ? Math.round(value).toLocaleString() : value.toFixed(1); } -function linePoints(values: number[], width: number, height: number, max: number) { - if (values.length === 1) { - const y = height - (values[0] / max) * height; - return `0,${y} ${width},${y}`; - } - return values.map((value, index) => { - const x = (index / (values.length - 1)) * width; - const y = height - (value / max) * height; - return `${x},${y}`; - }).join(' '); -} +type ThroughputWindow = Pick; -function RoutingChart({ windows }: { windows: PressureWindowJson[] }) { - if (windows.length === 0) return
Waiting for routing samples…
; +function RoutingChart({ windows }: { windows: ThroughputWindow[] }) { const arrivals = windows.map(window => window.arrivals_per_sec); const completions = windows.map(window => window.completions_per_sec); const max = Math.max(1, ...arrivals, ...completions); @@ -62,39 +52,61 @@ function RoutingChart({ windows }: { windows: PressureWindowJson[] }) {
Arrivals/s Completions/s - last {windows.length}s + last 60s +
+
+
{formatRate(max)}{formatRate(max / 2)}0
+ + + + window.arrivals_per_sec, 300, 76, max)} className="chart-line arrivals" /> + window.completions_per_sec, 300, 76, max)} className="chart-line completions" /> +
- - - - - - ); } -function QueueChart({ windows }: { windows: PressureWindowJson[] }) { - if (windows.length === 0) return null; - const waiters = windows.map(window => window.waiters); - const waits = windows.map(window => window.mean_wait_us / 1000); - const waiterMax = Math.max(1, ...waiters); - const waitMax = Math.max(1, ...waits); +function DetailCharts({ + windows, + pendingLabel, + latencyLabel, + pendingValue, + latencyUs, +}: { + windows: T[]; + pendingLabel: string; + latencyLabel: string; + pendingValue: (window: T) => number; + latencyUs: (window: T) => number; +}) { + const pending = windows.map(pendingValue); + const latencies = windows.map(window => latencyUs(window) / 1000); + const pendingMax = Math.max(1, ...pending); + const latencyMax = Math.max(1, ...latencies); return (
- Waiters - - - - {waiters.at(-1) ?? 0} + {pendingLabel} +
+
{formatRate(pendingMax)}0
+ + + + +
+ {pending.at(-1) ?? 0}
- Mean wait - - - - {formatWait(windows.at(-1)?.mean_wait_us)} + {latencyLabel} +
+
{formatWait(latencyMax * 1000)}0
+ + + latencyUs(window) / 1000, 220, 29, latencyMax)} className="chart-line wait" /> + +
+ {formatWait(windows.at(-1) ? latencyUs(windows.at(-1)!) : undefined)}
); @@ -130,24 +142,27 @@ export default function Nodes() {
{nodes?.map(node => { - const latest = node.pressure_windows.at(-1); + const latest = recentPressureWindows(node.setup_pressure_windows).at(-1); return (
select(node.ip)}>
{node.ip}
-
Observed {formatAge(node.pressure_age_ms)}
+
Setup observed {formatAge(node.setup_pressure_age_ms)}
+
+
+ {node.is_proxy && Proxy} + + {stateLabel(node.ingress_control.state)} +
- - {stateLabel(node.ingress_control.state)} -
-
Arrivals/s
{formatRate(latest?.arrivals_per_sec)}
-
Completions/s
{formatRate(latest?.completions_per_sec)}
-
Waiters
{latest?.waiters ?? '–'}
-
Mean wait
{formatWait(latest?.mean_wait_us)}
+
Setup/s
{formatRate(latest?.arrivals_per_sec)}
+
Completed/s
{formatRate(latest?.completions_per_sec)}
+
In flight
{latest?.in_flight ?? '–'}
+
Mean setup
{formatWait(latest?.mean_setup_us)}
@@ -172,7 +187,10 @@ export default function Nodes() {
- {selectedNode ? selectedNode.ip : '–'} +
+ {selectedNode ? selectedNode.ip : '–'} + {selectedNode?.is_proxy && Proxy} +
{selectedNode && }
@@ -192,12 +210,33 @@ export default function Nodes() {
-
Routing pressure
- - -
Latest proxy sample {formatAge(selectedNode.pressure_age_ms)}
+
Setup pressure
+ + + windows={recentPressureWindows(selectedNode.setup_pressure_windows)} + pendingLabel="In flight" + latencyLabel="Mean setup" + pendingValue={window => window.in_flight} + latencyUs={window => window.mean_setup_us} + /> +
Latest setup sample {formatAge(selectedNode.setup_pressure_age_ms)}
+ {selectedNode.is_proxy && ( +
+
Proxy routing pressure
+ + + windows={recentPressureWindows(selectedNode.pressure_windows)} + pendingLabel="Q waiters" + latencyLabel="Mean routing wait" + pendingValue={window => window.waiters} + latencyUs={window => window.mean_wait_us} + /> +
Latest proxy sample {formatAge(selectedNode.pressure_age_ms)}
+
+ )} +
Active ingress groups
diff --git a/members/nullnet-server/ui/src/types.ts b/members/nullnet-server/ui/src/types.ts index ef83bd3..fa54491 100644 --- a/members/nullnet-server/ui/src/types.ts +++ b/members/nullnet-server/ui/src/types.ts @@ -27,6 +27,7 @@ export interface HostedServiceJson { export interface NodeJson { ip: string; + is_proxy: boolean; hosted_services: HostedServiceJson[]; ingress_groups: { dedicated: number; @@ -44,6 +45,8 @@ export interface NodeJson { }; pressure_age_ms: number | null; pressure_windows: PressureWindowJson[]; + setup_pressure_age_ms: number | null; + setup_pressure_windows: SetupPressureWindowJson[]; } export type IngressControlState = 'normal' | 'multiplexing' | 'recovery'; @@ -66,6 +69,18 @@ export interface PressureWindowJson { trend: 'healthy' | 'growing' | 'neutral'; } +export interface SetupPressureWindowJson { + age_ms: number; + duration_ms: number; + arrivals: number; + completions: number; + arrivals_per_sec: number; + completions_per_sec: number; + in_flight: number; + mean_setup_us: number; + trend: 'healthy' | 'growing' | 'neutral'; +} + export interface SessionJson { id: number; network_id: number; diff --git a/members/nullnet-server/ui/tests/pressureChart.test.mjs b/members/nullnet-server/ui/tests/pressureChart.test.mjs new file mode 100644 index 0000000..e6d0eab --- /dev/null +++ b/members/nullnet-server/ui/tests/pressureChart.test.mjs @@ -0,0 +1,33 @@ +import assert from 'node:assert/strict'; +import { Buffer } from 'node:buffer'; +import { readFile } from 'node:fs/promises'; +import { test } from 'node:test'; +import ts from 'typescript'; + +const source = await readFile(new URL('../src/lib/pressureChart.ts', import.meta.url), 'utf8'); +const { outputText } = ts.transpileModule(source, { compilerOptions: { module: ts.ModuleKind.ESNext } }); +const module = await import(`data:text/javascript;base64,${Buffer.from(outputText).toString('base64')}`); +const { recentPressureWindows, timedLinePoints } = module; + +const windowAt = (age_ms, arrivals_per_sec = 1, duration_ms = 1000) => ({ age_ms, duration_ms, arrivals_per_sec }); + +test('the chart retains zero samples but never data older than 60 seconds', () => { + const recentZero = windowAt(10_000, 0); + const boundary = windowAt(60_000, 2); + const old = windowAt(60_001, 9); + + assert.deepEqual(recentPressureWindows([recentZero, old, boundary]), [boundary, recentZero]); +}); + +test('points use actual intervals and draw idle gaps as zero', () => { + const windows = [windowAt(40_000, 2, 20_000), windowAt(10_000, 5, 10_000)]; + + assert.equal( + timedLinePoints(windows, window => window.arrivals_per_sec, 300, 100, 10), + '0,100 100,80 200,100 250,50 300,100', + ); +}); + +test('a fully idle range is an explicit zero line', () => { + assert.equal(timedLinePoints([], () => 0, 300, 100, 1), '0,100 300,100'); +}); From 597faca3ff7ebeb3a9bbced0e91ee2240088abfd Mon Sep 17 00:00:00 2001 From: GyulyVGC Date: Wed, 21 Oct 2026 15:43:28 +0200 Subject: [PATCH 2/2] ui: scope event filters by subsystem --- .../nullnet-server/ui/src/lib/eventKinds.ts | 116 +++++++++++++----- 1 file changed, 83 insertions(+), 33 deletions(-) diff --git a/members/nullnet-server/ui/src/lib/eventKinds.ts b/members/nullnet-server/ui/src/lib/eventKinds.ts index e5addf7..0580400 100644 --- a/members/nullnet-server/ui/src/lib/eventKinds.ts +++ b/members/nullnet-server/ui/src/lib/eventKinds.ts @@ -27,67 +27,107 @@ export const EVENT_KIND_GROUPS = defineEventKindGroups([ ], }, { - label: 'Server', + label: 'Configuration & Observation', kinds: [ - 'all_replicas_removed', - 'backend_trigger_setup_bailed', - 'chain_owner_lost', 'config_reloaded', 'config_stack_removed', - 'edge_promotion_lost', - 'egress_reservation_reclaimed', 'file_watch_failed', - 'ingress_congestion_control_changed', - 'ingress_congestion_recovery_started', - 'max_networks_limit_enforced', - 'net_id_pool_exhausted', - 'net_teardown_unconfirmed', - 'node_connected', - 'node_disconnected', 'observation_changed', 'port_mapping_conflict', - 'proxy_chain_setup_failed', - 'proxy_client_timed_out', 'route_conflict', + ], + }, + { + label: 'Nodes & Services', + kinds: [ + 'all_replicas_removed', + 'node_connected', + 'node_disconnected', 'service_declaration_skipped', 'service_name_conflict', 'service_reachability_toggled', 'service_registered', 'service_unregistered', - 'session_created', - 'session_torn_down', + ], + }, + { + label: 'Ingress Control', + kinds: [ + 'ingress_congestion_control_changed', + 'ingress_congestion_recovery_started', + 'max_networks_limit_enforced', + 'proxy_chain_setup_failed', + 'proxy_client_timed_out', + 'stale_session_evicted', + ], + }, + { + label: 'Edge Lifecycle', + kinds: [ + 'backend_trigger_setup_bailed', + 'chain_owner_lost', + 'edge_promotion_lost', + 'egress_reservation_reclaimed', + 'net_id_pool_exhausted', + 'net_teardown_unconfirmed', 'setup_ack', 'setup_started', 'setup_timeout', - 'stale_session_evicted', ], }, { - label: 'Client', + label: 'Sessions', + kinds: [ + 'session_created', + 'session_torn_down', + ], + }, + { + label: 'Client Control', kinds: [ - 'backend_trigger_send_failed', - 'backend_trigger_setup_timed_out', - 'conntrack_flush_failed', - 'conntrack_subscribe_failed', - 'container_resume_failed', - 'container_suspend_failed', 'control_channel_ack_failed', 'control_channel_closed', 'control_channel_established', - 'device_event_bypass_changed', + 'services_list_update_failed', + 'services_list_updated', + ], + }, + { + label: 'Container Lifecycle', + kinds: [ + 'container_resume_failed', + 'container_suspend_failed', + 'host_mapping_failed', + ], + }, + { + label: 'Traffic Steering', + kinds: [ + 'backend_trigger_send_failed', + 'backend_trigger_setup_timed_out', 'dnat_install_failed', 'dnat_removal_failed', - 'egress_policy_check_failed', 'egress_steer_install_failed', 'egress_steer_setup_timed_out', 'egress_trigger_send_failed', - 'firewall_rules_load_failed', 'gateway_forward_install_failed', - 'host_mapping_failed', + ], + }, + { + label: 'Policy & Conntrack', + kinds: [ + 'conntrack_flush_failed', + 'conntrack_subscribe_failed', + 'egress_policy_check_failed', + 'firewall_rules_load_failed', 'mss_clamp_install_failed', 'nfqueue_bind_failed', - 'services_list_update_failed', - 'services_list_updated', + ], + }, + { + label: 'Network Devices', + kinds: [ + 'device_event_bypass_changed', 'vlan_setup_completed', 'vlan_setup_failed', 'vlan_teardown_failed', @@ -98,15 +138,25 @@ export const EVENT_KIND_GROUPS = defineEventKindGroups([ ], }, { - label: 'Proxy', + label: 'Proxy Connectivity', kinds: [ 'proxy_client_not_inet', 'proxy_connected', 'proxy_disconnected', + ], + }, + { + label: 'Proxy Listeners', + kinds: [ 'tcp_listener_bind_failed', - 'tcp_upstream_connect_failed', 'tls_certificate_invalid', 'udp_listener_bind_failed', + ], + }, + { + label: 'Proxy Upstreams', + kinds: [ + 'tcp_upstream_connect_failed', 'udp_upstream_connect_failed', 'upstream_ip_parse_failed', 'upstream_lookup_failed',