diff --git a/src/rust/lqos_config/src/network_json.rs b/src/rust/lqos_config/src/network_json.rs index 500584a3..3d7c0681 100644 --- a/src/rust/lqos_config/src/network_json.rs +++ b/src/rust/lqos_config/src/network_json.rs @@ -228,6 +228,62 @@ impl NetworkJson { } } + /// Carry forward matched nodes' live per-cycle counters from a previously loaded active tree. + /// + /// Matching prefers stable node identifiers when present. If no identifier is available, + /// this falls back to node name only when that name is unique in both trees to avoid + /// attaching live counters to the wrong node after a reload. + /// + /// Side effects: copies current counters and `rtt_buffer` values into `self`. + pub fn carry_forward_live_counters_from(&mut self, previous: &Self) { + let mut previous_by_id: HashMap = HashMap::new(); + let mut previous_name_counts: HashMap = HashMap::new(); + let mut current_name_counts: HashMap = HashMap::new(); + + for node in &previous.nodes { + if let Some(id) = node.id.as_deref() { + previous_by_id.insert(id.to_string(), node); + } + *previous_name_counts.entry(node.name.clone()).or_default() += 1; + } + + for node in &self.nodes { + *current_name_counts.entry(node.name.clone()).or_default() += 1; + } + + let mut previous_by_unique_name: HashMap = HashMap::new(); + for node in &previous.nodes { + if previous_name_counts.get(&node.name) == Some(&1) { + previous_by_unique_name.insert(node.name.clone(), node); + } + } + + for node in &mut self.nodes { + let preserved = node + .id + .as_deref() + .and_then(|id| previous_by_id.get(id).copied()) + .or_else(|| { + (current_name_counts.get(&node.name) == Some(&1)) + .then(|| node.name.clone()) + .and_then(|name| previous_by_unique_name.get(&name).copied()) + }); + + if let Some(previous_node) = preserved { + node.current_throughput = previous_node.current_throughput; + node.current_packets = previous_node.current_packets; + node.current_tcp_packets = previous_node.current_tcp_packets; + node.current_udp_packets = previous_node.current_udp_packets; + node.current_icmp_packets = previous_node.current_icmp_packets; + node.current_tcp_retransmits = previous_node.current_tcp_retransmits; + node.current_tcp_retransmit_packets = previous_node.current_tcp_retransmit_packets; + node.current_marks = previous_node.current_marks; + node.current_drops = previous_node.current_drops; + node.rtt_buffer = previous_node.rtt_buffer.clone(); + } + } + } + /// Sets all current throughput values to zero /// Note that due to interior mutability, this does not require mutable /// access. @@ -892,6 +948,93 @@ mod test { ); } + #[test] + fn carry_forward_live_counters_preserves_matched_runtime_state() { + let mut previous = parse_network_json_from_value(serde_json::json!({ + "Parent B": { + "children": { + "Site 1": { + "id": "site-1", + "children": {} + } + } + } + })); + let mut next = parse_network_json_from_value(serde_json::json!({ + "Parent B": { + "children": { + "Site 1": { + "id": "site-1", + "children": {} + } + } + } + })); + + let previous_site = previous + .nodes + .iter_mut() + .find(|node| node.id.as_deref() == Some("site-1")) + .expect("previous Site 1 should exist"); + previous_site.current_throughput.down = 12_500_000; + previous_site.current_throughput.up = 6_250_000; + previous_site.current_tcp_packets.down = 1200; + previous_site.current_tcp_retransmits.down = 12; + previous_site.current_tcp_retransmit_packets.down = 400; + previous_site.current_marks.up = 7; + previous_site.current_drops.down = 3; + previous_site.rtt_buffer = RttBuffer::new( + lqos_utils::rtt::RttData::from_nanos(18_000_000), + FlowbeeEffectiveDirection::Download, + 1, + ); + previous_site.rtt_buffer.push( + lqos_utils::rtt::RttData::from_nanos(18_000_000), + FlowbeeEffectiveDirection::Download, + 2, + ); + previous_site.rtt_buffer.push( + lqos_utils::rtt::RttData::from_nanos(27_000_000), + FlowbeeEffectiveDirection::Upload, + 3, + ); + previous_site.rtt_buffer.push( + lqos_utils::rtt::RttData::from_nanos(27_000_000), + FlowbeeEffectiveDirection::Upload, + 4, + ); + + next.carry_forward_live_counters_from(&previous); + + let next_site = next + .nodes + .iter() + .find(|node| node.id.as_deref() == Some("site-1")) + .expect("next Site 1 should exist"); + + assert_eq!(next_site.current_throughput.down, 12_500_000); + assert_eq!(next_site.current_throughput.up, 6_250_000); + assert_eq!(next_site.current_tcp_packets.down, 1200); + assert_eq!(next_site.current_tcp_retransmits.down, 12); + assert_eq!(next_site.current_tcp_retransmit_packets.down, 400); + assert_eq!(next_site.current_marks.up, 7); + assert_eq!(next_site.current_drops.down, 3); + assert_eq!( + next_site + .rtt_buffer + .percentile(RttBucket::Current, FlowbeeEffectiveDirection::Download, 50) + .map(|rtt| rtt.as_millis()), + Some(20.0) + ); + assert_eq!( + next_site + .rtt_buffer + .percentile(RttBucket::Current, FlowbeeEffectiveDirection::Upload, 50) + .map(|rtt| rtt.as_millis()), + Some(30.0) + ); + } + #[test] fn path_for_config_prefers_runtime_effective_tree_for_integration_ingress() { let temp = unique_temp_dir("network-json-integration-effective"); diff --git a/src/rust/lqos_network_devices/src/state.rs b/src/rust/lqos_network_devices/src/state.rs index 9b7b9c35..7a82977d 100644 --- a/src/rust/lqos_network_devices/src/state.rs +++ b/src/rust/lqos_network_devices/src/state.rs @@ -81,14 +81,18 @@ pub(crate) fn swap_shaped_devices_snapshot( old.shaped.clone() } -fn network_json_with_carried_heatmaps(previous: &NetworkJson, mut next: NetworkJson) -> NetworkJson { +fn network_json_with_carried_runtime_state( + previous: &NetworkJson, + mut next: NetworkJson, +) -> NetworkJson { next.carry_forward_heatmaps_from(previous); + next.carry_forward_live_counters_from(previous); next } pub(crate) fn publish_network_json(new_file: NetworkJson) { let mut writer = NETWORK_JSON.write(); - *writer = network_json_with_carried_heatmaps(&writer, new_file); + *writer = network_json_with_carried_runtime_state(&writer, new_file); } pub(crate) fn publish_dynamic_circuits_snapshot(new_snapshot: Vec) { @@ -125,7 +129,7 @@ pub(crate) fn refresh_dynamic_circuits_last_seen_for_hashes( #[cfg(test)] mod tests { - use super::network_json_with_carried_heatmaps; + use super::network_json_with_carried_runtime_state; use lqos_config::NetworkJsonNode; use lqos_utils::{ qoq_heatmap::TemporalQoqHeatmap, @@ -211,12 +215,36 @@ mod tests { ); let qoq_heatmap = site.qoq_heatmap.get_or_insert_with(TemporalQoqHeatmap::new); qoq_heatmap.add_sample(Some(91.0), Some(82.0)); + site.current_throughput.down = 12_500_000; + site.current_throughput.up = 6_250_000; + site.current_tcp_retransmits.down = 15; + site.current_tcp_retransmit_packets.down = 500; + site.rtt_buffer = RttBuffer::new( + lqos_utils::rtt::RttData::from_nanos(18_000_000), + lqos_utils::rtt::FlowbeeEffectiveDirection::Download, + 1, + ); + site.rtt_buffer.push( + lqos_utils::rtt::RttData::from_nanos(18_000_000), + lqos_utils::rtt::FlowbeeEffectiveDirection::Download, + 2, + ); + site.rtt_buffer.push( + lqos_utils::rtt::RttData::from_nanos(27_000_000), + lqos_utils::rtt::FlowbeeEffectiveDirection::Upload, + 3, + ); + site.rtt_buffer.push( + lqos_utils::rtt::RttData::from_nanos(27_000_000), + lqos_utils::rtt::FlowbeeEffectiveDirection::Upload, + 4, + ); let next = lqos_config::NetworkJson { nodes: vec![root_node(), site_node("Tower A Renamed", "tower-a")], }; - let carried = network_json_with_carried_heatmaps(&previous, next); + let carried = network_json_with_carried_runtime_state(&previous, next); let site = carried .nodes .iter() @@ -239,5 +267,29 @@ mod tests { assert_eq!(blocks.retransmit_up[14], Some(2.5)); assert_eq!(qoq_blocks.download_total[14], Some(91.0)); assert_eq!(qoq_blocks.upload_total[14], Some(82.0)); + assert_eq!(site.current_throughput.down, 12_500_000); + assert_eq!(site.current_throughput.up, 6_250_000); + assert_eq!(site.current_tcp_retransmits.down, 15); + assert_eq!(site.current_tcp_retransmit_packets.down, 500); + assert_eq!( + site.rtt_buffer + .percentile( + lqos_utils::rtt::RttBucket::Current, + lqos_utils::rtt::FlowbeeEffectiveDirection::Download, + 50, + ) + .map(|rtt| rtt.as_millis()), + Some(20.0) + ); + assert_eq!( + site.rtt_buffer + .percentile( + lqos_utils::rtt::RttBucket::Current, + lqos_utils::rtt::FlowbeeEffectiveDirection::Upload, + 50, + ) + .map(|rtt| rtt.as_millis()), + Some(30.0) + ); } }