diff --git a/src/rust/lqos_config/src/mikrotik_ipv6_credentials.rs b/src/rust/lqos_config/src/mikrotik_ipv6_credentials.rs index d257e2f8..c292309a 100644 --- a/src/rust/lqos_config/src/mikrotik_ipv6_credentials.rs +++ b/src/rust/lqos_config/src/mikrotik_ipv6_credentials.rs @@ -335,40 +335,44 @@ mod tests { }; use crate::Config; use std::fs; - use std::path::PathBuf; + use std::path::{Path, PathBuf}; use std::time::{SystemTime, UNIX_EPOCH}; fn temp_path(label: &str) -> PathBuf { let stamp = SystemTime::now() .duration_since(UNIX_EPOCH) - .unwrap() + .expect("system clock should be after UNIX_EPOCH") .as_nanos(); std::env::temp_dir().join(format!("libreqos-mikrotik-{label}-{stamp}")) } - fn test_config(root: &PathBuf) -> Config { - let mut config = Config::default(); - config.lqos_directory = root.join("src").display().to_string(); - config.mikrotik_ipv6.config_path = root - .join("etc/libreqos/mikrotik_ipv6.toml") - .display() - .to_string(); - config + fn test_config(root: &Path) -> Config { + Config { + lqos_directory: root.join("src").display().to_string(), + mikrotik_ipv6: crate::MikrotikIpv6Config { + config_path: root + .join("etc/libreqos/mikrotik_ipv6.toml") + .display() + .to_string(), + }, + ..Config::default() + } } #[test] fn migrates_legacy_csv_and_deletes_it() { let root = temp_path("migrate"); let src = root.join("src"); - fs::create_dir_all(&src).unwrap(); + fs::create_dir_all(&src).expect("test should create legacy src dir"); fs::write( src.join(LEGACY_MIKROTIK_CSV_FILENAME), "Router Name / ID,IP,API Username,API Password, API Port\nmain,100.64.0.1,admin,password,8728\n", ) - .unwrap(); + .expect("test should write legacy mikrotik csv"); let config = test_config(&root); - let routers = load_mikrotik_ipv6_router_credentials(&config).unwrap(); + let routers = load_mikrotik_ipv6_router_credentials(&config) + .expect("legacy csv should migrate into TOML"); assert_eq!(routers.len(), 1); assert_eq!(routers[0].name, "main"); @@ -376,7 +380,8 @@ mod tests { assert!(root.join("etc/libreqos/mikrotik_ipv6.toml").exists()); assert!(!src.join(LEGACY_MIKROTIK_CSV_FILENAME).exists()); - let saved = read_credentials_toml(&root.join("etc/libreqos/mikrotik_ipv6.toml")).unwrap(); + let saved = read_credentials_toml(&root.join("etc/libreqos/mikrotik_ipv6.toml")) + .expect("migrated TOML should be readable"); assert_eq!(saved.router, routers); let _ = fs::remove_dir_all(root); @@ -387,21 +392,22 @@ mod tests { let root = temp_path("prefer-toml"); let src = root.join("src"); let etc = root.join("etc/libreqos"); - fs::create_dir_all(&src).unwrap(); - fs::create_dir_all(&etc).unwrap(); + fs::create_dir_all(&src).expect("test should create src dir"); + fs::create_dir_all(&etc).expect("test should create etc dir"); fs::write( etc.join("mikrotik_ipv6.toml"), "version = 1\n\n[[router]]\nname = \"main\"\nhost = \"100.64.0.1\"\nport = 8728\nusername = \"admin\"\npassword = \"secret\"\nuse_ssl = false\nplaintext_login = true\n", ) - .unwrap(); + .expect("test should write current TOML"); fs::write( src.join(LEGACY_MIKROTIK_CSV_FILENAME), "Router Name / ID,IP,API Username,API Password, API Port\nlegacy,100.64.0.2,user,pass,8728\n", ) - .unwrap(); + .expect("test should write legacy csv"); let config = test_config(&root); - let routers = load_mikrotik_ipv6_router_credentials(&config).unwrap(); + let routers = load_mikrotik_ipv6_router_credentials(&config) + .expect("current TOML should take precedence"); assert_eq!(routers.len(), 1); assert_eq!(routers[0].name, "main"); diff --git a/src/rust/lqos_config/src/network_json.rs b/src/rust/lqos_config/src/network_json.rs index a6615689..a25b198a 100644 --- a/src/rust/lqos_config/src/network_json.rs +++ b/src/rust/lqos_config/src/network_json.rs @@ -6,7 +6,7 @@ use lqos_utils::{ qoo::{LossMeasurement, QoqScores, compute_qoq_scores}, qoq_heatmap::TemporalQoqHeatmap, rtt::{FlowbeeEffectiveDirection, RttBucket, RttBuffer}, - temporal_heatmap::TemporalHeatmap, + temporal_heatmap::{TemporalHeatmap, executive_retransmit_percent}, units::DownUpOrder, }; pub use network_json_node::NetworkJsonNode; @@ -367,11 +367,11 @@ impl NetworkJson { .rtt_buffer .percentile(RttBucket::Current, FlowbeeEffectiveDirection::Upload, 90) .map(|rtt| rtt.as_millis() as f32); - let retransmit_down = retransmit_percent( + let retransmit_down = executive_retransmit_percent( node.current_tcp_retransmits.down, node.current_tcp_retransmit_packets.down, ); - let retransmit_up = retransmit_percent( + let retransmit_up = executive_retransmit_percent( node.current_tcp_retransmits.up, node.current_tcp_retransmit_packets.up, ); @@ -516,14 +516,6 @@ fn utilization_percent_bytes(bytes: u64, max_mbps: f64) -> Option { Some(((bits_per_second / capacity_bps) * 100.0) as f32) } -fn retransmit_percent(retransmits: u64, packets: u64) -> Option { - // Ignore very small samples to avoid extreme ratios from tiny flows. - if retransmits == 0 || packets < 1_000 { - return None; - } - Some((retransmits as f32 / packets as f32) * 100.0) -} - fn tcp_retransmit_loss_proxy(retransmits: u64, packets: u64) -> Option { if packets == 0 { return None; @@ -922,7 +914,10 @@ mod test { config.uisp_integration.enable_uisp = true; let selected = NetworkJson::path_for_config(&config); - assert_eq!(selected, config.topology_state_file_path("network.effective.json")); + assert_eq!( + selected, + config.topology_state_file_path("network.effective.json") + ); } #[test] @@ -956,6 +951,70 @@ mod test { assert_eq!(selected, temp.join("network.effective.json")); } + #[test] + fn site_heatmap_suppresses_low_confidence_retransmits() { + let mut parsed = parse_network_json_from_value(serde_json::json!({ + "TinySite": { + "downloadBandwidthMbps": 100, + "uploadBandwidthMbps": 100, + "children": {} + } + })); + let site = parsed + .nodes + .iter_mut() + .find(|node| node.name == "TinySite") + .expect("TinySite should be present"); + site.current_throughput.down = 12_500_000; + site.current_tcp_retransmits.down = 1; + site.current_tcp_retransmit_packets.down = 99; + + parsed.record_site_heatmaps(true); + + let blocks = parsed + .nodes + .iter() + .find(|node| node.name == "TinySite") + .expect("TinySite should be present") + .heatmap + .as_ref() + .expect("site heatmap should exist") + .blocks(); + assert_eq!(blocks.retransmit_down.last().copied().flatten(), None); + } + + #[test] + fn site_heatmap_keeps_qualified_retransmits() { + let mut parsed = parse_network_json_from_value(serde_json::json!({ + "BusySite": { + "downloadBandwidthMbps": 100, + "uploadBandwidthMbps": 100, + "children": {} + } + })); + let site = parsed + .nodes + .iter_mut() + .find(|node| node.name == "BusySite") + .expect("BusySite should be present"); + site.current_throughput.down = 12_500_000; + site.current_tcp_retransmits.down = 2; + site.current_tcp_retransmit_packets.down = 100; + + parsed.record_site_heatmaps(true); + + let blocks = parsed + .nodes + .iter() + .find(|node| node.name == "BusySite") + .expect("BusySite should be present") + .heatmap + .as_ref() + .expect("site heatmap should exist") + .blocks(); + assert_eq!(blocks.retransmit_down.last().copied().flatten(), Some(2.0)); + } + fn unique_temp_dir(prefix: &str) -> PathBuf { let unique = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) diff --git a/src/rust/lqos_config/src/runtime_state_migration.rs b/src/rust/lqos_config/src/runtime_state_migration.rs index aff5f5fd..a08f6f0d 100644 --- a/src/rust/lqos_config/src/runtime_state_migration.rs +++ b/src/rust/lqos_config/src/runtime_state_migration.rs @@ -489,18 +489,18 @@ mod tests { use super::migrate_legacy_runtime_state; use crate::Config; use std::fs; - use std::path::PathBuf; + use std::path::{Path, PathBuf}; use std::time::{SystemTime, UNIX_EPOCH}; fn temp_path(label: &str) -> PathBuf { let stamp = SystemTime::now() .duration_since(UNIX_EPOCH) - .unwrap() + .expect("system clock should be after UNIX_EPOCH") .as_nanos(); std::env::temp_dir().join(format!("libreqos-state-migration-{label}-{stamp}")) } - fn test_config(root: &PathBuf) -> Config { + fn test_config(root: &Path) -> Config { Config { lqos_directory: root.join("src").display().to_string(), state_directory: Some(root.join("state").display().to_string()), @@ -512,24 +512,30 @@ mod tests { fn migrates_legacy_runtime_files_into_state_directory() { let root = temp_path("migrate-files"); let src = root.join("src"); - fs::create_dir_all(&src).unwrap(); - fs::write(src.join("lastGoodConfig.csv"), "legacy-csv\n").unwrap(); - fs::write(src.join("lastRun.txt"), "legacy-run\n").unwrap(); - fs::write(src.join("cpu_topology_cache.json"), "{}\n").unwrap(); + fs::create_dir_all(&src).expect("test should create src dir"); + fs::write(src.join("lastGoodConfig.csv"), "legacy-csv\n") + .expect("test should write legacy shaping csv"); + fs::write(src.join("lastRun.txt"), "legacy-run\n") + .expect("test should write legacy stats file"); + fs::write(src.join("cpu_topology_cache.json"), "{}\n") + .expect("test should write legacy cache file"); let config = test_config(&root); - migrate_legacy_runtime_state(&config).unwrap(); + migrate_legacy_runtime_state(&config).expect("legacy runtime files should migrate"); assert_eq!( - fs::read_to_string(root.join("state/shaping/lastGoodConfig.csv")).unwrap(), + fs::read_to_string(root.join("state/shaping/lastGoodConfig.csv")) + .expect("migrated shaping file should be readable"), "legacy-csv\n" ); assert_eq!( - fs::read_to_string(root.join("state/stats/lastRun.txt")).unwrap(), + fs::read_to_string(root.join("state/stats/lastRun.txt")) + .expect("migrated stats file should be readable"), "legacy-run\n" ); assert_eq!( - fs::read_to_string(root.join("state/cache/cpu_topology_cache.json")).unwrap(), + fs::read_to_string(root.join("state/cache/cpu_topology_cache.json")) + .expect("migrated cache file should be readable"), "{}\n" ); assert!(!src.join("lastGoodConfig.csv").exists()); @@ -544,17 +550,21 @@ mod tests { let root = temp_path("duplicate"); let src = root.join("src"); let state = root.join("state/shaping"); - fs::create_dir_all(&src).unwrap(); - fs::create_dir_all(&state).unwrap(); - fs::write(src.join("lastGoodConfig.csv"), "legacy\n").unwrap(); - fs::write(state.join("lastGoodConfig.csv"), "state\n").unwrap(); + fs::create_dir_all(&src).expect("test should create src dir"); + fs::create_dir_all(&state).expect("test should create state dir"); + fs::write(src.join("lastGoodConfig.csv"), "legacy\n") + .expect("test should write legacy file"); + fs::write(state.join("lastGoodConfig.csv"), "state\n") + .expect("test should write destination file"); let config = test_config(&root); - migrate_legacy_runtime_state(&config).unwrap(); + migrate_legacy_runtime_state(&config) + .expect("duplicate legacy file should be removed safely"); assert!(!src.join("lastGoodConfig.csv").exists()); assert_eq!( - fs::read_to_string(state.join("lastGoodConfig.csv")).unwrap(), + fs::read_to_string(state.join("lastGoodConfig.csv")) + .expect("state file should still be readable"), "state\n" ); @@ -565,17 +575,20 @@ mod tests { fn quarantines_obsolete_backups_and_moves_stale_directory() { let root = temp_path("quarantine"); let src = root.join("src"); - fs::create_dir_all(src.join(".topology_stale")).unwrap(); + fs::create_dir_all(src.join(".topology_stale")) + .expect("test should create stale topology dir"); fs::write( src.join("network.json.treeguard-polluted.20260322T014345Z.bak"), "polluted\n", ) - .unwrap(); - fs::write(src.join("network.json.pre_50k_generator.bak"), "backup\n").unwrap(); - fs::write(src.join(".topology_stale/stale.json"), "{}\n").unwrap(); + .expect("test should write polluted backup"); + fs::write(src.join("network.json.pre_50k_generator.bak"), "backup\n") + .expect("test should write backup file"); + fs::write(src.join(".topology_stale/stale.json"), "{}\n") + .expect("test should write stale topology file"); let config = test_config(&root); - migrate_legacy_runtime_state(&config).unwrap(); + migrate_legacy_runtime_state(&config).expect("obsolete backups should be quarantined"); assert!( root.join( @@ -605,18 +618,19 @@ mod tests { fn migrates_visp_token_cache_files() { let root = temp_path("visp-cache"); let src = root.join("src"); - fs::create_dir_all(&src).unwrap(); + fs::create_dir_all(&src).expect("test should create src dir"); fs::write( src.join(".visp_token_cache_deadbeef.json"), "{\"token\":1}\n", ) - .unwrap(); + .expect("test should write legacy visp cache"); let config = test_config(&root); - migrate_legacy_runtime_state(&config).unwrap(); + migrate_legacy_runtime_state(&config).expect("visp cache should migrate"); assert_eq!( - fs::read_to_string(root.join("state/cache/.visp_token_cache_deadbeef.json")).unwrap(), + fs::read_to_string(root.join("state/cache/.visp_token_cache_deadbeef.json")) + .expect("migrated visp cache should be readable"), "{\"token\":1}\n" ); assert!(!src.join(".visp_token_cache_deadbeef.json").exists()); diff --git a/src/rust/lqos_config/src/shaped_devices/mod.rs b/src/rust/lqos_config/src/shaped_devices/mod.rs index 8e576bfa..876920ed 100644 --- a/src/rust/lqos_config/src/shaped_devices/mod.rs +++ b/src/rust/lqos_config/src/shaped_devices/mod.rs @@ -499,13 +499,15 @@ mod test { assert_eq!(config.devices[0].circuit_hash, hash_to_i64("circuit-1")); assert_eq!(config.devices[0].device_hash, hash_to_i64("device-1")); assert_eq!(config.devices[0].parent_hash, hash_to_i64("Parent-A")); + let test_ip = "192.168.1.10" + .parse() + .expect("test IP literal should parse"); assert_eq!( - config - .get_circuit_hash_from_ip(&XdpIpAddress::from_ip("192.168.1.10".parse().unwrap())), + config.get_circuit_hash_from_ip(&XdpIpAddress::from_ip(test_ip)), Some(hash_to_i64("circuit-1")) ); let matched = config - .get_device_from_ip(&XdpIpAddress::from_ip("192.168.1.10".parse().unwrap())) + .get_device_from_ip(&XdpIpAddress::from_ip(test_ip)) .expect("device should resolve by IP"); assert_eq!(matched.device_hash, hash_to_i64("device-1")); assert_eq!(matched.circuit_id, "circuit-1"); diff --git a/src/rust/lqos_config/src/topology_canonical_state.rs b/src/rust/lqos_config/src/topology_canonical_state.rs index 16d5177d..3fa734c1 100644 --- a/src/rust/lqos_config/src/topology_canonical_state.rs +++ b/src/rust/lqos_config/src/topology_canonical_state.rs @@ -1320,8 +1320,8 @@ mod tests { let node = canonical .find_node("site-a") .expect("expected canonical node"); - assert_eq!(node.latitude, Some(31.86102867126465_f32)); - assert_eq!(node.longitude, Some(-106.5494613647461_f32)); + assert_eq!(node.latitude, Some(31.861_029_f32)); + assert_eq!(node.longitude, Some(-106.549_46_f32)); let logical = canonical.insight_topology_network_json(); let site = logical diff --git a/src/rust/lqos_utils/src/temporal_heatmap.rs b/src/rust/lqos_utils/src/temporal_heatmap.rs index 283d332a..a9a5e3a4 100644 --- a/src/rust/lqos_utils/src/temporal_heatmap.rs +++ b/src/rust/lqos_utils/src/temporal_heatmap.rs @@ -3,6 +3,10 @@ const RAW_SAMPLES: usize = 60; const SUMMARY_BLOCKS: usize = 14; const TOTAL_BLOCKS: usize = SUMMARY_BLOCKS + 1; +/// Minimum TCP packet count required before executive retransmit heatmaps color a sample. +pub const EXECUTIVE_RETRANSMIT_MIN_PACKETS: u64 = 100; +/// Minimum TCP retransmit count required before executive retransmit heatmaps color a sample. +pub const EXECUTIVE_RETRANSMIT_MIN_RETRANSMITS: u64 = 2; use allocative::Allocative; use serde::{Deserialize, Serialize}; @@ -270,6 +274,20 @@ impl TemporalHeatmap { } } +/// Compute the executive-summary retransmit percentage for a heatmap sample. +/// +/// Returns `None` when the sample is too small to be considered trustworthy for the +/// executive and queue-dynamics summary heatmaps. +pub fn executive_retransmit_percent(retransmits: u64, packets: u64) -> Option { + if retransmits < EXECUTIVE_RETRANSMIT_MIN_RETRANSMITS + || packets < EXECUTIVE_RETRANSMIT_MIN_PACKETS + { + return None; + } + let value = (retransmits as f32 / packets as f32) * 100.0; + if value > 50.0 { None } else { Some(value) } +} + impl Default for TemporalHeatmap { fn default() -> Self { Self::new() @@ -278,7 +296,10 @@ impl Default for TemporalHeatmap { #[cfg(test)] mod tests { - use super::{RAW_SAMPLES, SUMMARY_BLOCKS, TOTAL_BLOCKS, TemporalHeatmap}; + use super::{ + EXECUTIVE_RETRANSMIT_MIN_PACKETS, EXECUTIVE_RETRANSMIT_MIN_RETRANSMITS, RAW_SAMPLES, + SUMMARY_BLOCKS, TOTAL_BLOCKS, TemporalHeatmap, executive_retransmit_percent, + }; #[test] fn new_is_empty() { @@ -356,4 +377,48 @@ mod tests { assert_eq!(blocks.rtt_p50_down[TOTAL_BLOCKS - 1], Some(3.0)); assert_eq!(blocks.rtt_p50_up[TOTAL_BLOCKS - 1], Some(3.0)); } + + #[test] + fn executive_retransmit_percent_requires_confidence_floor() { + assert_eq!( + executive_retransmit_percent( + EXECUTIVE_RETRANSMIT_MIN_RETRANSMITS - 1, + EXECUTIVE_RETRANSMIT_MIN_PACKETS, + ), + None + ); + assert_eq!( + executive_retransmit_percent( + EXECUTIVE_RETRANSMIT_MIN_RETRANSMITS, + EXECUTIVE_RETRANSMIT_MIN_PACKETS - 1, + ), + None + ); + assert_eq!( + executive_retransmit_percent( + EXECUTIVE_RETRANSMIT_MIN_RETRANSMITS, + EXECUTIVE_RETRANSMIT_MIN_PACKETS, + ), + Some(2.0) + ); + } + + #[test] + fn executive_retransmit_percent_rejects_implausible_ratio() { + assert_eq!(executive_retransmit_percent(60, 100), None); + assert_eq!(executive_retransmit_percent(50, 100), Some(50.0)); + } + + #[test] + fn retransmit_median_ignores_suppressed_samples() { + let mut heatmap = TemporalHeatmap::new(); + heatmap.add_sample(10.0, 10.0, None, None, None, None, None, None); + heatmap.add_sample(10.0, 10.0, None, None, None, None, Some(2.0), Some(4.0)); + heatmap.add_sample(10.0, 10.0, None, None, None, None, Some(6.0), Some(8.0)); + + let blocks = heatmap.blocks(); + assert_eq!(blocks.retransmit_down[TOTAL_BLOCKS - 1], Some(4.0)); + assert_eq!(blocks.retransmit_up[TOTAL_BLOCKS - 1], Some(6.0)); + assert_eq!(blocks.retransmit[TOTAL_BLOCKS - 1], Some(5.0)); + } } diff --git a/src/rust/lqosd/src/throughput_tracker/flow_data/asn_heatmap.rs b/src/rust/lqosd/src/throughput_tracker/flow_data/asn_heatmap.rs index 1da4e912..29734b41 100644 --- a/src/rust/lqosd/src/throughput_tracker/flow_data/asn_heatmap.rs +++ b/src/rust/lqosd/src/throughput_tracker/flow_data/asn_heatmap.rs @@ -1,6 +1,6 @@ use fxhash::FxHashMap; use lqos_utils::{ - temporal_heatmap::{HeatmapBlocks, TemporalHeatmap}, + temporal_heatmap::{HeatmapBlocks, TemporalHeatmap, executive_retransmit_percent}, units::DownUpOrder, }; use once_cell::sync::Lazy; @@ -54,8 +54,9 @@ impl AsnHeatmapStore { for (asn, mut aggregate) in aggregates.drain() { let (p50, p90) = p50_p90(&mut aggregate.rtts); let retransmit_down = - retransmit_percent(aggregate.retransmits.down, aggregate.packets.down); - let retransmit_up = retransmit_percent(aggregate.retransmits.up, aggregate.packets.up); + executive_retransmit_percent(aggregate.retransmits.down, aggregate.packets.down); + let retransmit_up = + executive_retransmit_percent(aggregate.retransmits.up, aggregate.packets.up); let entry = self.entries.entry(asn).or_insert_with(|| AsnHeatmapEntry { heatmap: TemporalHeatmap::new(), last_updated_cycle: current_cycle, @@ -126,13 +127,6 @@ fn bytes_to_mbps(bytes: u64) -> f32 { (bytes as f64 * 8.0 / 1_000_000.0) as f32 } -fn retransmit_percent(retransmits: u64, packets: u64) -> Option { - if retransmits == 0 || packets == 0 { - return None; - } - Some((retransmits as f32 / packets as f32) * 100.0) -} - fn p50_p90(values: &mut [f32]) -> (Option, Option) { if values.is_empty() { return (None, None); @@ -155,3 +149,66 @@ fn p50_p90(values: &mut [f32]) -> (Option, Option) { (p50, p90) } + +#[cfg(test)] +mod tests { + use super::{AsnAggregate, AsnHeatmapStore}; + use fxhash::FxHashMap; + use lqos_utils::units::DownUpOrder; + + #[test] + fn asn_heatmaps_suppress_low_confidence_retransmits() { + let mut store = AsnHeatmapStore::new(); + let mut aggregates = FxHashMap::default(); + aggregates.insert( + 64512, + AsnAggregate { + bytes: DownUpOrder { + down: 12_500_000, + up: 0, + }, + packets: DownUpOrder { down: 99, up: 0 }, + retransmits: DownUpOrder { down: 1, up: 0 }, + rtts: vec![10.0, 12.0], + }, + ); + + store.update(aggregates, 1, true); + + let blocks = store + .entries + .get(&64512) + .expect("ASN heatmap should exist") + .heatmap + .blocks(); + assert_eq!(blocks.retransmit_down.last().copied().flatten(), None); + } + + #[test] + fn asn_heatmaps_keep_qualified_retransmits() { + let mut store = AsnHeatmapStore::new(); + let mut aggregates = FxHashMap::default(); + aggregates.insert( + 64513, + AsnAggregate { + bytes: DownUpOrder { + down: 12_500_000, + up: 0, + }, + packets: DownUpOrder { down: 100, up: 0 }, + retransmits: DownUpOrder { down: 2, up: 0 }, + rtts: vec![10.0, 12.0], + }, + ); + + store.update(aggregates, 1, true); + + let blocks = store + .entries + .get(&64513) + .expect("ASN heatmap should exist") + .heatmap + .blocks(); + assert_eq!(blocks.retransmit_down.last().copied().flatten(), Some(2.0)); + } +} diff --git a/src/rust/lqosd/src/throughput_tracker/tracking_data.rs b/src/rust/lqosd/src/throughput_tracker/tracking_data.rs index 3d0511be..3e00478a 100644 --- a/src/rust/lqosd/src/throughput_tracker/tracking_data.rs +++ b/src/rust/lqosd/src/throughput_tracker/tracking_data.rs @@ -23,7 +23,7 @@ use lqos_utils::{XdpIpAddress, unix_time::time_since_boot}; use lqos_utils::{ qoq_heatmap::TemporalQoqHeatmap, rtt::RttBucket, - temporal_heatmap::TemporalHeatmap, + temporal_heatmap::{TemporalHeatmap, executive_retransmit_percent}, units::{AtomicDownUp, DownUpOrder}, }; use parking_lot::Mutex; @@ -287,10 +287,14 @@ impl ThroughputTracker { rtt.percentile(RttBucket::Current, FlowbeeEffectiveDirection::Upload, 90) }) .map(|rtt| rtt.as_millis() as f32); - let retransmit_down = - retransmit_percent(aggregate.tcp_retransmits.down, aggregate.tcp_packets.down); - let retransmit_up = - retransmit_percent(aggregate.tcp_retransmits.up, aggregate.tcp_packets.up); + let retransmit_down = executive_retransmit_percent( + aggregate.tcp_retransmits.down, + aggregate.tcp_packets.down, + ); + let retransmit_up = executive_retransmit_percent( + aggregate.tcp_retransmits.up, + aggregate.tcp_packets.up, + ); let heatmap = heatmaps.entry(circuit_hash).or_default(); heatmap.add_sample( @@ -341,8 +345,9 @@ impl ThroughputTracker { let mut global_heatmap = self.global_heatmap.lock(); let global_retransmit_down = - retransmit_percent(total_retransmits.down, total_tcp_packets.down); - let global_retransmit_up = retransmit_percent(total_retransmits.up, total_tcp_packets.up); + executive_retransmit_percent(total_retransmits.down, total_tcp_packets.down); + let global_retransmit_up = + executive_retransmit_percent(total_retransmits.up, total_tcp_packets.up); global_heatmap.add_sample( utilization_percent(total_download_bytes, global_down_mbps).unwrap_or(0.0), utilization_percent(total_upload_bytes, global_up_mbps).unwrap_or(0.0), @@ -1271,14 +1276,6 @@ fn utilization_percent(bytes: u64, max_mbps: f32) -> Option { Some(((bits_per_second / capacity_bps) * 100.0) as f32) } -fn retransmit_percent(retransmits: u64, packets: u64) -> Option { - if retransmits == 0 || packets < 10 { - return None; - } - let value = (retransmits as f32 / packets as f32) * 100.0; - if value > 50.0 { None } else { Some(value) } -} - fn median(values: &mut [f32]) -> Option { if values.is_empty() { return None; @@ -1321,11 +1318,140 @@ fn tcp_retransmit_loss_proxy(retransmits: u64, packets: u64) -> Option &'static Mutex<()> { + static LOCK: OnceLock> = OnceLock::new(); + LOCK.get_or_init(|| Mutex::new(())) + } + + fn test_runtime_dir(name: &str) -> PathBuf { + let unique = format!( + "{}-{}-{}", + std::process::id(), + std::thread::current().name().unwrap_or("thread"), + name + ) + .replace(['/', ' '], "-"); + let dir = std::env::temp_dir().join(format!("libreqos-tracking-data-{unique}")); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).expect("temp runtime dir should be created"); + dir + } + + fn write_test_config(runtime_dir: &Path) -> PathBuf { + let config_path = runtime_dir.join("lqos.test.toml"); + let raw = include_str!("../../../lqos_config/src/etc/v15/example.toml").replace( + "lqos_directory = \"/opt/libreqos/src\"", + &format!("lqos_directory = \"{}\"", runtime_dir.display()), + ); + std::fs::write(&config_path, raw).expect("test config should be written"); + config_path + } + + struct HeatmapTestContext { + _guard: MutexGuard<'static, ()>, + old_lqos_config: Option, + old_lqos_directory: Option, + old_shaped: Arc, + } + + impl HeatmapTestContext { + fn new(name: &str) -> Self { + let guard = heatmap_test_lock() + .lock() + .expect("heatmap test lock should not be poisoned"); + let runtime_dir = test_runtime_dir(name); + let config_path = write_test_config(&runtime_dir); + let old_lqos_config = std::env::var_os("LQOS_CONFIG"); + let old_lqos_directory = std::env::var_os("LQOS_DIRECTORY"); + let old_shaped = SHAPED_DEVICES.load_full(); + unsafe { + std::env::set_var("LQOS_CONFIG", &config_path); + std::env::set_var("LQOS_DIRECTORY", &runtime_dir); + } + lqos_config::clear_cached_config(); + Self { + _guard: guard, + old_lqos_config, + old_lqos_directory, + old_shaped, + } + } + } + + impl Drop for HeatmapTestContext { + fn drop(&mut self) { + match &self.old_lqos_config { + Some(value) => unsafe { std::env::set_var("LQOS_CONFIG", value) }, + None => unsafe { std::env::remove_var("LQOS_CONFIG") }, + } + match &self.old_lqos_directory { + Some(value) => unsafe { std::env::set_var("LQOS_DIRECTORY", value) }, + None => unsafe { std::env::remove_var("LQOS_DIRECTORY") }, + } + SHAPED_DEVICES.store(self.old_shaped.clone()); + lqos_config::clear_cached_config(); + } + } + + fn make_entry(circuit_hash: i64, retransmits_down: u64, packets_down: u64) -> ThroughputEntry { + ThroughputEntry { + circuit_id: None, + circuit_hash: Some(circuit_hash), + device_hash: None, + network_json_parents: None, + first_cycle: 0, + most_recent_cycle: 0, + bytes: DownUpOrder { + down: 12_500_000, + up: 0, + }, + actual_bytes: DownUpOrder { + down: 12_500_000, + up: 0, + }, + packets: DownUpOrder::zeroed(), + tcp_packets: DownUpOrder::zeroed(), + udp_packets: DownUpOrder::zeroed(), + icmp_packets: DownUpOrder::zeroed(), + prev_bytes: DownUpOrder::zeroed(), + prev_actual_bytes: DownUpOrder::zeroed(), + prev_packets: DownUpOrder::zeroed(), + prev_tcp_packets: DownUpOrder::zeroed(), + prev_udp_packets: DownUpOrder::zeroed(), + prev_icmp_packets: DownUpOrder::zeroed(), + bytes_per_second: DownUpOrder::zeroed(), + actual_bytes_per_second: DownUpOrder::zeroed(), + packets_per_second: DownUpOrder::zeroed(), + tc_handle: TcHandle::from_u32(0), + rtt_buffer: RttBuffer::default(), + recent_rtt_data: [RttData::from_nanos(0); 60], + last_fresh_rtt_data_cycle: 0, + last_seen: 0, + tcp_retransmits: DownUpOrder { + down: retransmits_down, + up: 0, + }, + tcp_retransmit_packets: DownUpOrder { + down: packets_down, + up: 0, + }, + qoq: QoqScores::default(), + } + } #[test] fn shaped_device_lookup_falls_back_to_ip_when_hashes_missing() { @@ -1347,4 +1473,96 @@ mod tests { assert_eq!(matched.circuit_hash, hash_to_i64("circuit-1")); assert_eq!(matched.device_hash, hash_to_i64("device-1")); } + + #[test] + fn circuit_and_global_heatmaps_suppress_low_confidence_retransmits() { + let _ctx = HeatmapTestContext::new("retransmit-suppressed"); + let mut shaped = ConfigShapedDevices::default(); + shaped.replace_with_new_data(vec![ShapedDevice { + circuit_id: "circuit-1".to_string(), + device_id: "device-1".to_string(), + parent_node: "Parent-A".to_string(), + download_max_mbps: 100.0, + upload_max_mbps: 100.0, + ..Default::default() + }]); + let circuit_hash = shaped.devices[0].circuit_hash; + SHAPED_DEVICES.store(Arc::new(shaped)); + + let tracker = ThroughputTracker::new(); + tracker.raw_data.lock().insert( + XdpIpAddress::from_ip("192.168.1.11".parse().expect("test IP should parse")), + make_entry(circuit_hash, 1, 99), + ); + + tracker.record_circuit_heatmaps(); + + let circuit_blocks = tracker + .circuit_heatmaps + .lock() + .get(&circuit_hash) + .expect("circuit heatmap should exist") + .blocks(); + assert_eq!( + circuit_blocks.retransmit_down.last().copied().flatten(), + None + ); + assert_eq!( + tracker + .global_heatmap + .lock() + .blocks() + .retransmit_down + .last() + .copied() + .flatten(), + None + ); + } + + #[test] + fn circuit_and_global_heatmaps_keep_qualified_retransmits() { + let _ctx = HeatmapTestContext::new("retransmit-qualified"); + let mut shaped = ConfigShapedDevices::default(); + shaped.replace_with_new_data(vec![ShapedDevice { + circuit_id: "circuit-2".to_string(), + device_id: "device-2".to_string(), + parent_node: "Parent-B".to_string(), + download_max_mbps: 100.0, + upload_max_mbps: 100.0, + ..Default::default() + }]); + let circuit_hash = shaped.devices[0].circuit_hash; + SHAPED_DEVICES.store(Arc::new(shaped)); + + let tracker = ThroughputTracker::new(); + tracker.raw_data.lock().insert( + XdpIpAddress::from_ip("192.168.1.12".parse().expect("test IP should parse")), + make_entry(circuit_hash, 2, 100), + ); + + tracker.record_circuit_heatmaps(); + + let circuit_retransmit = tracker + .circuit_heatmaps + .lock() + .get(&circuit_hash) + .expect("circuit heatmap should exist") + .blocks() + .retransmit_down + .last() + .copied() + .flatten(); + let global_retransmit = tracker + .global_heatmap + .lock() + .blocks() + .retransmit_down + .last() + .copied() + .flatten(); + + assert_eq!(circuit_retransmit, Some(2.0)); + assert_eq!(global_retransmit, Some(2.0)); + } }