IntenseWebs/LibreQoS
3
0 Fork 0
mirror of https://github.com/LibreQoE/LibreQoS.git synced 2026-10-26 06:40:05 -05:00
Code Issues Packages Projects Releases Wiki Activity

changes

Browse Source
This commit is contained in:
rchac committed 2026-04-11 22:23:39 +00:00
1 parent 80244f89a1
commit 39375cdd6e
12 files changed
+1010 -232
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

docs/v2.0/topology-data-flow.md
+3
View File
@@ -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.
src/rust/lqos_bakery/src/utils.rs
+5 -1
View File
@@ -724,7 +724,11 @@ pub(crate) fn write_command_file(path: &Path, commands: &[Vec<String>]) -> 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 {
src/rust/lqos_setup/src/main.rs
+10 -2
View File
@@ -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?
src/rust/lqosd/src/lts2_sys/control_channel.rs
+27 -7
View File
@@ -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;
}
src/rust/lqosd/src/node_manager/local_api/circuit_activity.rs
+14 -6
View File
@@ -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<CircuitFlowSnapshotRow> {
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());
src/rust/lqosd/src/node_manager/local_api/ethernet_caps.rs
+1 -3
View File
@@ -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;
src/rust/lqosd/src/node_manager/local_api/flow_explorer.rs
+2 -28
View File
@@ -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<FlowTimeline> {
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();
}
src/rust/lqosd/src/node_manager/ws/ticker/ipstats_conversion.rs
+68 -18
View File
@@ -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::<IpAddr>().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");
}
}
src/rust/lqosd/src/shaped_devices_tracker/circuit_live.rs
+11 -11
View File
@@ -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<CircuitLiveSnapshot> {
let mut by_circuit_id: FxHashMap<String, CircuitAccumulator> = 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;
};
src/rust/lqosd/src/shaped_devices_tracker/mod.rs
+566 -53
View File
@@ -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<i64>,
circuit_hash: Option<i64>,
) -> 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<String> {
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<Option<PathBuf>> {
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<ConfigShapedDevices> {
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<Option<TopologyShapingInputsFile>> {
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<f32> = 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<f32> = 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");
}
}
src/rust/lqosd/src/throughput_tracker/mod.rs
+142 -54
View File
@@ -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<ThroughputTracker> = Lazy::new(ThroughputTra
pub(crate) static CIRCUIT_RTT_BUFFERS: Lazy<ArcSwap<FxHashMap<i64, RttBuffer>>> =
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::<IpAddr>() {
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");
}
}
src/rust/lqosd/src/throughput_tracker/stats_submission.rs
+161 -49
View File
@@ -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<i64> {
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<i64, DownUpOrder<u64>> = 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<i64, (u64, u64)> = 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<CircuitCakeDrops> = Vec::new();
let mut cake_marks: Vec<CircuitCakeMarks> = 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")));
}
}