From 935e22a3d90965fc40b489dba6c94c91bbfb8c99 Mon Sep 17 00:00:00 2001 From: rchac Date: Thu, 28 May 2026 21:28:35 -0600 Subject: [PATCH] changes --- docs-es/v2.0/troubleshooting-es.md | 1 + docs/v2.0/troubleshooting.md | 2 + src/rust/lqos_overrides/src/file_lock.rs | 568 ++++++++++++++++-- src/rust/lqos_overrides/src/overrides_file.rs | 21 +- .../src/overrides_file/file_lock.rs | 98 --- .../lqos_overrides/src/topology_overrides.rs | 2 +- src/rust/lqos_python/src/lib.rs | 64 +- src/rust/lqos_topology/src/lib.rs | 214 ++++++- src/scheduler.py | 94 ++- src/test_scheduler.py | 234 +++++++- 10 files changed, 1043 insertions(+), 255 deletions(-) delete mode 100644 src/rust/lqos_overrides/src/overrides_file/file_lock.rs diff --git a/docs-es/v2.0/troubleshooting-es.md b/docs-es/v2.0/troubleshooting-es.md index eb08e991..fe880d87 100644 --- a/docs-es/v2.0/troubleshooting-es.md +++ b/docs-es/v2.0/troubleshooting-es.md @@ -150,6 +150,7 @@ Use `LQOSD_MEMORY_WATCHDOG_DISABLED=1` solamente cuando esté observando activam Versiones recientes muestran estado/readiness del scheduler en WebUI. Si el modal del scheduler indica que se agotó el tiempo al cargar detalles, las versiones actuales mantienen visible la última instantánea buena del scheduler con su antigüedad en lugar de convertir ese problema de transporte en un error del scheduler. Trate primero esa advertencia como un problema de comunicación entre WebUI y `lqosd`, y confirme la salud real del scheduler en los logs antes de asumir que falló el shaping. Si falla un subproceso de integración pero el shaping puede continuar con la última topología válida, el scheduler puede seguir mostrando estado ready, pero el último error de integración permanece visible en el estado del scheduler hasta la siguiente integración exitosa. +Si el scheduler no puede leer `lqos_overrides.json` o sus capas materializadas porque otro proceso mantiene el lock de overrides, las versiones actuales reintentan brevemente y luego bloquean esa recarga. La topología anterior permanece en uso, y el error del scheduler incluye detalles del proceso que mantiene el lock, como PID, nombre de proceso, operación y hora de creación cuando estén disponibles. Si aparece caído/desactualizado: diff --git a/docs/v2.0/troubleshooting.md b/docs/v2.0/troubleshooting.md index 4a889d72..e4c3c7d9 100644 --- a/docs/v2.0/troubleshooting.md +++ b/docs/v2.0/troubleshooting.md @@ -202,6 +202,8 @@ If specific APs or switches appear multiple times with suffixed names such as `. If an integration subprocess fails, current builds keep the scheduler alive, publish a shortened output preview to the scheduler status/error surfaces, and save the full captured output to a timestamped file under `/tmp` such as `lqos_scheduler_uisp_integration_YYYYMMDD_HHMMSS.log`. If shaping can continue from the last-known-good topology, the scheduler may still report ready, but the latest integration failure remains visible in scheduler status until the next successful integration run. +If the scheduler cannot read `lqos_overrides.json` or its materialized override layers because another process holds the overrides lock, current builds retry briefly and then block that reload. The previous topology remains in use, and the scheduler error includes lock-holder details such as PID, process name, operation, and lock creation time when available. + ### Scheduler status in WebUI looks unhealthy Recent builds expose scheduler readiness/state in the WebUI (Node Manager). diff --git a/src/rust/lqos_overrides/src/file_lock.rs b/src/rust/lqos_overrides/src/file_lock.rs index 14dcc54c..996e7166 100644 --- a/src/rust/lqos_overrides/src/file_lock.rs +++ b/src/rust/lqos_overrides/src/file_lock.rs @@ -5,92 +5,548 @@ use nix::{ }; use std::{ ffi::CString, - fs::{File, remove_file}, - io::{Read, Write}, - path::Path, + fs::{File, OpenOptions, hard_link, remove_file}, + io::{ErrorKind, Read, Write}, + os::fd::AsRawFd, + path::{Path, PathBuf}, + sync::atomic::{AtomicU64, Ordering}, + time::{SystemTime, UNIX_EPOCH}, }; const LOCK_PATH: &str = "/run/lqos/lqos_overrides.lock"; const LOCK_DIR: &str = "/run/lqos"; const LOCK_DIR_PERMS: &str = "/run/lqos"; +const LOCK_GUARD_PATH: &str = "/run/lqos/lqos_overrides.lock.guard"; +const STALE_LOCK_REPLACE_ATTEMPTS: usize = 3; +const LOCK_TEMP_CREATE_ATTEMPTS: usize = 8; +const LOCK_CONTENTION_CODE: &str = "LQOS_OVERRIDES_LOCKED"; + +static TEMP_LOCK_SEQUENCE: AtomicU64 = AtomicU64::new(0); /// Cross-process lock used while mutating operator-owned override files. -pub struct FileLock {} +#[derive(Debug)] +pub struct FileLock { + lock_path: PathBuf, +} + +struct AcquisitionGuard { + file: File, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +struct LockMetadata { + pid: i32, + process: Option, + operation: Option, + created_unix: Option, +} + +impl LockMetadata { + fn current(operation: &str) -> Self { + Self { + pid: unsafe { getpid() }, + process: current_process_name(), + operation: Some(sanitize_lock_field(operation)), + created_unix: SystemTime::now() + .duration_since(UNIX_EPOCH) + .ok() + .map(|duration| duration.as_secs()), + } + } + + fn parse(contents: &str) -> Result { + let trimmed = contents.trim(); + if let Ok(pid) = trimmed.parse::() { + return Ok(Self { + pid, + process: None, + operation: None, + created_unix: None, + }); + } + + let mut pid = None; + let mut process = None; + let mut operation = None; + let mut created_unix = None; + for line in contents.lines() { + let Some((key, value)) = line.split_once('=') else { + continue; + }; + match key.trim() { + "pid" => pid = Some(value.trim().parse::()?), + "process" => process = Some(value.trim().to_string()), + "operation" => operation = Some(value.trim().to_string()), + "created_unix" => created_unix = Some(value.trim().parse::()?), + _ => {} + } + } + + let Some(pid) = pid else { + return Err(Error::msg( + "The LibreQoS overrides lock file does not contain a process id.", + )); + }; + + Ok(Self { + pid, + process, + operation, + created_unix, + }) + } + + fn serialize(&self) -> String { + let process = self.process.as_deref().unwrap_or("unknown"); + let operation = self.operation.as_deref().unwrap_or("unknown"); + let created_unix = self + .created_unix + .map(|seconds| seconds.to_string()) + .unwrap_or_else(|| "unknown".to_string()); + format!( + "pid={}\nprocess={}\noperation={}\ncreated_unix={}\n", + self.pid, process, operation, created_unix + ) + } + + fn describe_holder(&self) -> String { + let mut parts = vec![format!("pid={}", self.pid)]; + if let Some(process) = self.process.as_deref().filter(|value| !value.is_empty()) { + parts.push(format!("process={process}")); + } + if let Some(operation) = self.operation.as_deref().filter(|value| !value.is_empty()) { + parts.push(format!("operation={operation}")); + } + if let Some(created_unix) = self.created_unix { + parts.push(format!("created_unix={created_unix}")); + } + parts.join(", ") + } +} impl FileLock { - /// Acquires the lock or returns an error if another live process holds it. - pub fn new() -> Result { - Self::check_directory()?; - let lock_path = Path::new(LOCK_PATH); - if lock_path.exists() { - if Self::is_lock_valid()? { - return Err(Error::msg( - "The LibreQoS overrides files are locked by another process.", - )); + /// Acquires the lock and records the caller operation for diagnostics. + pub fn new_for_operation(operation: &str) -> Result { + Self::new_at( + operation, + Path::new(LOCK_PATH), + Path::new(LOCK_DIR), + Path::new(LOCK_DIR_PERMS), + Path::new(LOCK_GUARD_PATH), + ) + } + + fn new_at( + operation: &str, + lock_path: &Path, + lock_dir: &Path, + lock_dir_perms: &Path, + guard_path: &Path, + ) -> Result { + Self::check_directory_at(lock_dir, lock_dir_perms)?; + let _guard = Self::acquire_guard(guard_path)?; + for _ in 0..STALE_LOCK_REPLACE_ATTEMPTS { + if Self::create_lock(lock_path, lock_dir, operation)? { + return Ok(Self { + lock_path: lock_path.to_path_buf(), + }); } - Self::create_lock()?; - Ok(Self {}) - } else { - Self::create_lock()?; - Ok(Self {}) + let metadata = match Self::read_lock_metadata(lock_path) { + Ok(metadata) => metadata, + Err(err) if Self::is_not_found_error(&err) => continue, + Err(err) => { + return Err(Error::msg(format!( + "{LOCK_CONTENTION_CODE}: The LibreQoS overrides files are locked by another process (lock metadata unreadable: {err})." + ))); + } + }; + if Self::is_lock_valid(&metadata) { + return Err(Error::msg(format!( + "{LOCK_CONTENTION_CODE}: The LibreQoS overrides files are locked by another process ({}).", + metadata.describe_holder() + ))); + } + + match remove_file(lock_path) { + Ok(()) => {} + Err(err) if err.kind() == ErrorKind::NotFound => {} + Err(err) => return Err(err.into()), + } } + + Err(Error::msg( + "Unable to acquire the LibreQoS overrides lock after replacing a stale lock.", + )) } - fn is_lock_valid() -> Result { - let mut f = File::open(LOCK_PATH)?; - let mut contents = String::new(); - f.read_to_string(&mut contents)?; - let pid: i32 = contents.trim().parse()?; - - let ret = unsafe { nix::libc::kill(pid, 0) }; - if ret == 0 { - return Ok(true); - } - let err = Errno::last(); - if err == Errno::ESRCH { - Ok(false) - } else { - Ok(true) - } - } - - fn create_lock() -> Result<()> { - let pid = unsafe { getpid() }; - let pid_format = format!("{pid}"); - { - let mut f = File::create(LOCK_PATH)?; - f.write_all(pid_format.as_bytes())?; - } - let unix_path = CString::new(LOCK_PATH)?; + fn acquire_guard(guard_path: &Path) -> Result { + let file = OpenOptions::new() + .read(true) + .write(true) + .create(true) + .truncate(false) + .open(guard_path)?; + let unix_path = CString::new(guard_path.to_string_lossy().as_bytes())?; unsafe { nix::libc::chmod(unix_path.as_ptr(), mode_t::from_le(0o666)); } - Ok(()) + let ret = unsafe { nix::libc::flock(file.as_raw_fd(), nix::libc::LOCK_EX) }; + if ret == 0 { + Ok(AcquisitionGuard { file }) + } else { + Err(Error::msg(format!( + "Unable to acquire LibreQoS overrides lock guard: {}", + Errno::last() + ))) + } } - fn check_directory() -> Result<()> { - let dir_path = std::path::Path::new(LOCK_DIR); - if dir_path.exists() && dir_path.is_dir() { + fn read_lock_metadata(lock_path: &Path) -> Result { + let mut f = File::open(lock_path)?; + let mut contents = String::new(); + f.read_to_string(&mut contents)?; + LockMetadata::parse(&contents) + } + + fn is_not_found_error(err: &Error) -> bool { + err.downcast_ref::() + .is_some_and(|io_err| io_err.kind() == ErrorKind::NotFound) + } + + fn is_lock_valid(metadata: &LockMetadata) -> bool { + let ret = unsafe { nix::libc::kill(metadata.pid, 0) }; + if ret == 0 { + return true; + } + let err = Errno::last(); + err != Errno::ESRCH + } + + fn create_lock(lock_path: &Path, lock_dir: &Path, operation: &str) -> Result { + let metadata = LockMetadata::current(operation); + let (mut f, temp_path) = Self::create_temp_lock_file(lock_dir)?; + f.write_all(metadata.serialize().as_bytes())?; + f.sync_all()?; + drop(f); + let link_result = hard_link(&temp_path, lock_path); + let _ = remove_file(&temp_path); + match link_result { + Ok(()) => {} + Err(err) if err.kind() == ErrorKind::AlreadyExists => return Ok(false), + Err(err) => return Err(err.into()), + } + let unix_path = CString::new(lock_path.to_string_lossy().as_bytes())?; + unsafe { + nix::libc::chmod(unix_path.as_ptr(), mode_t::from_le(0o666)); + } + Ok(true) + } + + fn create_temp_lock_file(lock_dir: &Path) -> Result<(File, PathBuf)> { + let nanos = SystemTime::now() + .duration_since(UNIX_EPOCH) + .ok() + .map(|duration| duration.as_nanos()) + .unwrap_or_default(); + for _ in 0..LOCK_TEMP_CREATE_ATTEMPTS { + let sequence = TEMP_LOCK_SEQUENCE.fetch_add(1, Ordering::Relaxed); + let temp_path = lock_dir.join(format!( + ".lqos_overrides.lock.{}.{}.{}.tmp", + std::process::id(), + nanos, + sequence + )); + match OpenOptions::new() + .write(true) + .create_new(true) + .open(&temp_path) + { + Ok(file) => return Ok((file, temp_path)), + Err(err) if err.kind() == ErrorKind::AlreadyExists => continue, + Err(err) => return Err(err.into()), + } + } + + Err(Error::msg(format!( + "Unable to create a unique temporary LibreQoS overrides lock file after {LOCK_TEMP_CREATE_ATTEMPTS} attempts." + ))) + } + + fn check_directory_at(lock_dir: &Path, lock_dir_perms: &Path) -> Result<()> { + if lock_dir.exists() && lock_dir.is_dir() { Ok(()) } else { - std::fs::create_dir(dir_path)?; - let unix_path = CString::new(LOCK_DIR_PERMS)?; + std::fs::create_dir(lock_dir)?; + let unix_path = CString::new(lock_dir_perms.to_string_lossy().as_bytes())?; unsafe { nix::libc::chmod(unix_path.as_ptr(), 0o777); } Ok(()) } } - - /// Removes the current lock file, if present. - pub fn remove_lock() { - let _ = remove_file(LOCK_PATH); - } } impl Drop for FileLock { fn drop(&mut self) { - Self::remove_lock(); + let _ = remove_file(&self.lock_path); + } +} + +impl Drop for AcquisitionGuard { + fn drop(&mut self) { + let _ = unsafe { nix::libc::flock(self.file.as_raw_fd(), nix::libc::LOCK_UN) }; + } +} + +fn current_process_name() -> Option { + std::fs::read_to_string("/proc/self/comm") + .ok() + .and_then(|name| { + let trimmed = sanitize_lock_field(name.trim()); + if trimmed.is_empty() { + None + } else { + Some(trimmed) + } + }) + .or_else(|| { + std::env::args().next().and_then(|arg| { + let trimmed = sanitize_lock_field(&arg); + if trimmed.is_empty() { + None + } else { + Some(trimmed) + } + }) + }) +} + +fn sanitize_lock_field(value: &str) -> String { + value + .chars() + .filter(|character| *character != '\n' && *character != '\r') + .collect() +} + +#[cfg(test)] +mod tests { + use super::{FileLock, LockMetadata}; + use std::{ + fs::{create_dir_all, read_to_string, remove_dir_all, write}, + path::PathBuf, + time::{SystemTime, UNIX_EPOCH}, + }; + + fn unique_test_dir() -> PathBuf { + let nanos = SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("system clock before UNIX_EPOCH") + .as_nanos(); + std::env::temp_dir().join(format!( + "lqos-overrides-lock-test-{}-{nanos}", + std::process::id() + )) + } + + #[test] + fn parses_legacy_pid_only_lock_file() { + let metadata = LockMetadata::parse("12345\n").expect("legacy pid lock should parse"); + + assert_eq!(metadata.pid, 12345); + assert_eq!(metadata.process, None); + assert_eq!(metadata.operation, None); + assert_eq!(metadata.created_unix, None); + } + + #[test] + fn parses_metadata_lock_file() { + let metadata = LockMetadata::parse( + "pid=12345\nprocess=lqosd\noperation=load effective overrides\ncreated_unix=1800000000\n", + ) + .expect("metadata lock should parse"); + + assert_eq!(metadata.pid, 12345); + assert_eq!(metadata.process.as_deref(), Some("lqosd")); + assert_eq!( + metadata.operation.as_deref(), + Some("load effective overrides") + ); + assert_eq!(metadata.created_unix, Some(1_800_000_000)); + } + + #[test] + fn holder_description_includes_available_metadata() { + let metadata = LockMetadata::parse( + "pid=12345\nprocess=node_manager\noperation=save overrides\ncreated_unix=1800000000\n", + ) + .expect("metadata lock should parse"); + + assert_eq!( + metadata.describe_holder(), + "pid=12345, process=node_manager, operation=save overrides, created_unix=1800000000" + ); + } + + #[test] + fn lock_file_records_operation_metadata_and_cleans_up_on_drop() { + let dir = unique_test_dir(); + create_dir_all(&dir).expect("failed to create temp test dir"); + let path = dir.join("lqos_overrides.lock"); + + { + let _lock = FileLock::new_at( + "load effective overrides", + &path, + &dir, + &dir, + &dir.join("lqos_overrides.lock.guard"), + ) + .expect("failed to create lock in temp dir"); + let contents = read_to_string(&path).expect("failed to read temp lock"); + + assert!(contents.contains(&format!("pid={}", std::process::id()))); + assert!(contents.contains("operation=load effective overrides")); + assert!(contents.contains("created_unix=")); + } + + assert!(!path.exists()); + remove_dir_all(&dir).expect("failed to clean up temp test dir"); + } + + #[test] + fn live_lock_error_reports_existing_holder_metadata() { + let dir = unique_test_dir(); + create_dir_all(&dir).expect("failed to create temp test dir"); + let path = dir.join("lqos_overrides.lock"); + write( + &path, + format!( + "pid={}\nprocess=test-holder\noperation=save overrides\ncreated_unix=1800000000\n", + std::process::id() + ), + ) + .expect("failed to write temp lock"); + + let error = FileLock::new_at( + "load effective overrides", + &path, + &dir, + &dir, + &dir.join("lqos_overrides.lock.guard"), + ) + .expect_err("live lock should reject a second holder"); + let message = error.to_string(); + + assert!(message.contains("pid=")); + assert!(message.contains("process=test-holder")); + assert!(message.contains("operation=save overrides")); + assert!(message.contains("created_unix=1800000000")); + + remove_dir_all(&dir).expect("failed to clean up temp test dir"); + } + + #[test] + fn malformed_lock_file_is_reported_as_lock_contention() { + let dir = unique_test_dir(); + create_dir_all(&dir).expect("failed to create temp test dir"); + let path = dir.join("lqos_overrides.lock"); + write(&path, "pid=").expect("failed to write malformed temp lock"); + + let error = FileLock::new_at( + "load effective overrides", + &path, + &dir, + &dir, + &dir.join("lqos_overrides.lock.guard"), + ) + .expect_err("malformed lock metadata should be reported as contention"); + let message = error.to_string(); + + assert!(message.contains("locked by another process")); + assert!(message.contains("lock metadata unreadable")); + + remove_dir_all(&dir).expect("failed to clean up temp test dir"); + } + + #[test] + fn truncated_metadata_lock_missing_pid_is_reported_as_contention() { + let dir = unique_test_dir(); + create_dir_all(&dir).expect("failed to create temp test dir"); + let path = dir.join("lqos_overrides.lock"); + write( + &path, + "process=test-holder\noperation=save overrides\ncreated_unix=1800000000\n", + ) + .expect("failed to write truncated temp lock"); + + let error = FileLock::new_at( + "load effective overrides", + &path, + &dir, + &dir, + &dir.join("lqos_overrides.lock.guard"), + ) + .expect_err("truncated lock metadata should be reported as contention"); + let message = error.to_string(); + + assert!(message.contains("locked by another process")); + assert!(message.contains("lock metadata unreadable")); + + remove_dir_all(&dir).expect("failed to clean up temp test dir"); + } + + #[test] + fn stale_legacy_lock_is_replaced_with_metadata_lock() { + let dir = unique_test_dir(); + create_dir_all(&dir).expect("failed to create temp test dir"); + let path = dir.join("lqos_overrides.lock"); + write(&path, "2147483647\n").expect("failed to write stale temp lock"); + + { + let _lock = FileLock::new_at( + "load effective overrides", + &path, + &dir, + &dir, + &dir.join("lqos_overrides.lock.guard"), + ) + .expect("stale legacy lock should be replaceable"); + let contents = read_to_string(&path).expect("failed to read temp lock"); + + assert!(contents.contains(&format!("pid={}", std::process::id()))); + assert!(contents.contains("operation=load effective overrides")); + } + + remove_dir_all(&dir).expect("failed to clean up temp test dir"); + } + + #[test] + fn stale_structured_lock_is_replaced_with_metadata_lock() { + let dir = unique_test_dir(); + create_dir_all(&dir).expect("failed to create temp test dir"); + let path = dir.join("lqos_overrides.lock"); + write( + &path, + "pid=2147483647\nprocess=old-holder\noperation=save overrides\ncreated_unix=1800000000\n", + ) + .expect("failed to write stale structured temp lock"); + + { + let _lock = FileLock::new_at( + "load effective overrides", + &path, + &dir, + &dir, + &dir.join("lqos_overrides.lock.guard"), + ) + .expect("stale structured lock should be replaceable"); + let contents = read_to_string(&path).expect("failed to read temp lock"); + + assert!(contents.contains(&format!("pid={}", std::process::id()))); + assert!(contents.contains("operation=load effective overrides")); + } + + remove_dir_all(&dir).expect("failed to clean up temp test dir"); } } diff --git a/src/rust/lqos_overrides/src/overrides_file.rs b/src/rust/lqos_overrides/src/overrides_file.rs index 3d4eaebc..d19607b2 100644 --- a/src/rust/lqos_overrides/src/overrides_file.rs +++ b/src/rust/lqos_overrides/src/overrides_file.rs @@ -228,11 +228,14 @@ fn save_to_path(path: &Path, overrides: &OverrideFile) -> Result<()> { fn write_atomic(path: &Path, contents: &[u8]) -> Result<()> { let parent = path.parent().ok_or_else(|| { - anyhow::anyhow!("Overrides path '{}' has no parent directory", path.display()) - })?; - let file_name = path.file_name().ok_or_else(|| { - anyhow::anyhow!("Overrides path '{}' has no file name", path.display()) + anyhow::anyhow!( + "Overrides path '{}' has no parent directory", + path.display() + ) })?; + let file_name = path + .file_name() + .ok_or_else(|| anyhow::anyhow!("Overrides path '{}' has no file name", path.display()))?; let pid = std::process::id(); let nanos = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) @@ -604,7 +607,7 @@ impl OverrideFile { /// Loads the operator-owned overrides file, creating an empty file if it does not exist. pub fn load() -> Result { - let lock = FileLock::new()?; + let lock = FileLock::new_for_operation("load operator overrides")?; let config = lqos_config::load_config()?; let path = overrides_path(&config, OverrideLayer::Operator); ensure_exists_default(&path)?; @@ -615,7 +618,7 @@ impl OverrideFile { /// Saves this value to the operator-owned overrides file. pub fn save(&self) -> Result<()> { - let lock = FileLock::new()?; + let lock = FileLock::new_for_operation("save operator overrides")?; let config = lqos_config::load_config()?; let path = overrides_path(&config, OverrideLayer::Operator); save_to_path(&path, self)?; @@ -1080,7 +1083,7 @@ impl OverrideStore { /// /// Side effects: acquires the global overrides lock and may create the operator overrides file. pub fn load_layer(layer: OverrideLayer) -> Result { - let lock = FileLock::new()?; + let lock = FileLock::new_for_operation(&format!("load {layer:?} overrides"))?; let config = lqos_config::load_config()?; let path = Self::path_for_layer_config(&config, layer); let overrides = match layer { @@ -1104,7 +1107,7 @@ impl OverrideStore { /// /// Side effects: acquires the global overrides lock and writes the selected overrides file. pub fn save_layer(layer: OverrideLayer, overrides: &OverrideFile) -> Result<()> { - let lock = FileLock::new()?; + let lock = FileLock::new_for_operation(&format!("save {layer:?} overrides"))?; let config = lqos_config::load_config()?; let path = overrides_path(&config, layer); save_to_path(&path, overrides)?; @@ -1118,7 +1121,7 @@ impl OverrideStore { /// /// Side effects: acquires the global overrides lock and may create the operator overrides file. pub fn load_effective(apply_stormguard: bool, apply_treeguard: bool) -> Result { - let lock = FileLock::new()?; + let lock = FileLock::new_for_operation("load effective overrides")?; let config = lqos_config::load_config()?; let merged = Self::load_effective_for_config_inner(&config, apply_stormguard, apply_treeguard)?; diff --git a/src/rust/lqos_overrides/src/overrides_file/file_lock.rs b/src/rust/lqos_overrides/src/overrides_file/file_lock.rs deleted file mode 100644 index 85f57937..00000000 --- a/src/rust/lqos_overrides/src/overrides_file/file_lock.rs +++ /dev/null @@ -1,98 +0,0 @@ -use anyhow::{Error, Result}; -use nix::{ - errno::Errno, - libc::{getpid, mode_t}, -}; -use std::{ - ffi::CString, - fs::{File, remove_file}, - io::{Read, Write}, - path::Path, -}; - -const LOCK_PATH: &str = "/run/lqos/lqos_overrides.lock"; -const LOCK_DIR: &str = "/run/lqos"; -const LOCK_DIR_PERMS: &str = "/run/lqos"; - -pub struct FileLock {} - -impl FileLock { - pub fn new() -> Result { - Self::check_directory()?; - let lock_path = Path::new(LOCK_PATH); - if lock_path.exists() { - if Self::is_lock_valid()? { - return Err(Error::msg( - "The lqos_overrides file is locked by another process.", - )); - } - - // It's a stale pid, so we need to replace it - Self::create_lock()?; - - Ok(Self {}) - } else { - Self::create_lock()?; - Ok(Self {}) - } - } - - fn is_lock_valid() -> Result { - let mut f = File::open(LOCK_PATH)?; - let mut contents = String::new(); - f.read_to_string(&mut contents)?; - let pid: i32 = contents.trim().parse()?; - - // Check if the process exists by sending signal 0. - // Returns 0 if the process exists; -1 with ESRCH if it doesn't; - // other errors (e.g., EPERM) mean the process exists but we lack permission. - let ret = unsafe { nix::libc::kill(pid, 0) }; - if ret == 0 { - return Ok(true); - } - let err = Errno::last(); - if err == Errno::ESRCH { - Ok(false) - } else { - Ok(true) - } - } - - fn create_lock() -> Result<()> { - let pid = unsafe { getpid() }; - let pid_format = format!("{pid}"); - { - let mut f = File::create(LOCK_PATH)?; - f.write_all(pid_format.as_bytes())?; - } - let unix_path = CString::new(LOCK_PATH)?; - unsafe { - nix::libc::chmod(unix_path.as_ptr(), mode_t::from_le(666)); - } - Ok(()) - } - - fn check_directory() -> Result<()> { - let dir_path = std::path::Path::new(LOCK_DIR); - if dir_path.exists() && dir_path.is_dir() { - Ok(()) - } else { - std::fs::create_dir(dir_path)?; - let unix_path = CString::new(LOCK_DIR_PERMS)?; - unsafe { - nix::libc::chmod(unix_path.as_ptr(), 777); - } - Ok(()) - } - } - - pub fn remove_lock() { - let _ = remove_file(LOCK_PATH); // Ignore result - } -} - -impl Drop for FileLock { - fn drop(&mut self) { - Self::remove_lock(); - } -} diff --git a/src/rust/lqos_overrides/src/topology_overrides.rs b/src/rust/lqos_overrides/src/topology_overrides.rs index 8d321193..ed48bfb0 100644 --- a/src/rust/lqos_overrides/src/topology_overrides.rs +++ b/src/rust/lqos_overrides/src/topology_overrides.rs @@ -200,7 +200,7 @@ impl TopologyOverridesFile { /// Saves this value to the operator-owned topology-manager overrides file. pub fn save(&self) -> Result<()> { - let lock = FileLock::new()?; + let lock = FileLock::new_for_operation("save topology overrides")?; let config = lqos_config::load_config()?; let path = topology_overrides_path(&config); save_to_path(&path, self)?; diff --git a/src/rust/lqos_python/src/lib.rs b/src/rust/lqos_python/src/lib.rs index 788b7cef..ea46c9bc 100644 --- a/src/rust/lqos_python/src/lib.rs +++ b/src/rust/lqos_python/src/lib.rs @@ -1067,6 +1067,7 @@ fn liblqos_python(m: &Bound<'_, PyModule>) -> PyResult<()> { overrides_network_adjustments_materialized, m )?)?; + m.add_function(wrap_pyfunction!(overrides_materialized, m)?)?; m.add_function(wrap_pyfunction!(is_network_flat, m)?)?; m.add_function(wrap_pyfunction!(blackboard_finish, m)?)?; m.add_function(wrap_pyfunction!(blackboard_submit, m)?)?; @@ -1564,13 +1565,17 @@ fn overrides_persistent_devices_effective(py: Python<'_>) -> PyResult) -> PyResult> { - let overrides = match lqos_overrides::OverrideStore::load_effective(false, false) { - Ok(o) => o, - Err(e) => return Err(PyOSError::new_err(e.to_string())), - }; + let overrides = load_materialized_overrides()?; + persistent_devices_to_py(py, overrides.persistent_devices()) +} + +fn persistent_devices_to_py( + py: Python<'_>, + devices: &[lqos_config::ShapedDevice], +) -> PyResult> { let mut out: Vec = Vec::new(); - for dev in overrides.persistent_devices().iter() { + for dev in devices.iter() { let ipv4s: Vec = dev .ipv4 .iter() @@ -1813,13 +1818,17 @@ fn overrides_circuit_adjustments_effective(py: Python<'_>) -> PyResult) -> PyResult> { - let overrides = match lqos_overrides::OverrideStore::load_effective(false, false) { - Ok(o) => o, - Err(e) => return Err(PyOSError::new_err(e.to_string())), - }; + let overrides = load_materialized_overrides()?; + circuit_adjustments_to_py(py, overrides.circuit_adjustments()) +} + +fn circuit_adjustments_to_py( + py: Python<'_>, + adjustments: &[lqos_overrides::CircuitAdjustment], +) -> PyResult> { let mut out: Vec = Vec::new(); - for adj in overrides.circuit_adjustments().iter() { + for adj in adjustments.iter() { let d = PyDict::new(py); match adj { lqos_overrides::CircuitAdjustment::CircuitAdjustSpeed { @@ -1940,14 +1949,41 @@ fn overrides_network_adjustments_effective(py: Python<'_>) -> PyResult) -> PyResult> { - let overrides = match lqos_overrides::OverrideStore::load_effective(false, false) { - Ok(o) => o, - Err(e) => return Err(PyOSError::new_err(e.to_string())), - }; + let overrides = load_materialized_overrides()?; network_adjustments_to_py(py, overrides.network_adjustments()) } +/// Returns all operator-owned override sections that scheduler materializes into +/// source-of-truth compatibility files. +/// +/// Side effects: acquires the overrides file lock once and reads the effective +/// operator-only override snapshot. +#[pyfunction] +fn overrides_materialized(py: Python<'_>) -> PyResult { + let overrides = load_materialized_overrides()?; + + let out = PyDict::new(py); + out.set_item( + "persistent_devices", + persistent_devices_to_py(py, overrides.persistent_devices())?, + )?; + out.set_item( + "circuit_adjustments", + circuit_adjustments_to_py(py, overrides.circuit_adjustments())?, + )?; + out.set_item( + "network_adjustments", + network_adjustments_to_py(py, overrides.network_adjustments())?, + )?; + Ok(out.unbind().into()) +} + +fn load_materialized_overrides() -> PyResult { + lqos_overrides::OverrideStore::load_effective(false, false) + .map_err(|e| PyOSError::new_err(e.to_string())) +} + fn network_adjustments_to_py( py: Python<'_>, adjustments: &[lqos_overrides::NetworkAdjustment], diff --git a/src/rust/lqos_topology/src/lib.rs b/src/rust/lqos_topology/src/lib.rs index 29e698f5..4518aff7 100644 --- a/src/rust/lqos_topology/src/lib.rs +++ b/src/rust/lqos_topology/src/lib.rs @@ -10,13 +10,14 @@ use lqos_config::{ TopLevelPlannerItem, TopLevelPlannerMode, TopLevelPlannerParams, TopologyAllowedParent, TopologyAttachmentHealthStateFile, TopologyAttachmentHealthStatus, TopologyAttachmentOption, TopologyAttachmentRateSource, TopologyAttachmentRole, TopologyCanonicalIngressKind, - TopologyCanonicalNode, TopologyCanonicalStateFile, TopologyEditorNode, TopologyEditorStateFile, - TopologyEffectiveAttachmentState, TopologyEffectiveNodeState, TopologyEffectiveStateFile, - TopologyQueueVisibilityPolicy, TopologyRuntimeStatusFile, TopologyShapingCircuitInput, - TopologyShapingDeviceInput, TopologyShapingInputsFile, TopologyShapingResolutionSource, - circuit_anchors_path, compute_effective_network_generation, detect_shaping_cpus, - plan_top_level_assignments, topology_effective_network_path, topology_effective_state_path, - topology_runtime_status_path, topology_shaping_inputs_path, + TopologyCanonicalNode, TopologyCanonicalRateInputSource, TopologyCanonicalStateFile, + TopologyEditorNode, TopologyEditorStateFile, TopologyEffectiveAttachmentState, + TopologyEffectiveNodeState, TopologyEffectiveStateFile, TopologyQueueVisibilityPolicy, + TopologyRuntimeStatusFile, TopologyShapingCircuitInput, TopologyShapingDeviceInput, + TopologyShapingInputsFile, TopologyShapingResolutionSource, circuit_anchors_path, + compute_effective_network_generation, detect_shaping_cpus, plan_top_level_assignments, + topology_effective_network_path, topology_effective_state_path, topology_runtime_status_path, + topology_shaping_inputs_path, }; use lqos_overrides::{ CircuitAdjustment, NetworkAdjustment, OverrideStore, TopologyAttachmentMode, @@ -3269,7 +3270,16 @@ fn rate_pair_from_value(node: &Map) -> CompiledRatePair { ) } -fn rate_pair_from_canonical_node(node: &TopologyCanonicalNode) -> CompiledRatePair { +fn rate_pair_from_canonical_node( + node: &TopologyCanonicalNode, + use_compatibility_export_rates: bool, +) -> CompiledRatePair { + if node.rate_input.source == TopologyCanonicalRateInputSource::CompatibilityExport + && !use_compatibility_export_rates + { + return CompiledRatePair::default(); + } + compiled_rate_pair( node.rate_input .intrinsic_download_mbps @@ -3352,16 +3362,21 @@ fn recompile_effective_bandwidths_for_value( canonical_nodes: &HashMap<&str, &TopologyCanonicalNode>, selected_attachment_caps: &HashMap, inherited_parent_rates: Option, + use_compatibility_export_rates: bool, ) { let Some(node) = value.as_object_mut() else { return; }; let existing_rates = rate_pair_from_value(node); let node_id = node.get("id").and_then(Value::as_str); - let mut compiled = node_id - .and_then(|node_id| canonical_nodes.get(node_id).copied()) - .map(rate_pair_from_canonical_node) + let canonical_node = node_id.and_then(|node_id| canonical_nodes.get(node_id).copied()); + let mut compiled = canonical_node + .map(|node| rate_pair_from_canonical_node(node, use_compatibility_export_rates)) .unwrap_or(existing_rates); + let skip_existing_compatibility_rates = canonical_node.is_some_and(|node| { + node.rate_input.source == TopologyCanonicalRateInputSource::CompatibilityExport + && !use_compatibility_export_rates + }); if let Some(node_id) = node_id && let Some(attachment_rates) = selected_attachment_caps.get(node_id) { @@ -3371,14 +3386,22 @@ fn recompile_effective_bandwidths_for_value( compiled = intersect_rate_pairs(compiled, parent_rates); } if compiled.download.is_none() { - compiled.download = existing_rates - .download - .or(inherited_parent_rates.and_then(|pair| pair.download)); + compiled.download = if skip_existing_compatibility_rates { + inherited_parent_rates.and_then(|pair| pair.download) + } else { + existing_rates + .download + .or(inherited_parent_rates.and_then(|pair| pair.download)) + }; } if compiled.upload.is_none() { - compiled.upload = existing_rates - .upload - .or(inherited_parent_rates.and_then(|pair| pair.upload)); + compiled.upload = if skip_existing_compatibility_rates { + inherited_parent_rates.and_then(|pair| pair.upload) + } else { + existing_rates + .upload + .or(inherited_parent_rates.and_then(|pair| pair.upload)) + }; } write_rate_pair(node, compiled); let next_parent_rates = Some(rate_pair_from_value(node)); @@ -3389,6 +3412,7 @@ fn recompile_effective_bandwidths_for_value( canonical_nodes, selected_attachment_caps, next_parent_rates, + use_compatibility_export_rates, ); } } @@ -3406,12 +3430,15 @@ fn recompile_effective_network_bandwidths( .map(|node| (node.node_id.as_str(), node)) .collect::>(); let selected_attachment_caps = selected_attachment_rate_caps(ui_state, effective); + let use_compatibility_export_rates = + canonical.ingress_kind != TopologyCanonicalIngressKind::NativeIntegration; for node in root.values_mut() { recompile_effective_bandwidths_for_value( node, &canonical_nodes, &selected_attachment_caps, None, + use_compatibility_export_rates, ); } } @@ -4376,10 +4403,11 @@ mod tests { CircuitAnchor, CircuitAnchorsFile, Config, ConfigShapedDevices, ShapedDevice, TopologyAllowedParent, TopologyAttachmentHealthStateFile, TopologyAttachmentHealthStatus, TopologyAttachmentOption, TopologyAttachmentRateSource, TopologyAttachmentRole, - TopologyCanonicalIngressKind, TopologyCanonicalNode, TopologyCanonicalStateFile, - TopologyEditorNode, TopologyEditorStateFile, TopologyEffectiveAttachmentState, - TopologyEffectiveNodeState, TopologyEffectiveStateFile, TopologyQueueVisibilityPolicy, - TopologyRuntimeStatusFile, topology_effective_network_path, topology_effective_state_path, + TopologyCanonicalIngressKind, TopologyCanonicalNode, TopologyCanonicalRateInput, + TopologyCanonicalRateInputSource, TopologyCanonicalStateFile, TopologyEditorNode, + TopologyEditorStateFile, TopologyEffectiveAttachmentState, TopologyEffectiveNodeState, + TopologyEffectiveStateFile, TopologyQueueVisibilityPolicy, TopologyRuntimeStatusFile, + topology_effective_network_path, topology_effective_state_path, topology_runtime_status_path, topology_shaping_inputs_path, }; use lqos_overrides::{TopologyAttachmentMode, TopologyOverridesFile}; @@ -4399,6 +4427,150 @@ mod tests { path } + fn canonical_node_with_rate_source( + node_id: &str, + node_name: &str, + download: u64, + upload: u64, + source: TopologyCanonicalRateInputSource, + ) -> TopologyCanonicalNode { + TopologyCanonicalNode { + node_id: node_id.to_string(), + node_name: node_name.to_string(), + latitude: None, + longitude: None, + node_kind: "Site".to_string(), + is_virtual: false, + current_parent_node_id: None, + current_parent_node_name: None, + current_attachment_id: None, + current_attachment_name: None, + can_move: true, + allowed_parents: Vec::new(), + queue_visibility_policy: TopologyQueueVisibilityPolicy::QueueVisible, + rate_input: TopologyCanonicalRateInput { + intrinsic_download_mbps: Some(download), + intrinsic_upload_mbps: Some(upload), + legacy_imported_download_mbps: Some(download), + legacy_imported_upload_mbps: Some(upload), + source, + }, + } + } + + fn sample_effective_bandwidth_tree() -> serde_json::Map { + json!({ + "Hoodoo Hill": { + "children": { + "HoodooHill-Thunderhill": { + "children": {}, + "downloadBandwidthMbps": 214, + "id": "ap-thunderhill", + "name": "HoodooHill-Thunderhill", + "type": "AP", + "uploadBandwidthMbps": 773 + } + }, + "downloadBandwidthMbps": 774, + "id": "site-hoodoo", + "name": "Hoodoo Hill", + "type": "Site", + "uploadBandwidthMbps": 774 + } + }) + .as_object() + .expect("sample tree should be an object") + .clone() + } + + #[test] + fn native_compatibility_export_rates_do_not_clamp_auto_child_nodes() { + let mut root = sample_effective_bandwidth_tree(); + let canonical = TopologyCanonicalStateFile { + ingress_kind: TopologyCanonicalIngressKind::NativeIntegration, + nodes: vec![ + canonical_node_with_rate_source( + "site-hoodoo", + "Hoodoo Hill", + 774, + 774, + TopologyCanonicalRateInputSource::AttachmentMax, + ), + canonical_node_with_rate_source( + "ap-thunderhill", + "HoodooHill-Thunderhill", + 214, + 773, + TopologyCanonicalRateInputSource::CompatibilityExport, + ), + ], + ..Default::default() + }; + + super::recompile_effective_network_bandwidths( + &mut root, + &canonical, + &TopologyEditorStateFile::default(), + &TopologyEffectiveStateFile::default(), + ); + + let ap = root["Hoodoo Hill"]["children"]["HoodooHill-Thunderhill"] + .as_object() + .expect("AP node should exist"); + assert_eq!( + super::node_bandwidth_mbps(ap, "downloadBandwidthMbps"), + Some(774) + ); + assert_eq!( + super::node_bandwidth_mbps(ap, "uploadBandwidthMbps"), + Some(774) + ); + } + + #[test] + fn legacy_imported_rates_still_cap_child_nodes() { + let mut root = sample_effective_bandwidth_tree(); + let canonical = TopologyCanonicalStateFile { + ingress_kind: TopologyCanonicalIngressKind::LegacyNetworkJson, + nodes: vec![ + canonical_node_with_rate_source( + "site-hoodoo", + "Hoodoo Hill", + 774, + 774, + TopologyCanonicalRateInputSource::ImportedNetworkJson, + ), + canonical_node_with_rate_source( + "ap-thunderhill", + "HoodooHill-Thunderhill", + 214, + 773, + TopologyCanonicalRateInputSource::ImportedNetworkJson, + ), + ], + ..Default::default() + }; + + super::recompile_effective_network_bandwidths( + &mut root, + &canonical, + &TopologyEditorStateFile::default(), + &TopologyEffectiveStateFile::default(), + ); + + let ap = root["Hoodoo Hill"]["children"]["HoodooHill-Thunderhill"] + .as_object() + .expect("AP node should exist"); + assert_eq!( + super::node_bandwidth_mbps(ap, "downloadBandwidthMbps"), + Some(214) + ); + assert_eq!( + super::node_bandwidth_mbps(ap, "uploadBandwidthMbps"), + Some(773) + ); + } + fn write_runtime_json_fixture(path: PathBuf, value: &Value, label: &str) { let parent = path .parent() diff --git a/src/scheduler.py b/src/scheduler.py index 7798a15a..551b441c 100644 --- a/src/scheduler.py +++ b/src/scheduler.py @@ -14,8 +14,8 @@ from liblqos_python import automatic_import_uisp, automatic_import_splynx, queue blackboard_finish, blackboard_submit, automatic_import_wispgate, enable_insight_topology, insight_topology_role, \ automatic_import_netzur, automatic_import_visp, calculate_shaping_runtime_hash, efficiency_core_ids, scheduler_alive as _scheduler_alive_native, scheduler_error as _scheduler_error_native, \ calculate_topology_source_generation, topology_import_ingress_enabled, \ - scheduler_progress as _scheduler_progress_native, overrides_persistent_devices_materialized, overrides_circuit_adjustments_materialized, \ - overrides_network_adjustments_materialized, \ + scheduler_progress as _scheduler_progress_native, overrides_network_adjustments_materialized, \ + overrides_materialized, \ scheduler_output as _scheduler_output_native, wait_for_bus_ready from apscheduler.schedulers.background import BlockingScheduler @@ -37,6 +37,13 @@ STARTUP_TOPOLOGY_RUNTIME_MAX_WAIT_SECONDS = 120 PARTIAL_TOPOLOGY_RUNTIME_MAX_WAIT_SECONDS = 120 SCHEDULER_STARTUP_STEP_COUNT = 5 SCHEDULER_REFRESH_STEP_COUNT = 4 +OVERRIDE_SECTION_READ_ATTEMPTS = 5 +OVERRIDE_SECTION_READ_RETRY_SECONDS = 0.25 +OVERRIDE_LOCK_ERROR_MARKERS = ( + "lqos_overrides_locked", + "overrides files are locked by another process", + "locked by another process", +) scheduler_status_bus_enabled = True startup_topology_runtime_pending = False startup_topology_runtime_generation = None @@ -49,6 +56,10 @@ partial_topology_runtime_last_report_state = None last_integration_failure_message = "" +class RequiredOverrideReadError(RuntimeError): + """Raised when scheduler reloads cannot safely materialize operator overrides.""" + + def configure_scheduler_stdio(): """ Enable line-buffered scheduler logs when running under systemd. @@ -493,8 +504,19 @@ def importFromCRM( ) apply_lqos_overrides() except Exception as e: - scheduler_error(f"Failed to apply lqos_overrides: {e}") - print(f"Failed to apply lqos_overrides: {e}") + message = f"Failed to apply lqos_overrides; preserving last-known-good topology: {e}" + scheduler_error(message) + scheduler_output(message) + publish_scheduler_progress( + False, + "degraded", + "Scheduler blocked by override failure", + overrides_step, + progress_step_count, + percent=100, + ) + print(message) + raise # --------------- Overrides Handling --------------- @@ -704,23 +726,23 @@ def operator_requires_sqm_column(devices, adjustments): def apply_lqos_overrides(): """Apply overrides to source topology files without mutating integration-owned shaping CSV.""" + materialized = read_required_override_section( + "materialized overrides", + overrides_materialized, + ) + persistent_devices = materialized.get("persistent_devices", []) + circuit_adjustments = materialized.get("circuit_adjustments", []) + network_adjustments = materialized.get("network_adjustments", []) + if not topology_import_ingress_enabled(): path = shaped_devices_csv_path() header, rows = read_shaped_devices_csv(path) # 1) Persistent devices: replace by device_id or append - try: - extra = overrides_persistent_devices_materialized() - except Exception as e: - print(f"Skipping persistent device overrides: {e}") - extra = [] + extra = persistent_devices # 2) Circuit adjustments: speed changes, removals, reparenting - try: - adjustments = overrides_circuit_adjustments_materialized() - except Exception as e: - print(f"Failed to read circuit adjustments: {e}") - adjustments = [] + adjustments = circuit_adjustments need_sqm_column = header_has_sqm(header) or operator_requires_sqm_column(extra, adjustments) if need_sqm_column and not header_has_sqm(header): @@ -806,11 +828,7 @@ def apply_lqos_overrides(): print("Updated ShapedDevices.csv with overrides") # 3) Load, adjust, and optionally save topology compatibility state - try: - adjustments = overrides_network_adjustments_materialized() - except Exception as e: - print(f"Failed to read network adjustments: {e}") - adjustments = [] + adjustments = network_adjustments canonical_path = topology_canonical_state_path() canonical_state = load_topology_canonical_state(canonical_path) if topology_import_ingress_enabled(): @@ -870,6 +888,35 @@ def write_topology_canonical_state(path: str, canonical_state: dict): f.write(json.dumps(canonical_state, indent=4)) +def _format_attempt_count(attempts: int) -> str: + if attempts == 1: + return "1 attempt" + return f"{attempts} attempts" + + +def _is_override_lock_error(exc: Exception) -> bool: + message = str(exc or "").lower() + return any(marker in message for marker in OVERRIDE_LOCK_ERROR_MARKERS) + + +def read_required_override_section(section_name: str, reader): + attempts = max(1, int(OVERRIDE_SECTION_READ_ATTEMPTS)) + last_error = None + for attempt_index in range(attempts): + try: + return reader() + except Exception as exc: + last_error = exc + if not _is_override_lock_error(exc): + raise + if attempt_index + 1 < attempts: + time.sleep(OVERRIDE_SECTION_READ_RETRY_SECONDS) + + raise RequiredOverrideReadError( + f"failed to read {section_name} after {_format_attempt_count(attempts)}: {last_error}" + ) from last_error + + def apply_network_adjustments(network: dict, adjustments=None) -> bool: """Apply network adjustments from overrides to the network JSON structure. @@ -886,11 +933,10 @@ def apply_network_adjustments(network: dict, adjustments=None) -> bool: Returns True if any changes were applied. """ if adjustments is None: - try: - adjustments = overrides_network_adjustments_materialized() - except Exception as e: - print(f"Failed to read network adjustments: {e}") - return False + adjustments = read_required_override_section( + "network adjustments", + overrides_network_adjustments_materialized, + ) if not adjustments: return False diff --git a/src/test_scheduler.py b/src/test_scheduler.py index 371fbb3f..31146d82 100644 --- a/src/test_scheduler.py +++ b/src/test_scheduler.py @@ -59,11 +59,14 @@ def install_scheduler_stubs(): lqlib.scheduler_progress = Mock() lqlib.wait_for_bus_ready = Mock(return_value=True) lqlib.overrides_persistent_devices_effective = lambda: [] - lqlib.overrides_persistent_devices_materialized = lambda: [] lqlib.overrides_circuit_adjustments_effective = lambda: [] - lqlib.overrides_circuit_adjustments_materialized = lambda: [] lqlib.overrides_network_adjustments_effective = lambda: [] lqlib.overrides_network_adjustments_materialized = lambda: [] + lqlib.overrides_materialized = lambda: { + "persistent_devices": [], + "circuit_adjustments": [], + "network_adjustments": [], + } sys.modules["liblqos_python"] = lqlib apscheduler_pkg = types.ModuleType("apscheduler") @@ -1235,6 +1238,165 @@ class TestSchedulerOverrideMerge(unittest.TestCase): self.assertEqual(len(merged), 2) self.assertEqual(merged[1][2], "legacy_device_1") + def test_required_override_section_retries_transient_lock_failure(self): + reader = Mock(side_effect=[ + RuntimeError("locked by another process"), + RuntimeError("still locked by another process"), + [{"type": "adjust_site_speed"}], + ]) + + with patch.object(scheduler, "OVERRIDE_SECTION_READ_ATTEMPTS", 5): + with patch.object(scheduler, "OVERRIDE_SECTION_READ_RETRY_SECONDS", 0.01): + with patch.object(scheduler.time, "sleep") as mock_sleep: + adjustments = scheduler.read_required_override_section( + "network adjustments", + reader, + ) + + self.assertEqual(adjustments, [{"type": "adjust_site_speed"}]) + self.assertEqual(reader.call_count, 3) + self.assertEqual(mock_sleep.call_count, 2) + + def test_required_override_section_raises_after_retry_budget(self): + reader = Mock(side_effect=RuntimeError("locked by another process: pid 42")) + + with patch.object(scheduler, "OVERRIDE_SECTION_READ_ATTEMPTS", 2): + with patch.object(scheduler.time, "sleep"): + with self.assertRaises(scheduler.RequiredOverrideReadError) as raised: + scheduler.read_required_override_section( + "network adjustments", + reader, + ) + + self.assertEqual(reader.call_count, 2) + self.assertIn("failed to read network adjustments after 2 attempts", str(raised.exception)) + self.assertIn("locked by another process: pid 42", str(raised.exception)) + + def test_required_override_section_does_not_retry_non_lock_failure(self): + reader = Mock(side_effect=RuntimeError("invalid overrides json")) + + with patch.object(scheduler, "OVERRIDE_SECTION_READ_ATTEMPTS", 5): + with patch.object(scheduler.time, "sleep") as mock_sleep: + with self.assertRaises(RuntimeError) as raised: + scheduler.read_required_override_section( + "network adjustments", + reader, + ) + + self.assertEqual(reader.call_count, 1) + mock_sleep.assert_not_called() + self.assertIn("invalid overrides json", str(raised.exception)) + + def test_apply_lqos_overrides_aborts_when_network_adjustments_are_unavailable(self): + with patch.object(scheduler, "topology_import_ingress_enabled", return_value=True): + with patch.object( + scheduler, + "overrides_materialized", + side_effect=RuntimeError("locked by another process: pid 42"), + ): + with patch.object(scheduler, "OVERRIDE_SECTION_READ_ATTEMPTS", 2): + with patch.object(scheduler.time, "sleep"): + with patch.object(scheduler, "load_topology_canonical_state") as mock_load_canonical: + with patch.object(scheduler, "write_topology_canonical_state") as mock_write_canonical: + with self.assertRaises(scheduler.RequiredOverrideReadError): + scheduler.apply_lqos_overrides() + + mock_load_canonical.assert_not_called() + mock_write_canonical.assert_not_called() + + def test_apply_lqos_overrides_reads_all_sections_before_writing_non_ingress_files(self): + with patch.object(scheduler, "topology_import_ingress_enabled", return_value=False): + with patch.object( + scheduler, + "overrides_materialized", + side_effect=RuntimeError("locked by another process: pid 42"), + ): + with patch.object(scheduler, "OVERRIDE_SECTION_READ_ATTEMPTS", 2): + with patch.object(scheduler.time, "sleep"): + with patch.object(scheduler, "read_shaped_devices_csv") as mock_read_sd: + with patch.object(scheduler, "write_shaped_devices_csv") as mock_write_sd: + with patch.object(scheduler, "load_network_json") as mock_load_network: + with patch.object(scheduler, "write_network_json") as mock_write_network: + with self.assertRaises(scheduler.RequiredOverrideReadError): + scheduler.apply_lqos_overrides() + + mock_read_sd.assert_not_called() + mock_write_sd.assert_not_called() + mock_load_network.assert_not_called() + mock_write_network.assert_not_called() + + def test_apply_lqos_overrides_uses_single_materialized_snapshot(self): + header = [ + "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", + ] + rows = [] + materialized = { + "persistent_devices": [], + "circuit_adjustments": [], + "network_adjustments": [], + } + + with patch.object(scheduler, "topology_import_ingress_enabled", return_value=False): + with patch.object(scheduler, "overrides_materialized", return_value=materialized) as mock_materialized: + with patch.object( + scheduler, + "overrides_persistent_devices_materialized", + side_effect=AssertionError("legacy persistent reader should not be used"), + create=True, + ) as mock_persistent: + with patch.object( + scheduler, + "overrides_circuit_adjustments_materialized", + side_effect=AssertionError("legacy circuit reader should not be used"), + create=True, + ) as mock_circuit: + with patch.object( + scheduler, + "overrides_network_adjustments_materialized", + side_effect=AssertionError("legacy network reader should not be used"), + ) as mock_network: + with patch.object(scheduler, "read_shaped_devices_csv", return_value=(header, rows)): + with patch.object(scheduler, "load_network_json", return_value={}): + with patch.object(scheduler, "load_topology_canonical_state", return_value=None): + scheduler.apply_lqos_overrides() + + mock_materialized.assert_called_once_with() + mock_persistent.assert_not_called() + mock_circuit.assert_not_called() + mock_network.assert_not_called() + + def test_import_from_crm_does_not_continue_after_override_failure(self): + with patch.object( + scheduler, + "apply_lqos_overrides", + side_effect=scheduler.RequiredOverrideReadError("locked by pid 42"), + ): + with patch.object(scheduler.os.path, "isfile", return_value=False): + with patch.object(scheduler, "scheduler_error") as mock_scheduler_error: + with patch.object(scheduler, "scheduler_output") as mock_scheduler_output: + with patch.object(scheduler, "publish_scheduler_progress") as mock_progress: + with patch("builtins.print"): + with self.assertRaises(scheduler.RequiredOverrideReadError): + scheduler.importFromCRM() + + self.assertIn( + "preserving last-known-good topology", + mock_scheduler_error.call_args.args[0], + ) + self.assertEqual(mock_scheduler_output.call_args.args, mock_scheduler_error.call_args.args) + self.assertTrue( + any( + call.args[:3] == ( + False, + "degraded", + "Scheduler blocked by override failure", + ) + for call in mock_progress.call_args_list + ) + ) + def test_apply_lqos_overrides_device_adjust_sqm_only_updates_sqm_column(self): header = [ "Circuit ID", "Circuit Name", "Device ID", "Device Name", "Parent Node", "MAC", @@ -1249,18 +1411,21 @@ class TestSchedulerOverrideMerge(unittest.TestCase): # Test-only fake csv path. with patch.object(scheduler, "shaped_devices_csv_path", return_value="/tmp/ShapedDevices.csv"): # nosec B108 with patch.object(scheduler, "read_shaped_devices_csv", return_value=(header, rows)): - with patch.object(scheduler, "overrides_persistent_devices_materialized", return_value=[]): - with patch.object( - scheduler, - "overrides_circuit_adjustments_materialized", - return_value=[{ + with patch.object( + scheduler, + "overrides_materialized", + return_value={ + "persistent_devices": [], + "circuit_adjustments": [{ "type": "device_adjust_sqm", "device_id": "splynx_service_93", "sqm_override": "fq_codel/fq_codel", }], - ): - with patch.object(scheduler, "write_shaped_devices_csv") as mock_write: - scheduler.apply_lqos_overrides() + "network_adjustments": [], + }, + ): + with patch.object(scheduler, "write_shaped_devices_csv") as mock_write: + scheduler.apply_lqos_overrides() written_rows = mock_write.call_args.args[2] self.assertEqual(written_rows[0][10], "330") @@ -1281,18 +1446,21 @@ class TestSchedulerOverrideMerge(unittest.TestCase): with patch.object(scheduler, "shaped_devices_csv_path", return_value="/tmp/ShapedDevices.csv"): # nosec B108 with patch.object(scheduler, "read_shaped_devices_csv", return_value=(header, rows)): - with patch.object(scheduler, "overrides_persistent_devices_materialized", return_value=[]): - with patch.object( - scheduler, - "overrides_circuit_adjustments_materialized", - return_value=[{ + with patch.object( + scheduler, + "overrides_materialized", + return_value={ + "persistent_devices": [], + "circuit_adjustments": [{ "type": "reparent_circuit", "circuit_id": "93", "parent_node": "AP-Updated", }], - ): - with patch.object(scheduler, "write_shaped_devices_csv") as mock_write: - scheduler.apply_lqos_overrides() + "network_adjustments": [], + }, + ): + with patch.object(scheduler, "write_shaped_devices_csv") as mock_write: + scheduler.apply_lqos_overrides() written_rows = mock_write.call_args.args[2] self.assertEqual(written_rows[0][4], "AP-Updated") @@ -1330,26 +1498,28 @@ class TestSchedulerOverrideMerge(unittest.TestCase): with patch.object(scheduler, "shaped_devices_csv_path", return_value="/tmp/ShapedDevices.csv"): # nosec B108 with patch.object(scheduler, "read_shaped_devices_csv", return_value=(header, rows)): - with patch.object(scheduler, "overrides_persistent_devices_materialized", return_value=[]): - with patch.object(scheduler, "overrides_circuit_adjustments_materialized", return_value=[]): - with patch.object( - scheduler, - "overrides_network_adjustments_materialized", - return_value=[{ + with patch.object( + scheduler, + "overrides_materialized", + return_value={ + "persistent_devices": [], + "circuit_adjustments": [], + "network_adjustments": [{ "type": "adjust_site_speed", "node_id": "node-b", "site_name": "NodeB", "download_bandwidth_mbps": 80, "upload_bandwidth_mbps": 40, }], - ): - with patch.object(scheduler, "topology_import_ingress_enabled", return_value=True): - with patch.object(scheduler, "load_topology_canonical_state", return_value=canonical_state): - with patch.object(scheduler, "write_topology_canonical_state") as mock_write_canonical: - with patch.object(scheduler, "load_network_json") as mock_load_network: - with patch.object(scheduler, "write_network_json") as mock_write_network: - with patch.object(scheduler, "write_shaped_devices_csv") as mock_write_sd: - scheduler.apply_lqos_overrides() + }, + ): + with patch.object(scheduler, "topology_import_ingress_enabled", return_value=True): + with patch.object(scheduler, "load_topology_canonical_state", return_value=canonical_state): + with patch.object(scheduler, "write_topology_canonical_state") as mock_write_canonical: + with patch.object(scheduler, "load_network_json") as mock_load_network: + with patch.object(scheduler, "write_network_json") as mock_write_network: + with patch.object(scheduler, "write_shaped_devices_csv") as mock_write_sd: + scheduler.apply_lqos_overrides() mock_load_network.assert_not_called() mock_write_network.assert_not_called()