From 4bea5eb05738dbb1a117785e1364f9ff418a7339 Mon Sep 17 00:00:00 2001 From: rchac Date: Wed, 6 May 2026 18:40:43 +0000 Subject: [PATCH] changes --- src/LibreQoS.py | 103 ++++++-- src/rust/lqos_bus/src/bus/protocol.rs | 13 + src/rust/lqos_bus/src/bus/request.rs | 8 + src/rust/lqos_python/src/lib.rs | 55 +++++ src/rust/lqos_sys/src/lib.rs | 1 + src/rust/lqos_sys/src/lqos_kernel.rs | 223 +++++++++++++----- src/rust/lqosd/src/main.rs | 34 +++ .../lqosd/src/shaped_devices_tracker/mod.rs | 15 +- src/rust/lqosd/src/urgent.rs | 80 ++++++- src/test_xdp_mapping_recovery.py | 124 ++++++++++ 10 files changed, 577 insertions(+), 79 deletions(-) create mode 100644 src/test_xdp_mapping_recovery.py diff --git a/src/LibreQoS.py b/src/LibreQoS.py index 93f17ac4..67e04899 100755 --- a/src/LibreQoS.py +++ b/src/LibreQoS.py @@ -51,13 +51,37 @@ from liblqos_python import is_lqosd_alive, clear_ip_mappings, delete_ip_mapping, shaping_cpu_count, \ Bakery -# Optional: urgent issue submission (available in newer liblqos_python) +# Optional: urgent issue helpers (available in newer liblqos_python) try: from liblqos_python import submit_urgent_issue # type: ignore except Exception: def submit_urgent_issue(*_args, **_kwargs): return False +try: + from liblqos_python import clear_urgent_issue_by_identity # type: ignore +except ImportError: + _missing_clear_urgent_issue_by_identity_warned = False + + def clear_urgent_issue_by_identity(*_args, **_kwargs): + global _missing_clear_urgent_issue_by_identity_warned + if not _missing_clear_urgent_issue_by_identity_warned: + logging.warning("Unable to clear recovered urgent issues: liblqos_python helper is unavailable") + _missing_clear_urgent_issue_by_identity_warned = True + return False + +try: + from liblqos_python import xdp_ip_mapping_ready # type: ignore +except ImportError: + _missing_xdp_ip_mapping_ready_warned = False + + def xdp_ip_mapping_ready(): + global _missing_xdp_ip_mapping_ready_warned + if not _missing_xdp_ip_mapping_ready_warned: + logging.warning("Unable to check XDP IP mapping readiness: liblqos_python helper is unavailable") + _missing_xdp_ip_mapping_ready_warned = True + return True + try: from liblqos_python import calculate_topology_source_generation # type: ignore except Exception: @@ -78,6 +102,11 @@ class ValidationFailure(RefreshFailure): pass +XDP_IP_MAPPING_APPLY_FAILED = "XDP_IP_MAPPING_APPLY_FAILED" +XDP_MAPPING_READY_TIMEOUT_SECONDS = 5.0 +XDP_MAPPING_READY_INTERVAL_SECONDS = 0.1 + + def report_refresh_failure(code, message, context=None, dedupe_key=None): logging.error(message) print("ERROR: " + message) @@ -98,6 +127,56 @@ def report_refresh_failure(code, message, context=None, dedupe_key=None): pass raise RefreshFailure(message) + +def wait_for_xdp_ip_mapping_ready( + ready=xdp_ip_mapping_ready, + sleep=time.sleep, + now=time.monotonic, + timeout_seconds=XDP_MAPPING_READY_TIMEOUT_SECONDS, + interval_seconds=XDP_MAPPING_READY_INTERVAL_SECONDS, +): + deadline = now() + timeout_seconds + while now() < deadline: + if ready(): + return + sleep(interval_seconds) + raise TimeoutError("XDP IP mapping BPF maps are not ready") + + +def apply_xdp_ip_mappings( + ip_map_batch, + context, + ready=xdp_ip_mapping_ready, + sleep=time.sleep, + report_failure=report_refresh_failure, + clear_recovered_issue=clear_urgent_issue_by_identity, +): + try: + wait_for_xdp_ip_mapping_ready(ready=ready, sleep=sleep) + except (TimeoutError, OSError) as exc: + report_failure( + XDP_IP_MAPPING_APPLY_FAILED, + "Failed to apply XDP IP mappings: " + str(exc), + context, + XDP_IP_MAPPING_APPLY_FAILED, + ) + return + try: + ip_map_batch.finish_ip_mappings() + ip_map_batch.submit() + except OSError as exc: + report_failure( + XDP_IP_MAPPING_APPLY_FAILED, + "Failed to apply XDP IP mappings: " + str(exc), + context, + XDP_IP_MAPPING_APPLY_FAILED, + ) + return + try: + clear_recovered_issue(XDP_IP_MAPPING_APPLY_FAILED, XDP_IP_MAPPING_APPLY_FAILED) + except OSError as exc: + logging.warning("Unable to clear recovered XDP IP mapping urgent issue: %s", exc) + R2Q = 10 #MAX_R2Q = 200_000 MAX_R2Q = 60_000 # See https://lartc.vger.kernel.narkive.com/NKaH1ZNG/htb-quantum-of-class-100001-is-small-consider-r2q-change @@ -2757,20 +2836,14 @@ def refreshShapers(): print("Observe mode active; skipping XDP-CPUMAP-TC IP filter apply after clearing mappings") numXdpCommands = 0 else: - ipMapBatch.finish_ip_mappings() - try: - ipMapBatch.submit() - except Exception as e: - report_refresh_failure( - "XDP_IP_MAPPING_APPLY_FAILED", - "Failed to apply XDP IP mappings: " + str(e), - { - "required_ip_mappings": requiredIpMappings, - "queued_requests": numXdpCommands, - "on_a_stick": on_a_stick(), - }, - "XDP_IP_MAPPING_APPLY_FAILED", - ) + apply_xdp_ip_mappings( + ipMapBatch, + { + "required_ip_mappings": requiredIpMappings, + "queued_requests": numXdpCommands, + "on_a_stick": on_a_stick(), + }, + ) #for command in xdpCPUmapCommands: # logging.info(command) # commands = command.split(' ') diff --git a/src/rust/lqos_bus/src/bus/protocol.rs b/src/rust/lqos_bus/src/bus/protocol.rs index ce68d635..6d48ac66 100644 --- a/src/rust/lqos_bus/src/bus/protocol.rs +++ b/src/rust/lqos_bus/src/bus/protocol.rs @@ -171,6 +171,19 @@ mod tests { assert_eq!(decoded.requests, session.requests); } + #[test] + fn cbor_round_trip_clear_urgent_issue_by_identity_request() { + let session = BusSession { + requests: vec![BusRequest::ClearUrgentIssueByIdentity { + code: "XDP_IP_MAPPING_APPLY_FAILED".to_string(), + dedupe_key: "XDP_IP_MAPPING_APPLY_FAILED".to_string(), + }], + }; + let bytes = encode_session_cbor(&session).expect("encode_session_cbor"); + let decoded = decode_session_cbor(&bytes).expect("decode_session_cbor"); + assert_eq!(decoded.requests, session.requests); + } + #[test] fn cbor_round_trip_reply() { let reply = BusReply { diff --git a/src/rust/lqos_bus/src/bus/request.rs b/src/rust/lqos_bus/src/bus/request.rs index e2561fd9..1209ed21 100644 --- a/src/rust/lqos_bus/src/bus/request.rs +++ b/src/rust/lqos_bus/src/bus/request.rs @@ -539,6 +539,14 @@ pub enum BusRequest { /// Clear a specific urgent issue by ID ClearUrgentIssue(u64), + /// Clear urgent issues matching a code and dedupe key + ClearUrgentIssueByIdentity { + /// Machine-readable issue code + code: String, + /// Dedupe key used when the issue was submitted + dedupe_key: String, + }, + /// Clear all urgent issues ClearAllUrgentIssues, diff --git a/src/rust/lqos_python/src/lib.rs b/src/rust/lqos_python/src/lib.rs index 4c2c9484..788b7cef 100644 --- a/src/rust/lqos_python/src/lib.rs +++ b/src/rust/lqos_python/src/lib.rs @@ -942,6 +942,7 @@ fn liblqos_python(m: &Bound<'_, PyModule>) -> PyResult<()> { m.add_function(wrap_pyfunction!(add_ip_mapping, m)?)?; m.add_function(wrap_pyfunction!(validate_shaped_devices, m)?)?; m.add_function(wrap_pyfunction!(wait_for_bus_ready, m)?)?; + m.add_function(wrap_pyfunction!(xdp_ip_mapping_ready, m)?)?; m.add_function(wrap_pyfunction!(is_libre_already_running, m)?)?; m.add_function(wrap_pyfunction!(create_lock_file, m)?)?; m.add_function(wrap_pyfunction!(free_lock_file, m)?)?; @@ -1084,6 +1085,7 @@ fn liblqos_python(m: &Bound<'_, PyModule>) -> PyResult<()> { m.add_function(wrap_pyfunction!(scheduler_error, m)?)?; m.add_function(wrap_pyfunction!(scheduler_output, m)?)?; m.add_function(wrap_pyfunction!(submit_urgent_issue, m)?)?; + m.add_function(wrap_pyfunction!(clear_urgent_issue_by_identity, m)?)?; m.add_function(wrap_pyfunction!(xdp_ip_mapping_capacity, m)?)?; m.add_function(wrap_pyfunction!(is_insight_enabled, m)?)?; m.add_function(wrap_pyfunction!(log_info, m)?)?; @@ -1379,6 +1381,13 @@ fn xdp_ip_mapping_capacity() -> PyResult { Ok(lqos_sys::ip_mapping_capacity()) } +/// Returns whether the pinned BPF maps required for XDP IP mapping updates are ready. +#[pyfunction] +fn xdp_ip_mapping_ready() -> PyResult { + lqos_sys::ip_mapping_subsystem_ready() + .map_err(|e| PyOSError::new_err(format!("Unable to inspect XDP IP mapping maps: {e}"))) +} + /// Requests Rust-side validation of `ShapedDevices.csv` #[pyfunction] fn validate_shaped_devices() -> PyResult { @@ -3865,6 +3874,52 @@ fn submit_urgent_issue( Ok(false) } +/// Clear urgent issues matching a code and dedupe key. +#[pyfunction] +fn clear_urgent_issue_by_identity(_py: Python, code: String, dedupe_key: String) -> PyResult { + if let Ok(reply) = run_query(vec![BusRequest::ClearUrgentIssueByIdentity { + code: code.clone(), + dedupe_key: dedupe_key.clone(), + }]) { + for resp in reply.iter() { + if let BusResponse::Ack = resp { + return Ok(true); + } + } + } + + let Ok(reply) = run_query(vec![BusRequest::GetUrgentIssues]) else { + return Ok(false); + }; + let mut ids_to_clear = Vec::new(); + for resp in reply.iter() { + if let BusResponse::UrgentIssues(issues) = resp { + ids_to_clear.extend( + issues + .iter() + .filter(|issue| { + issue.code == code + && issue.dedupe_key.as_deref().unwrap_or(&issue.code) == dedupe_key + }) + .map(|issue| issue.id), + ); + } + } + + if ids_to_clear.is_empty() { + return Ok(false); + } + + let requests = ids_to_clear + .into_iter() + .map(BusRequest::ClearUrgentIssue) + .collect::>(); + let Ok(reply) = run_query(requests) else { + return Ok(false); + }; + Ok(reply.iter().any(|resp| matches!(resp, BusResponse::Ack))) +} + /// Log an informational message via the lqosd bus (appears in lqosd logs). #[pyfunction] fn log_info(_py: Python, message: String) -> PyResult { diff --git a/src/rust/lqos_sys/src/lib.rs b/src/rust/lqos_sys/src/lib.rs index f1e10074..a284b0ce 100644 --- a/src/rust/lqos_sys/src/lib.rs +++ b/src/rust/lqos_sys/src/lib.rs @@ -34,6 +34,7 @@ pub use kernel_wrapper::LibreQoSKernels; pub use linux::num_possible_cpus; pub use lqos_kernel::interface_name_to_index; pub use lqos_kernel::ip_mapping_capacity; +pub use lqos_kernel::ip_mapping_subsystem_ready; pub use lqos_kernel::max_tracked_ips; pub use lqos_kernel::unload_xdp_from_interface; pub use tc_classify_control::{initialize_tc_classify_bypass, set_tc_classify_bypass}; diff --git a/src/rust/lqos_sys/src/lqos_kernel.rs b/src/rust/lqos_sys/src/lqos_kernel.rs index 809d05dc..c5ecfaf8 100644 --- a/src/rust/lqos_sys/src/lqos_kernel.rs +++ b/src/rust/lqos_sys/src/lqos_kernel.rs @@ -31,6 +31,13 @@ pub(crate) mod bpf { include!(concat!(env!("OUT_DIR"), "/bindings.rs")); } +#[derive(Clone, Copy)] +struct PinnedMapAbiSpec { + path: &'static str, + expected_key_size: u32, + expected_value_size: u32, +} + /// Returns the value set in the C XDP system's MAX_TRACKED_IPS /// constant. pub fn max_tracked_ips() -> usize { @@ -42,27 +49,7 @@ fn pinned_map_max_entries(path: &str) -> Result> { return Ok(None); } - let path_c = CString::new(path)?; - let fd = unsafe { bpf_obj_get(path_c.as_ptr()) }; - if fd < 0 { - warn!("Unable to open pinned BPF map '{path}' for capacity check (fd={fd})"); - return Ok(None); - } - - let mut info = MaybeUninit::::zeroed(); - let mut len: u32 = std::mem::size_of::() as u32; - let err = unsafe { bpf_obj_get_info_by_fd(fd, info.as_mut_ptr() as *mut c_void, &mut len) }; - unsafe { - close(fd); - } - if err != 0 { - return Err(Error::msg(format!( - "Unable to query pinned BPF map '{path}' capacity (err={err})." - ))); - } - - let info = unsafe { info.assume_init() }; - Ok(Some(info.max_entries)) + Ok(pinned_map_info(path)?.map(|info| info.max_entries)) } /// Returns the currently available IP-mapping capacity. @@ -80,6 +67,155 @@ pub fn ip_mapping_capacity() -> usize { } } +fn ip_mapping_pinned_map_abi_specs() -> [PinnedMapAbiSpec; 3] { + let expected_key_size = std::mem::size_of::() as u32; + let expected_value_size = std::mem::size_of::() as u32; + [ + PinnedMapAbiSpec { + path: "/sys/fs/bpf/map_ip_to_cpu_and_tc", + expected_key_size, + expected_value_size, + }, + PinnedMapAbiSpec { + path: "/sys/fs/bpf/ip_to_cpu_and_tc_hotcache", + expected_key_size: std::mem::size_of::() as u32, + expected_value_size, + }, + PinnedMapAbiSpec { + path: "/sys/fs/bpf/ip_mapping_epoch", + expected_key_size: std::mem::size_of::() as u32, + expected_value_size: std::mem::size_of::() as u32, + }, + ] +} + +fn pinned_map_info(path: &str) -> Result> { + let path_c = CString::new(path)?; + let fd = unsafe { bpf_obj_get(path_c.as_ptr()) }; + if fd < 0 { + return Ok(None); + } + + let mut info = MaybeUninit::::zeroed(); + let mut len: u32 = std::mem::size_of::() as u32; + let err = unsafe { bpf_obj_get_info_by_fd(fd, info.as_mut_ptr() as *mut c_void, &mut len) }; + unsafe { + close(fd); + } + if err != 0 { + return Err(Error::msg(format!( + "Unable to query pinned BPF map '{path}' info (err={err})." + ))); + } + let info = unsafe { info.assume_init() }; + Ok(Some(info)) +} + +fn pinned_map_key_value_sizes(path: &str) -> Result> { + Ok(pinned_map_info(path)?.map(|info| (info.key_size, info.value_size))) +} + +fn ip_mapping_subsystem_ready_with(mut map_sizes: F) -> Result +where + F: FnMut(&str) -> Result>, +{ + for spec in ip_mapping_pinned_map_abi_specs() { + let Some((key_size, value_size)) = map_sizes(spec.path)? else { + return Ok(false); + }; + if key_size != spec.expected_key_size || value_size != spec.expected_value_size { + return Ok(false); + } + } + Ok(true) +} + +/// Returns whether the pinned maps required for XDP IP mapping updates are ready. +/// +/// This function has side effects: it opens the relevant pinned BPF maps and +/// queries their ABI metadata. It returns `Ok(false)` for maps that are not yet +/// pinned or have an incompatible ABI, and `Err` for unexpected inspection +/// failures. +pub fn ip_mapping_subsystem_ready() -> Result { + ip_mapping_subsystem_ready_with(pinned_map_key_value_sizes) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::collections::HashMap; + + #[test] + fn ip_mapping_ready_requires_all_expected_maps() { + let specs = ip_mapping_pinned_map_abi_specs(); + let sizes: HashMap<&str, (u32, u32)> = specs + .iter() + .map(|spec| { + ( + spec.path, + (spec.expected_key_size, spec.expected_value_size), + ) + }) + .collect(); + + let ready = ip_mapping_subsystem_ready_with(|path| Ok(sizes.get(path).copied())) + .expect("readiness check should succeed"); + + assert!(ready); + } + + #[test] + fn ip_mapping_ready_is_false_when_a_map_is_missing() { + let ready = ip_mapping_subsystem_ready_with(|path| { + if path == "/sys/fs/bpf/ip_mapping_epoch" { + Ok(None) + } else { + let spec = ip_mapping_pinned_map_abi_specs() + .into_iter() + .find(|spec| spec.path == path) + .expect("test path should be part of IP mapping ABI specs"); + Ok(Some((spec.expected_key_size, spec.expected_value_size))) + } + }) + .expect("missing map should not be an inspection error"); + + assert!(!ready); + } + + #[test] + fn ip_mapping_ready_is_false_when_a_map_abi_differs() { + let ready = ip_mapping_subsystem_ready_with(|path| { + let spec = ip_mapping_pinned_map_abi_specs() + .into_iter() + .find(|spec| spec.path == path) + .expect("test path should be part of IP mapping ABI specs"); + let value_size = if path == "/sys/fs/bpf/map_ip_to_cpu_and_tc" { + spec.expected_value_size + 1 + } else { + spec.expected_value_size + }; + Ok(Some((spec.expected_key_size, value_size))) + }) + .expect("ABI mismatch should not be an inspection error"); + + assert!(!ready); + } + + #[test] + fn ip_mapping_ready_returns_inspection_error() { + let err = ip_mapping_subsystem_ready_with(|path| { + if path == "/sys/fs/bpf/map_ip_to_cpu_and_tc" { + Err(Error::msg("map info failed")) + } else { + Ok(None) + } + }) + .expect_err("inspection errors should be returned to the caller"); + + assert!(err.to_string().contains("map info failed")); + } +} + pub fn check_root() -> Result<()> { unsafe { if geteuid() == 0 { @@ -99,25 +235,10 @@ fn remove_incompatible_pinned_map( return Ok(()); } - let path_c = CString::new(path)?; - let fd = unsafe { bpf_obj_get(path_c.as_ptr()) }; - if fd < 0 { - warn!("Unable to open pinned BPF map '{path}' for ABI check (fd={fd})"); + let Some(info) = pinned_map_info(path)? else { + warn!("Unable to open pinned BPF map '{path}' for ABI check"); return Ok(()); - } - - let mut info = MaybeUninit::::zeroed(); - let mut len: u32 = std::mem::size_of::() as u32; - let err = unsafe { bpf_obj_get_info_by_fd(fd, info.as_mut_ptr() as *mut c_void, &mut len) }; - unsafe { - close(fd); - } - if err != 0 { - return Err(Error::msg(format!( - "Unable to query pinned BPF map '{path}' info (err={err})." - ))); - } - let info = unsafe { info.assume_init() }; + }; if info.key_size != expected_key_size || info.value_size != expected_value_size { warn!( @@ -135,23 +256,13 @@ fn remove_incompatible_pinned_map( } fn ensure_ip_mapping_maps_abi() -> Result<()> { - let expected_key_size = std::mem::size_of::() as u32; - let expected_value_size = std::mem::size_of::() as u32; - remove_incompatible_pinned_map( - "/sys/fs/bpf/map_ip_to_cpu_and_tc", - expected_key_size, - expected_value_size, - )?; - remove_incompatible_pinned_map( - "/sys/fs/bpf/ip_to_cpu_and_tc_hotcache", - std::mem::size_of::() as u32, - expected_value_size, - )?; - remove_incompatible_pinned_map( - "/sys/fs/bpf/ip_mapping_epoch", - std::mem::size_of::() as u32, - std::mem::size_of::() as u32, - )?; + for spec in ip_mapping_pinned_map_abi_specs() { + remove_incompatible_pinned_map( + spec.path, + spec.expected_key_size, + spec.expected_value_size, + )?; + } remove_incompatible_pinned_map( "/sys/fs/bpf/tc_classify_control", std::mem::size_of::() as u32, diff --git a/src/rust/lqosd/src/main.rs b/src/rust/lqosd/src/main.rs index 6976a159..5e745fb2 100644 --- a/src/rust/lqosd/src/main.rs +++ b/src/rust/lqosd/src/main.rs @@ -1047,6 +1047,10 @@ fn handle_bus_requests(requests: &[BusRequest], responses: &mut Vec urgent::clear(*id); BusResponse::Ack } + BusRequest::ClearUrgentIssueByIdentity { code, dedupe_key } => { + urgent::clear_by_identity(code, dedupe_key); + BusResponse::Ack + } BusRequest::ClearAllUrgentIssues => { urgent::clear_all(); BusResponse::Ack @@ -1390,3 +1394,33 @@ fn search_result_to_bus(entry: node_manager::SearchResult) -> lqos_bus::SearchRe } } } + +#[cfg(test)] +mod tests { + use super::*; + use lqos_bus::{UrgentSeverity, UrgentSource}; + + #[test] + fn bus_clear_urgent_issue_by_identity_reaches_urgent_store() { + let code = "TEST_BUS_XDP_IP_MAPPING_APPLY_FAILED"; + urgent::clear_by_identity(code, code); + urgent::submit( + UrgentSource::LibreQoS, + UrgentSeverity::Error, + code.to_string(), + "mapping failed".to_string(), + None, + Some(code.to_string()), + ); + + let requests = [BusRequest::ClearUrgentIssueByIdentity { + code: code.to_string(), + dedupe_key: code.to_string(), + }]; + let mut responses = Vec::new(); + handle_bus_requests(&requests, &mut responses); + + assert!(matches!(responses.as_slice(), [BusResponse::Ack])); + assert!(!urgent::list().iter().any(|issue| issue.code == code)); + } +} diff --git a/src/rust/lqosd/src/shaped_devices_tracker/mod.rs b/src/rust/lqosd/src/shaped_devices_tracker/mod.rs index 20959e14..e1a64021 100644 --- a/src/rust/lqosd/src/shaped_devices_tracker/mod.rs +++ b/src/rust/lqosd/src/shaped_devices_tracker/mod.rs @@ -4,14 +4,13 @@ use anyhow::Result; use arc_swap::ArcSwap; use fxhash::{FxHashMap, FxHashSet}; use lqos_bus::{BusResponse, Circuit}; +#[cfg(test)] +use lqos_config::load_active_runtime_shaping_inputs; use lqos_config::{ NetworkJsonNode, NetworkJsonTransport, TopologyRuntimeShapingPayloadIdentity, TopologyRuntimeStatusFile, TopologyShapingInputsFile, - load_active_runtime_shaping_inputs_from_status, load_config, - topology_runtime_status_path, + load_active_runtime_shaping_inputs_from_status, load_config, topology_runtime_status_path, }; -#[cfg(test)] -use lqos_config::load_active_runtime_shaping_inputs; use lqos_queue_tracker::EFFECTIVE_NODE_RATES; use lqos_utils::file_watcher::FileWatcher; use lqos_utils::hash_to_i64; @@ -106,7 +105,9 @@ fn publish_shaping_inputs(shaping_inputs: TopologyShapingInputsFile) { invalidate_executive_cache_snapshot(); } -fn topology_status_identity(status: &TopologyRuntimeStatusFile) -> TopologyRuntimeShapingPayloadIdentity { +fn topology_status_identity( + status: &TopologyRuntimeStatusFile, +) -> TopologyRuntimeShapingPayloadIdentity { status.shaping_payload_identity() } @@ -286,7 +287,9 @@ fn load_topology_runtime_status_payload() { let status = match TopologyRuntimeStatusFile::load(config.as_ref()) { Ok(status) => status, Err(err) => { - warn!("Unable to load topology_runtime_status.json: {err}; keeping last-known-good effective parent cache"); + warn!( + "Unable to load topology_runtime_status.json: {err}; keeping last-known-good effective parent cache" + ); return; } }; diff --git a/src/rust/lqosd/src/urgent.rs b/src/rust/lqosd/src/urgent.rs index d8a76da0..f38b87c4 100644 --- a/src/rust/lqosd/src/urgent.rs +++ b/src/rust/lqosd/src/urgent.rs @@ -17,6 +17,10 @@ fn now_unix() -> u64 { unix_now().unwrap_or_default() } +fn urgent_identity_key(code: &str, dedupe_key: Option<&str>) -> String { + dedupe_key.unwrap_or(code).to_string() +} + fn prune_expired(q: &mut VecDeque) { let now = now_unix(); while let Some(front) = q.front() { @@ -44,8 +48,8 @@ pub fn submit( let mut guard = URGENT.lock(); prune_expired(&mut guard); - // Dedupe: same (code, dedupe_key) within window → update timestamp/message - let key = dedupe_key.clone().unwrap_or_else(|| code.clone()); + // Dedupe: same (code, dedupe_key) within window updates timestamp/message. + let key = urgent_identity_key(&code, dedupe_key.as_deref()); if let Some(existing) = guard.iter_mut().rev().find(|i| { (i.code == code) && (i.dedupe_key.as_deref().unwrap_or("") == key) @@ -90,7 +94,79 @@ pub fn clear(id: u64) -> bool { } } +/// Clears urgent issues matching the same code and dedupe key used for submission. +pub fn clear_by_identity(code: &str, dedupe_key: &str) -> usize { + let key = urgent_identity_key(code, Some(dedupe_key)); + let mut guard = URGENT.lock(); + let before = guard.len(); + guard.retain(|issue| { + issue.code != code || issue.dedupe_key.as_deref().unwrap_or("") != key.as_str() + }); + before.saturating_sub(guard.len()) +} + pub fn clear_all() { let mut guard = URGENT.lock(); guard.clear(); } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn clear_by_identity_removes_only_matching_issues() { + let code = "TEST_XDP_IP_MAPPING_APPLY_FAILED"; + let other_code = "TEST_OTHER"; + let other_dedupe = "TEST_XDP_IP_MAPPING_APPLY_FAILED_OTHER_DEDUPE"; + clear_by_identity(code, code); + clear_by_identity(code, other_dedupe); + clear_by_identity(other_code, code); + submit( + UrgentSource::LibreQoS, + UrgentSeverity::Error, + code.to_string(), + "mapping failed".to_string(), + None, + Some(code.to_string()), + ); + submit( + UrgentSource::LibreQoS, + UrgentSeverity::Error, + code.to_string(), + "same code different dedupe".to_string(), + None, + Some(other_dedupe.to_string()), + ); + submit( + UrgentSource::LibreQoS, + UrgentSeverity::Error, + other_code.to_string(), + "other failed".to_string(), + None, + Some(code.to_string()), + ); + + let cleared = clear_by_identity(code, code); + let matching_issue_survived = list() + .into_iter() + .any(|issue| issue.code == code && issue.dedupe_key.as_deref() == Some(other_dedupe)); + + assert_eq!(cleared, 1); + assert!(matching_issue_survived); + assert_eq!(clear_by_identity(code, other_dedupe), 1); + assert_eq!(clear_by_identity(other_code, code), 1); + } + + #[test] + fn urgent_identity_key_falls_back_to_code() { + assert_eq!( + urgent_identity_key("CODE_ONLY", None), + "CODE_ONLY".to_string() + ); + assert_eq!( + urgent_identity_key("CODE", Some("DEDUP")), + "DEDUP".to_string() + ); + } +} diff --git a/src/test_xdp_mapping_recovery.py b/src/test_xdp_mapping_recovery.py new file mode 100644 index 00000000..650f4bdd --- /dev/null +++ b/src/test_xdp_mapping_recovery.py @@ -0,0 +1,124 @@ +import unittest + +import LibreQoS + + +class FakeIpMapBatch: + def __init__(self, submit_error=None): + self.finished = False + self.submitted = False + self.submit_error = submit_error + + def finish_ip_mappings(self): + self.finished = True + + def submit(self): + self.submitted = True + if self.submit_error is not None: + raise self.submit_error + + +class XdpMappingRecoveryTests(unittest.TestCase): + def test_ready_on_first_check_does_not_sleep(self): + sleeps = [] + + LibreQoS.wait_for_xdp_ip_mapping_ready( + ready=lambda: True, + sleep=lambda seconds: sleeps.append(seconds), + now=lambda: 0.0, + ) + + self.assertEqual(sleeps, []) + + def test_readiness_after_one_false_check_sleeps_once(self): + checks = iter([False, True]) + sleeps = [] + times = iter([0.0, 0.0, 0.0]) + + LibreQoS.wait_for_xdp_ip_mapping_ready( + ready=lambda: next(checks), + sleep=lambda seconds: sleeps.append(seconds), + now=lambda: next(times), + timeout_seconds=5.0, + interval_seconds=0.1, + ) + + self.assertEqual(sleeps, [0.1]) + + def test_persistent_false_readiness_times_out(self): + times = iter([0.0, 0.0, 0.2]) + + with self.assertRaises(TimeoutError): + LibreQoS.wait_for_xdp_ip_mapping_ready( + ready=lambda: False, + sleep=lambda _seconds: None, + now=lambda: next(times), + timeout_seconds=0.1, + interval_seconds=0.1, + ) + + def test_readiness_os_error_reports_before_submit(self): + batch = FakeIpMapBatch() + reports = [] + + def raise_os_error(): + raise OSError("Unable to inspect XDP IP mapping maps") + + LibreQoS.apply_xdp_ip_mappings( + batch, + {"queued_requests": 1}, + ready=raise_os_error, + sleep=lambda _seconds: None, + report_failure=lambda *args: reports.append(args), + clear_recovered_issue=lambda *_args: True, + ) + + self.assertFalse(batch.finished) + self.assertFalse(batch.submitted) + self.assertEqual(reports[0][0], LibreQoS.XDP_IP_MAPPING_APPLY_FAILED) + + def test_submit_failure_reports_mapping_apply_failure(self): + batch = FakeIpMapBatch(submit_error=OSError("Unable to open BPF map")) + reports = [] + + LibreQoS.apply_xdp_ip_mappings( + batch, + {"queued_requests": 1}, + ready=lambda: True, + sleep=lambda _seconds: None, + report_failure=lambda *args: reports.append(args), + clear_recovered_issue=lambda *_args: True, + ) + + self.assertTrue(batch.finished) + self.assertTrue(batch.submitted) + self.assertEqual(reports[0][0], LibreQoS.XDP_IP_MAPPING_APPLY_FAILED) + + def test_success_clears_recovered_mapping_issue(self): + batch = FakeIpMapBatch() + cleared = [] + + LibreQoS.apply_xdp_ip_mappings( + batch, + {"queued_requests": 1}, + ready=lambda: True, + sleep=lambda _seconds: None, + report_failure=lambda *_args: self.fail("unexpected failure report"), + clear_recovered_issue=lambda code, dedupe_key: cleared.append((code, dedupe_key)) or True, + ) + + self.assertTrue(batch.finished) + self.assertTrue(batch.submitted) + self.assertEqual( + cleared, + [ + ( + LibreQoS.XDP_IP_MAPPING_APPLY_FAILED, + LibreQoS.XDP_IP_MAPPING_APPLY_FAILED, + ) + ], + ) + + +if __name__ == "__main__": + unittest.main()