diff --git a/src/rust/lqos_bus/src/bus/request.rs b/src/rust/lqos_bus/src/bus/request.rs index 7a0f47c2..c669ebca 100644 --- a/src/rust/lqos_bus/src/bus/request.rs +++ b/src/rust/lqos_bus/src/bus/request.rs @@ -116,14 +116,14 @@ pub enum BusRequest { /// Hashed circuit identifier (from ShapedDevices.csv). /// - /// Defaults to `0` for backward compatibility. - #[serde(default)] + /// This is required because lqosd treats kernel-provided hashes as the + /// authoritative source for circuit attribution. circuit_id: u64, /// Hashed device identifier (from ShapedDevices.csv). /// - /// Defaults to `0` for backward compatibility. - #[serde(default)] + /// This is required because lqosd treats kernel-provided hashes as the + /// authoritative source for device attribution. device_id: u64, /// If true, this is a *second* flow for the same IP range on diff --git a/src/rust/lqos_python/src/lib.rs b/src/rust/lqos_python/src/lib.rs index 1979545c..1d991cf3 100644 --- a/src/rust/lqos_python/src/lib.rs +++ b/src/rust/lqos_python/src/lib.rs @@ -775,48 +775,43 @@ fn parse_add_ip( classid: &str, cpu: &str, upload: bool, - circuit_id: Option<&str>, - device_id: Option<&str>, + circuit_id: &str, + device_id: &str, ) -> Result { if !classid.contains(':') { return Err(Error::msg(format!( "Class id must be in the format (major):(minor), e.g. 1:12. Provided string: {classid}" ))); } - let circuit_id = circuit_id - .and_then(|s| (!s.is_empty()).then(|| lqos_utils::hash_to_i64(s) as u64)) - .unwrap_or(0); - let device_id = device_id - .and_then(|s| (!s.is_empty()).then(|| lqos_utils::hash_to_i64(s) as u64)) - .unwrap_or(0); + let circuit_id = circuit_id.trim(); + if circuit_id.is_empty() { + return Err(Error::msg("circuit_id is required")); + } + let device_id = device_id.trim(); + if device_id.is_empty() { + return Err(Error::msg("device_id is required")); + } Ok(BusRequest::MapIpToFlow { ip_address: ip.to_string(), tc_handle: TcHandle::from_string(classid)?, cpu: read_hex_string(cpu)?, // Force HEX representation - circuit_id, - device_id, + circuit_id: lqos_utils::hash_to_i64(circuit_id) as u64, + device_id: lqos_utils::hash_to_i64(device_id) as u64, upload, }) } /// Adds an IP address mapping -#[pyfunction(signature = (ip, classid, cpu, upload, circuit_id=None, device_id=None))] +#[pyfunction(signature = (ip, classid, cpu, upload, circuit_id, device_id))] fn add_ip_mapping( ip: String, classid: String, cpu: String, // In HEX upload: bool, - circuit_id: Option, - device_id: Option, + circuit_id: String, + device_id: String, ) -> PyResult<()> { - let request = parse_add_ip( - &ip, - &classid, - &cpu, - upload, - circuit_id.as_deref(), - device_id.as_deref(), - ); + let request = parse_add_ip(&ip, &classid, &cpu, upload, &circuit_id, &device_id); if let Ok(request) = request { run_query(vec![request]).unwrap(); Ok(()) @@ -837,24 +832,17 @@ impl BatchedCommands { Ok(Self { batch: Vec::new() }) } - #[pyo3(signature = (ip, classid, cpu, upload, circuit_id=None, device_id=None))] + #[pyo3(signature = (ip, classid, cpu, upload, circuit_id, device_id))] pub fn add_ip_mapping( &mut self, ip: String, classid: String, cpu: String, upload: bool, - circuit_id: Option, - device_id: Option, + circuit_id: String, + device_id: String, ) -> PyResult<()> { - let request = parse_add_ip( - &ip, - &classid, - &cpu, - upload, - circuit_id.as_deref(), - device_id.as_deref(), - ); + let request = parse_add_ip(&ip, &classid, &cpu, upload, &circuit_id, &device_id); if let Ok(request) = request { self.batch.push(request); Ok(()) @@ -971,21 +959,49 @@ fn overrides_circuit_adjustments(py: Python<'_>) -> PyResult> { for adj in overrides.circuit_adjustments().iter() { let d = PyDict::new(py); match adj { - lqos_overrides::CircuitAdjustment::CircuitAdjustSpeed { circuit_id, min_download_bandwidth, max_download_bandwidth, min_upload_bandwidth, max_upload_bandwidth } => { + lqos_overrides::CircuitAdjustment::CircuitAdjustSpeed { + circuit_id, + min_download_bandwidth, + max_download_bandwidth, + min_upload_bandwidth, + max_upload_bandwidth, + } => { d.set_item("type", "circuit_adjust_speed")?; d.set_item("circuit_id", circuit_id.clone())?; - if let Some(v) = min_download_bandwidth { d.set_item("min_download_bandwidth", *v)?; } - if let Some(v) = max_download_bandwidth { d.set_item("max_download_bandwidth", *v)?; } - if let Some(v) = min_upload_bandwidth { d.set_item("min_upload_bandwidth", *v)?; } - if let Some(v) = max_upload_bandwidth { d.set_item("max_upload_bandwidth", *v)?; } + if let Some(v) = min_download_bandwidth { + d.set_item("min_download_bandwidth", *v)?; + } + if let Some(v) = max_download_bandwidth { + d.set_item("max_download_bandwidth", *v)?; + } + if let Some(v) = min_upload_bandwidth { + d.set_item("min_upload_bandwidth", *v)?; + } + if let Some(v) = max_upload_bandwidth { + d.set_item("max_upload_bandwidth", *v)?; + } } - lqos_overrides::CircuitAdjustment::DeviceAdjustSpeed { device_id, min_download_bandwidth, max_download_bandwidth, min_upload_bandwidth, max_upload_bandwidth } => { + lqos_overrides::CircuitAdjustment::DeviceAdjustSpeed { + device_id, + min_download_bandwidth, + max_download_bandwidth, + min_upload_bandwidth, + max_upload_bandwidth, + } => { d.set_item("type", "device_adjust_speed")?; d.set_item("device_id", device_id.clone())?; - if let Some(v) = min_download_bandwidth { d.set_item("min_download_bandwidth", *v)?; } - if let Some(v) = max_download_bandwidth { d.set_item("max_download_bandwidth", *v)?; } - if let Some(v) = min_upload_bandwidth { d.set_item("min_upload_bandwidth", *v)?; } - if let Some(v) = max_upload_bandwidth { d.set_item("max_upload_bandwidth", *v)?; } + if let Some(v) = min_download_bandwidth { + d.set_item("min_download_bandwidth", *v)?; + } + if let Some(v) = max_download_bandwidth { + d.set_item("max_download_bandwidth", *v)?; + } + if let Some(v) = min_upload_bandwidth { + d.set_item("min_upload_bandwidth", *v)?; + } + if let Some(v) = max_upload_bandwidth { + d.set_item("max_upload_bandwidth", *v)?; + } } lqos_overrides::CircuitAdjustment::RemoveCircuit { circuit_id } => { d.set_item("type", "remove_circuit")?; @@ -995,7 +1011,10 @@ fn overrides_circuit_adjustments(py: Python<'_>) -> PyResult> { d.set_item("type", "remove_device")?; d.set_item("device_id", device_id.clone())?; } - lqos_overrides::CircuitAdjustment::ReparentCircuit { circuit_id, parent_node } => { + lqos_overrides::CircuitAdjustment::ReparentCircuit { + circuit_id, + parent_node, + } => { d.set_item("type", "reparent_circuit")?; d.set_item("circuit_id", circuit_id.clone())?; d.set_item("parent_node", parent_node.clone())?; @@ -1020,13 +1039,24 @@ fn overrides_network_adjustments(py: Python<'_>) -> PyResult> { for adj in overrides.network_adjustments().iter() { let d = PyDict::new(py); match adj { - lqos_overrides::NetworkAdjustment::AdjustSiteSpeed { site_name, download_bandwidth_mbps, upload_bandwidth_mbps } => { + lqos_overrides::NetworkAdjustment::AdjustSiteSpeed { + site_name, + download_bandwidth_mbps, + upload_bandwidth_mbps, + } => { d.set_item("type", "adjust_site_speed")?; d.set_item("site_name", site_name.clone())?; - if let Some(v) = download_bandwidth_mbps { d.set_item("download_bandwidth_mbps", *v)?; } - if let Some(v) = upload_bandwidth_mbps { d.set_item("upload_bandwidth_mbps", *v)?; } + if let Some(v) = download_bandwidth_mbps { + d.set_item("download_bandwidth_mbps", *v)?; + } + if let Some(v) = upload_bandwidth_mbps { + d.set_item("upload_bandwidth_mbps", *v)?; + } } - lqos_overrides::NetworkAdjustment::SetNodeVirtual { node_name, virtual_node } => { + lqos_overrides::NetworkAdjustment::SetNodeVirtual { + node_name, + virtual_node, + } => { d.set_item("type", "set_node_virtual")?; d.set_item("node_name", node_name.clone())?; d.set_item("virtual", *virtual_node)?; diff --git a/src/rust/lqosd/src/ip_mapping.rs b/src/rust/lqosd/src/ip_mapping.rs index 57de5bdd..953f9389 100644 --- a/src/rust/lqosd/src/ip_mapping.rs +++ b/src/rust/lqosd/src/ip_mapping.rs @@ -18,6 +18,11 @@ pub(crate) fn map_ip_to_flow( circuit_id: u64, device_id: u64, ) -> BusResponse { + if circuit_id == 0 || device_id == 0 { + return BusResponse::Fail( + "MapIpToFlow requires non-zero circuit_id and device_id hashes".to_string(), + ); + } expect_ack(lqos_sys::add_ip_to_tc( ip_address, *tc_handle, cpu, upload, circuit_id, device_id, )) diff --git a/src/rust/lqosd/src/throughput_tracker/tracking_data.rs b/src/rust/lqosd/src/throughput_tracker/tracking_data.rs index 7e2597be..d1d5c1b4 100644 --- a/src/rust/lqosd/src/throughput_tracker/tracking_data.rs +++ b/src/rust/lqosd/src/throughput_tracker/tracking_data.rs @@ -61,6 +61,60 @@ struct CircuitHeatmapAggregate { tcp_packets: DownUpOrder, } +struct ReducedHostCounters { + bytes: DownUpOrder, + packets: DownUpOrder, + tcp_packets: DownUpOrder, + udp_packets: DownUpOrder, + icmp_packets: DownUpOrder, + last_seen: u64, + tc_handle: TcHandle, + circuit_hash: Option, + device_hash: Option, +} + +impl ReducedHostCounters { + fn from_counters(counts: &[lqos_sys::HostCounter]) -> Self { + let mut bytes = DownUpOrder::zeroed(); + let mut packets = DownUpOrder::zeroed(); + let mut tcp_packets = DownUpOrder::zeroed(); + let mut udp_packets = DownUpOrder::zeroed(); + let mut icmp_packets = DownUpOrder::zeroed(); + let mut last_seen = 0u64; + let mut meta_last_seen = 0u64; + let mut meta_tc_handle = 0u32; + let mut meta_circuit_id = 0u64; + let mut meta_device_id = 0u64; + + for c in counts { + bytes.checked_add_direct(c.download_bytes, c.upload_bytes); + packets.checked_add_direct(c.download_packets, c.upload_packets); + tcp_packets.checked_add_direct(c.tcp_download_packets, c.tcp_upload_packets); + udp_packets.checked_add_direct(c.udp_download_packets, c.udp_upload_packets); + icmp_packets.checked_add_direct(c.icmp_download_packets, c.icmp_upload_packets); + last_seen = u64::max(last_seen, c.last_seen); + if c.last_seen > meta_last_seen { + meta_last_seen = c.last_seen; + meta_tc_handle = c.tc_handle; + meta_circuit_id = c.circuit_id; + meta_device_id = c.device_id; + } + } + + Self { + bytes, + packets, + tcp_packets, + udp_packets, + icmp_packets, + last_seen, + tc_handle: TcHandle::from_u32(meta_tc_handle), + circuit_hash: (meta_circuit_id != 0).then_some(meta_circuit_id as i64), + device_hash: (meta_device_id != 0).then_some(meta_device_id as i64), + } + } +} + impl ThroughputTracker { pub(crate) fn new() -> Self { // The capacity used to be taken from MAX_TRACKED_IPs, but @@ -402,63 +456,21 @@ impl ThroughputTracker { let cache = SHAPED_DEVICE_HASH_CACHE.load(); let mut raw_data = self.raw_data.lock(); throughput_for_each(&mut |xdp_ip, counts| { + let reduced = ReducedHostCounters::from_counters(counts); if let Some(entry) = raw_data.get_mut(xdp_ip) { // Zero the counter, we have to do a per-CPU sum - entry.bytes = DownUpOrder::zeroed(); - entry.packets = DownUpOrder::zeroed(); - entry.tcp_packets = DownUpOrder::zeroed(); - entry.udp_packets = DownUpOrder::zeroed(); - entry.icmp_packets = DownUpOrder::zeroed(); - let mut last_seen = 0u64; - // Choose the most recent metadata to reduce staleness when mappings change. - let mut meta_last_seen = 0u64; - let mut meta_tc_handle = 0u32; - let mut meta_circuit_id = 0u64; - let mut meta_device_id = 0u64; - // Sum the counts across CPUs (it's a per-CPU map) - for c in counts { - entry - .bytes - .checked_add_direct(c.download_bytes, c.upload_bytes); - entry - .packets - .checked_add_direct(c.download_packets, c.upload_packets); - entry - .tcp_packets - .checked_add_direct(c.tcp_download_packets, c.tcp_upload_packets); - entry - .udp_packets - .checked_add_direct(c.udp_download_packets, c.udp_upload_packets); - entry - .icmp_packets - .checked_add_direct(c.icmp_download_packets, c.icmp_upload_packets); - last_seen = u64::max(last_seen, c.last_seen); - if c.last_seen > meta_last_seen { - meta_last_seen = c.last_seen; - meta_tc_handle = c.tc_handle; - meta_circuit_id = c.circuit_id; - meta_device_id = c.device_id; - } - } - entry.last_seen = last_seen; + entry.bytes = reduced.bytes; + entry.packets = reduced.packets; + entry.tcp_packets = reduced.tcp_packets; + entry.udp_packets = reduced.udp_packets; + entry.icmp_packets = reduced.icmp_packets; + entry.last_seen = reduced.last_seen; - let new_tc_handle = TcHandle::from_u32(meta_tc_handle); - let new_circuit_hash = if meta_circuit_id != 0 { - Some(meta_circuit_id as i64) - } else { - None - }; - let new_device_hash = if meta_device_id != 0 { - Some(meta_device_id as i64) - } else { - None - }; - - let hashes_changed = - entry.circuit_hash != new_circuit_hash || entry.device_hash != new_device_hash; - entry.tc_handle = new_tc_handle; - entry.circuit_hash = new_circuit_hash; - entry.device_hash = new_device_hash; + let hashes_changed = entry.circuit_hash != reduced.circuit_hash + || entry.device_hash != reduced.device_hash; + entry.tc_handle = reduced.tc_handle; + entry.circuit_hash = reduced.circuit_hash; + entry.device_hash = reduced.device_hash; if hashes_changed { let shaped_device = Self::shaped_device_for_hashes( &shaped, @@ -523,45 +535,8 @@ impl ThroughputTracker { } } } else { - let mut last_seen = 0u64; - let mut meta_last_seen = 0u64; - let mut meta_tc_handle = 0u32; - let mut meta_circuit_id = 0u64; - let mut meta_device_id = 0u64; - let mut total_bytes: DownUpOrder = DownUpOrder::zeroed(); - let mut total_packets: DownUpOrder = DownUpOrder::zeroed(); - let mut total_tcp_packets: DownUpOrder = DownUpOrder::zeroed(); - let mut total_udp_packets: DownUpOrder = DownUpOrder::zeroed(); - let mut total_icmp_packets: DownUpOrder = DownUpOrder::zeroed(); - for c in counts { - total_bytes.checked_add_direct(c.download_bytes, c.upload_bytes); - total_packets.checked_add_direct(c.download_packets, c.upload_packets); - total_tcp_packets - .checked_add_direct(c.tcp_download_packets, c.tcp_upload_packets); - total_udp_packets - .checked_add_direct(c.udp_download_packets, c.udp_upload_packets); - total_icmp_packets - .checked_add_direct(c.icmp_download_packets, c.icmp_upload_packets); - last_seen = u64::max(last_seen, c.last_seen); - if c.last_seen > meta_last_seen { - meta_last_seen = c.last_seen; - meta_tc_handle = c.tc_handle; - meta_circuit_id = c.circuit_id; - meta_device_id = c.device_id; - } - } - - let tc_handle = TcHandle::from_u32(meta_tc_handle); - let circuit_hash = if meta_circuit_id != 0 { - Some(meta_circuit_id as i64) - } else { - None - }; - let device_hash = if meta_device_id != 0 { - Some(meta_device_id as i64) - } else { - None - }; + let circuit_hash = reduced.circuit_hash; + let device_hash = reduced.device_hash; let shaped_device = Self::shaped_device_for_hashes(&shaped, &cache, device_hash, circuit_hash); let circuit_id = shaped_device.map(|d| d.circuit_id.clone()); @@ -573,7 +548,7 @@ impl ThroughputTracker { if config.queues.lazy_threshold_bytes.is_some() { let threshold = config.queues.lazy_threshold_bytes.unwrap_or(0); - if total_bytes.down.saturating_add(total_bytes.up) < threshold { + if reduced.bytes.down.saturating_add(reduced.bytes.up) < threshold { add = false; } } @@ -597,23 +572,23 @@ impl ThroughputTracker { ), first_cycle: self_cycle, most_recent_cycle: 0, - bytes: total_bytes, - packets: total_packets, + bytes: reduced.bytes, + packets: reduced.packets, prev_bytes: DownUpOrder::zeroed(), prev_packets: DownUpOrder::zeroed(), bytes_per_second: DownUpOrder::zeroed(), packets_per_second: DownUpOrder::zeroed(), - tcp_packets: total_tcp_packets, - udp_packets: total_udp_packets, - icmp_packets: total_icmp_packets, + tcp_packets: reduced.tcp_packets, + udp_packets: reduced.udp_packets, + icmp_packets: reduced.icmp_packets, prev_tcp_packets: DownUpOrder::zeroed(), prev_udp_packets: DownUpOrder::zeroed(), prev_icmp_packets: DownUpOrder::zeroed(), - tc_handle, + tc_handle: reduced.tc_handle, rtt_buffer: RttBuffer::default(), recent_rtt_data: [RttData::from_nanos(0); 60], last_fresh_rtt_data_cycle: 0, - last_seen, + last_seen: reduced.last_seen, tcp_retransmits: DownUpOrder::zeroed(), prev_tcp_retransmits: DownUpOrder::zeroed(), qoq: QoqScores::default(), diff --git a/src/rust/xdp_iphash_to_cpu_cmdline/src/main.rs b/src/rust/xdp_iphash_to_cpu_cmdline/src/main.rs index ca4953e8..ca0dfa45 100644 --- a/src/rust/xdp_iphash_to_cpu_cmdline/src/main.rs +++ b/src/rust/xdp_iphash_to_cpu_cmdline/src/main.rs @@ -29,11 +29,11 @@ enum Commands { /// Circuit ID (raw string). Hashed before being stored. #[arg(long)] - circuit_id: Option, + circuit_id: String, /// Device ID (raw string). Hashed before being stored. #[arg(long)] - device_id: Option, + device_id: String, /// Add "--upload 1" if you are using on-a-stick and need to map upload separately #[arg(long)] @@ -98,8 +98,8 @@ fn parse_add_ip( classid: &str, cpu: &str, upload: &Option, - circuit_id: &Option, - device_id: &Option, + circuit_id: &str, + device_id: &str, ) -> Result { //if ip.parse::().is_err() { // return Err(Error::msg(format!("Unable to parse IP address: {ip}"))); @@ -109,22 +109,20 @@ fn parse_add_ip( "Class id must be in the format (major):(minor), e.g. 1:12. Provided string: {classid}" ))); } - let circuit_id = circuit_id - .as_deref() - .filter(|s| !s.is_empty()) - .map(|s| lqos_utils::hash_to_i64(s) as u64) - .unwrap_or(0); - let device_id = device_id - .as_deref() - .filter(|s| !s.is_empty()) - .map(|s| lqos_utils::hash_to_i64(s) as u64) - .unwrap_or(0); + let circuit_id = circuit_id.trim(); + if circuit_id.is_empty() { + return Err(Error::msg("--circuit_id is required")); + } + let device_id = device_id.trim(); + if device_id.is_empty() { + return Err(Error::msg("--device_id is required")); + } Ok(BusRequest::MapIpToFlow { ip_address: ip.to_string(), tc_handle: TcHandle::from_string(classid)?, cpu: read_hex_string(cpu)?, // Force HEX representation - circuit_id, - device_id, + circuit_id: lqos_utils::hash_to_i64(circuit_id) as u64, + device_id: lqos_utils::hash_to_i64(device_id) as u64, upload: upload.is_some(), }) } @@ -142,8 +140,15 @@ pub async fn main() -> Result<()> { device_id, upload, }) => { - talk_to_server(parse_add_ip(&ip, &classid, &cpu, &upload, &circuit_id, &device_id)?) - .await?; + talk_to_server(parse_add_ip( + &ip, + &classid, + &cpu, + &upload, + &circuit_id, + &device_id, + )?) + .await?; } Some(Commands::Del { ip, upload }) => { talk_to_server(BusRequest::DelIpFlow {