diff --git a/docs-es/v2.0/libreqos-backend-architecture-es.md b/docs-es/v2.0/libreqos-backend-architecture-es.md index 741bceff..1ca79e39 100644 --- a/docs-es/v2.0/libreqos-backend-architecture-es.md +++ b/docs-es/v2.0/libreqos-backend-architecture-es.md @@ -270,14 +270,15 @@ flowchart TD 3. Esperar mayor riesgo cuando coinciden muchos circuitos y muchos cambios estructurales. 4. Diseñar la cadencia operativa para priorizar cambios incrementalmente seguros. -### 7.4 Guardrails de seguridad para recargas completas +### 7.4 Salvaguardas de seguridad para recargas completas -Las recargas completas actuales de Bakery aplican dos verificaciones conservadoras de seguridad antes y durante reconstrucciones grandes de colas: +Las recargas completas actuales de Bakery aplican verificaciones conservadoras de seguridad antes y durante reconstrucciones grandes de colas: 1. Un preflight de qdisc estima los qdisc planificados por interfaz y además separa qdisc de infraestructura, hojas `cake` y hojas `fq_codel`. 2. Ese mismo preflight aplica una proyección conservadora de memoria y bloquea de forma estricta las recargas completas claramente inseguras antes de arrancar `tc -batch`. -3. Durante la aplicación chunked de una recarga completa, Bakery vuelve a revisar la memoria del host en los límites de chunk y aborta el resto de la aplicación si la memoria disponible cae por debajo de su piso de seguridad. -4. Estos guardrails están sesgados intencionalmente hacia falsos positivos en recargas grandes para fallar temprano con diagnósticos en lugar de entrar en una espiral OOM. +3. El piso de memoria de Bakery escala con la RAM del host: mantiene al menos 2 GiB disponibles, o un octavo de la RAM total en sistemas más grandes. +4. Durante la aplicación por fragmentos de una recarga completa, Bakery vuelve a revisar la memoria del host en los límites de cada fragmento y aborta el resto de la aplicación si la memoria disponible cae por debajo del piso escalado más la memoria qdisc proyectada para el lote. +5. Estas salvaguardas están sesgadas intencionalmente hacia falsos positivos en recargas grandes para fallar temprano con diagnósticos en lugar de entrar en una espiral OOM. ## 8) Límites de Diseño para Operadores diff --git a/docs-es/v2.0/troubleshooting-es.md b/docs-es/v2.0/troubleshooting-es.md index c9883eec..c5daff47 100644 --- a/docs-es/v2.0/troubleshooting-es.md +++ b/docs-es/v2.0/troubleshooting-es.md @@ -253,7 +253,7 @@ WebUI muestra códigos legibles por máquina para triage rápido. | `MAPPED_CIRCUIT_LIMIT` | Bakery está forzando límite de circuitos mapeados. | Estado de licencia Insight y `journalctl -u lqosd` con requested/allowed/dropped. | Reducir circuitos mapeados o actualizar licencia/límites. | | `TC_U16_OVERFLOW` | IDs minor de clases/colas excedieron rango u16 de tc en una cola CPU. | `journalctl -u lqos_scheduler -u lqosd`, profundidad topológica y distribución por colas. | Aumentar paralelismo de colas y/o simplificar/rebalancear jerarquía. | | `TC_QDISC_CAPACITY` | Los qdisc autoasignados planificados exceden el presupuesto seguro por interfaz o el preflight conservador de seguridad de memoria de Bakery antes de aplicar. | Conteos estimados por interfaz, desglose por tipo de qdisc y campos de memoria en el contexto del urgent issue, `journalctl -u lqos_scheduler -u lqosd`, configuración `on_a_stick` y `monitor_only`. | Reducir la carga planificada de qdisc para esta ejecución (por ejemplo menos circuitos/dispositivos en la forma de prueba) antes de reintentar; no confiar en una aplicación parcial. | -| `BAKERY_MEMORY_GUARD` | Un full reload chunked de Bakery fue detenido a mitad de aplicación porque la memoria disponible del host cayó por debajo del piso de seguridad. | `journalctl -u lqosd`, memoria disponible/total en el contexto del urgent issue y progreso reciente de aplicación de Bakery. | Tratar la ejecución como fallida, reducir presión de memoria o huella de colas y reintentar solo cuando el host esté estable. | +| `BAKERY_MEMORY_GUARD` | Una recarga completa por fragmentos de Bakery fue detenida a mitad de aplicación porque la memoria disponible del host cayó por debajo del piso de seguridad escalado. | `journalctl -u lqosd`, memoria disponible/total en el contexto del urgent issue y progreso reciente de aplicación de Bakery. | Tratar la ejecución como fallida, reducir presión de memoria o huella de colas y reintentar solo cuando el host esté estable. | | `XDP_IP_MAPPING_CAPACITY` | Los mapeos IP requeridos exceden la capacidad actual del mapa XDP en el kernel. | Forma de `ShapedDevices.csv`, mezcla IPv4/IPv6, supuesto de un dispositivo frente a varios dispositivos, `journalctl -u lqos_scheduler -u lqosd`. | Reducir mapeos requeridos de inmediato (por ejemplo menos dispositivos o prueba IPv4-only), o aumentar la capacidad del mapa del kernel en un cambio coordinado. | | `XDP_IP_MAPPING_APPLY_FAILED` | Uno o más inserts de mapeo IP fallaron durante la aplicación. | `journalctl -u lqos_scheduler -u lqosd` para ejemplos resumidos y conteos de fallo. | Corregir la causa del fallo y volver a ejecutar; no confiar en shaping parcial. | diff --git a/docs/v2.0/libreqos-backend-architecture.md b/docs/v2.0/libreqos-backend-architecture.md index 13e58fbb..195317b7 100644 --- a/docs/v2.0/libreqos-backend-architecture.md +++ b/docs/v2.0/libreqos-backend-architecture.md @@ -343,12 +343,13 @@ This retained-root strategy reduces avoidable root-level churn, while still pref ### 8.6 Full-reload safety guards -Current Bakery full reloads apply two conservative safety checks before and during large queue rebuilds: +Current Bakery full reloads apply conservative safety checks before and during large queue rebuilds: 1. A qdisc preflight estimates planned qdiscs per interface and also separates infrastructure, `cake`, and `fq_codel` leaf qdiscs. 2. That same preflight applies a conservative memory forecast and hard-blocks clearly unsafe full reloads before `tc -batch` starts. -3. During chunked full reload apply, Bakery re-checks host memory at chunk boundaries and aborts the remaining apply if available memory drops below its safety floor. -4. These guards are intentionally biased toward false positives on large reloads so the system fails early with diagnostics instead of spiraling into an OOM event. +3. Bakery's memory floor scales with host RAM: it keeps at least 2 GiB available, or one eighth of total RAM on larger systems. +4. During chunked full reload apply, Bakery re-checks host memory at chunk boundaries and aborts the remaining apply if available memory drops below the scaled floor plus the projected qdisc memory for the batch. +5. These guards are intentionally biased toward false positives on large reloads so the system fails early with diagnostics instead of spiraling into an OOM event. ### 8.7 Runtime safety model diff --git a/docs/v2.0/troubleshooting.md b/docs/v2.0/troubleshooting.md index e4da9fc3..058d56a2 100644 --- a/docs/v2.0/troubleshooting.md +++ b/docs/v2.0/troubleshooting.md @@ -330,7 +330,7 @@ WebUI urgent issues include machine-readable codes. Use them to triage quickly. | `MAPPED_CIRCUIT_LIMIT` | Bakery is enforcing a mapped-circuit limit. | Insight license status, `journalctl -u lqosd` for requested/allowed/dropped counts. | Reduce mapped circuits immediately or update license/limits. | | `TC_U16_OVERFLOW` | Queue/class minor IDs exceeded the Linux tc u16 range on a CPU queue. | `journalctl -u lqos_scheduler -u lqosd`, topology depth/queue distribution. | Increase queue count and/or simplify/rebalance hierarchy (for example with integration strategy or root promotion changes). | | `TC_QDISC_CAPACITY` | Planned auto-allocated qdiscs exceed the per-interface safe budget or Bakery's conservative memory-safety preflight before apply. | Estimated per-interface qdisc counts, qdisc-kind breakdown, and memory fields in the urgent issue context, `journalctl -u lqos_scheduler -u lqosd`, `on_a_stick` and `queue_mode` config. | Reduce the planned qdisc load for this run (for example fewer circuits/devices in the test shape) before retrying; do not trust partial apply. | -| `BAKERY_MEMORY_GUARD` | A chunked Bakery full reload was stopped mid-apply because available host memory fell below the safety floor. | `journalctl -u lqosd`, available/total memory in the urgent issue context, and recent Bakery apply progress. | Treat the run as failed, reduce memory pressure or queue footprint, and retry only after the host is stable. | +| `BAKERY_MEMORY_GUARD` | A chunked Bakery full reload was stopped mid-apply because available host memory fell below the scaled safety floor. | `journalctl -u lqosd`, available/total memory in the urgent issue context, and recent Bakery apply progress. | Treat the run as failed, reduce memory pressure or queue footprint, and retry only after the host is stable. | | `XDP_IP_MAPPING_CAPACITY` | Required IP mappings exceed the current XDP kernel map capacity. | `ShapedDevices.csv` row shape, IPv4/IPv6 mix, one-device-vs-many-device assumptions, `journalctl -u lqos_scheduler -u lqosd`. | Reduce required mappings immediately (for example fewer devices or IPv4-only test shape), or raise kernel map capacity in a coordinated change. | | `XDP_IP_MAPPING_APPLY_FAILED` | One or more IP mapping inserts failed during apply. | `journalctl -u lqos_scheduler -u lqosd` for summarized failure examples and counts. | Fix the underlying mapping failure, then rerun; do not trust partial shaping. | diff --git a/src/rust/lqos_bakery/src/lib.rs b/src/rust/lqos_bakery/src/lib.rs index 9264b986..7d4c34f1 100644 --- a/src/rust/lqos_bakery/src/lib.rs +++ b/src/rust/lqos_bakery/src/lib.rs @@ -1513,7 +1513,14 @@ pub const FQ_CODEL_QDISC_ESTIMATED_MEMORY_BYTES: u64 = 64 * 1024; /// substantially exceeding the earlier heuristic during busy periods. pub const CAKE_QDISC_ESTIMATED_MEMORY_BYTES: u64 = 512 * 1024; /// Minimum memory headroom Bakery tries to leave unused after a projected or in-flight apply. -pub const BAKERY_MEMORY_GUARD_MIN_AVAILABLE_BYTES: u64 = 768 * 1024 * 1024; +pub const BAKERY_MEMORY_GUARD_MIN_AVAILABLE_BYTES: u64 = 2 * 1024 * 1024 * 1024; +/// Divisor used to scale Bakery's memory guard floor with installed RAM. +pub const BAKERY_MEMORY_GUARD_TOTAL_RAM_DIVISOR: u64 = 8; + +/// Returns Bakery's effective memory guard floor for a host with `total_bytes` RAM. +pub fn bakery_memory_guard_min_available_bytes(total_bytes: u64) -> u64 { + BAKERY_MEMORY_GUARD_MIN_AVAILABLE_BYTES.max(total_bytes / BAKERY_MEMORY_GUARD_TOTAL_RAM_DIVISOR) +} /// Maximum number of mapped circuits allowed without Insight. const DEFAULT_MAPPED_CIRCUITS_LIMIT: usize = 1000; /// Minimum interval between repeated mapped-circuit-limit urgent issues. @@ -2055,22 +2062,69 @@ fn telemetry_state() -> &'static RwLock { BAKERY_TELEMETRY.get_or_init(|| RwLock::new(BakeryTelemetryState::default())) } +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum TcCommandType { + Class, + Qdisc, +} + +fn tc_command_type(argv: &[String]) -> Option { + match argv.first()?.as_str() { + "class" => Some(TcCommandType::Class), + "qdisc" => Some(TcCommandType::Qdisc), + _ => None, + } +} + fn count_tc_command_types(commands: &[Vec]) -> (usize, usize, usize) { let total = commands.len(); let mut class_count = 0usize; let mut qdisc_count = 0usize; for argv in commands { - if let Some(kind) = argv.first() { - match kind.as_str() { - "class" => class_count += 1, - "qdisc" => qdisc_count += 1, - _ => {} - } + match tc_command_type(argv) { + Some(TcCommandType::Class) => class_count += 1, + Some(TcCommandType::Qdisc) => qdisc_count += 1, + None => {} } } (total, class_count, qdisc_count) } +fn count_tc_command_types_and_estimate_qdisc_memory( + commands: &[Vec], +) -> (usize, usize, usize, u64) { + let total = commands.len(); + let mut class_count = 0usize; + let mut qdisc_count = 0usize; + let mut estimated_qdisc_memory_bytes = 0u64; + let mut seen_qdiscs = HashSet::new(); + + for argv in commands { + match tc_command_type(argv) { + Some(TcCommandType::Class) => class_count += 1, + Some(TcCommandType::Qdisc) => qdisc_count += 1, + None => {} + } + + let Some(qdisc_identity) = planned_qdisc_identity(argv) else { + continue; + }; + if !seen_qdiscs.insert(qdisc_identity) { + continue; + } + let kind = planned_qdisc_kind(argv).unwrap_or(PlannedQdiscKind::Infra); + estimated_qdisc_memory_bytes = + estimated_qdisc_memory_bytes.saturating_add(qdisc_kind_estimated_memory_bytes(kind)); + } + + ( + total, + class_count, + qdisc_count, + estimated_qdisc_memory_bytes, + ) +} + fn push_bakery_event(event: &str, status: &str, summary: String) { push_bakery_event_with_site(event, status, None, summary); } @@ -2505,6 +2559,8 @@ pub struct QdiscBudgetEstimate { pub estimated_total_memory_bytes: u64, /// Current host memory snapshot used for preflight, if available. pub memory_snapshot: Option, + /// Effective memory floor used for the memory preflight. + pub memory_guard_min_available_bytes: u64, /// Whether the memory preflight passed. pub memory_ok: bool, } @@ -2649,10 +2705,14 @@ pub fn estimate_full_reload_auto_qdisc_budget( acc.saturating_add(detail.estimated_memory_bytes) }); let memory_snapshot = read_memory_snapshot().ok(); + let memory_guard_min_available_bytes = memory_snapshot + .as_ref() + .map(|snapshot| bakery_memory_guard_min_available_bytes(snapshot.total_bytes)) + .unwrap_or(BAKERY_MEMORY_GUARD_MIN_AVAILABLE_BYTES); let memory_ok = memory_snapshot.as_ref().is_none_or(|snapshot| { snapshot .available_bytes - .saturating_sub(BAKERY_MEMORY_GUARD_MIN_AVAILABLE_BYTES) + .saturating_sub(memory_guard_min_available_bytes) >= estimated_total_memory_bytes }); @@ -2663,6 +2723,7 @@ pub fn estimate_full_reload_auto_qdisc_budget( hard_limit: HARD_QDISC_HANDLE_LIMIT_PER_INTERFACE, estimated_total_memory_bytes, memory_snapshot, + memory_guard_min_available_bytes, memory_ok, } } @@ -3889,16 +3950,26 @@ fn maybe_emit_mapped_circuit_limit_urgent(stats: &MappedLimitStats) { }); } -fn maybe_emit_memory_guard_urgent(summary: &str) { +fn maybe_emit_memory_guard_urgent( + summary: &str, + memory_guard_required_available_bytes: u64, + estimated_qdisc_memory_bytes: u64, +) { if !summary.contains("Bakery memory guard stopped") { return; } let message = summary.to_string(); let context = read_memory_snapshot().ok().map(|snapshot| { + let memory_guard_floor_bytes = + bakery_memory_guard_min_available_bytes(snapshot.total_bytes); format!( - "{{\"available_bytes\":{},\"total_bytes\":{},\"memory_guard_floor_bytes\":{}}}", - snapshot.available_bytes, snapshot.total_bytes, BAKERY_MEMORY_GUARD_MIN_AVAILABLE_BYTES + "{{\"available_bytes\":{},\"total_bytes\":{},\"memory_guard_floor_bytes\":{},\"estimated_qdisc_memory_bytes\":{},\"memory_guard_required_available_bytes\":{}}}", + snapshot.available_bytes, + snapshot.total_bytes, + memory_guard_floor_bytes, + estimated_qdisc_memory_bytes, + memory_guard_required_available_bytes ) }); @@ -9059,6 +9130,34 @@ fn site_commands_observed_live( }) } +fn read_optional_restore_class_snapshots( + config: &Arc, + site_label: &str, +) -> Option<( + HashMap, + HashMap, +)> { + let down_snapshot = match read_live_class_snapshot(&config.isp_interface()) { + Ok(snapshot) => snapshot, + Err(summary) => { + debug!( + "Bakery: skipping live restore shortcut for {site_label}; unable to snapshot downlink classes: {summary}" + ); + return None; + } + }; + let up_snapshot = match read_live_class_snapshot(&config.internet_interface()) { + Ok(snapshot) => snapshot, + Err(summary) => { + debug!( + "Bakery: skipping live restore shortcut for {site_label}; unable to snapshot uplink classes: {summary}" + ); + return None; + } + }; + Some((down_snapshot, up_snapshot)) +} + fn live_parent_matches(observed: Option, expected: TcHandle) -> bool { if observed == Some(expected) { return true; @@ -10593,28 +10692,27 @@ fn handle_treeguard_set_node_virtual_live( } let reversible_standby = saved_state.active_branch_hides_original_site(); - let live_restore_snapshots = if reversible_standby { - Some(( - read_live_class_snapshot(&config.isp_interface())?, - read_live_class_snapshot(&config.internet_interface())?, - )) - } else { - None - }; + let mut live_restore_snapshots = None; if !saved_state.pending_prune && let Some(cmds) = saved_state .site .to_commands(&config, ExecutionMode::Builder) { - let root_already_live = + let root_already_live = if reversible_standby { live_restore_snapshots + .get_or_insert_with(|| { + read_optional_restore_class_snapshots(&config, &site_label) + }) .as_ref() .is_some_and(|(down_snapshot, up_snapshot)| { let mut only_root = HashMap::new(); only_root.insert(site_hash, saved_state.site.clone()); site_commands_observed_live(&only_root, down_snapshot, up_snapshot) - }); + }) + } else { + false + }; if !root_already_live { let result = execute_and_record_live_change(&cmds, "TreeGuard runtime hidden site restore"); @@ -10643,8 +10741,9 @@ fn handle_treeguard_set_node_virtual_live( ) }) .collect(); - let originals_already_live = + let originals_already_live = if reversible_standby && !saved_state.saved_sites.is_empty() { live_restore_snapshots + .get_or_insert_with(|| read_optional_restore_class_snapshots(&config, &site_label)) .as_ref() .is_some_and(|(down_snapshot, up_snapshot)| { site_commands_observed_live( @@ -10652,7 +10751,10 @@ fn handle_treeguard_set_node_virtual_live( down_snapshot, up_snapshot, ) - }); + }) + } else { + false + }; if reversible_standby && originals_already_live { for (hash, command) in &saved_state.saved_sites { sites.insert(*hash, Arc::clone(command)); @@ -11037,7 +11139,12 @@ fn process_batch( let path = config.debug_state_file_path("linux_tc_rust.txt"); write_command_file(&path, &commands); let build_duration_ms = build_started.elapsed().as_millis() as u64; - let (total_tc_commands, class_commands, qdisc_commands) = count_tc_command_types(&commands); + let (total_tc_commands, class_commands, qdisc_commands, estimated_qdisc_memory_bytes) = + count_tc_command_types_and_estimate_qdisc_memory(&commands); + let memory_guard_required_available_bytes = read_memory_snapshot() + .map(|snapshot| bakery_memory_guard_min_available_bytes(snapshot.total_bytes)) + .unwrap_or(BAKERY_MEMORY_GUARD_MIN_AVAILABLE_BYTES) + .saturating_add(estimated_qdisc_memory_bytes); let total_chunks = if total_tc_commands == 0 { 0 } else { @@ -11054,7 +11161,7 @@ fn process_batch( &commands, "processing batch", FULL_RELOAD_TC_CHUNK_SIZE, - Some(BAKERY_MEMORY_GUARD_MIN_AVAILABLE_BYTES), + Some(memory_guard_required_available_bytes), |completed_tc_commands, total_tc_commands, completed_chunks, total_chunks| { update_bakery_apply_progress( Some("Applying tc command chunks"), @@ -11066,7 +11173,11 @@ fn process_batch( }, ); let summary = summarize_apply_result("processing batch", &result); - maybe_emit_memory_guard_urgent(&summary); + maybe_emit_memory_guard_urgent( + &summary, + memory_guard_required_available_bytes, + estimated_qdisc_memory_bytes, + ); BakeryFullReloadBatchResult { result, summary, @@ -13363,6 +13474,65 @@ mod tests { assert!(estimate.ok()); } + #[test] + fn bakery_memory_guard_floor_scales_with_host_ram() { + assert_eq!( + bakery_memory_guard_min_available_bytes(8 * 1024 * 1024 * 1024), + BAKERY_MEMORY_GUARD_MIN_AVAILABLE_BYTES + ); + assert_eq!( + bakery_memory_guard_min_available_bytes(64 * 1024 * 1024 * 1024), + 8 * 1024 * 1024 * 1024 + ); + } + + #[test] + fn command_qdisc_memory_estimate_deduplicates_qdisc_identity() { + let commands = vec![ + vec![ + "qdisc".to_string(), + "add".to_string(), + "dev".to_string(), + "eth0".to_string(), + "parent".to_string(), + "1:10".to_string(), + "cake".to_string(), + ], + vec![ + "qdisc".to_string(), + "replace".to_string(), + "dev".to_string(), + "eth0".to_string(), + "parent".to_string(), + "1:10".to_string(), + "cake".to_string(), + ], + vec![ + "qdisc".to_string(), + "add".to_string(), + "dev".to_string(), + "eth1".to_string(), + "parent".to_string(), + "1:10".to_string(), + "cake".to_string(), + ], + vec![ + "qdisc".to_string(), + "add".to_string(), + "dev".to_string(), + "eth0".to_string(), + "parent".to_string(), + "1:11".to_string(), + "fq_codel".to_string(), + ], + ]; + + assert_eq!( + count_tc_command_types_and_estimate_qdisc_memory(&commands).3, + (CAKE_QDISC_ESTIMATED_MEMORY_BYTES * 2) + FQ_CODEL_QDISC_ESTIMATED_MEMORY_BYTES + ); + } + #[test] fn qdisc_budget_estimate_counts_only_root_mq_in_observe_mode() { let _guard = bakery_test_lock().lock().expect("lock"); diff --git a/src/rust/lqos_bakery/src/utils.rs b/src/rust/lqos_bakery/src/utils.rs index a0aa61d2..350fc435 100644 --- a/src/rust/lqos_bakery/src/utils.rs +++ b/src/rust/lqos_bakery/src/utils.rs @@ -269,7 +269,7 @@ fn memory_guard_failure_summary( phase: &str, ) -> String { format!( - "Bakery memory guard stopped {purpose} {phase} chunk {}/{}: available memory {} bytes is below safety floor {} bytes (total RAM {} bytes).", + "Bakery memory guard stopped {purpose} {phase} chunk {}/{}: available memory {} bytes is below required available memory {} bytes (total RAM {} bytes).", chunk_number, total_chunks, snapshot.available_bytes, diff --git a/src/rust/lqos_python/src/device_weights.rs b/src/rust/lqos_python/src/device_weights.rs index 770975ae..7d0076ae 100644 --- a/src/rust/lqos_python/src/device_weights.rs +++ b/src/rust/lqos_python/src/device_weights.rs @@ -16,7 +16,13 @@ use lqos_bus::{BusRequest, BusResponse}; use lqos_config::{ShapedDevice, load_config}; use pyo3::pyclass; use serde::{Deserialize, Serialize}; -use std::time::Duration; +use std::{ + sync::{Mutex, OnceLock}, + time::{Duration, Instant}, +}; + +const REMOTE_WEIGHT_SUCCESS_TTL: Duration = Duration::from_secs(300); +const REMOTE_WEIGHT_FAILURE_BACKOFF: Duration = Duration::from_secs(300); /// This struct is used to send a request to the Long Term Stats API #[derive(Serialize, Deserialize)] @@ -30,7 +36,7 @@ pub struct DeviceWeightRequest { /// This struct is used to receive a response from the Long Term Stats API /// It contains the circuit_id and the weight of the device -#[derive(Serialize, Deserialize, Debug)] +#[derive(Serialize, Deserialize, Debug, Clone)] #[pyclass] pub struct DeviceWeightResponse { #[pyo3(get)] @@ -39,6 +45,57 @@ pub struct DeviceWeightResponse { pub weight: i64, } +#[derive(Clone)] +struct CachedRemoteWeights { + fetched_at: Instant, + weights: Vec, +} + +#[derive(Default)] +struct RemoteWeightCache { + last_success: Option, + last_failure: Option, +} + +fn remote_weight_cache() -> &'static Mutex { + static CACHE: OnceLock> = OnceLock::new(); + CACHE.get_or_init(|| Mutex::new(RemoteWeightCache::default())) +} + +fn cached_remote_weights() -> Option> { + let cache = remote_weight_cache().lock().ok()?; + let cached = cache.last_success.as_ref()?; + if cached.fetched_at.elapsed() <= REMOTE_WEIGHT_SUCCESS_TTL { + return Some(cached.weights.clone()); + } + None +} + +fn recent_remote_weight_failure() -> bool { + let Ok(cache) = remote_weight_cache().lock() else { + return false; + }; + cache + .last_failure + .is_some_and(|failed_at| failed_at.elapsed() <= REMOTE_WEIGHT_FAILURE_BACKOFF) +} + +fn remember_remote_weight_success(weights: &[DeviceWeightResponse]) { + if let Ok(mut cache) = remote_weight_cache().lock() { + cache.last_success = Some(CachedRemoteWeights { + fetched_at: Instant::now(), + weights: weights.to_vec(), + }); + cache.last_failure = None; + } +} + +fn remember_remote_weight_failure() { + if let Ok(mut cache) = remote_weight_cache().lock() { + cache.last_failure = Some(Instant::now()); + } +} + /// This function is used to get the device weights from the Long Term Stats API fn get_weights_from_lts( org_key: &str, @@ -147,10 +204,28 @@ pub(crate) fn get_weights_rust() -> Result> { let duration_seconds = 60 * 60 * 24; // 1 day let percentile = 0.95; - eprintln!("Getting weights from Insight"); - let weights = get_weights_from_lts(&org_key, &node_id, start, duration_seconds, percentile); - if let Ok(weights) = weights { - eprintln!("Retrieved {} weights from Insight", weights.len()); + let remote_weights = if let Some(weights) = cached_remote_weights() { + eprintln!("Using cached Insight weights"); + Some(weights) + } else if recent_remote_weight_failure() { + eprintln!("Skipping Insight weight fetch; recent request failed"); + None + } else { + eprintln!("Getting weights from Insight"); + match get_weights_from_lts(&org_key, &node_id, start, duration_seconds, percentile) { + Ok(weights) => { + eprintln!("Retrieved {} weights from Insight", weights.len()); + remember_remote_weight_success(&weights); + Some(weights) + } + Err(err) => { + remember_remote_weight_failure(); + eprintln!("Failed to get weights from Insight: {:?}", err); + None + } + } + }; + if let Some(weights) = remote_weights { // Merge them for weight in weights.iter() { if let Some(existing) = shaped_devices_weights @@ -160,8 +235,6 @@ pub(crate) fn get_weights_rust() -> Result> { existing.weight = weight.weight; } } - } else { - eprintln!("Failed to get weights from Insight: {:?}", weights); } } diff --git a/src/rust/lqos_python/src/lib.rs b/src/rust/lqos_python/src/lib.rs index aafb385d..4c2c9484 100644 --- a/src/rust/lqos_python/src/lib.rs +++ b/src/rust/lqos_python/src/lib.rs @@ -3571,15 +3571,15 @@ impl Bakery { (false, false) => "exceeds qdisc-count preflight and failed memory preflight", }; let memory_summary = if let Some(snapshot) = estimate.memory_snapshot.as_ref() { - let memory_floor = lqos_bakery::BAKERY_MEMORY_GUARD_MIN_AVAILABLE_BYTES; - let required_available_bytes = - memory_floor.saturating_add(estimate.estimated_total_memory_bytes); + let required_available_bytes = estimate + .memory_guard_min_available_bytes + .saturating_add(estimate.estimated_total_memory_bytes); format!( "estimated qdisc memory {} bytes; available memory {} bytes; required minimum {} bytes (safety floor {} bytes plus estimate)", estimate.estimated_total_memory_bytes, snapshot.available_bytes, required_available_bytes, - memory_floor + estimate.memory_guard_min_available_bytes ) } else { format!( @@ -3625,7 +3625,7 @@ impl Bakery { .memory_snapshot .as_ref() .map(|snapshot| snapshot.available_bytes), - memory_guard_min_available_bytes: lqos_bakery::BAKERY_MEMORY_GUARD_MIN_AVAILABLE_BYTES, + memory_guard_min_available_bytes: estimate.memory_guard_min_available_bytes, memory_ok: estimate.memory_ok, interfaces: interface_reports, }]); @@ -3658,7 +3658,7 @@ impl Bakery { result.set_item("memory_ok", estimate.memory_ok)?; result.set_item( "memory_guard_min_available_bytes", - lqos_bakery::BAKERY_MEMORY_GUARD_MIN_AVAILABLE_BYTES, + estimate.memory_guard_min_available_bytes, )?; if let Some(snapshot) = estimate.memory_snapshot { result.set_item("memory_total_bytes", snapshot.total_bytes)?; diff --git a/src/rust/lqos_queue_tracker/src/queue_diff.rs b/src/rust/lqos_queue_tracker/src/queue_diff.rs index c343c0e0..d5e3c207 100644 --- a/src/rust/lqos_queue_tracker/src/queue_diff.rs +++ b/src/rust/lqos_queue_tracker/src/queue_diff.rs @@ -1,7 +1,7 @@ use crate::queue_types::QueueType; use serde::Serialize; use thiserror::Error; -use tracing::error; +use tracing::warn; #[derive(Debug, Clone, Serialize)] pub enum QueueDiff { @@ -21,27 +21,46 @@ pub(crate) fn make_queue_diff( QueueType::FqCodel(..) => match current { QueueType::FqCodel(..) => Ok(fq_codel_diff(previous, current)?), _ => { - error!( - "Queue diffs are not implemented for FqCodel to {:?}", - current + warn!( + previous = queue_kind(previous), + current = queue_kind(current), + "Queue type changed; resetting queue diff sample" ); - Err(QueueDiffError::NotImplemented) + Ok(QueueDiff::None) } }, QueueType::Cake(..) => match current { QueueType::Cake(..) => Ok(cake_diff(previous, current)?), _ => { - error!("Queue diffs are not implemented for Cake to {:?}", current); - Err(QueueDiffError::NotImplemented) + warn!( + previous = queue_kind(previous), + current = queue_kind(current), + "Queue type changed; resetting queue diff sample" + ); + Ok(QueueDiff::None) } }, _ => { - error!("Queue diffs are not implemented for {:?}", current); - Err(QueueDiffError::NotImplemented) + warn!( + previous = queue_kind(previous), + current = queue_kind(current), + "Queue diff unavailable for queue type; resetting queue diff sample" + ); + Ok(QueueDiff::None) } } } +fn queue_kind(queue: &QueueType) -> &'static str { + match queue { + QueueType::FqCodel(_) => "fq_codel", + QueueType::Cake(_) => "cake", + QueueType::Mq(_) => "mq", + QueueType::Htb(_) => "htb", + QueueType::ClsAct => "clsact", + } +} + #[derive(Serialize, Clone, Debug)] pub struct CakeDiff { pub bytes: u64, @@ -116,3 +135,42 @@ fn fq_codel_diff(previous: &QueueType, current: &QueueType) -> Result 0) { @@ -339,7 +347,7 @@ export class WsClient { dashboardWsInteresting("unsubscribe-wire-interesting", [channel], { desiredChannels: Array.from(this.desiredChannels.keys()), }); - this.ws.send(encoder.encode({ Unsubscribe: { channel } })); + this._sendControl({ Unsubscribe: { channel } }); } } else { this.desiredChannels.set(channel, current - 1); @@ -355,11 +363,16 @@ export class WsClient { if (!this.ws) { this.connect(); } - if (!this.handshake_done) { + if (!this.handshake_done || !this._socketIsOpen()) { this.pending.push(normalized); + if (this.handshake_done && this.ws && !this._socketIsOpen()) { + this._retireSocket(); + } return; } - this.ws.send(encoder.encode(normalized)); + if (!this._sendControl(normalized)) { + this.pending.push(normalized); + } } on(event_name, handler) { @@ -406,28 +419,58 @@ export class WsClient { dashboardWsDebug("handshake-complete", { desiredChannels: Array.from(this.desiredChannels.keys()), }); - this.ws.send( - encoder.encode({ - HelloReply: { - ack: ACK_TEXT, - token: get_user_token(), - }, - }), - ); - for (let i = 0; i < this.pending.length; i++) { - this.ws.send(encoder.encode(this.pending[i])); + if (!this._sendControl({ + HelloReply: { + ack: ACK_TEXT, + token: get_user_token(), + }, + })) { + return; } + const pending = this.pending; this.pending = []; + for (let i = 0; i < pending.length; i++) { + if (!this._sendControl(pending[i])) { + this.pending = pending.slice(i).concat(this.pending); + return; + } + } for (const channel of this.desiredChannels.keys()) { dashboardWsInteresting("subscribe-wire-interesting", [channel], { desiredChannels: Array.from(this.desiredChannels.keys()), fromHandshake: true, }); - this.ws.send(encoder.encode({ Subscribe: { channel } })); + if (!this._sendControl({ Subscribe: { channel } })) { + return; + } } this.reconnectDelayMs = 1000; } + _socketIsOpen() { + return this.ws && this.ws.readyState === WebSocket.OPEN; + } + + _retireSocket() { + this._dropSocket("retire-socket"); + this._scheduleReconnect(); + } + + _sendControl(request_obj) { + if (!this._socketIsOpen()) { + this._retireSocket(); + return false; + } + try { + this.ws.send(encoder.encode(request_obj)); + return true; + } catch (err) { + console.warn("Websocket send failed; reconnecting", err); + this._retireSocket(); + return false; + } + } + _dispatch(msg) { if (!msg || !msg.event) { return; diff --git a/src/rust/lqosd/src/node_manager/ws.rs b/src/rust/lqosd/src/node_manager/ws.rs index 4e359bd8..6b5bb6eb 100644 --- a/src/rust/lqosd/src/node_manager/ws.rs +++ b/src/rust/lqosd/src/node_manager/ws.rs @@ -55,6 +55,7 @@ mod ticker; const WS_VERSION: &str = include_str!("../../../../VERSION_STRING"); const HANDSHAKE_TIMEOUT_SECS: u64 = 10; +const REQUEST_TIMEOUT_SECS: u64 = 15; /// Provides an Axum router for the websocket system. Exposes a single /ws route that supports /// pubsub subscriptions and private commands. @@ -182,19 +183,29 @@ async fn handle_socket( // Received a websocket message match inbound { Some(Ok(msg)) => { - let should_close = receive_channel_message( - msg, - channels.clone(), - tx.clone(), - &mut subscribed_channels, - &mut handshake_complete, - &mut WsRequestState { - private_state: &mut private_state, - login: &mut login, - shaper_query: shaper_query.clone(), - }, + let should_close = match tokio::time::timeout( + std::time::Duration::from_secs(REQUEST_TIMEOUT_SECS), + receive_channel_message( + msg, + channels.clone(), + tx.clone(), + &mut subscribed_channels, + &mut handshake_complete, + &mut WsRequestState { + private_state: &mut private_state, + login: &mut login, + shaper_query: shaper_query.clone(), + }, + ), ) - .await; + .await + { + Ok(should_close) => should_close, + Err(_) => { + warn!("Websocket request timed out; closing connection"); + true + } + }; if should_close { break; } @@ -212,6 +223,11 @@ async fn handle_socket( } } } + for channel in subscribed_channels.drain() { + channels.unsubscribe(channel, tx.clone()).await; + } + drop(private_state); + drop(tx); outbound_handle.abort(); let _ = outbound_handle.await; info!("Websocket disconnected");