diff --git a/docs-es/v2.0/configuration-advanced-es.md b/docs-es/v2.0/configuration-advanced-es.md index 88c44f46..6f26f0ab 100644 --- a/docs-es/v2.0/configuration-advanced-es.md +++ b/docs-es/v2.0/configuration-advanced-es.md @@ -71,11 +71,11 @@ LibreQoS puede mantener contadores de bytes acumulados por circuito y por nodo d enabled = true ``` -- Los contadores acumulan el tráfico medido por el rastreador de throughput de un segundo. - Los valores son bytes realmente transmitidos, no bytes descartados por el shaping. -- Los valores son acumulados desde que `lqosd` inició y se reinician cuando `lqosd` se reinicia. +- Los valores se acumulan mientras `lqosd` se ejecuta y se reinician cuando `lqosd` se reinicia. - Los valores de un nodo incluyen el tráfico de todos sus nodos descendientes. - Solo se listan los circuitos con tráfico registrado. +- Los circuitos eliminados de `ShapedDevices.csv` siguen aportando sus bytes acumulados a su último nodo principal conocido mientras `lqosd` se ejecuta. Con los contadores habilitados, `GET /byte_counters` en la API del nodo devuelve los valores actuales. Consulte [API del nodo LibreQoS](api-es.md). diff --git a/docs/v2.0/configuration-advanced.md b/docs/v2.0/configuration-advanced.md index 2c075db4..02508517 100644 --- a/docs/v2.0/configuration-advanced.md +++ b/docs/v2.0/configuration-advanced.md @@ -73,11 +73,11 @@ LibreQoS can keep cumulative byte counters per circuit and per network node for enabled = true ``` -- Counters accumulate the traffic measured by the one-second throughput tracker. - Values are actual transmitted bytes, not bytes dropped by shaping. -- Values are cumulative since `lqosd` started and reset when `lqosd` restarts. +- Values accumulate while `lqosd` runs and reset when `lqosd` restarts. - Node values include traffic from all descendant nodes. - Only circuits with recorded traffic are listed. +- Circuits removed from `ShapedDevices.csv` keep contributing their accumulated bytes to their last known parent while `lqosd` runs. With the counters enabled, `GET /byte_counters` on the Node API returns the current values. See [LibreQoS Node API](api.md). diff --git a/src/rust/lqos_config/src/etc/v15/byte_counters.rs b/src/rust/lqos_config/src/etc/v15/byte_counters.rs index 73c96fc2..6c6677c5 100644 --- a/src/rust/lqos_config/src/etc/v15/byte_counters.rs +++ b/src/rust/lqos_config/src/etc/v15/byte_counters.rs @@ -23,7 +23,7 @@ mod tests { use super::ByteCountersConfig; #[test] - fn absent_section_defaults_to_disabled() { + fn empty_section_defaults_to_disabled() { let config: ByteCountersConfig = toml::from_str("").expect("empty byte_counters section should parse"); assert!(!config.enabled); diff --git a/src/rust/lqos_config/src/etc/v15/top_config.rs b/src/rust/lqos_config/src/etc/v15/top_config.rs index 9d9feefb..6cb4294d 100644 --- a/src/rust/lqos_config/src/etc/v15/top_config.rs +++ b/src/rust/lqos_config/src/etc/v15/top_config.rs @@ -722,6 +722,16 @@ mod test { assert_eq!(config.local_api, Default::default()); } + #[test] + fn byte_counters_config_defaults_when_section_is_absent() { + let legacy = remove_sections(include_str!("example.toml"), &["byte_counters"]); + let config = + Config::load_from_string(&legacy).expect("Cannot read config without byte_counters"); + + assert_eq!(config.byte_counters, Default::default()); + assert!(!config.byte_counters.enabled); + } + #[test] fn load_example_legacy_spylnx() { let legacy = include_str!("example.toml") diff --git a/src/rust/lqos_network_devices/src/combined_catalog.rs b/src/rust/lqos_network_devices/src/combined_catalog.rs index 83c92660..62688327 100644 --- a/src/rust/lqos_network_devices/src/combined_catalog.rs +++ b/src/rust/lqos_network_devices/src/combined_catalog.rs @@ -225,6 +225,19 @@ mod tests { ); } + #[test] + fn device_by_hashes_falls_back_to_dynamic_circuit() { + let mut dynamic = device(42); + dynamic.circuit_id = "dyn-circuit".to_string(); + + let catalog = catalog(vec![], vec![dynamic]); + let resolved = catalog + .device_by_hashes(None, Some(42)) + .expect("dynamic circuit should resolve by circuit hash"); + + assert_eq!(resolved.circuit_id, "dyn-circuit"); + } + #[test] fn mapped_circuit_count_ignores_unmapped_rows_and_counts_ipv6() { let mut unmapped = device(20); diff --git a/src/rust/lqosd/src/byte_counters.rs b/src/rust/lqosd/src/byte_counters.rs index 12a4b99b..347fbbf7 100644 --- a/src/rust/lqosd/src/byte_counters.rs +++ b/src/rust/lqosd/src/byte_counters.rs @@ -5,20 +5,24 @@ //! module. Counters are cumulative since `lqosd` started and are not //! persisted across restarts. //! -//! Only per-circuit totals are stored. Node totals are derived when a -//! snapshot is requested by walking each circuit's effective parent node -//! through the runtime network tree, so node totals include descendants. +//! Only per-circuit totals and their last-known identity are stored. Node +//! totals are derived when a snapshot is requested by walking each circuit's +//! parent node through the runtime network tree, so node totals include +//! descendants. Circuit identity is captured when a counter first receives +//! traffic, so circuits removed from configuration keep contributing their +//! accumulated bytes to their last known parent while `lqosd` runs. use fxhash::FxHashMap; use lqos_bus::{ByteCountersSnapshot, CircuitByteCounters, NodeByteCounters}; use lqos_config::NetworkJson; +use lqos_network_devices::NetworkDevicesCatalog; use lqos_utils::units::DownUpOrder; use once_cell::sync::Lazy; use parking_lot::RwLock; use std::sync::atomic::{AtomicBool, Ordering}; -/// Cumulative bytes keyed by circuit hash. -static CIRCUIT_BYTES: Lazy>>> = +/// Cumulative bytes and last-known identity keyed by circuit hash. +static CIRCUIT_BYTES: Lazy>> = Lazy::new(|| RwLock::new(FxHashMap::default())); /// Whether byte counter collection is enabled. @@ -39,28 +43,76 @@ pub(crate) fn is_enabled() -> bool { /// Adds one cycle of per-circuit byte deltas to the cumulative counters. /// +/// New circuits capture their identity from `catalog` on first sight. +/// /// Side effects: mutates the shared cumulative counter map. -pub(crate) fn add_batch(deltas: FxHashMap>) { +pub(crate) fn add_batch(deltas: FxHashMap>, catalog: &NetworkDevicesCatalog) { if deltas.is_empty() { return; } - merge(&mut CIRCUIT_BYTES.write(), deltas); + apply_deltas(&mut CIRCUIT_BYTES.write(), deltas, |circuit_hash| { + describe_circuit(catalog, circuit_hash) + }); } -/// Merges per-circuit deltas into a cumulative counter map. -fn merge( - counters: &mut FxHashMap>, +/// Merges per-circuit deltas into a cumulative counter map, capturing identity +/// through `describe` when a circuit is first seen. +fn apply_deltas( + counters: &mut FxHashMap, deltas: FxHashMap>, + describe: impl Fn(i64) -> Option, ) { for (circuit_hash, bytes) in deltas { - counters.entry(circuit_hash).or_default().checked_add(bytes); + let counter = counters.entry(circuit_hash).or_default(); + if counter.descriptor.is_none() { + counter.descriptor = describe(circuit_hash); + } + counter.bytes.checked_add(bytes); } } +/// Fills descriptors for counters that could not resolve identity when they +/// were first recorded, for example because the catalog snapshot was stale. +/// +/// Side effects: mutates the shared cumulative counter map. +fn fill_missing_descriptors() { + if CIRCUIT_BYTES + .read() + .values() + .all(|counter| counter.descriptor.is_some()) + { + return; + } + + let catalog = lqos_network_devices::network_devices_catalog(); + let mut counters = CIRCUIT_BYTES.write(); + for (circuit_hash, counter) in counters.iter_mut() { + if counter.descriptor.is_none() { + counter.descriptor = describe_circuit(&catalog, *circuit_hash); + } + } +} + +/// Resolves one circuit's identity fields from the combined device catalog. +fn describe_circuit( + catalog: &NetworkDevicesCatalog, + circuit_hash: i64, +) -> Option { + let device = catalog.device_by_hashes(None, Some(circuit_hash))?; + Some(CircuitDescriptor { + circuit_id: device.circuit_id.clone(), + circuit_name: device.circuit_name.clone(), + parent_node: device.parent_node.clone(), + parent_node_id: device.parent_node_id.clone(), + }) +} + /// Builds a cumulative byte counter snapshot for the bus. /// -/// Side effects: acquires the counter read lock and the runtime network tree -/// read lock. +/// Side effects: acquires the counter lock and the runtime network tree read +/// lock. Descriptors are resolved before the tree lock is taken so the lock +/// order stays `CIRCUIT_BYTES` then `NETWORK_JSON`, matching the tracker's +/// `NETWORK_JSON` write followed by `CIRCUIT_BYTES` write. pub(crate) fn snapshot() -> ByteCountersSnapshot { if !is_enabled() { return ByteCountersSnapshot { @@ -70,65 +122,70 @@ pub(crate) fn snapshot() -> ByteCountersSnapshot { }; } - let circuit_bytes: Vec<(i64, DownUpOrder)> = CIRCUIT_BYTES - .read() - .iter() - .map(|(circuit_hash, bytes)| (*circuit_hash, *bytes)) - .collect(); + fill_missing_descriptors(); - let catalog = lqos_network_devices::network_devices_catalog(); - lqos_network_devices::with_network_json_read(|tree| { - rollup(tree, &circuit_bytes, |circuit_hash| { - let device = catalog.device_by_hashes(None, Some(circuit_hash))?; - Some(CircuitDescriptor { - circuit_id: device.circuit_id.clone(), - circuit_name: device.circuit_name.clone(), - parent_node: device.parent_node.clone(), - }) - }) - }) + let counters: Vec = CIRCUIT_BYTES.read().values().cloned().collect(); + + lqos_network_devices::with_network_json_read(|tree| rollup(tree, &counters)) } -/// Identity fields for one circuit, resolved from the shaped-devices catalog. +/// Cumulative bytes plus the identity captured when the circuit was first seen. +#[derive(Clone, Default)] +struct CircuitCounter { + bytes: DownUpOrder, + descriptor: Option, +} + +/// Identity fields for one circuit, captured from the device catalog. +#[derive(Clone)] struct CircuitDescriptor { circuit_id: String, circuit_name: String, parent_node: String, + parent_node_id: Option, } /// Rolls circuit totals up the network tree and formats a bus snapshot. /// -/// Node totals include every descendant circuit. Circuits whose device or -/// parent node cannot be resolved are omitted from both lists. -fn rollup( - tree: &NetworkJson, - circuits: &[(i64, DownUpOrder)], - mut lookup: impl FnMut(i64) -> Option, -) -> ByteCountersSnapshot { +/// Node totals include every descendant circuit. Circuits keep a row even when +/// their parent node can no longer be resolved, but those bytes contribute to +/// no node total until the parent resolves again. +fn rollup(tree: &NetworkJson, counters: &[CircuitCounter]) -> ByteCountersSnapshot { let mut node_totals = vec![DownUpOrder::zeroed(); tree.nodes.len()]; - let mut node_index: FxHashMap<&str, usize> = FxHashMap::default(); + let mut node_by_name: FxHashMap<&str, usize> = FxHashMap::default(); + let mut node_by_id: FxHashMap<&str, usize> = FxHashMap::default(); for (index, node) in tree.nodes.iter().enumerate() { - node_index.entry(node.name.as_str()).or_insert(index); + node_by_name.entry(node.name.as_str()).or_insert(index); + if let Some(id) = node.id.as_deref() { + node_by_id.entry(id).or_insert(index); + } } let mut circuit_rows = Vec::new(); - for (circuit_hash, bytes) in circuits { - let Some(descriptor) = lookup(*circuit_hash) else { + for counter in counters { + let Some(descriptor) = counter.descriptor.as_ref() else { continue; }; - let Some(&parent_index) = node_index.get(descriptor.parent_node.as_str()) else { - continue; - }; - for &index in &tree.nodes[parent_index].parents { - if let Some(total) = node_totals.get_mut(index) { - total.checked_add(*bytes); + if let Some(parent_index) = resolve_parent_node(descriptor, &node_by_id, &node_by_name) { + let parent_node = &tree.nodes[parent_index]; + if parent_node.parents.is_empty() { + // The root node carries no parent list, so add its own traffic. + if let Some(total) = node_totals.get_mut(parent_index) { + total.checked_add(counter.bytes); + } + } else { + for &index in &parent_node.parents { + if let Some(total) = node_totals.get_mut(index) { + total.checked_add(counter.bytes); + } + } } } circuit_rows.push(CircuitByteCounters { - circuit_id: descriptor.circuit_id, - circuit_name: descriptor.circuit_name, - parent_node: descriptor.parent_node, - bytes: *bytes, + circuit_id: descriptor.circuit_id.clone(), + circuit_name: descriptor.circuit_name.clone(), + parent_node: descriptor.parent_node.clone(), + bytes: counter.bytes, }); } circuit_rows.sort_by(|a, b| a.circuit_id.cmp(&b.circuit_id)); @@ -151,6 +208,19 @@ fn rollup( } } +/// Resolves a circuit's parent node, preferring the stable node identifier. +fn resolve_parent_node( + descriptor: &CircuitDescriptor, + node_by_id: &FxHashMap<&str, usize>, + node_by_name: &FxHashMap<&str, usize>, +) -> Option { + descriptor + .parent_node_id + .as_deref() + .and_then(|id| node_by_id.get(id).copied()) + .or_else(|| node_by_name.get(descriptor.parent_node.as_str()).copied()) +} + #[cfg(test)] mod tests { use super::*; @@ -190,18 +260,26 @@ mod tests { } } - fn descriptor(circuit_id: &str, parent_node: &str) -> Option { - Some(CircuitDescriptor { + fn descriptor(circuit_id: &str, parent_node: &str) -> CircuitDescriptor { + CircuitDescriptor { circuit_id: circuit_id.to_string(), circuit_name: format!("{circuit_id} name"), parent_node: parent_node.to_string(), - }) + parent_node_id: None, + } + } + + fn counter(descriptor: CircuitDescriptor, bytes: DownUpOrder) -> CircuitCounter { + CircuitCounter { + bytes, + descriptor: Some(descriptor), + } } fn test_tree() -> NetworkJson { NetworkJson { nodes: vec![ - node("Root", None, vec![0], None), + node("Root", None, vec![], None), node("Site A", Some("site-a"), vec![0, 1], Some(0)), node("Site A1", None, vec![0, 1, 2], Some(1)), node("Site B", None, vec![0, 3], Some(0)), @@ -210,28 +288,34 @@ mod tests { } #[test] - fn merge_accumulates_repeated_circuit_deltas() { + fn apply_deltas_accumulates_repeated_circuit_deltas() { let mut counters = FxHashMap::default(); - merge( + apply_deltas( &mut counters, FxHashMap::from_iter([(7, DownUpOrder::new(100, 40))]), + |_| Some(descriptor("circuit-7", "Site A")), ); - merge( + apply_deltas( &mut counters, FxHashMap::from_iter([(7, DownUpOrder::new(10, 5))]), + |_| None, ); - assert_eq!(counters.get(&7), Some(&DownUpOrder::new(110, 45))); + let counter = counters.get(&7).expect("counter should exist"); + assert_eq!(counter.bytes, DownUpOrder::new(110, 45)); + assert_eq!( + counter.descriptor.as_ref().map(|d| d.circuit_id.as_str()), + Some("circuit-7") + ); } #[test] fn rollup_includes_descendant_bytes_in_ancestor_totals() { - let circuits = vec![(1, DownUpOrder::new(100, 0)), (2, DownUpOrder::new(0, 50))]; - let snapshot = rollup(&test_tree(), &circuits, |circuit_hash| match circuit_hash { - 1 => descriptor("circuit-1", "Site A1"), - 2 => descriptor("circuit-2", "Site A"), - _ => None, - }); + let counters = vec![ + counter(descriptor("circuit-1", "Site A1"), DownUpOrder::new(100, 0)), + counter(descriptor("circuit-2", "Site A"), DownUpOrder::new(0, 50)), + ]; + let snapshot = rollup(&test_tree(), &counters); assert!(snapshot.enabled); assert_eq!(snapshot.nodes[0].bytes, DownUpOrder::new(100, 50)); @@ -246,21 +330,86 @@ mod tests { } #[test] - fn rollup_skips_unresolved_circuits_and_parents() { - let circuits = vec![ - (1, DownUpOrder::new(100, 0)), - (2, DownUpOrder::new(10, 10)), - (3, DownUpOrder::new(5, 5)), + fn rollup_keeps_circuit_rows_when_the_parent_node_is_missing() { + let counters = vec![ + counter(descriptor("circuit-1", "Site B"), DownUpOrder::new(100, 0)), + counter( + descriptor("circuit-2", "Missing Node"), + DownUpOrder::new(10, 10), + ), ]; - let snapshot = rollup(&test_tree(), &circuits, |circuit_hash| match circuit_hash { - 1 => descriptor("circuit-1", "Site B"), - 2 => None, - _ => descriptor("circuit-3", "Missing Node"), - }); + let snapshot = rollup(&test_tree(), &counters); - assert_eq!(snapshot.circuits.len(), 1); + assert_eq!(snapshot.circuits.len(), 2); assert_eq!(snapshot.circuits[0].circuit_id, "circuit-1"); + assert_eq!(snapshot.circuits[1].circuit_id, "circuit-2"); assert_eq!(snapshot.nodes[0].bytes, DownUpOrder::new(100, 0)); assert_eq!(snapshot.nodes[3].bytes, DownUpOrder::new(100, 0)); } + + #[test] + fn rollup_skips_counters_without_identity() { + let counters = vec![ + counter(descriptor("circuit-1", "Site B"), DownUpOrder::new(100, 0)), + CircuitCounter::default(), + ]; + let snapshot = rollup(&test_tree(), &counters); + + assert_eq!(snapshot.circuits.len(), 1); + assert_eq!(snapshot.circuits[0].circuit_id, "circuit-1"); + } + + #[test] + fn rollup_prefers_stable_node_ids_over_names() { + let mut tree = test_tree(); + tree.nodes + .push(node("Site B", Some("site-b-2"), vec![0, 4], Some(0))); + let mut identified = descriptor("circuit-1", "Site B"); + identified.parent_node_id = Some("site-b-2".to_string()); + + let snapshot = rollup(&tree, &[counter(identified, DownUpOrder::new(5, 5))]); + + assert_eq!(snapshot.nodes[0].bytes, DownUpOrder::new(5, 5)); + assert_eq!(snapshot.nodes[3].bytes, DownUpOrder::new(0, 0)); + assert_eq!(snapshot.nodes[4].bytes, DownUpOrder::new(5, 5)); + } + + #[test] + fn rollup_counts_circuits_attached_to_root() { + let counters = vec![counter( + descriptor("circuit-1", "Root"), + DownUpOrder::new(7, 3), + )]; + let snapshot = rollup(&test_tree(), &counters); + + assert_eq!(snapshot.nodes[0].bytes, DownUpOrder::new(7, 3)); + } + + #[test] + fn rollup_sorts_circuit_rows_by_circuit_id() { + let counters = vec![ + counter(descriptor("circuit-b", "Site A"), DownUpOrder::new(1, 0)), + counter(descriptor("circuit-a", "Site A"), DownUpOrder::new(1, 0)), + ]; + let snapshot = rollup(&test_tree(), &counters); + + let ids: Vec<&str> = snapshot + .circuits + .iter() + .map(|row| row.circuit_id.as_str()) + .collect(); + assert_eq!(ids, vec!["circuit-a", "circuit-b"]); + } + + #[test] + fn disabled_snapshot_reports_empty_lists() { + let previous = is_enabled(); + set_enabled(false); + let snapshot = snapshot(); + set_enabled(previous); + + assert!(!snapshot.enabled); + assert!(snapshot.nodes.is_empty()); + assert!(snapshot.circuits.is_empty()); + } } diff --git a/src/rust/lqosd/src/throughput_tracker/tracking_data.rs b/src/rust/lqosd/src/throughput_tracker/tracking_data.rs index 364c4956..96cd76be 100644 --- a/src/rust/lqosd/src/throughput_tracker/tracking_data.rs +++ b/src/rust/lqosd/src/throughput_tracker/tracking_data.rs @@ -771,20 +771,12 @@ impl ThroughputTracker { } if entry.packets != entry.prev_packets { entry.most_recent_cycle = self_cycle; - let actual_bytes_delta = DownUpOrder::new( - entry - .actual_bytes - .down - .saturating_sub(entry.prev_actual_bytes.down), - entry - .actual_bytes - .up - .saturating_sub(entry.prev_actual_bytes.up), - ); + let actual_bytes_delta = + entry.actual_bytes.checked_sub_or_zero(entry.prev_actual_bytes); // Call to Bakery Update for existing traffic if let Some(circuit_hash) = entry.circuit_hash { changed_circuits.insert(circuit_hash); - if byte_counters_enabled { + if byte_counters_enabled && actual_bytes_delta.not_zero() { byte_counter_deltas .entry(circuit_hash) .or_default() @@ -945,7 +937,7 @@ impl ThroughputTracker { }); if byte_counters_enabled { - crate::byte_counters::add_batch(byte_counter_deltas); + crate::byte_counters::add_batch(byte_counter_deltas, &catalog); } if !observations.is_empty() {