From 39375cdd6ec448a92f086788e1eb4111c8cdcc6f Mon Sep 17 00:00:00 2001 From: rchac Date: Sat, 11 Apr 2026 22:23:39 +0000 Subject: [PATCH] changes --- docs/v2.0/topology-data-flow.md | 3 + src/rust/lqos_bakery/src/utils.rs | 6 +- src/rust/lqos_setup/src/main.rs | 12 +- .../lqosd/src/lts2_sys/control_channel.rs | 34 +- .../local_api/circuit_activity.rs | 20 +- .../node_manager/local_api/ethernet_caps.rs | 4 +- .../node_manager/local_api/flow_explorer.rs | 30 +- .../ws/ticker/ipstats_conversion.rs | 86 ++- .../shaped_devices_tracker/circuit_live.rs | 22 +- .../lqosd/src/shaped_devices_tracker/mod.rs | 619 ++++++++++++++++-- src/rust/lqosd/src/throughput_tracker/mod.rs | 196 ++++-- .../throughput_tracker/stats_submission.rs | 210 ++++-- 12 files changed, 1010 insertions(+), 232 deletions(-) diff --git a/docs/v2.0/topology-data-flow.md b/docs/v2.0/topology-data-flow.md index 2ba623a2..aedb8ad1 100644 --- a/docs/v2.0/topology-data-flow.md +++ b/docs/v2.0/topology-data-flow.md @@ -179,6 +179,9 @@ What this means: - Topology Manager uses logical topology, not the queue tree. - `network.effective.json` is the queue-visible runtime tree. - `shaping_inputs.json` is the shaping-ready runtime contract for scheduler/Bakery/TC. +- Runtime consumers should prefer active runtime outputs such as `shaping_inputs.json`; ingress + files such as `ShapedDevices.csv` and `topology_import.json` are upstream inputs and fallback + sources when runtime outputs are not yet ready. - In built-in integration mode, `network.json` and `ShapedDevices.csv` are not the working source of truth. - Compatibility or legacy tree artifacts are downstream-only in integration mode. - Static virtual nodes remain visible logically/runtime for monitoring, but they do not consume physical HTB classes. diff --git a/src/rust/lqos_bakery/src/utils.rs b/src/rust/lqos_bakery/src/utils.rs index 1c911a50..a0aa61d2 100644 --- a/src/rust/lqos_bakery/src/utils.rs +++ b/src/rust/lqos_bakery/src/utils.rs @@ -724,7 +724,11 @@ pub(crate) fn write_command_file(path: &Path, commands: &[Vec]) -> Optio if let Some(parent) = path.parent() && let Err(e) = std::fs::create_dir_all(parent) { - error!("Failed to create output directory {}: {}", parent.display(), e); + error!( + "Failed to create output directory {}: {}", + parent.display(), + e + ); return None; } let Ok(file) = File::create(path) else { diff --git a/src/rust/lqos_setup/src/main.rs b/src/rust/lqos_setup/src/main.rs index 31c9bbef..84a8c928 100644 --- a/src/rust/lqos_setup/src/main.rs +++ b/src/rust/lqos_setup/src/main.rs @@ -375,8 +375,16 @@ fn continue_finalize(ui: &mut cursive::Cursive) { event_log.push("Configuration updated".to_string()); let state_root = config.resolved_state_directory(); - for category in ["topology", "shaping", "stats", "cache", "debug", "quarantine"] { - std::fs::create_dir_all(state_root.join(category)).expect("Unable to create state directory"); + for category in [ + "topology", + "shaping", + "stats", + "cache", + "debug", + "quarantine", + ] { + std::fs::create_dir_all(state_root.join(category)) + .expect("Unable to create state directory"); } // Does network.json exist? diff --git a/src/rust/lqosd/src/lts2_sys/control_channel.rs b/src/rust/lqosd/src/lts2_sys/control_channel.rs index be610f57..2ce468c0 100644 --- a/src/rust/lqosd/src/lts2_sys/control_channel.rs +++ b/src/rust/lqosd/src/lts2_sys/control_channel.rs @@ -1471,16 +1471,27 @@ async fn circuit_snapshot_streaming( let raw = crate::throughput_tracker::THROUGHPUT_TRACKER .raw_data .lock(); - for (_xdp_ip, te) in raw.iter() { - // Only consider entries known to belong to this circuit and are fresh enough - if te.circuit_hash != Some(circuit_hash) { + for (xdp_ip, te) in raw.iter() { + let device = crate::shaped_devices_tracker::shaped_device_from_hashes_or_ip( + &shaped, + &shaped_cache, + xdp_ip, + te.device_hash, + te.circuit_hash, + ); + let matches_desired = te.circuit_hash == Some(circuit_hash) + || device.is_some_and(|device| device.circuit_hash == circuit_hash); + if !matches_desired { continue; } // retire_check is local; use the same heuristic: require most_recent_cycle >= tp_cycle - RETIRE_AFTER_SECONDS // We don't have RETIRE_AFTER_SECONDS here; accept all entries for snapshot. - if let Some(device_hash) = te.device_hash - && let Some(id) = shaped_cache.index_by_device_hash(&shaped, device_hash) - && let Some(agg) = aggregates.get_mut(&id) + if let Some(id) = device.and_then(|device| { + shaped + .devices + .iter() + .position(|candidate| std::ptr::eq(candidate, device)) + }) && let Some(agg) = aggregates.get_mut(&id) { // bytes_per_second -> later convert to bits agg.bps_bytes += te.bytes_per_second; @@ -1567,7 +1578,16 @@ async fn circuit_snapshot_streaming( if local.last_seen < five_minutes_ago { continue; } - if local.circuit_hash != Some(circuit_hash) { + let device = crate::shaped_devices_tracker::shaped_device_from_hashes_or_ip( + &shaped, + &shaped_cache, + &key.local_ip, + local.device_hash, + local.circuit_hash, + ); + let matches_desired = local.circuit_hash == Some(circuit_hash) + || device.is_some_and(|device| device.circuit_hash == circuit_hash); + if !matches_desired { continue; } diff --git a/src/rust/lqosd/src/node_manager/local_api/circuit_activity.rs b/src/rust/lqosd/src/node_manager/local_api/circuit_activity.rs index 0abc0e61..38d7f2ef 100644 --- a/src/rust/lqosd/src/node_manager/local_api/circuit_activity.rs +++ b/src/rust/lqosd/src/node_manager/local_api/circuit_activity.rs @@ -1,4 +1,6 @@ -use crate::shaped_devices_tracker::{SHAPED_DEVICE_HASH_CACHE, SHAPED_DEVICES}; +use crate::shaped_devices_tracker::{ + SHAPED_DEVICE_HASH_CACHE, SHAPED_DEVICES, shaped_device_from_hashes_or_ip, +}; use crate::throughput_tracker::flow_data::{ALL_FLOWS, FlowbeeLocalData, get_asn_name_and_country}; use lqos_utils::hash_to_i64; use lqos_utils::units::{DownUpOrder, TcpRetransmitSample}; @@ -237,14 +239,20 @@ fn flow_snapshot_rows(circuit_id: &str) -> Vec { if local.last_seen < recent_cutoff { return None; } - if local.circuit_hash != Some(circuit_hash) { + let device = shaped_device_from_hashes_or_ip( + &shaped, + &cache, + &key.local_ip, + local.device_hash, + local.circuit_hash, + ); + let matches_desired = local.circuit_hash == Some(circuit_hash) + || device.is_some_and(|device| device.circuit_id == circuit_id); + if !matches_desired { return None; } - let device_name = local - .device_hash - .and_then(|hash| cache.index_by_device_hash(&shaped, hash)) - .and_then(|idx| shaped.devices.get(idx)) + let device_name = device .map(|d| d.device_name.clone()) .unwrap_or_else(|| "Unknown".to_string()); diff --git a/src/rust/lqosd/src/node_manager/local_api/ethernet_caps.rs b/src/rust/lqosd/src/node_manager/local_api/ethernet_caps.rs index 72268478..1066ed96 100644 --- a/src/rust/lqosd/src/node_manager/local_api/ethernet_caps.rs +++ b/src/rust/lqosd/src/node_manager/local_api/ethernet_caps.rs @@ -1,7 +1,5 @@ use crate::shaped_devices_tracker::SHAPED_DEVICES; -use lqos_config::{ - CircuitEthernetMetadata, CircuitEthernetMetadataFile, load_config, -}; +use lqos_config::{CircuitEthernetMetadata, CircuitEthernetMetadataFile, load_config}; use serde::{Deserialize, Deserializer, Serialize}; use std::collections::HashMap; diff --git a/src/rust/lqosd/src/node_manager/local_api/flow_explorer.rs b/src/rust/lqosd/src/node_manager/local_api/flow_explorer.rs index 2e6621a2..87444806 100644 --- a/src/rust/lqosd/src/node_manager/local_api/flow_explorer.rs +++ b/src/rust/lqosd/src/node_manager/local_api/flow_explorer.rs @@ -2,10 +2,7 @@ use crate::throughput_tracker::flow_data::{ AsnCountryListEntry, AsnListEntry, AsnProtocolListEntry, FlowAnalysis, FlowbeeLocalData, RECENT_FLOWS, RttData, }; -use crate::{ - shaped_devices_tracker::{SHAPED_DEVICE_HASH_CACHE, SHAPED_DEVICES}, - throughput_tracker::THROUGHPUT_TRACKER, -}; +use crate::throughput_tracker::resolve_circuit_metadata_for_ip; use lqos_sys::flowbee_data::FlowbeeKey; use lqos_utils::units::DownUpOrder; use lqos_utils::unix_time::{time_since_boot, unix_now}; @@ -57,9 +54,6 @@ fn all_flows_to_transport( boot_time: u64, all_flows_for_asn: Vec<(FlowbeeKey, FlowbeeLocalData, FlowAnalysis)>, ) -> Vec { - let shaped = SHAPED_DEVICES.load(); - let shaped_cache = SHAPED_DEVICE_HASH_CACHE.load(); - let throughput = THROUGHPUT_TRACKER.raw_data.lock(); all_flows_for_asn .iter() .filter(|flow| { @@ -67,27 +61,7 @@ fn all_flows_to_transport( flow.1.last_seen - flow.1.start_time > 2_000_000_000 }) .map(|flow| { - let mut circuit_id = String::new(); - let mut circuit_name = String::new(); - if let Some(te) = throughput.get(&flow.0.local_ip) { - if let Some(id) = &te.circuit_id { - circuit_id = id.clone(); - } - let shaped_device = te - .device_hash - .and_then(|hash| shaped_cache.index_by_device_hash(&shaped, hash)) - .or_else(|| { - te.circuit_hash - .and_then(|hash| shaped_cache.index_by_circuit_hash(&shaped, hash)) - }) - .and_then(|idx| shaped.devices.get(idx)); - if let Some(device) = shaped_device { - if circuit_id.is_empty() { - circuit_id = device.circuit_id.clone(); - } - circuit_name = device.circuit_name.clone(); - } - } + let (circuit_id, mut circuit_name) = resolve_circuit_metadata_for_ip(&flow.0.local_ip); if circuit_name.is_empty() { circuit_name = flow.0.local_ip.as_ip().to_string(); } diff --git a/src/rust/lqosd/src/node_manager/ws/ticker/ipstats_conversion.rs b/src/rust/lqosd/src/node_manager/ws/ticker/ipstats_conversion.rs index 644c0357..802133a9 100644 --- a/src/rust/lqosd/src/node_manager/ws/ticker/ipstats_conversion.rs +++ b/src/rust/lqosd/src/node_manager/ws/ticker/ipstats_conversion.rs @@ -1,12 +1,32 @@ use crate::shaped_devices_tracker::SHAPED_DEVICES; use lqos_bus::{IpStats, TcHandle}; +use lqos_config::{ConfigShapedDevices, ShapedDevice}; +use lqos_utils::XdpIpAddress; use lqos_utils::units::{DownUpOrder, TcpRetransmitSample}; use serde::{Deserialize, Serialize}; +use std::net::IpAddr; fn truncate_by_chars(input: &str, max_chars: usize) -> String { input.chars().take(max_chars).collect() } +fn shaped_device_for_ip_stats<'a>( + stat: &IpStats, + shaped_devices: &'a ConfigShapedDevices, +) -> Option<&'a ShapedDevice> { + if !stat.circuit_id.is_empty() + && let Some(circuit) = shaped_devices + .devices + .iter() + .find(|sd| sd.circuit_id == stat.circuit_id) + { + return Some(circuit); + } + + let ip = stat.ip_address.parse::().ok()?; + shaped_devices.get_device_from_ip(&XdpIpAddress::from_ip(ip)) +} + #[derive(Serialize, Deserialize, Clone, Debug)] pub struct IpStatsWithPlan { pub ip_address: String, @@ -32,24 +52,21 @@ impl From<&IpStats> for IpStatsWithPlan { tcp_retransmit_sample: i.tcp_retransmit_sample, }; - if !result.circuit_id.is_empty() { - let shaped_devices_reader = SHAPED_DEVICES.load(); - if let Some(circuit) = shaped_devices_reader - .devices - .iter() - .find(|sd| sd.circuit_id == result.circuit_id) - { - let name = if circuit.circuit_name.chars().count() > 20 { - truncate_by_chars(&circuit.circuit_name, 20) - } else { - circuit.circuit_name.clone() - }; - result.ip_address = name.to_string(); - result.plan = DownUpOrder { - down: circuit.download_max_mbps, - up: circuit.upload_max_mbps, - }; + let shaped_devices_reader = SHAPED_DEVICES.load(); + if let Some(circuit) = shaped_device_for_ip_stats(i, &shaped_devices_reader) { + if result.circuit_id.is_empty() { + result.circuit_id = circuit.circuit_id.clone(); } + let name = if circuit.circuit_name.chars().count() > 20 { + truncate_by_chars(&circuit.circuit_name, 20) + } else { + circuit.circuit_name.clone() + }; + result.ip_address = name.to_string(); + result.plan = DownUpOrder { + down: circuit.download_max_mbps, + up: circuit.upload_max_mbps, + }; } result @@ -58,7 +75,11 @@ impl From<&IpStats> for IpStatsWithPlan { #[cfg(test)] mod tests { - use super::truncate_by_chars; + use super::{shaped_device_for_ip_stats, truncate_by_chars}; + use lqos_bus::{IpStats, TcHandle}; + use lqos_config::{ConfigShapedDevices, ShapedDevice}; + use lqos_utils::units::{DownUpOrder, TcpRetransmitSample}; + use std::net::Ipv4Addr; #[test] fn truncates_ascii_to_exact_length() { @@ -80,4 +101,33 @@ mod tests { fn keeps_short_strings_unchanged() { assert_eq!(truncate_by_chars("Věra", 20), "Věra"); } + + #[test] + fn shaped_device_lookup_falls_back_to_ip_when_circuit_id_is_blank() { + let mut shaped = ConfigShapedDevices::default(); + shaped.replace_with_new_data(vec![ShapedDevice { + circuit_id: "circuit-1".to_string(), + circuit_name: "Circuit Alpha".to_string(), + device_id: "device-1".to_string(), + parent_node: "Parent-A".to_string(), + ipv4: vec![(Ipv4Addr::new(192, 168, 1, 10), 32)], + ..Default::default() + }]); + + let stat = IpStats { + ip_address: "192.168.1.10".to_string(), + circuit_id: String::new(), + bits_per_second: DownUpOrder::zeroed(), + packets_per_second: DownUpOrder::zeroed(), + median_tcp_rtt: 0.0, + tc_handle: TcHandle::from_u32(0), + tcp_retransmit_sample: DownUpOrder::new( + TcpRetransmitSample::new(0, 0), + TcpRetransmitSample::new(0, 0), + ), + }; + + let matched = shaped_device_for_ip_stats(&stat, &shaped).expect("lookup should resolve"); + assert_eq!(matched.circuit_id, "circuit-1"); + } } diff --git a/src/rust/lqosd/src/shaped_devices_tracker/circuit_live.rs b/src/rust/lqosd/src/shaped_devices_tracker/circuit_live.rs index 3d805d29..8cdec384 100644 --- a/src/rust/lqosd/src/shaped_devices_tracker/circuit_live.rs +++ b/src/rust/lqosd/src/shaped_devices_tracker/circuit_live.rs @@ -8,8 +8,10 @@ use std::net::IpAddr; use std::sync::Arc; use std::time::{SystemTime, UNIX_EPOCH}; -use super::{CIRCUIT_LIVE_LAST_REFRESH_SECS, CIRCUIT_LIVE_REFRESH_LOCK, CIRCUIT_LIVE_SNAPSHOT}; -use super::{SHAPED_DEVICE_HASH_CACHE, SHAPED_DEVICES}; +use super::{ + CIRCUIT_LIVE_LAST_REFRESH_SECS, CIRCUIT_LIVE_REFRESH_LOCK, CIRCUIT_LIVE_SNAPSHOT, + SHAPED_DEVICE_HASH_CACHE, SHAPED_DEVICES, shaped_device_from_hashes_or_ip, +}; /// Per-circuit live metrics aggregated from the device-level throughput tracker. #[derive(Clone, Debug, Serialize, Deserialize, PartialEq)] @@ -108,15 +110,13 @@ pub fn rebuild_circuit_live_snapshot() -> Arc { let mut by_circuit_id: FxHashMap = FxHashMap::default(); for (ip_key, data) in THROUGHPUT_TRACKER.raw_data.lock().iter() { - let device = data - .device_hash - .and_then(|device_hash| cache.index_by_device_hash(&shaped_devices, device_hash)) - .or_else(|| { - data.circuit_hash.and_then(|circuit_hash| { - cache.index_by_circuit_hash(&shaped_devices, circuit_hash) - }) - }) - .and_then(|idx| shaped_devices.devices.get(idx)); + let device = shaped_device_from_hashes_or_ip( + &shaped_devices, + &cache, + ip_key, + data.device_hash, + data.circuit_hash, + ); let Some(device) = device else { continue; }; diff --git a/src/rust/lqosd/src/shaped_devices_tracker/mod.rs b/src/rust/lqosd/src/shaped_devices_tracker/mod.rs index e5dc12e9..8f75eaec 100644 --- a/src/rust/lqosd/src/shaped_devices_tracker/mod.rs +++ b/src/rust/lqosd/src/shaped_devices_tracker/mod.rs @@ -6,10 +6,12 @@ use fxhash::{FxHashMap, FxHashSet}; use lqos_bus::{BusResponse, Circuit}; use lqos_config::{ ConfigShapedDevices, NetworkJsonNode, NetworkJsonTransport, ShapedDevice, - TopologyShapingInputsFile, load_config, topology_import_path, topology_shaping_inputs_path, + TopologyRuntimeStatusFile, TopologyShapingInputsFile, load_config, topology_import_path, + topology_runtime_status_path, }; use lqos_queue_tracker::EFFECTIVE_NODE_RATES; use lqos_topology_compile::TopologyImportFile; +use lqos_utils::XdpIpAddress; use lqos_utils::file_watcher::FileWatcher; use lqos_utils::hash_to_i64; use lqos_utils::rtt::{FlowbeeEffectiveDirection, RttBucket}; @@ -17,6 +19,8 @@ use lqos_utils::units::{DownUpOrder, down_up_retransmit_sample}; use lqos_utils::unix_time::time_since_boot; use once_cell::sync::Lazy; use parking_lot::Mutex; +use std::net::{Ipv4Addr, Ipv6Addr}; +use std::path::PathBuf; use std::sync::Arc; use std::sync::atomic::AtomicU64; use std::time::Duration; @@ -119,6 +123,20 @@ pub(crate) fn invalidate_executive_cache_snapshot() { crate::node_manager::invalidate_executive_cache_snapshot(); } +pub(crate) fn shaped_device_from_hashes_or_ip<'a>( + shaped: &'a ConfigShapedDevices, + cache: &ShapedDeviceHashCache, + ip: &XdpIpAddress, + device_hash: Option, + circuit_hash: Option, +) -> Option<&'a ShapedDevice> { + device_hash + .and_then(|hash| cache.index_by_device_hash(shaped, hash)) + .or_else(|| circuit_hash.and_then(|hash| cache.index_by_circuit_hash(shaped, hash))) + .and_then(|idx| shaped.devices.get(idx)) + .or_else(|| shaped.get_device_from_ip(ip)) +} + fn normalize_circuit_id_key(circuit_id: &str) -> Option { let trimmed = circuit_id.trim(); if trimmed.is_empty() { @@ -166,13 +184,114 @@ fn publish_shaping_inputs(shaping_inputs: TopologyShapingInputsFile) { invalidate_executive_cache_snapshot(); } +fn active_runtime_shaping_inputs_path(config: &lqos_config::Config) -> Result> { + let status = TopologyRuntimeStatusFile::load(config)?; + if !status.ready { + return Ok(None); + } + let path = status.shaping_inputs_path.trim(); + if path.is_empty() { + return Ok(None); + } + Ok(Some(PathBuf::from(path))) +} + +fn parse_ipv4_entry(value: &str) -> Option<(Ipv4Addr, u32)> { + let trimmed = value.trim(); + if trimmed.is_empty() { + return None; + } + let (ip, cidr) = if let Some((ip, cidr)) = trimmed.split_once('/') { + (ip.trim(), cidr.trim().parse().ok()?) + } else { + (trimmed, 32) + }; + Some((ip.parse().ok()?, cidr)) +} + +fn parse_ipv6_entry(value: &str) -> Option<(Ipv6Addr, u32)> { + let trimmed = value.trim(); + if trimmed.is_empty() { + return None; + } + let (ip, cidr) = if let Some((ip, cidr)) = trimmed.split_once('/') { + (ip.trim(), cidr.trim().parse().ok()?) + } else { + (trimmed, 128) + }; + Some((ip.parse().ok()?, cidr)) +} + +fn parse_ipv4_list(values: &[String]) -> Vec<(Ipv4Addr, u32)> { + values + .iter() + .filter_map(|value| parse_ipv4_entry(value)) + .collect() +} + +fn parse_ipv6_list(values: &[String]) -> Vec<(Ipv6Addr, u32)> { + values + .iter() + .filter_map(|value| parse_ipv6_entry(value)) + .collect() +} + +fn shaped_devices_from_runtime_inputs( + shaping_inputs: &TopologyShapingInputsFile, +) -> ConfigShapedDevices { + let mut devices = Vec::new(); + for circuit in &shaping_inputs.circuits { + let parent_node = optional_trimmed_string(&circuit.effective_parent_node_name) + .or_else(|| circuit.logical_parent_node_name.clone()) + .unwrap_or_default(); + let parent_node_id = optional_trimmed_string(&circuit.effective_parent_node_id) + .or_else(|| circuit.logical_parent_node_id.clone()); + for device in &circuit.devices { + devices.push(ShapedDevice { + circuit_id: circuit.circuit_id.clone(), + circuit_name: circuit.circuit_name.clone(), + device_id: device.device_id.clone(), + device_name: device.device_name.clone(), + parent_node: parent_node.clone(), + parent_node_id: parent_node_id.clone(), + anchor_node_id: circuit.anchor_node_id.clone(), + mac: device.mac.clone(), + ipv4: parse_ipv4_list(&device.ipv4), + ipv6: parse_ipv6_list(&device.ipv6), + download_min_mbps: circuit.download_min_mbps, + upload_min_mbps: circuit.upload_min_mbps, + download_max_mbps: circuit.download_max_mbps, + upload_max_mbps: circuit.upload_max_mbps, + comment: if device.comment.trim().is_empty() { + circuit.comment.clone() + } else { + device.comment.clone() + }, + sqm_override: circuit.sqm_override.clone(), + ..ShapedDevice::default() + }); + } + } + + let mut shaped = ConfigShapedDevices::default(); + shaped.replace_with_new_data(devices); + shaped +} + fn load_shaping_inputs() { let Ok(config) = load_config() else { warn!("Unable to load LibreQoS config while loading shaping_inputs.json"); return; }; - match TopologyShapingInputsFile::load(config.as_ref()) { - Ok(shaping_inputs) => publish_shaping_inputs(shaping_inputs), + match load_active_runtime_shaping_inputs(config.as_ref()) { + Ok(Some(shaping_inputs)) => { + debug!("Loaded shaping inputs from active runtime status"); + publish_shaping_inputs(shaping_inputs); + } + Ok(None) => { + debug!("No active runtime shaping inputs published; clearing effective parent cache"); + publish_shaping_inputs(TopologyShapingInputsFile::default()); + } Err(err) => { warn!("Unable to load shaping_inputs.json: {err}"); } @@ -198,15 +317,15 @@ fn watch_for_shaping_inputs_changing() -> Result<()> { "Unable to load LibreQoS config for shaping_inputs.json", )); }; - let watch_path = topology_shaping_inputs_path(config.as_ref()); + let watch_path = topology_runtime_status_path(config.as_ref()); - let mut watcher = FileWatcher::new("shaping_inputs.json", watch_path); + let mut watcher = FileWatcher::new("topology_runtime_status.json", watch_path); watcher.set_file_exists_callback(load_shaping_inputs); watcher.set_file_created_callback(load_shaping_inputs); watcher.set_file_changed_callback(load_shaping_inputs); loop { let result = watcher.watch(); - info!("shaping_inputs.json watcher returned: {result:?}"); + info!("topology_runtime_status.json watcher returned: {result:?}"); } } @@ -298,6 +417,23 @@ fn publish_shaped_devices(new_file: ConfigShapedDevices) { crate::throughput_tracker::THROUGHPUT_TRACKER.refresh_circuit_ids(&nj); } +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum ShapedDevicesLoadSource { + RuntimeShapingInputs, + TopologyImport, + ShapedDevicesCsv, +} + +impl ShapedDevicesLoadSource { + const fn label(self) -> &'static str { + match self { + Self::RuntimeShapingInputs => "runtime shaping inputs", + Self::TopologyImport => "topology_import.json", + Self::ShapedDevicesCsv => "ShapedDevices.csv", + } + } +} + fn integration_ingress_enabled(config: &lqos_config::Config) -> bool { config.uisp_integration.enable_uisp || config.splynx_integration.enable_splynx @@ -317,21 +453,55 @@ fn integration_ingress_enabled(config: &lqos_config::Config) -> bool { .is_some_and(|integration| integration.enable_wispgate) } -fn load_shaped_devices_from_active_ingress() -> Result { - let config = load_config()?; - if integration_ingress_enabled(config.as_ref()) - && let Some(topology_import) = TopologyImportFile::load(config.as_ref())? - { - return Ok(topology_import.into_imported_bundle().shaped_devices); +fn load_active_runtime_shaping_inputs( + config: &lqos_config::Config, +) -> Result> { + let Some(active_path) = active_runtime_shaping_inputs_path(config)? else { + return Ok(None); + }; + if !active_path.exists() { + return Ok(None); } - Ok(ConfigShapedDevices::load()?) + + let raw = std::fs::read_to_string(&active_path)?; + let shaping_inputs = serde_json::from_str(&raw)?; + Ok(Some(shaping_inputs)) +} + +fn load_shaped_devices_from_preferred_source( + config: &lqos_config::Config, +) -> Result<(ConfigShapedDevices, ShapedDevicesLoadSource)> { + if let Some(shaping_inputs) = load_active_runtime_shaping_inputs(config)? { + return Ok(( + shaped_devices_from_runtime_inputs(&shaping_inputs), + ShapedDevicesLoadSource::RuntimeShapingInputs, + )); + } + + if integration_ingress_enabled(config) + && let Some(topology_import) = TopologyImportFile::load(config)? + { + return Ok(( + topology_import.into_imported_bundle().shaped_devices, + ShapedDevicesLoadSource::TopologyImport, + )); + } + Ok(( + ConfigShapedDevices::load_for_config(config)?, + ShapedDevicesLoadSource::ShapedDevicesCsv, + )) } fn load_shaped_devices() { debug!("Shaped-device ingress has changed. Attempting to load it."); + let Ok(config) = load_config() else { + warn!("Unable to load LibreQoS config while loading shaped devices"); + return; + }; for attempt in 1..=SHAPED_DEVICES_RELOAD_ATTEMPTS { - match load_shaped_devices_from_active_ingress() { - Ok(new_file) => { + match load_shaped_devices_from_preferred_source(config.as_ref()) { + Ok((new_file, source)) => { + debug!("Loaded shaped devices from {}", source.label()); publish_shaped_devices(new_file); return; } @@ -367,29 +537,46 @@ pub fn shaped_devices_watcher() -> Result<()> { Ok(()) } -/// Fires up a Linux file system watcher than notifies -/// when `ShapedDevices.csv` changes, and triggers a reload. -fn watch_for_shaped_devices_changing() -> Result<()> { - let config = load_config()?; - let (watch_name, watch_path) = if integration_ingress_enabled(config.as_ref()) { - ( - "topology_import.json", - topology_import_path(config.as_ref()), - ) - } else { - ("ShapedDevices.csv", ConfigShapedDevices::path()?) - }; - +fn watch_shaped_devices_hint_path(watch_name: &str, watch_path: PathBuf) -> Result<()> { let mut watcher = FileWatcher::new(watch_name, watch_path); watcher.set_file_exists_callback(load_shaped_devices); watcher.set_file_created_callback(load_shaped_devices); watcher.set_file_changed_callback(load_shaped_devices); loop { let result = watcher.watch(); - info!("ShapedDevices watcher returned: {result:?}"); + info!("{watch_name} watcher returned: {result:?}"); } } +/// Watches runtime status and source-ingress hints, then reevaluates the preferred shaped-device +/// source. The reload path always prefers active runtime shaping inputs over source ingress files. +fn watch_for_shaped_devices_changing() -> Result<()> { + let config = load_config()?; + let runtime_status_path = topology_runtime_status_path(config.as_ref()); + let csv_path = ConfigShapedDevices::path_for_config(config.as_ref()); + let topology_import = topology_import_path(config.as_ref()); + + std::thread::Builder::new() + .name("ShapedDevices Runtime Watcher".to_string()) + .spawn(move || { + if let Err(err) = + watch_shaped_devices_hint_path("topology_runtime_status.json", runtime_status_path) + { + error!("Error watching topology_runtime_status.json: {err:?}"); + } + })?; + + std::thread::Builder::new() + .name("ShapedDevices CSV Hint Watcher".to_string()) + .spawn(move || { + if let Err(err) = watch_shaped_devices_hint_path("ShapedDevices.csv", csv_path) { + error!("Error watching ShapedDevices.csv: {err:?}"); + } + })?; + + watch_shaped_devices_hint_path("topology_import.json", topology_import) +} + pub fn get_one_network_map_layer(parent_idx: usize) -> BusResponse { let net_json = NETWORK_JSON.read(); let nodes_ref = net_json.get_nodes_when_ready(); @@ -733,24 +920,24 @@ pub fn get_all_circuits() -> BusResponse { }; // Map to circuit et al - let mut circuit_id = None; + let mut circuit_id = v.circuit_id.clone(); let mut circuit_name = None; let mut device_id = None; let mut device_name = None; let mut parent_node = None; // Plan is expressed in Mbps as f32 let mut plan: DownUpOrder = DownUpOrder { down: 0.0, up: 0.0 }; - let device = v - .device_hash - .and_then(|device_hash| cache.index_by_device_hash(&devices, device_hash)) - .or_else(|| { - v.circuit_hash.and_then(|circuit_hash| { - cache.index_by_circuit_hash(&devices, circuit_hash) - }) - }) - .and_then(|idx| devices.devices.get(idx)); + let device = shaped_device_from_hashes_or_ip( + &devices, + &cache, + k, + v.device_hash, + v.circuit_hash, + ); if let Some(device) = device { - circuit_id = Some(device.circuit_id.clone()); + if circuit_id.as_deref().unwrap_or_default().is_empty() { + circuit_id = Some(device.circuit_id.clone()); + } circuit_name = Some(device.circuit_name.clone()); device_id = Some(device.device_id.clone()); device_name = Some(device.device_name.clone()); @@ -843,7 +1030,17 @@ pub fn get_circuit_by_id(desired_circuit_id: String) -> BusResponse { .lock() .iter() .filter_map(|(k, v)| { - if v.circuit_hash != Some(desired_hash) { + let device = shaped_device_from_hashes_or_ip( + &devices, + &cache, + k, + v.device_hash, + v.circuit_hash, + ); + let matches_desired = v.circuit_hash == Some(desired_hash) + || v.circuit_id.as_deref() == Some(desired_circuit_id.as_str()) + || device.is_some_and(|device| device.circuit_id == desired_circuit_id); + if !matches_desired { return None; } let last_seen_nanos = if v.last_seen > 0 { @@ -856,24 +1053,17 @@ pub fn get_circuit_by_id(desired_circuit_id: String) -> BusResponse { }; // Map to circuit et al - let mut circuit_id = None; + let mut circuit_id = v.circuit_id.clone(); let mut circuit_name = None; let mut device_id = None; let mut device_name = None; let mut parent_node = None; // Plan is expressed in Mbps as f32 let mut plan: DownUpOrder = DownUpOrder { down: 0.0, up: 0.0 }; - let device = v - .device_hash - .and_then(|device_hash| cache.index_by_device_hash(&devices, device_hash)) - .or_else(|| { - v.circuit_hash.and_then(|circuit_hash| { - cache.index_by_circuit_hash(&devices, circuit_hash) - }) - }) - .and_then(|idx| devices.devices.get(idx)); if let Some(device) = device { - circuit_id = Some(device.circuit_id.clone()); + if circuit_id.as_deref().unwrap_or_default().is_empty() { + circuit_id = Some(device.circuit_id.clone()); + } circuit_name = Some(device.circuit_name.clone()); device_id = Some(device.device_id.clone()); device_name = Some(device.device_name.clone()); @@ -960,7 +1150,63 @@ pub fn get_circuit_by_id(desired_circuit_id: String) -> BusResponse { #[cfg(test)] mod tests { use super::*; - use lqos_config::{TopologyShapingCircuitInput, TopologyShapingInputsFile}; + use lqos_config::{ + CircuitAnchorsFile, Config, TOPOLOGY_RUNTIME_STATUS_FILENAME, TopologyShapingCircuitInput, + TopologyShapingDeviceInput, TopologyShapingInputsFile, + }; + use lqos_topology_compile::{ImportedTopologyBundle, TopologyImportFile}; + use serde_json::json; + use std::fs; + use std::net::Ipv4Addr; + use std::time::{SystemTime, UNIX_EPOCH}; + + fn unique_temp_dir(prefix: &str) -> PathBuf { + let unique = SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("clock should be monotonic enough for tests") + .as_nanos(); + let path = std::env::temp_dir().join(format!("{prefix}-{unique}")); + fs::create_dir_all(&path).expect("temp directory should be creatable"); + path + } + + fn write_shaped_devices_csv(path: &std::path::Path, circuit_id: &str, ip: &str) { + fs::write( + path, + format!( + concat!( + "Circuit ID,Circuit Name,Device ID,Device Name,Parent Node,MAC,IPv4,IPv6,", + "Download Min Mbps,Upload Min Mbps,Download Max Mbps,Upload Max Mbps,Comment\n", + "\"{}\",\"Circuit {}\",\"device-{}\",", + "\"Device {}\",\"Tower 1\",\"aa:bb:cc:dd:ee:ff\",\"{}\",\"\",", + "\"10\",\"10\",\"100\",\"100\",\"\"\n", + ), + circuit_id, circuit_id, circuit_id, circuit_id, ip, + ), + ) + .expect("ShapedDevices.csv should write"); + } + + fn write_runtime_status( + path: &std::path::Path, + ready: bool, + shaping_inputs_path: &std::path::Path, + ) { + fs::write( + path, + serde_json::json!({ + "schema_version": 1, + "ready": ready, + "shaping_inputs_path": shaping_inputs_path, + "effective_state_path": "", + "effective_network_path": "", + "source_generation": "gen-1", + "shaping_generation": "shape-1", + }) + .to_string(), + ) + .expect("status should write"); + } #[test] fn effective_circuit_parent_map_uses_effective_parent_fields() { @@ -997,4 +1243,271 @@ mod tests { let map = build_effective_circuit_parent_map(&shaping_inputs); assert!(map.is_empty()); } + + #[test] + fn shaped_device_from_hashes_or_ip_falls_back_to_ip() { + let mut shaped = ConfigShapedDevices::default(); + shaped.replace_with_new_data(vec![ShapedDevice { + circuit_id: "circuit-1".to_string(), + circuit_name: "Circuit Alpha".to_string(), + device_id: "device-1".to_string(), + parent_node: "Parent-A".to_string(), + ipv4: vec![(Ipv4Addr::new(192, 168, 1, 10), 32)], + ..Default::default() + }]); + let cache = ShapedDeviceHashCache::default(); + let ip = XdpIpAddress::from_ip("192.168.1.10".parse().expect("test IP should parse")); + + let device = + shaped_device_from_hashes_or_ip(&shaped, &cache, &ip, None, None).expect("match"); + + assert_eq!(device.circuit_id, "circuit-1"); + assert_eq!(device.circuit_name, "Circuit Alpha"); + } + + #[test] + fn shaped_devices_from_runtime_inputs_uses_effective_parent_and_ip_lookup() { + let shaping_inputs = TopologyShapingInputsFile { + circuits: vec![TopologyShapingCircuitInput { + circuit_id: "circuit-1".to_string(), + circuit_name: "Circuit Alpha".to_string(), + anchor_node_id: Some("anchor-1".to_string()), + effective_parent_node_name: "Parent-A".to_string(), + effective_parent_node_id: "parent-id-1".to_string(), + download_min_mbps: 10.0, + upload_min_mbps: 5.0, + download_max_mbps: 50.0, + upload_max_mbps: 10.0, + devices: vec![TopologyShapingDeviceInput { + device_id: "device-1".to_string(), + device_name: "Device Alpha".to_string(), + mac: "aa:bb:cc:dd:ee:ff".to_string(), + ipv4: vec!["192.168.1.10/32".to_string()], + comment: "device-comment".to_string(), + ..TopologyShapingDeviceInput::default() + }], + comment: "circuit-comment".to_string(), + ..TopologyShapingCircuitInput::default() + }], + ..TopologyShapingInputsFile::default() + }; + + let shaped = shaped_devices_from_runtime_inputs(&shaping_inputs); + let ip = XdpIpAddress::from_ip("192.168.1.10".parse().expect("test IP should parse")); + let device = shaped + .get_device_from_ip(&ip) + .expect("lookup should resolve"); + + assert_eq!(device.parent_node, "Parent-A"); + assert_eq!(device.parent_node_id.as_deref(), Some("parent-id-1")); + assert_eq!(device.anchor_node_id.as_deref(), Some("anchor-1")); + assert_eq!(device.comment, "device-comment"); + assert_eq!(device.circuit_id, "circuit-1"); + } + + #[test] + fn load_active_runtime_shaping_inputs_prefers_runtime_status_path_over_state_fallback() { + let lqos_directory = unique_temp_dir("lqosd-shaped-devices-runtime-status"); + let state_directory = lqos_directory.join("state"); + fs::create_dir_all(state_directory.join("topology")).expect("topology dir should exist"); + fs::create_dir_all(state_directory.join("shaping")).expect("shaping dir should exist"); + + let active_shaping_path = lqos_directory.join("shaping_inputs.json"); + let stale_state_path = state_directory.join("shaping").join("shaping_inputs.json"); + let status_path = state_directory + .join("topology") + .join(TOPOLOGY_RUNTIME_STATUS_FILENAME); + + let active_inputs = TopologyShapingInputsFile { + circuits: vec![TopologyShapingCircuitInput { + circuit_id: "active-circuit".to_string(), + effective_parent_node_name: "Parent-A".to_string(), + devices: vec![TopologyShapingDeviceInput { + device_id: "device-1".to_string(), + device_name: "Device Alpha".to_string(), + ipv4: vec!["192.168.10.5/32".to_string()], + ..TopologyShapingDeviceInput::default() + }], + ..TopologyShapingCircuitInput::default() + }], + ..TopologyShapingInputsFile::default() + }; + let stale_inputs = TopologyShapingInputsFile { + circuits: vec![TopologyShapingCircuitInput { + circuit_id: "stale-circuit".to_string(), + effective_parent_node_name: "Stale Parent".to_string(), + ..TopologyShapingCircuitInput::default() + }], + ..TopologyShapingInputsFile::default() + }; + + fs::write( + &active_shaping_path, + serde_json::to_string_pretty(&active_inputs).expect("active shaping should encode"), + ) + .expect("active shaping should write"); + fs::write( + &stale_state_path, + serde_json::to_string_pretty(&stale_inputs).expect("stale shaping should encode"), + ) + .expect("stale shaping should write"); + write_runtime_status(&status_path, true, &active_shaping_path); + + let config = Config { + lqos_directory: lqos_directory.to_string_lossy().to_string(), + state_directory: Some(state_directory.to_string_lossy().to_string()), + ..Config::default() + }; + + let loaded = load_active_runtime_shaping_inputs(&config) + .expect("runtime shaping should load") + .expect("runtime shaping inputs should be active"); + assert_eq!(loaded.circuits.len(), 1); + assert_eq!(loaded.circuits[0].circuit_id, "active-circuit"); + } + + #[test] + fn load_shaped_devices_prefers_ready_runtime_even_when_empty() { + let lqos_directory = unique_temp_dir("lqosd-shaped-devices-runtime-empty"); + let state_directory = lqos_directory.join("state"); + fs::create_dir_all(state_directory.join("topology")).expect("topology dir should exist"); + fs::create_dir_all(state_directory.join("shaping")).expect("shaping dir should exist"); + + let runtime_path = lqos_directory.join("runtime_shaping_inputs.json"); + let status_path = state_directory + .join("topology") + .join(TOPOLOGY_RUNTIME_STATUS_FILENAME); + write_shaped_devices_csv( + &lqos_directory.join("ShapedDevices.csv"), + "csv-circuit", + "192.0.2.10/32", + ); + fs::write( + &runtime_path, + serde_json::to_string_pretty(&TopologyShapingInputsFile::default()) + .expect("empty shaping inputs should encode"), + ) + .expect("runtime shaping inputs should write"); + write_runtime_status(&status_path, true, &runtime_path); + + let config = Config { + lqos_directory: lqos_directory.to_string_lossy().to_string(), + state_directory: Some(state_directory.to_string_lossy().to_string()), + ..Config::default() + }; + + let (loaded, source) = load_shaped_devices_from_preferred_source(&config) + .expect("preferred shaped-device source should load"); + assert_eq!(source, ShapedDevicesLoadSource::RuntimeShapingInputs); + assert!(loaded.devices.is_empty()); + } + + #[test] + fn load_shaped_devices_uses_csv_when_runtime_is_not_ready() { + let lqos_directory = unique_temp_dir("lqosd-shaped-devices-csv-fallback"); + let state_directory = lqos_directory.join("state"); + fs::create_dir_all(state_directory.join("topology")).expect("topology dir should exist"); + + let runtime_path = lqos_directory.join("runtime_shaping_inputs.json"); + let status_path = state_directory + .join("topology") + .join(TOPOLOGY_RUNTIME_STATUS_FILENAME); + write_shaped_devices_csv( + &lqos_directory.join("ShapedDevices.csv"), + "csv-circuit", + "192.0.2.10/32", + ); + fs::write( + &runtime_path, + serde_json::to_string_pretty(&TopologyShapingInputsFile { + circuits: vec![TopologyShapingCircuitInput { + circuit_id: "runtime-circuit".to_string(), + devices: vec![TopologyShapingDeviceInput { + device_id: "device-1".to_string(), + device_name: "Device 1".to_string(), + ipv4: vec!["198.51.100.10/32".to_string()], + ..TopologyShapingDeviceInput::default() + }], + ..TopologyShapingCircuitInput::default() + }], + ..TopologyShapingInputsFile::default() + }) + .expect("runtime shaping inputs should encode"), + ) + .expect("runtime shaping inputs should write"); + write_runtime_status(&status_path, false, &runtime_path); + + let config = Config { + lqos_directory: lqos_directory.to_string_lossy().to_string(), + state_directory: Some(state_directory.to_string_lossy().to_string()), + ..Config::default() + }; + + let (loaded, source) = + load_shaped_devices_from_preferred_source(&config).expect("CSV fallback should load"); + assert_eq!(source, ShapedDevicesLoadSource::ShapedDevicesCsv); + assert_eq!(loaded.devices.len(), 1); + assert_eq!(loaded.devices[0].circuit_id, "csv-circuit"); + } + + #[test] + fn load_shaped_devices_uses_topology_import_when_runtime_is_not_ready() { + let lqos_directory = unique_temp_dir("lqosd-shaped-devices-topology-import-fallback"); + let state_directory = lqos_directory.join("state"); + fs::create_dir_all(state_directory.join("topology")).expect("topology dir should exist"); + + let runtime_path = lqos_directory.join("runtime_shaping_inputs.json"); + let status_path = state_directory + .join("topology") + .join(TOPOLOGY_RUNTIME_STATUS_FILENAME); + fs::write( + &runtime_path, + serde_json::to_string_pretty(&TopologyShapingInputsFile::default()) + .expect("runtime shaping inputs should encode"), + ) + .expect("runtime shaping inputs should write"); + write_runtime_status(&status_path, false, &runtime_path); + + let mut import_devices = ConfigShapedDevices::default(); + import_devices.replace_with_new_data(vec![ShapedDevice { + circuit_id: "import-circuit".to_string(), + circuit_name: "Import Circuit".to_string(), + device_id: "device-import".to_string(), + parent_node: "Tower Import".to_string(), + ipv4: vec![(Ipv4Addr::new(203, 0, 113, 10), 32)], + ..Default::default() + }]); + let imported = ImportedTopologyBundle { + source: "test/import".to_string(), + generated_unix: Some(123), + ingress_identity: Some("import-base".to_string()), + native_canonical: None, + native_editor: None, + parent_candidates: None, + compatibility_network_json: json!({}), + shaped_devices: import_devices, + circuit_anchors: CircuitAnchorsFile::default(), + ethernet_advisories: Vec::new(), + }; + TopologyImportFile::from_imported_bundle(&imported, "full") + .save(&Config { + lqos_directory: lqos_directory.to_string_lossy().to_string(), + state_directory: Some(state_directory.to_string_lossy().to_string()), + ..Config::default() + }) + .expect("topology import should save"); + + let mut config = Config { + lqos_directory: lqos_directory.to_string_lossy().to_string(), + state_directory: Some(state_directory.to_string_lossy().to_string()), + ..Config::default() + }; + config.uisp_integration.enable_uisp = true; + + let (loaded, source) = load_shaped_devices_from_preferred_source(&config) + .expect("topology import fallback should load"); + assert_eq!(source, ShapedDevicesLoadSource::TopologyImport); + assert_eq!(loaded.devices.len(), 1); + assert_eq!(loaded.devices[0].circuit_id, "import-circuit"); + } } diff --git a/src/rust/lqosd/src/throughput_tracker/mod.rs b/src/rust/lqosd/src/throughput_tracker/mod.rs index d83898f0..ea61fe2c 100644 --- a/src/rust/lqosd/src/throughput_tracker/mod.rs +++ b/src/rust/lqosd/src/throughput_tracker/mod.rs @@ -7,6 +7,7 @@ use self::flow_data::{ ALL_FLOWS, FlowAnalysis, FlowbeeLocalData, get_asn_name_and_country, get_asn_name_by_id, snapshot_asn_heatmaps, }; +use self::throughput_entry::ThroughputEntry; use crate::system_stats::SystemStats; use crate::throughput_tracker::flow_data::FlowbeeEffectiveDirection; use crate::{ @@ -42,6 +43,53 @@ pub static THROUGHPUT_TRACKER: Lazy = Lazy::new(ThroughputTra pub(crate) static CIRCUIT_RTT_BUFFERS: Lazy>> = Lazy::new(|| ArcSwap::new(Arc::new(FxHashMap::default()))); +fn shaped_device_for_entry<'a>( + shaped: &'a lqos_config::ConfigShapedDevices, + cache: &crate::shaped_devices_tracker::ShapedDeviceHashCache, + ip: &XdpIpAddress, + entry: &ThroughputEntry, +) -> Option<&'a lqos_config::ShapedDevice> { + entry + .device_hash + .and_then(|hash| cache.index_by_device_hash(shaped, hash)) + .or_else(|| { + entry + .circuit_hash + .and_then(|hash| cache.index_by_circuit_hash(shaped, hash)) + }) + .and_then(|idx| shaped.devices.get(idx)) + .or_else(|| shaped.get_device_from_ip(ip)) +} + +fn resolve_circuit_metadata_for_entry( + shaped: &lqos_config::ConfigShapedDevices, + cache: &crate::shaped_devices_tracker::ShapedDeviceHashCache, + ip: &XdpIpAddress, + entry: &ThroughputEntry, +) -> (String, String) { + let mut circuit_id = entry.circuit_id.clone().unwrap_or_default(); + let mut circuit_name = String::new(); + + if let Some(device) = shaped_device_for_entry(shaped, cache, ip, entry) { + if circuit_id.is_empty() { + circuit_id = device.circuit_id.clone(); + } + circuit_name = device.circuit_name.clone(); + } + + (circuit_id, circuit_name) +} + +pub(crate) fn resolve_circuit_metadata_for_ip(ip: &XdpIpAddress) -> (String, String) { + let shaped = SHAPED_DEVICES.load(); + let cache = SHAPED_DEVICE_HASH_CACHE.load(); + let throughput = THROUGHPUT_TRACKER.raw_data.lock(); + let Some(entry) = throughput.get(ip) else { + return (String::new(), String::new()); + }; + resolve_circuit_metadata_for_entry(&shaped, &cache, ip, entry) +} + /// Create the throughput monitor thread, and begin polling for /// throughput data every second. /// @@ -361,6 +409,8 @@ pub fn top_n(start: u32, end: u32) -> BusResponse { let tp_cycle = THROUGHPUT_TRACKER .cycle .load(std::sync::atomic::Ordering::Relaxed); + let shaped = SHAPED_DEVICES.load(); + let shaped_cache = SHAPED_DEVICE_HASH_CACHE.load(); THROUGHPUT_TRACKER .raw_data .lock() @@ -368,13 +418,15 @@ pub fn top_n(start: u32, end: u32) -> BusResponse { .filter(|(k, _v)| !k.as_ip().is_loopback()) .filter(|(_k, d)| retire_check(tp_cycle, d.most_recent_cycle)) .map(|(k, te)| { + let (circuit_id, _circuit_name) = + resolve_circuit_metadata_for_entry(&shaped, &shaped_cache, k, te); ( *k, te.actual_bytes_per_second, te.packets_per_second, te.median_latency().unwrap_or(0.0), te.tc_handle, - te.circuit_id.as_ref().unwrap_or(&String::new()).clone(), + circuit_id, down_up_retransmit_sample(te.tcp_retransmits, te.tcp_packets), ) }) @@ -405,6 +457,8 @@ pub fn top_n_up(start: u32, end: u32) -> BusResponse { let tp_cycle = THROUGHPUT_TRACKER .cycle .load(std::sync::atomic::Ordering::Relaxed); + let shaped = SHAPED_DEVICES.load(); + let shaped_cache = SHAPED_DEVICE_HASH_CACHE.load(); THROUGHPUT_TRACKER .raw_data .lock() @@ -412,13 +466,15 @@ pub fn top_n_up(start: u32, end: u32) -> BusResponse { .filter(|(k, _v)| !k.as_ip().is_loopback()) .filter(|(_k, d)| retire_check(tp_cycle, d.most_recent_cycle)) .map(|(k, te)| { + let (circuit_id, _circuit_name) = + resolve_circuit_metadata_for_entry(&shaped, &shaped_cache, k, te); ( *k, te.actual_bytes_per_second, te.packets_per_second, te.median_latency().unwrap_or(0.0), te.tc_handle, - te.circuit_id.as_ref().unwrap_or(&String::new()).clone(), + circuit_id, down_up_retransmit_sample(te.tcp_retransmits, te.tcp_packets), ) }) @@ -552,6 +608,8 @@ pub fn worst_n(start: u32, end: u32) -> BusResponse { let tp_cycle = THROUGHPUT_TRACKER .cycle .load(std::sync::atomic::Ordering::Relaxed); + let shaped = SHAPED_DEVICES.load(); + let shaped_cache = SHAPED_DEVICE_HASH_CACHE.load(); THROUGHPUT_TRACKER .raw_data .lock() @@ -565,13 +623,15 @@ pub fn worst_n(start: u32, end: u32) -> BusResponse { }) .filter(|(_k, te)| te.median_latency().is_some()) .map(|(k, te)| { + let (circuit_id, _circuit_name) = + resolve_circuit_metadata_for_entry(&shaped, &shaped_cache, k, te); ( *k, te.actual_bytes_per_second, te.packets_per_second, te.median_latency().unwrap_or(0.0), te.tc_handle, - te.circuit_id.as_ref().unwrap_or(&String::new()).clone(), + circuit_id, down_up_retransmit_sample(te.tcp_retransmits, te.tcp_packets), ) }) @@ -602,6 +662,8 @@ pub fn worst_n_retransmits(start: u32, end: u32) -> BusResponse { let tp_cycle = THROUGHPUT_TRACKER .cycle .load(std::sync::atomic::Ordering::Relaxed); + let shaped = SHAPED_DEVICES.load(); + let shaped_cache = SHAPED_DEVICE_HASH_CACHE.load(); THROUGHPUT_TRACKER .raw_data .lock() @@ -610,13 +672,15 @@ pub fn worst_n_retransmits(start: u32, end: u32) -> BusResponse { .filter(|(_k, d)| retire_check(tp_cycle, d.most_recent_cycle)) .filter(|(_k, te)| te.median_latency().is_some()) .map(|(k, te)| { + let (circuit_id, _circuit_name) = + resolve_circuit_metadata_for_entry(&shaped, &shaped_cache, k, te); ( *k, te.actual_bytes_per_second, te.packets_per_second, te.median_latency().unwrap_or(0.0), te.tc_handle, - te.circuit_id.as_ref().unwrap_or(&String::new()).clone(), + circuit_id, down_up_retransmit_sample(te.tcp_retransmits, te.tcp_packets), ) }) @@ -655,6 +719,8 @@ pub fn best_n(start: u32, end: u32) -> BusResponse { let tp_cycle = THROUGHPUT_TRACKER .cycle .load(std::sync::atomic::Ordering::Relaxed); + let shaped = SHAPED_DEVICES.load(); + let shaped_cache = SHAPED_DEVICE_HASH_CACHE.load(); THROUGHPUT_TRACKER .raw_data .lock() @@ -668,13 +734,15 @@ pub fn best_n(start: u32, end: u32) -> BusResponse { }) .filter(|(_k, te)| te.median_latency().is_some()) .map(|(k, te)| { + let (circuit_id, _circuit_name) = + resolve_circuit_metadata_for_entry(&shaped, &shaped_cache, k, te); ( *k, te.actual_bytes_per_second, te.packets_per_second, te.median_latency().unwrap_or(0.0), te.tc_handle, - te.circuit_id.as_ref().unwrap_or(&String::new()).clone(), + circuit_id, down_up_retransmit_sample(te.tcp_retransmits, te.tcp_packets), ) }) @@ -1119,27 +1187,12 @@ pub fn top_flows(n: u32, flow_type: TopFlowType) -> BusResponse { .map(|(ip, flow)| { let geo = get_asn_name_and_country(ip.remote_ip.as_ip()); - let mut circuit_id = String::new(); - let mut circuit_name = String::new(); - if let Some(te) = throughput.get(&ip.local_ip) { - if let Some(id) = &te.circuit_id { - circuit_id = id.clone(); - } - let shaped_device = te - .device_hash - .and_then(|hash| shaped_cache.index_by_device_hash(&shaped, hash)) - .or_else(|| { - te.circuit_hash - .and_then(|hash| shaped_cache.index_by_circuit_hash(&shaped, hash)) - }) - .and_then(|idx| shaped.devices.get(idx)); - if let Some(device) = shaped_device { - if circuit_id.is_empty() { - circuit_id = device.circuit_id.clone(); - } - circuit_name = device.circuit_name.clone(); - } - } + let (circuit_id, circuit_name) = throughput + .get(&ip.local_ip) + .map(|te| { + resolve_circuit_metadata_for_entry(&shaped, &shaped_cache, &ip.local_ip, te) + }) + .unwrap_or_default(); lqos_bus::FlowbeeSummaryData { remote_ip: ip.remote_ip.as_ip().to_string(), @@ -1180,33 +1233,7 @@ pub fn flows_by_ip(ip: &str) -> BusResponse { if let Ok(ip) = ip.parse::() { let ip = XdpIpAddress::from_ip(ip); let lock = ALL_FLOWS.lock(); - let throughput = THROUGHPUT_TRACKER.raw_data.lock(); - let shaped = SHAPED_DEVICES.load(); - let shaped_cache = SHAPED_DEVICE_HASH_CACHE.load(); - let (circuit_id, circuit_name) = { - let mut circuit_id = String::new(); - let mut circuit_name = String::new(); - if let Some(te) = throughput.get(&ip) { - if let Some(id) = &te.circuit_id { - circuit_id = id.clone(); - } - let shaped_device = te - .device_hash - .and_then(|hash| shaped_cache.index_by_device_hash(&shaped, hash)) - .or_else(|| { - te.circuit_hash - .and_then(|hash| shaped_cache.index_by_circuit_hash(&shaped, hash)) - }) - .and_then(|idx| shaped.devices.get(idx)); - if let Some(device) = shaped_device { - if circuit_id.is_empty() { - circuit_id = device.circuit_id.clone(); - } - circuit_name = device.circuit_name.clone(); - } - } - (circuit_id, circuit_name) - }; + let (circuit_id, circuit_name) = resolve_circuit_metadata_for_ip(&ip); let matching_flows: Vec<_> = lock .flow_data .iter() @@ -1364,8 +1391,17 @@ pub struct Lts2Device { #[cfg(test)] mod compatibility_tests { - use super::Lts2Circuit; + use super::{Lts2Circuit, resolve_circuit_metadata_for_entry}; + use crate::shaped_devices_tracker::ShapedDeviceHashCache; + use crate::throughput_tracker::flow_data::RttData; + use crate::throughput_tracker::throughput_entry::ThroughputEntry; + use lqos_bus::TcHandle; + use lqos_config::{ConfigShapedDevices, ShapedDevice}; + use lqos_utils::XdpIpAddress; + use lqos_utils::qoo::QoqScores; + use lqos_utils::units::DownUpOrder; use serde::Deserialize; + use std::net::Ipv4Addr; #[allow(dead_code)] #[derive(Debug, Deserialize)] @@ -1420,4 +1456,56 @@ mod compatibility_tests { assert_eq!(decoded.download_max_mbps, 7); assert_eq!(decoded.upload_max_mbps, 4); } + + #[test] + fn circuit_metadata_falls_back_to_ip_when_entry_circuit_id_is_blank() { + let mut shaped = ConfigShapedDevices::default(); + shaped.replace_with_new_data(vec![ShapedDevice { + circuit_id: "circuit-1".to_string(), + circuit_name: "Circuit Alpha".to_string(), + device_id: "device-1".to_string(), + parent_node: "Parent-A".to_string(), + ipv4: vec![(Ipv4Addr::new(192, 168, 1, 10), 32)], + ..Default::default() + }]); + let cache = ShapedDeviceHashCache::default(); + let ip = XdpIpAddress::from_ip("192.168.1.10".parse().expect("test IP should parse")); + let entry = ThroughputEntry { + circuit_id: None, + circuit_hash: None, + device_hash: None, + network_json_parents: None, + first_cycle: 0, + most_recent_cycle: 0, + bytes: DownUpOrder::zeroed(), + actual_bytes: DownUpOrder::zeroed(), + 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: Default::default(), + recent_rtt_data: [RttData::from_nanos(0); 60], + last_fresh_rtt_data_cycle: 0, + last_seen: 0, + tcp_retransmits: DownUpOrder::zeroed(), + tcp_retransmit_packets: DownUpOrder::zeroed(), + qoq: QoqScores::default(), + }; + + let (circuit_id, circuit_name) = + resolve_circuit_metadata_for_entry(&shaped, &cache, &ip, &entry); + + assert_eq!(circuit_id, "circuit-1"); + assert_eq!(circuit_name, "Circuit Alpha"); + } } diff --git a/src/rust/lqosd/src/throughput_tracker/stats_submission.rs b/src/rust/lqosd/src/throughput_tracker/stats_submission.rs index 8c48f888..2a452033 100644 --- a/src/rust/lqosd/src/throughput_tracker/stats_submission.rs +++ b/src/rust/lqosd/src/throughput_tracker/stats_submission.rs @@ -1,9 +1,12 @@ use crate::lts2_sys::get_lts_license_status; use crate::lts2_sys::shared_types::{CircuitCakeDrops, CircuitCakeMarks, LtsStatus}; -use crate::shaped_devices_tracker::{NETWORK_JSON, SHAPED_DEVICES}; +use crate::shaped_devices_tracker::{ + NETWORK_JSON, SHAPED_DEVICE_HASH_CACHE, SHAPED_DEVICES, shaped_device_from_hashes_or_ip, +}; use crate::system_stats::SystemStats; use crate::throughput_tracker::flow_data::ALL_FLOWS; use crate::throughput_tracker::flow_data::FlowbeeEffectiveDirection; +use crate::throughput_tracker::throughput_entry::ThroughputEntry; use crate::throughput_tracker::{ CIRCUIT_RTT_BUFFERS, Lts2Circuit, Lts2Device, RawNetJs, THROUGHPUT_TRACKER, min_max_median_rtt, min_max_median_tcp_retransmits, @@ -11,8 +14,10 @@ use crate::throughput_tracker::{ use csv::ReaderBuilder; use fxhash::{FxHashMap, FxHashSet}; use lqos_config::{ShapedDevice, TopologyCanonicalStateFile, load_config}; +use lqos_queue_tracker::queue_stats_stale; use lqos_queue_tracker::{ALL_QUEUE_SUMMARY, TOTAL_QUEUE_STATS}; use lqos_topology_compile::TopologyImportFile; +use lqos_utils::XdpIpAddress; use lqos_utils::hash_to_i64; use lqos_utils::units::DownUpOrder; use lqos_utils::unix_time::unix_now; @@ -70,6 +75,22 @@ fn rate_for_submission(rate_mbps: f32) -> u32 { } } +fn queue_metrics_available(config: &lqos_config::Config) -> bool { + !config.queues.queue_mode.is_observe() && !queue_stats_stale() +} + +fn resolved_circuit_hash_for_submission( + shaped: &lqos_config::ConfigShapedDevices, + cache: &crate::shaped_devices_tracker::ShapedDeviceHashCache, + ip: &XdpIpAddress, + entry: &ThroughputEntry, +) -> Option { + entry.circuit_hash.or_else(|| { + shaped_device_from_hashes_or_ip(shaped, cache, ip, entry.device_hash, entry.circuit_hash) + .map(|device| device.circuit_hash) + }) +} + pub(crate) fn submit_throughput_stats( scale: f64, counter: u8, @@ -79,6 +100,7 @@ pub(crate) fn submit_throughput_stats( let Ok(config) = load_config() else { return; }; + let queue_metrics_available = queue_metrics_available(config.as_ref()); // Bail out if we don't have gather stats or a license key if !config.long_term_stats.gather_stats { @@ -126,15 +148,13 @@ pub(crate) fn submit_throughput_stats( let bits_per_second = THROUGHPUT_TRACKER.actual_bits_per_second(); // Check that the stats haven't gone wonky and don't submit obviously bad data. - if let Ok(config) = load_config() { - if bits_per_second.down > (config.queues.downlink_bandwidth_mbps * 1_000_000) { - debug!("Spike detected - not submitting LTS"); - return; // Do not submit these stats - } - if bits_per_second.up > (config.queues.uplink_bandwidth_mbps * 1_000_000) { - debug!("Spike detected - not submitting LTS"); - return; // Do not submit these stats - } + if bits_per_second.down > (config.queues.downlink_bandwidth_mbps * 1_000_000) { + debug!("Spike detected - not submitting LTS"); + return; // Do not submit these stats + } + if bits_per_second.up > (config.queues.uplink_bandwidth_mbps * 1_000_000) { + debug!("Spike detected - not submitting LTS"); + return; // Do not submit these stats } ///////////////////////////////////////////////////////////////// @@ -265,10 +285,26 @@ pub(crate) fn submit_throughput_stats( median_rtt, tcp_retransmits_down: tcp_retransmits.down, tcp_retransmits_up: tcp_retransmits.up, - cake_marks_down: TOTAL_QUEUE_STATS.marks.get_down() as i32, - cake_marks_up: TOTAL_QUEUE_STATS.marks.get_up() as i32, - cake_drops_down: TOTAL_QUEUE_STATS.drops.get_down() as i32, - cake_drops_up: TOTAL_QUEUE_STATS.drops.get_up() as i32, + cake_marks_down: if queue_metrics_available { + TOTAL_QUEUE_STATS.marks.get_down() as i32 + } else { + 0 + }, + cake_marks_up: if queue_metrics_available { + TOTAL_QUEUE_STATS.marks.get_up() as i32 + } else { + 0 + }, + cake_drops_down: if queue_metrics_available { + TOTAL_QUEUE_STATS.drops.get_down() as i32 + } else { + 0 + }, + cake_drops_up: if queue_metrics_available { + TOTAL_QUEUE_STATS.drops.get_up() as i32 + } else { + 0 + }, }) .is_err() { @@ -293,6 +329,7 @@ pub(crate) fn submit_throughput_stats( let mut circuit_retransmits: FxHashMap> = FxHashMap::default(); let shaped_devices = SHAPED_DEVICES.load(); + let shaped_cache = SHAPED_DEVICE_HASH_CACHE.load(); const CRAZY_LIMIT: u64 = 8; // 8x the max bandwidth let plan_lookup: FxHashMap = shaped_devices .devices @@ -313,20 +350,26 @@ pub(crate) fn submit_throughput_stats( .raw_data .lock() .iter() - .filter(|(_k, h)| h.circuit_id.is_some() && h.actual_bytes_per_second.not_zero()) - .for_each(|(_k, h)| { + .filter_map(|(ip, h)| { + let circuit_hash = + resolved_circuit_hash_for_submission(&shaped_devices, &shaped_cache, ip, h)?; + h.actual_bytes_per_second + .not_zero() + .then_some((circuit_hash, h)) + }) + .for_each(|(circuit_hash, h)| { let mut crazy = false; - if let Some((dl, ul)) = plan_lookup.get(&h.circuit_hash.unwrap_or(0)) + if let Some((dl, ul)) = plan_lookup.get(&circuit_hash) && (h.actual_bytes_per_second.down > *dl || h.actual_bytes_per_second.up > *ul) { - crazy_values.insert(h.circuit_hash.unwrap_or(0)); + crazy_values.insert(circuit_hash); crazy = true; } if crazy { return; } - if let Some(c) = circuit_throughput.get_mut(&h.circuit_hash.unwrap_or(0)) { + if let Some(c) = circuit_throughput.get_mut(&circuit_hash) { c.bytes += h.actual_bytes_per_second; c.packets += h.packets_per_second; c.tcp_packets += h.tcp_packets; @@ -334,7 +377,7 @@ pub(crate) fn submit_throughput_stats( c.icmp_packets += h.icmp_packets; } else { circuit_throughput.insert( - h.circuit_hash.unwrap_or(0), + circuit_hash, CircuitThroughputTemp { bytes: h.actual_bytes_per_second, packets: h.packets_per_second, @@ -350,16 +393,17 @@ pub(crate) fn submit_throughput_stats( .raw_data .lock() .iter() - .filter(|(_k, h)| { - h.circuit_id.is_some() - && h.tcp_retransmits.not_zero() - && !crazy_values.contains(&h.circuit_hash.unwrap_or(0)) + .filter_map(|(ip, h)| { + let circuit_hash = + resolved_circuit_hash_for_submission(&shaped_devices, &shaped_cache, ip, h)?; + (h.tcp_retransmits.not_zero() && !crazy_values.contains(&circuit_hash)) + .then_some((circuit_hash, h)) }) - .for_each(|(_k, h)| { - if let Some(c) = circuit_retransmits.get_mut(&h.circuit_hash.unwrap_or(0)) { + .for_each(|(circuit_hash, h)| { + if let Some(c) = circuit_retransmits.get_mut(&circuit_hash) { *c += h.tcp_retransmits; } else { - circuit_retransmits.insert(h.circuit_hash.unwrap_or(0), h.tcp_retransmits); + circuit_retransmits.insert(circuit_hash, h.tcp_retransmits); } }); @@ -431,24 +475,26 @@ pub(crate) fn submit_throughput_stats( // Per host CAKE stats let mut cake_drops: Vec = Vec::new(); let mut cake_marks: Vec = Vec::new(); - ALL_QUEUE_SUMMARY.iterate_queues(|circuit_hash, drops, marks| { - if drops.not_zero() { - cake_drops.push(CircuitCakeDrops { - timestamp: now, - circuit_hash, - cake_drops_down: drops.get_down() as u32, - cake_drops_up: drops.get_up() as u32, - }); - } - if marks.not_zero() { - cake_marks.push(CircuitCakeMarks { - timestamp: now, - circuit_hash, - cake_marks_down: marks.get_down() as u32, - cake_marks_up: marks.get_up() as u32, - }); - } - }); + if queue_metrics_available { + ALL_QUEUE_SUMMARY.iterate_queues(|circuit_hash, drops, marks| { + if drops.not_zero() { + cake_drops.push(CircuitCakeDrops { + timestamp: now, + circuit_hash, + cake_drops_down: drops.get_down() as u32, + cake_drops_up: drops.get_up() as u32, + }); + } + if marks.not_zero() { + cake_marks.push(CircuitCakeMarks { + timestamp: now, + circuit_hash, + cake_marks_down: marks.get_down() as u32, + cake_marks_up: marks.get_up() as u32, + }); + } + }); + } if !cake_drops.is_empty() && crate::lts2_sys::circuit_cake_drops(&cake_drops).is_err() { warn!("Error sending message to LTS2."); } @@ -492,7 +538,7 @@ pub(crate) fn submit_throughput_stats( tcp_retransmits_up: node.current_tcp_retransmits.up as u32, }); } - if node.current_drops.not_zero() { + if queue_metrics_available && node.current_drops.not_zero() { site_cake_drops.push(crate::lts2_sys::shared_types::SiteCakeDrops { timestamp: now, site_hash, @@ -500,7 +546,7 @@ pub(crate) fn submit_throughput_stats( cake_drops_up: node.current_drops.get_up() as u32, }); } - if node.current_marks.not_zero() { + if queue_metrics_available && node.current_marks.not_zero() { site_cake_marks.push(crate::lts2_sys::shared_types::SiteCakeMarks { timestamp: now, site_hash, @@ -544,12 +590,14 @@ pub(crate) fn submit_throughput_stats( if !site_rtt.is_empty() && crate::lts2_sys::site_rtt(&site_rtt).is_err() { warn!("Error sending message to LTS2."); } - if !site_cake_drops.is_empty() + if queue_metrics_available + && !site_cake_drops.is_empty() && crate::lts2_sys::site_cake_drops(&site_cake_drops).is_err() { warn!("Error sending message to LTS2."); } - if !site_cake_marks.is_empty() + if queue_metrics_available + && !site_cake_marks.is_empty() && crate::lts2_sys::site_cake_marks(&site_cake_marks).is_err() { warn!("Error sending message to LTS2."); @@ -679,6 +727,13 @@ fn integration_ingress_enabled(config: &lqos_config::Config) -> bool { #[cfg(test)] mod tests { use super::*; + use crate::throughput_tracker::throughput_entry::ThroughputEntry; + use lqos_bus::TcHandle; + use lqos_config::{ConfigShapedDevices, QueueMode, ShapedDevice}; + use lqos_utils::XdpIpAddress; + use lqos_utils::qoo::QoqScores; + use lqos_utils::units::DownUpOrder; + use std::net::Ipv4Addr; #[test] fn test_rate_for_submission_small_rates() { @@ -728,4 +783,61 @@ mod tests { assert_eq!(rate_for_submission(10.0), 10); assert_eq!(rate_for_submission(100.0), 100); } + + #[test] + fn queue_metrics_disabled_in_observe_mode() { + let mut config = lqos_config::Config::default(); + config.queues.set_queue_mode(QueueMode::Observe); + assert!(!queue_metrics_available(&config)); + } + + #[test] + fn resolved_circuit_hash_falls_back_to_ip_lookup() { + let mut shaped = ConfigShapedDevices::default(); + shaped.replace_with_new_data(vec![ShapedDevice { + circuit_id: "circuit-1".to_string(), + circuit_name: "Circuit Alpha".to_string(), + device_id: "device-1".to_string(), + parent_node: "Parent-A".to_string(), + ipv4: vec![(Ipv4Addr::new(192, 168, 1, 10), 32)], + ..Default::default() + }]); + let cache = crate::shaped_devices_tracker::ShapedDeviceHashCache::default(); + let ip = XdpIpAddress::from_ip("192.168.1.10".parse().expect("test IP should parse")); + let entry = ThroughputEntry { + circuit_id: Some("circuit-1".to_string()), + circuit_hash: None, + device_hash: None, + network_json_parents: None, + first_cycle: 0, + most_recent_cycle: 0, + bytes: DownUpOrder::zeroed(), + actual_bytes: DownUpOrder::zeroed(), + 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: Default::default(), + recent_rtt_data: [crate::throughput_tracker::flow_data::RttData::from_nanos(0); 60], + last_fresh_rtt_data_cycle: 0, + last_seen: 0, + tcp_retransmits: DownUpOrder::zeroed(), + tcp_retransmit_packets: DownUpOrder::zeroed(), + qoq: QoqScores::default(), + }; + + let circuit_hash = resolved_circuit_hash_for_submission(&shaped, &cache, &ip, &entry); + + assert_eq!(circuit_hash, Some(hash_to_i64("circuit-1"))); + } }