mirror of
https://github.com/LibreQoE/LibreQoS.git
synced 2026-11-02 20:30:15 -06:00
changes
This commit is contained in:
1 parent
7f9eb9fc56
commit
f0c008342d
8 files changed
+831
-346
No files matched your search
@@ -40,6 +40,7 @@ from liblqos_python import is_lqosd_alive, clear_ip_mappings, delete_ip_mapping,
|
||||
run_shell_commands_as_sudo, generated_pn_download_mbps, generated_pn_upload_mbps, queues_available_override, \
|
||||
on_a_stick, get_tree_weights, get_weights, is_network_flat, get_libreqos_directory, enable_insight_topology, \
|
||||
is_insight_enabled, scheduler_error, xdp_ip_mapping_capacity, \
|
||||
overrides_circuit_adjustments_effective, \
|
||||
plan_top_level_cpu_bins, \
|
||||
plan_class_identities, \
|
||||
fast_queues_fq_codel, \
|
||||
@@ -686,6 +687,66 @@ def loadSubscriberCircuits(shapedDevicesFile):
|
||||
)
|
||||
return (subscriberCircuits, dictForCircuitsWithoutParentNodes)
|
||||
|
||||
|
||||
def normalize_sqm_override_token(raw_token):
|
||||
token = (raw_token or '').strip().lower()
|
||||
if token == '':
|
||||
return ''
|
||||
if '/' in token:
|
||||
left, right = token.split('/', 1)
|
||||
token = left.strip() + '/' + right.strip()
|
||||
return token
|
||||
|
||||
|
||||
def apply_effective_runtime_circuit_overrides(subscriberCircuits):
|
||||
"""
|
||||
Overlay adaptive runtime circuit adjustments in memory without mutating
|
||||
ShapedDevices.csv, so rebuilds can honor TreeGuard/StormGuard state.
|
||||
"""
|
||||
try:
|
||||
adjustments = overrides_circuit_adjustments_effective()
|
||||
except Exception as e:
|
||||
warnings.warn(f"Unable to load effective runtime circuit overrides: {e}", stacklevel=2)
|
||||
return 0
|
||||
|
||||
sqm_by_device_id = {}
|
||||
for adj in adjustments:
|
||||
if adj.get('type') != 'device_adjust_sqm':
|
||||
continue
|
||||
device_id = (adj.get('device_id') or '').strip()
|
||||
sqm_override = normalize_sqm_override_token(adj.get('sqm_override'))
|
||||
if device_id == '' or sqm_override == '':
|
||||
continue
|
||||
sqm_by_device_id[device_id] = sqm_override
|
||||
|
||||
if not sqm_by_device_id:
|
||||
return 0
|
||||
|
||||
overlay_count = 0
|
||||
for circuit in subscriberCircuits:
|
||||
circuit_override = None
|
||||
for device in circuit.get('devices', []):
|
||||
override = sqm_by_device_id.get(device.get('deviceID', ''))
|
||||
if override is None:
|
||||
continue
|
||||
if circuit_override is not None and circuit_override != override:
|
||||
warnings.warn(
|
||||
"Effective runtime SQM override conflict on circuit "
|
||||
+ circuit.get('circuitID', 'unknown')
|
||||
+ ". Will instead use the first runtime SQM override discovered.",
|
||||
stacklevel=2,
|
||||
)
|
||||
continue
|
||||
circuit_override = override
|
||||
|
||||
if circuit_override is None:
|
||||
continue
|
||||
if circuit.get('sqm') != circuit_override:
|
||||
overlay_count += 1
|
||||
circuit['sqm'] = circuit_override
|
||||
|
||||
return overlay_count
|
||||
|
||||
def refreshShapers():
|
||||
|
||||
# Starting
|
||||
@@ -733,6 +794,13 @@ def refreshShapers():
|
||||
|
||||
# Load Subscriber Circuits & Devices
|
||||
subscriberCircuits, dictForCircuitsWithoutParentNodes = loadSubscriberCircuits(shapedDevicesFile)
|
||||
runtime_override_count = apply_effective_runtime_circuit_overrides(subscriberCircuits)
|
||||
if runtime_override_count > 0:
|
||||
print(
|
||||
"Applied "
|
||||
+ str(runtime_override_count)
|
||||
+ " effective runtime circuit SQM override(s) in memory"
|
||||
)
|
||||
|
||||
# Preserve the logical parent (as configured in ShapedDevices.csv) before any shaping-time rewrites.
|
||||
for circuit in subscriberCircuits:
|
||||
|
||||
@@ -4,7 +4,7 @@ use std::sync::Arc;
|
||||
use tracing::{debug, warn};
|
||||
|
||||
pub(crate) enum SiteDiffResult {
|
||||
RebuildRequired,
|
||||
RebuildRequired { summary: String },
|
||||
SpeedChanges { changes: Vec<BakeryCommands> },
|
||||
NoChange,
|
||||
}
|
||||
@@ -29,6 +29,7 @@ pub(crate) fn diff_sites(
|
||||
// Therefore, we must rebuild the entire site configuration.
|
||||
let old_keys: Vec<i64> = old_sites.keys().cloned().collect();
|
||||
let new_keys: Vec<i64> = new_sites.keys().copied().collect();
|
||||
let (added_site_hashes, removed_site_hashes) = site_set_delta(old_sites, &new_sites);
|
||||
warn!(
|
||||
"Site count mismatch: old {} vs new {}",
|
||||
old_sites.len(),
|
||||
@@ -36,7 +37,15 @@ pub(crate) fn diff_sites(
|
||||
);
|
||||
debug!("Old site hashes: {:?}", old_keys);
|
||||
debug!("New site hashes: {:?}", new_keys);
|
||||
return SiteDiffResult::RebuildRequired;
|
||||
return SiteDiffResult::RebuildRequired {
|
||||
summary: format!(
|
||||
"Bakery full reload triggered by site diff: site_count_mismatch old_count={} new_count={} added_site_hashes={:?} removed_site_hashes={:?}",
|
||||
old_sites.len(),
|
||||
new_sites.len(),
|
||||
added_site_hashes,
|
||||
removed_site_hashes
|
||||
),
|
||||
};
|
||||
}
|
||||
|
||||
// Compare each site in the old and new maps for structural differences.
|
||||
@@ -83,7 +92,12 @@ pub(crate) fn diff_sites(
|
||||
"Site hash {} change detail: parent={}→{}, up_parent={}→{}, minor=0x{:x}→0x{:x}",
|
||||
site_hash, opar, npar, oup, nup, omin, nmin
|
||||
);
|
||||
return SiteDiffResult::RebuildRequired;
|
||||
return SiteDiffResult::RebuildRequired {
|
||||
summary: format!(
|
||||
"Bakery full reload triggered by site diff: site_hash={} parent={}→{} up_parent={}→{} minor=0x{:x}→0x{:x}",
|
||||
site_hash, opar, npar, oup, nup, omin, nmin
|
||||
),
|
||||
};
|
||||
}
|
||||
// If the speeds have changed, we need to store the change.
|
||||
if let Some(speed_change) = site_speeds_changed(old_cmd.as_ref(), new_cmd.as_ref()) {
|
||||
@@ -93,7 +107,17 @@ pub(crate) fn diff_sites(
|
||||
} else {
|
||||
// If a site is missing in the new batch, we need to rebuild.
|
||||
debug!("Site hash {} is missing in the new batch", site_hash);
|
||||
return SiteDiffResult::RebuildRequired;
|
||||
let (added_site_hashes, removed_site_hashes) = site_set_delta(old_sites, &new_sites);
|
||||
return SiteDiffResult::RebuildRequired {
|
||||
summary: format!(
|
||||
"Bakery full reload triggered by site diff: site_set_changed missing_old_site_hash={} old_count={} new_count={} added_site_hashes={:?} removed_site_hashes={:?}",
|
||||
site_hash,
|
||||
old_sites.len(),
|
||||
new_sites.len(),
|
||||
added_site_hashes,
|
||||
removed_site_hashes
|
||||
),
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
@@ -107,6 +131,27 @@ pub(crate) fn diff_sites(
|
||||
SiteDiffResult::NoChange
|
||||
}
|
||||
|
||||
fn site_set_delta(
|
||||
old_sites: &HashMap<i64, Arc<BakeryCommands>>,
|
||||
new_sites: &HashMap<i64, &Arc<BakeryCommands>>,
|
||||
) -> (Vec<i64>, Vec<i64>) {
|
||||
let mut added_site_hashes: Vec<i64> = new_sites
|
||||
.keys()
|
||||
.filter(|hash| !old_sites.contains_key(hash))
|
||||
.copied()
|
||||
.collect();
|
||||
added_site_hashes.sort_unstable();
|
||||
|
||||
let mut removed_site_hashes: Vec<i64> = old_sites
|
||||
.keys()
|
||||
.filter(|hash| !new_sites.contains_key(hash))
|
||||
.copied()
|
||||
.collect();
|
||||
removed_site_hashes.sort_unstable();
|
||||
|
||||
(added_site_hashes, removed_site_hashes)
|
||||
}
|
||||
|
||||
fn is_structurally_different(a: &BakeryCommands, b: &BakeryCommands) -> bool {
|
||||
let BakeryCommands::AddSite {
|
||||
site_hash,
|
||||
@@ -211,3 +256,71 @@ fn site_speeds_changed(a: &BakeryCommands, b: &BakeryCommands) -> Option<BakeryC
|
||||
None
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::{SiteDiffResult, diff_sites};
|
||||
use crate::BakeryCommands;
|
||||
use lqos_bus::TcHandle;
|
||||
use std::collections::HashMap;
|
||||
use std::sync::Arc;
|
||||
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
fn add_site(
|
||||
site_hash: i64,
|
||||
parent_class_id: u32,
|
||||
up_parent_class_id: u32,
|
||||
class_minor: u16,
|
||||
dl_min: f32,
|
||||
ul_min: f32,
|
||||
dl_max: f32,
|
||||
ul_max: f32,
|
||||
) -> Arc<BakeryCommands> {
|
||||
Arc::new(BakeryCommands::AddSite {
|
||||
site_hash,
|
||||
parent_class_id: TcHandle::from_u32(parent_class_id),
|
||||
up_parent_class_id: TcHandle::from_u32(up_parent_class_id),
|
||||
class_minor,
|
||||
download_bandwidth_min: dl_min,
|
||||
upload_bandwidth_min: ul_min,
|
||||
download_bandwidth_max: dl_max,
|
||||
upload_bandwidth_max: ul_max,
|
||||
})
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn site_count_mismatch_reports_added_and_removed_hashes() {
|
||||
let old_site = add_site(10, 0x10010, 0x20010, 0x11, 1.0, 1.0, 10.0, 10.0);
|
||||
let new_site = add_site(20, 0x10020, 0x20020, 0x21, 1.0, 1.0, 10.0, 10.0);
|
||||
let old_sites = HashMap::from([(10, old_site)]);
|
||||
let batch = vec![
|
||||
new_site,
|
||||
add_site(30, 0x10030, 0x20030, 0x31, 1.0, 1.0, 10.0, 10.0),
|
||||
];
|
||||
|
||||
let SiteDiffResult::RebuildRequired { summary } = diff_sites(&batch, &old_sites) else {
|
||||
panic!("expected rebuild-required site diff");
|
||||
};
|
||||
|
||||
assert!(summary.contains("site_count_mismatch"));
|
||||
assert!(summary.contains("added_site_hashes=[20, 30]"));
|
||||
assert!(summary.contains("removed_site_hashes=[10]"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn structural_site_change_reports_parent_transition() {
|
||||
let old_site = add_site(10, 0x10010, 0x20010, 0x11, 1.0, 1.0, 10.0, 10.0);
|
||||
let new_site = add_site(10, 0x10020, 0x20020, 0x21, 1.0, 1.0, 10.0, 10.0);
|
||||
let old_sites = HashMap::from([(10, old_site)]);
|
||||
let batch = vec![new_site];
|
||||
|
||||
let SiteDiffResult::RebuildRequired { summary } = diff_sites(&batch, &old_sites) else {
|
||||
panic!("expected rebuild-required site diff");
|
||||
};
|
||||
|
||||
assert!(summary.contains("site_hash=10"));
|
||||
assert!(summary.contains("parent=0x1:0x10→0x1:0x20"));
|
||||
assert!(summary.contains("up_parent=0x2:0x10→0x2:0x20"));
|
||||
assert!(summary.contains("minor=0x11→0x21"));
|
||||
}
|
||||
}
|
||||
+139
-16
@@ -813,6 +813,11 @@ fn push_bakery_event(event: &str, status: &str, summary: String) {
|
||||
}
|
||||
}
|
||||
|
||||
fn announce_full_reload(summary: &str) {
|
||||
warn!("{summary}");
|
||||
push_bakery_event("full_reload_trigger", "warning", summary.to_string());
|
||||
}
|
||||
|
||||
fn clear_bakery_apply_progress(state: &mut BakeryTelemetryState) {
|
||||
state.current_apply_phase = None;
|
||||
state.current_apply_total_tc_commands = 0;
|
||||
@@ -1976,6 +1981,7 @@ fn bakery_main(rx: Receiver<BakeryCommands>, tx: Sender<BakeryCommands>) {
|
||||
}
|
||||
|
||||
fn process_pending_migrations(
|
||||
config: &Arc<Config>,
|
||||
circuits: &mut HashMap<i64, Arc<BakeryCommands>>,
|
||||
live_circuits: &mut HashMap<i64, u64>,
|
||||
sites: &HashMap<i64, Arc<BakeryCommands>>,
|
||||
@@ -1984,15 +1990,6 @@ fn bakery_main(rx: Receiver<BakeryCommands>, tx: Sender<BakeryCommands>) {
|
||||
virtualized_sites: &mut HashMap<i64, VirtualizedSiteState>,
|
||||
runtime_node_operations: &mut HashMap<i64, RuntimeNodeOperation>,
|
||||
) {
|
||||
if migrations.is_empty() {
|
||||
return;
|
||||
}
|
||||
|
||||
let Ok(config) = lqos_config::load_config() else {
|
||||
error!("Failed to load configuration while processing pending migrations.");
|
||||
return;
|
||||
};
|
||||
|
||||
let mut advanced = 0usize;
|
||||
let mut to_remove = Vec::new();
|
||||
let mut effective_state_changed = false;
|
||||
@@ -2321,7 +2318,12 @@ fn bakery_main(rx: Receiver<BakeryCommands>, tx: Sender<BakeryCommands>) {
|
||||
let command = match rx.recv_timeout(Duration::from_millis(BAKERY_BACKGROUND_INTERVAL_MS)) {
|
||||
Ok(command) => command,
|
||||
Err(RecvTimeoutError::Timeout) => {
|
||||
let Ok(config) = lqos_config::load_config() else {
|
||||
error!("Failed to load configuration while processing pending migrations.");
|
||||
continue;
|
||||
};
|
||||
process_pending_migrations(
|
||||
&config,
|
||||
&mut circuits,
|
||||
&mut live_circuits,
|
||||
&sites,
|
||||
@@ -2473,7 +2475,12 @@ fn bakery_main(rx: Receiver<BakeryCommands>, tx: Sender<BakeryCommands>) {
|
||||
&mut virtualized_sites,
|
||||
&mut runtime_node_operations,
|
||||
);
|
||||
let Ok(config) = lqos_config::load_config() else {
|
||||
error!("Failed to load configuration while processing pending migrations.");
|
||||
continue;
|
||||
};
|
||||
process_pending_migrations(
|
||||
&config,
|
||||
&mut circuits,
|
||||
&mut live_circuits,
|
||||
&sites,
|
||||
@@ -2504,7 +2511,12 @@ fn bakery_main(rx: Receiver<BakeryCommands>, tx: Sender<BakeryCommands>) {
|
||||
BakeryCommands::Tick => {
|
||||
// Reset per-cycle counters at the start of the tick
|
||||
handle_tick(&mut circuits, &mut live_circuits, &mut sites);
|
||||
let Ok(config) = lqos_config::load_config() else {
|
||||
error!("Failed to load configuration while processing pending migrations.");
|
||||
continue;
|
||||
};
|
||||
process_pending_migrations(
|
||||
&config,
|
||||
&mut circuits,
|
||||
&mut live_circuits,
|
||||
&sites,
|
||||
@@ -2654,6 +2666,10 @@ fn handle_commit_batch(
|
||||
let limit_label = format_mapped_limit(effective_limit);
|
||||
|
||||
if let Some(reason) = bakery_reload_required_reason() {
|
||||
let summary = format!(
|
||||
"Bakery full reload triggered by reload-required state: {}",
|
||||
reason
|
||||
);
|
||||
let (new_batch, mapped_limit_stats) =
|
||||
filter_batch_by_mapped_circuit_limit(new_batch, circuits, effective_limit);
|
||||
log_mapped_limit_decision("reload-required rebuild", mapped_limit, mapped_limit_stats);
|
||||
@@ -2669,10 +2685,7 @@ fn handle_commit_batch(
|
||||
);
|
||||
maybe_emit_mapped_circuit_limit_urgent(&mapped_limit_stats);
|
||||
}
|
||||
warn!(
|
||||
"Bakery: full reload required before further incremental topology mutation: {}",
|
||||
reason
|
||||
);
|
||||
announce_full_reload(&summary);
|
||||
full_reload(
|
||||
batch,
|
||||
sites,
|
||||
@@ -2686,6 +2699,7 @@ fn handle_commit_batch(
|
||||
stormguard_overrides,
|
||||
virtualized_sites,
|
||||
runtime_node_operations,
|
||||
summary,
|
||||
);
|
||||
MQ_CREATED.store(true, std::sync::atomic::Ordering::Relaxed);
|
||||
return;
|
||||
@@ -2714,7 +2728,10 @@ fn handle_commit_batch(
|
||||
maybe_emit_mapped_circuit_limit_urgent(&mapped_limit_stats);
|
||||
}
|
||||
// If the MQ hasn't been created, we need to do this as a full, unadjusted run.
|
||||
let summary = "Bakery full reload triggered by baseline rebuild after restart/cold start."
|
||||
.to_string();
|
||||
info!("Bakery baseline rebuild after restart/cold start: performing explicit full reload.");
|
||||
announce_full_reload(&summary);
|
||||
full_reload(
|
||||
batch,
|
||||
sites,
|
||||
@@ -2728,13 +2745,14 @@ fn handle_commit_batch(
|
||||
stormguard_overrides,
|
||||
virtualized_sites,
|
||||
runtime_node_operations,
|
||||
summary,
|
||||
);
|
||||
MQ_CREATED.store(true, std::sync::atomic::Ordering::Relaxed);
|
||||
return;
|
||||
}
|
||||
|
||||
let site_change_mode = diff_sites(&new_batch, sites);
|
||||
if matches!(site_change_mode, SiteDiffResult::RebuildRequired) {
|
||||
if let SiteDiffResult::RebuildRequired { summary } = &site_change_mode {
|
||||
let (new_batch, mapped_limit_stats) =
|
||||
filter_batch_by_mapped_circuit_limit(new_batch, circuits, effective_limit);
|
||||
log_mapped_limit_decision("site-structure rebuild", mapped_limit, mapped_limit_stats);
|
||||
@@ -2752,6 +2770,7 @@ fn handle_commit_batch(
|
||||
}
|
||||
// If the site structure has changed, we need to rebuild everything.
|
||||
info!("Bakery full reload: site_struct=1, circuit_struct=0");
|
||||
announce_full_reload(summary);
|
||||
full_reload(
|
||||
batch,
|
||||
sites,
|
||||
@@ -2765,6 +2784,7 @@ fn handle_commit_batch(
|
||||
stormguard_overrides,
|
||||
virtualized_sites,
|
||||
runtime_node_operations,
|
||||
summary.clone(),
|
||||
);
|
||||
MQ_CREATED.store(true, std::sync::atomic::Ordering::Relaxed);
|
||||
return;
|
||||
@@ -2786,6 +2806,10 @@ fn handle_commit_batch(
|
||||
if let CircuitDiffResult::Categorized(categories) = &circuit_change_mode
|
||||
&& !categories.structural_changed.is_empty()
|
||||
{
|
||||
let summary = format!(
|
||||
"Bakery full reload triggered by circuit structural diff: structural_changed_count={}",
|
||||
categories.structural_changed.len()
|
||||
);
|
||||
let (new_batch, mapped_limit_stats) =
|
||||
filter_batch_by_mapped_circuit_limit(new_batch.clone(), circuits, effective_limit);
|
||||
log_mapped_limit_decision(
|
||||
@@ -2809,6 +2833,7 @@ fn handle_commit_batch(
|
||||
"Bakery full reload: site_struct=0, circuit_struct={}",
|
||||
categories.structural_changed.len()
|
||||
);
|
||||
announce_full_reload(&summary);
|
||||
full_reload(
|
||||
batch,
|
||||
sites,
|
||||
@@ -2822,6 +2847,7 @@ fn handle_commit_batch(
|
||||
stormguard_overrides,
|
||||
virtualized_sites,
|
||||
runtime_node_operations,
|
||||
summary,
|
||||
);
|
||||
MQ_CREATED.store(true, std::sync::atomic::Ordering::Relaxed);
|
||||
return;
|
||||
@@ -2857,6 +2883,11 @@ fn handle_commit_batch(
|
||||
}) {
|
||||
error!("Channel full, falling back to full rebuild: {}", e);
|
||||
info!("Bakery full reload: site_struct=0, circuit_struct=0");
|
||||
let summary = format!(
|
||||
"Bakery full reload triggered because site speed live-change enqueue failed: {}",
|
||||
e
|
||||
);
|
||||
announce_full_reload(&summary);
|
||||
full_reload(
|
||||
batch,
|
||||
sites,
|
||||
@@ -2870,6 +2901,7 @@ fn handle_commit_batch(
|
||||
stormguard_overrides,
|
||||
virtualized_sites,
|
||||
runtime_node_operations,
|
||||
summary,
|
||||
);
|
||||
return; // Skip the rest of this CommitBatch processing
|
||||
}
|
||||
@@ -3095,6 +3127,11 @@ fn handle_commit_batch(
|
||||
mapped_limit,
|
||||
mapped_limit_stats,
|
||||
);
|
||||
let summary = format!(
|
||||
"Bakery full reload triggered because live migration setup failed for active migrated circuit {}.",
|
||||
circuit_hash
|
||||
);
|
||||
announce_full_reload(&summary);
|
||||
full_reload(
|
||||
batch,
|
||||
sites,
|
||||
@@ -3108,6 +3145,7 @@ fn handle_commit_batch(
|
||||
stormguard_overrides,
|
||||
virtualized_sites,
|
||||
runtime_node_operations,
|
||||
summary,
|
||||
);
|
||||
MQ_CREATED.store(true, std::sync::atomic::Ordering::Relaxed);
|
||||
return;
|
||||
@@ -6242,13 +6280,13 @@ fn full_reload(
|
||||
stormguard_overrides: &HashMap<StormguardOverrideKey, u64>,
|
||||
virtualized_sites: &mut HashMap<i64, VirtualizedSiteState>,
|
||||
runtime_node_operations: &mut HashMap<i64, RuntimeNodeOperation>,
|
||||
trigger_summary: String,
|
||||
) {
|
||||
warn!("Bakery: Full reload triggered due to site or circuit changes.");
|
||||
FULL_RELOAD_IN_PROGRESS.store(true, Ordering::Relaxed);
|
||||
mark_bakery_action_started(
|
||||
BakeryMode::ApplyingFullReload,
|
||||
"full_reload_started",
|
||||
"Full reload triggered due to site or circuit changes.".to_string(),
|
||||
trigger_summary,
|
||||
);
|
||||
let _reload_scope = FullReloadScope;
|
||||
let previous_sites = sites.clone();
|
||||
@@ -7275,6 +7313,91 @@ mod tests {
|
||||
assert!(!telemetry.reload_required);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn deferred_runtime_site_prune_advances_without_pending_migrations() {
|
||||
let _guard = bakery_test_lock().lock().expect("test lock");
|
||||
reset_bakery_test_state();
|
||||
|
||||
let mut cfg = Config::default();
|
||||
cfg.bridge = Some(lqos_config::BridgeConfig {
|
||||
use_xdp_bridge: false,
|
||||
to_internet: "__bakery-missing-wan__".to_string(),
|
||||
to_network: "__bakery-missing-lan__".to_string(),
|
||||
});
|
||||
let config = Arc::new(cfg);
|
||||
|
||||
let site_hash = 20;
|
||||
let site = mk_add_site(site_hash, 0x10020, 0x20020, 0x21);
|
||||
let mut virtualized_sites = HashMap::from([(
|
||||
site_hash,
|
||||
VirtualizedSiteState {
|
||||
site,
|
||||
saved_sites: HashMap::new(),
|
||||
saved_circuits: HashMap::new(),
|
||||
active_sites: HashMap::new(),
|
||||
active_circuits: HashMap::new(),
|
||||
qdisc_handles: VirtualizedSiteQdiscHandles {
|
||||
down: None,
|
||||
up: None,
|
||||
},
|
||||
pending_prune: true,
|
||||
next_prune_attempt_unix: 0,
|
||||
},
|
||||
)]);
|
||||
let mut runtime_node_operations = HashMap::new();
|
||||
let submitted_at = unix_now().saturating_sub(5);
|
||||
let mut operation = RuntimeNodeOperation::new(
|
||||
1,
|
||||
site_hash,
|
||||
RuntimeNodeOperationAction::Virtualize,
|
||||
submitted_at,
|
||||
);
|
||||
operation.attempt_count = 1;
|
||||
operation.update_status(
|
||||
RuntimeNodeOperationStatus::AppliedAwaitingCleanup,
|
||||
submitted_at,
|
||||
None,
|
||||
Some(0),
|
||||
);
|
||||
runtime_node_operations.insert(site_hash, operation);
|
||||
|
||||
flush_deferred_runtime_site_prunes(
|
||||
&config,
|
||||
&HashMap::new(),
|
||||
&HashMap::new(),
|
||||
&mut virtualized_sites,
|
||||
&HashMap::new(),
|
||||
&mut runtime_node_operations,
|
||||
);
|
||||
|
||||
let state = virtualized_sites
|
||||
.get(&site_hash)
|
||||
.expect("virtualized site state");
|
||||
let operation = runtime_node_operations
|
||||
.get(&site_hash)
|
||||
.expect("runtime operation");
|
||||
|
||||
assert!(state.pending_prune);
|
||||
assert_eq!(
|
||||
operation.status,
|
||||
RuntimeNodeOperationStatus::AppliedAwaitingCleanup
|
||||
);
|
||||
assert_eq!(operation.attempt_count, 2);
|
||||
assert!(operation.updated_at_unix >= submitted_at);
|
||||
assert_eq!(
|
||||
operation.next_retry_at_unix,
|
||||
Some(state.next_prune_attempt_unix)
|
||||
);
|
||||
assert!(operation.next_retry_at_unix.expect("retry time") >= operation.updated_at_unix);
|
||||
assert!(
|
||||
operation
|
||||
.last_error
|
||||
.as_deref()
|
||||
.unwrap_or_default()
|
||||
.contains("Failed to snapshot live classes on __bakery-missing-lan__")
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn dirty_runtime_subtrees_below_threshold_do_not_require_full_reload() {
|
||||
let _guard = bakery_test_lock().lock().expect("test lock");
|
||||
|
||||
@@ -13,6 +13,7 @@ use nix::libc::getpid;
|
||||
use pyo3::exceptions::PyOSError;
|
||||
use pyo3::prelude::*;
|
||||
use pyo3::types::{PyDict, PyList};
|
||||
use std::fmt::Write as _;
|
||||
use std::{
|
||||
fs::{File, read_to_string, remove_file},
|
||||
io::Write,
|
||||
@@ -973,6 +974,7 @@ fn liblqos_python(m: &Bound<'_, PyModule>) -> PyResult<()> {
|
||||
m.add_function(wrap_pyfunction!(promote_to_root_list, m)?)?;
|
||||
m.add_function(wrap_pyfunction!(client_bandwidth_multiplier, m)?)?;
|
||||
m.add_function(wrap_pyfunction!(calculate_hash, m)?)?;
|
||||
m.add_function(wrap_pyfunction!(calculate_shaping_runtime_hash, m)?)?;
|
||||
m.add_function(wrap_pyfunction!(scheduler_alive, m)?)?;
|
||||
m.add_function(wrap_pyfunction!(scheduler_error, m)?)?;
|
||||
m.add_function(wrap_pyfunction!(scheduler_output, m)?)?;
|
||||
@@ -2538,6 +2540,73 @@ fn calculate_hash() -> PyResult<i64> {
|
||||
Ok(hash)
|
||||
}
|
||||
|
||||
fn shaping_runtime_sqm_fingerprint() -> Result<String> {
|
||||
let config = lqos_config::load_config()?;
|
||||
let apply_stormguard = config
|
||||
.stormguard
|
||||
.as_ref()
|
||||
.is_some_and(|sg| sg.enabled && !sg.dry_run);
|
||||
let apply_treeguard = config.treeguard.enabled;
|
||||
let overrides =
|
||||
lqos_overrides::OverrideStore::load_effective(apply_stormguard, apply_treeguard)?;
|
||||
|
||||
let mut sqm_entries: Vec<(String, String)> = overrides
|
||||
.circuit_adjustments()
|
||||
.iter()
|
||||
.filter_map(|adj| match adj {
|
||||
lqos_overrides::CircuitAdjustment::DeviceAdjustSqm {
|
||||
device_id,
|
||||
sqm_override,
|
||||
} => {
|
||||
let token = sqm_override
|
||||
.as_deref()
|
||||
.map(str::trim)
|
||||
.map(str::to_lowercase)
|
||||
.unwrap_or_default();
|
||||
if token.is_empty() {
|
||||
None
|
||||
} else {
|
||||
Some((device_id.clone(), token))
|
||||
}
|
||||
}
|
||||
_ => None,
|
||||
})
|
||||
.collect();
|
||||
sqm_entries.sort();
|
||||
|
||||
let mut fingerprint = String::new();
|
||||
for (device_id, token) in sqm_entries {
|
||||
let _ = writeln!(&mut fingerprint, "{device_id}={token}");
|
||||
}
|
||||
|
||||
Ok(fingerprint)
|
||||
}
|
||||
|
||||
#[pyfunction]
|
||||
fn calculate_shaping_runtime_hash() -> PyResult<i64> {
|
||||
let Ok(config) = lqos_config::load_config() else {
|
||||
return Ok(0);
|
||||
};
|
||||
let nj_path = Path::new(&config.lqos_directory).join("network.json");
|
||||
let sd_path = Path::new(&config.lqos_directory).join("ShapedDevices.csv");
|
||||
|
||||
let Ok(nj_as_string) = read_to_string(nj_path) else {
|
||||
return Ok(0);
|
||||
};
|
||||
let Ok(sd_as_string) = read_to_string(sd_path) else {
|
||||
return Ok(0);
|
||||
};
|
||||
let Ok(runtime_sqm_fingerprint) = shaping_runtime_sqm_fingerprint() else {
|
||||
return Ok(0);
|
||||
};
|
||||
|
||||
let combined = format!(
|
||||
"{}\n{}\n{}",
|
||||
nj_as_string, sd_as_string, runtime_sqm_fingerprint
|
||||
);
|
||||
Ok(lqos_utils::hash_to_i64(&combined))
|
||||
}
|
||||
|
||||
////////////////////////////// The Bakery class //////////////////////////////
|
||||
|
||||
#[derive(Clone)]
|
||||
|
||||
@@ -2465,7 +2465,6 @@ fn apply_circuit_sqm_change<P, C, L>(
|
||||
}
|
||||
|
||||
let mut persisted_ok = false;
|
||||
let mut can_apply_live = true;
|
||||
let device_ids: Vec<String> = devices
|
||||
.iter()
|
||||
.map(|device| device.device_id.clone())
|
||||
@@ -2481,7 +2480,6 @@ fn apply_circuit_sqm_change<P, C, L>(
|
||||
persisted_ok = true;
|
||||
}
|
||||
Err(e) => {
|
||||
can_apply_live = false;
|
||||
status.warnings.push(format!(
|
||||
"TreeGuard circuits: failed to {} SQM overrides for circuit '{circuit_id}': {e}",
|
||||
if returning_to_base { "clear" } else { "persist" }
|
||||
@@ -2505,33 +2503,29 @@ fn apply_circuit_sqm_change<P, C, L>(
|
||||
}
|
||||
}
|
||||
|
||||
let live_ok = if can_apply_live {
|
||||
match live_apply(circuit_id, devices, live_token) {
|
||||
Ok(()) => true,
|
||||
Err(e) => {
|
||||
status.warnings.push(format!(
|
||||
"TreeGuard circuits: live SQM apply failed for circuit '{circuit_id}': {e}"
|
||||
));
|
||||
push_activity(
|
||||
activity,
|
||||
TreeguardActivityEntry {
|
||||
time: now_unix.to_string(),
|
||||
entity_type: "circuit".to_string(),
|
||||
entity_id: circuit_entity_id.to_string(),
|
||||
action: if returning_to_base {
|
||||
"clear_sqm_live_failed".to_string()
|
||||
} else {
|
||||
format!("apply_sqm_live_failed:{token}")
|
||||
},
|
||||
persisted: persisted_ok,
|
||||
reason: format!("Bakery live apply failed: {e}"),
|
||||
let live_ok = match live_apply(circuit_id, devices, live_token) {
|
||||
Ok(()) => true,
|
||||
Err(e) => {
|
||||
status.warnings.push(format!(
|
||||
"TreeGuard circuits: live SQM apply failed for circuit '{circuit_id}': {e}"
|
||||
));
|
||||
push_activity(
|
||||
activity,
|
||||
TreeguardActivityEntry {
|
||||
time: now_unix.to_string(),
|
||||
entity_type: "circuit".to_string(),
|
||||
entity_id: circuit_entity_id.to_string(),
|
||||
action: if returning_to_base {
|
||||
"clear_sqm_live_failed".to_string()
|
||||
} else {
|
||||
format!("apply_sqm_live_failed:{token}")
|
||||
},
|
||||
);
|
||||
false
|
||||
}
|
||||
persisted: persisted_ok,
|
||||
reason: format!("Bakery live apply failed: {e}"),
|
||||
},
|
||||
);
|
||||
false
|
||||
}
|
||||
} else {
|
||||
false
|
||||
};
|
||||
|
||||
if live_ok || persisted_ok {
|
||||
|
||||
@@ -777,307 +777,148 @@ fn min_capacity_along_route(graph: &GraphType, path: &[NodeIndex]) -> u64 {
|
||||
#[derive(Debug)]
|
||||
struct SquashCandidate {
|
||||
endpoint_a: NodeIndex,
|
||||
endpoint_a_id: String,
|
||||
endpoint_a_name: String,
|
||||
relay_a: NodeIndex,
|
||||
relay_a_id: String,
|
||||
relay_a_name: String,
|
||||
relay_b: NodeIndex,
|
||||
relay_b_id: String,
|
||||
relay_b_name: String,
|
||||
endpoint_b: NodeIndex,
|
||||
endpoint_b_id: String,
|
||||
endpoint_b_name: String,
|
||||
}
|
||||
|
||||
impl SquashCandidate {
|
||||
fn new(
|
||||
graph: &GraphType,
|
||||
endpoint_a: NodeIndex,
|
||||
relay_a: NodeIndex,
|
||||
relay_b: NodeIndex,
|
||||
endpoint_b: NodeIndex,
|
||||
) -> Self {
|
||||
let candidate = Self {
|
||||
endpoint_a,
|
||||
endpoint_a_id: graph[endpoint_a].network_json_id(),
|
||||
endpoint_a_name: graph[endpoint_a].name(),
|
||||
relay_a,
|
||||
relay_a_id: graph[relay_a].network_json_id(),
|
||||
relay_a_name: graph[relay_a].name(),
|
||||
relay_b,
|
||||
relay_b_id: graph[relay_b].network_json_id(),
|
||||
relay_b_name: graph[relay_b].name(),
|
||||
endpoint_b,
|
||||
endpoint_b_id: graph[endpoint_b].network_json_id(),
|
||||
endpoint_b_name: graph[endpoint_b].name(),
|
||||
};
|
||||
|
||||
candidate.canonicalized()
|
||||
}
|
||||
|
||||
fn canonicalized(self) -> Self {
|
||||
if self.reverse_key() < self.forward_key() {
|
||||
self.reversed()
|
||||
} else {
|
||||
self
|
||||
}
|
||||
}
|
||||
|
||||
fn canonical_key(&self) -> (String, String, String, String) {
|
||||
let forward = self.forward_key();
|
||||
let reverse = self.reverse_key();
|
||||
if reverse < forward { reverse } else { forward }
|
||||
}
|
||||
|
||||
fn forward_key(&self) -> (String, String, String, String) {
|
||||
(
|
||||
self.endpoint_a_id.clone(),
|
||||
self.relay_a_id.clone(),
|
||||
self.relay_b_id.clone(),
|
||||
self.endpoint_b_id.clone(),
|
||||
)
|
||||
}
|
||||
|
||||
fn reverse_key(&self) -> (String, String, String, String) {
|
||||
(
|
||||
self.endpoint_b_id.clone(),
|
||||
self.relay_b_id.clone(),
|
||||
self.relay_a_id.clone(),
|
||||
self.endpoint_a_id.clone(),
|
||||
)
|
||||
}
|
||||
|
||||
fn reversed(self) -> Self {
|
||||
Self {
|
||||
endpoint_a: self.endpoint_b,
|
||||
endpoint_a_id: self.endpoint_b_id,
|
||||
endpoint_a_name: self.endpoint_b_name,
|
||||
relay_a: self.relay_b,
|
||||
relay_a_id: self.relay_b_id,
|
||||
relay_a_name: self.relay_b_name,
|
||||
relay_b: self.relay_a,
|
||||
relay_b_id: self.relay_a_id,
|
||||
relay_b_name: self.relay_a_name,
|
||||
endpoint_b: self.endpoint_a,
|
||||
endpoint_b_id: self.endpoint_a_id,
|
||||
endpoint_b_name: self.endpoint_a_name,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn stable_node_key(graph: &GraphType, node: NodeIndex) -> (String, String) {
|
||||
(graph[node].network_json_id(), graph[node].name())
|
||||
}
|
||||
|
||||
fn sorted_unique_neighbors(graph: &GraphType, node: NodeIndex) -> Vec<NodeIndex> {
|
||||
let mut neighbors: Vec<NodeIndex> = graph
|
||||
.neighbors_directed(node, petgraph::Incoming)
|
||||
.chain(graph.neighbors_directed(node, petgraph::Outgoing))
|
||||
.collect();
|
||||
neighbors.sort_unstable_by_key(|neighbor| stable_node_key(graph, *neighbor));
|
||||
neighbors.dedup();
|
||||
neighbors
|
||||
}
|
||||
|
||||
fn total_degree(graph: &GraphType, node: NodeIndex) -> usize {
|
||||
graph.neighbors_directed(node, petgraph::Incoming).count()
|
||||
+ graph.neighbors_directed(node, petgraph::Outgoing).count()
|
||||
}
|
||||
|
||||
fn is_relay_node(graph: &GraphType, node: NodeIndex) -> bool {
|
||||
total_degree(graph, node) == 4 && sorted_unique_neighbors(graph, node).len() == 2
|
||||
}
|
||||
|
||||
fn is_meaningful_endpoint(graph: &GraphType, node: NodeIndex) -> bool {
|
||||
let unique_neighbors = sorted_unique_neighbors(graph, node);
|
||||
!(total_degree(graph, node) == 4 && unique_neighbors.len() == 2)
|
||||
&& unique_neighbors.len() >= 3
|
||||
&& !matches!(graph[node], GraphMapping::AccessPoint { .. })
|
||||
}
|
||||
|
||||
fn should_skip_squash(
|
||||
graph: &GraphType,
|
||||
config: &Arc<Config>,
|
||||
chain_nodes: [NodeIndex; 4],
|
||||
) -> bool {
|
||||
let do_not_squash = config
|
||||
.uisp_integration
|
||||
.do_not_squash_sites
|
||||
.clone()
|
||||
.unwrap_or_default();
|
||||
|
||||
chain_nodes
|
||||
.into_iter()
|
||||
.map(|node| graph[node].name())
|
||||
.any(|name| do_not_squash.contains(&name))
|
||||
}
|
||||
|
||||
fn find_point_to_point_squash_candidates(
|
||||
graph: &mut GraphType,
|
||||
aps_with_clients: &HashSet<String>,
|
||||
config: &Arc<Config>,
|
||||
) {
|
||||
let mut candidates = Vec::new();
|
||||
|
||||
// Find all nodes with exactly total degree 4 in bidirectional graph (2 unique neighbors)
|
||||
// This accounts for bidirectional edges: A->B and B->A both exist
|
||||
let relay_nodes: Vec<NodeIndex> = graph
|
||||
.node_indices()
|
||||
.filter(|&node| {
|
||||
let incoming = graph.neighbors_directed(node, petgraph::Incoming).count();
|
||||
let outgoing = graph.neighbors_directed(node, petgraph::Outgoing).count();
|
||||
let total_degree = incoming + outgoing;
|
||||
|
||||
// In bidirectional graph, relay nodes have degree 4 (2 in, 2 out)
|
||||
// but only 2 unique neighbors
|
||||
if total_degree == 4 {
|
||||
let mut unique_neighbors = std::collections::HashSet::new();
|
||||
for neighbor in graph.neighbors_directed(node, petgraph::Incoming) {
|
||||
unique_neighbors.insert(neighbor);
|
||||
}
|
||||
for neighbor in graph.neighbors_directed(node, petgraph::Outgoing) {
|
||||
unique_neighbors.insert(neighbor);
|
||||
}
|
||||
unique_neighbors.len() == 2
|
||||
} else {
|
||||
false
|
||||
}
|
||||
})
|
||||
.collect();
|
||||
|
||||
// For each potential relay node, check if it's part of a 2-relay chain
|
||||
for &relay_node in &relay_nodes {
|
||||
// Skip if this relay node is an AP with clients
|
||||
if let GraphMapping::AccessPoint { id, .. } = &graph[relay_node]
|
||||
&& aps_with_clients.contains(id)
|
||||
{
|
||||
continue;
|
||||
}
|
||||
// Get the unique neighbors for this relay node
|
||||
let mut unique_neighbors = std::collections::HashSet::new();
|
||||
for neighbor in graph.neighbors_directed(relay_node, petgraph::Incoming) {
|
||||
unique_neighbors.insert(neighbor);
|
||||
}
|
||||
for neighbor in graph.neighbors_directed(relay_node, petgraph::Outgoing) {
|
||||
unique_neighbors.insert(neighbor);
|
||||
}
|
||||
|
||||
if unique_neighbors.len() != 2 {
|
||||
continue;
|
||||
}
|
||||
|
||||
let neighbors: Vec<NodeIndex> = unique_neighbors.into_iter().collect();
|
||||
let node_a = neighbors[0];
|
||||
let node_b = neighbors[1];
|
||||
|
||||
// Check if each neighbor is a relay or endpoint in bidirectional context
|
||||
let is_node_a_relay = {
|
||||
let mut a_neighbors = std::collections::HashSet::new();
|
||||
for neighbor in graph.neighbors_directed(node_a, petgraph::Incoming) {
|
||||
a_neighbors.insert(neighbor);
|
||||
}
|
||||
for neighbor in graph.neighbors_directed(node_a, petgraph::Outgoing) {
|
||||
a_neighbors.insert(neighbor);
|
||||
}
|
||||
let a_incoming = graph.neighbors_directed(node_a, petgraph::Incoming).count();
|
||||
let a_outgoing = graph.neighbors_directed(node_a, petgraph::Outgoing).count();
|
||||
(a_incoming + a_outgoing) == 4 && a_neighbors.len() == 2
|
||||
};
|
||||
|
||||
let is_node_b_relay = {
|
||||
let mut b_neighbors = std::collections::HashSet::new();
|
||||
for neighbor in graph.neighbors_directed(node_b, petgraph::Incoming) {
|
||||
b_neighbors.insert(neighbor);
|
||||
}
|
||||
for neighbor in graph.neighbors_directed(node_b, petgraph::Outgoing) {
|
||||
b_neighbors.insert(neighbor);
|
||||
}
|
||||
let b_incoming = graph.neighbors_directed(node_b, petgraph::Incoming).count();
|
||||
let b_outgoing = graph.neighbors_directed(node_b, petgraph::Outgoing).count();
|
||||
(b_incoming + b_outgoing) == 4 && b_neighbors.len() == 2
|
||||
};
|
||||
|
||||
// We want exactly one of the neighbors to be a relay and one to be an endpoint
|
||||
// Case 1: node_a is endpoint, node_b is relay
|
||||
if !is_node_a_relay && is_node_b_relay {
|
||||
// First check that node_a is a meaningful endpoint
|
||||
let mut node_a_neighbors = std::collections::HashSet::new();
|
||||
for neighbor in graph.neighbors_directed(node_a, petgraph::Incoming) {
|
||||
node_a_neighbors.insert(neighbor);
|
||||
}
|
||||
for neighbor in graph.neighbors_directed(node_a, petgraph::Outgoing) {
|
||||
node_a_neighbors.insert(neighbor);
|
||||
}
|
||||
let is_node_a_meaningful = node_a_neighbors.len() >= 3
|
||||
&& !matches!(graph[node_a], GraphMapping::AccessPoint { .. });
|
||||
|
||||
if !is_node_a_meaningful {
|
||||
continue;
|
||||
}
|
||||
|
||||
// Also verify that node_b is a pure relay (exactly 2 unique neighbors, no more)
|
||||
let mut node_b_neighbors = std::collections::HashSet::new();
|
||||
for neighbor in graph.neighbors_directed(node_b, petgraph::Incoming) {
|
||||
node_b_neighbors.insert(neighbor);
|
||||
}
|
||||
for neighbor in graph.neighbors_directed(node_b, petgraph::Outgoing) {
|
||||
node_b_neighbors.insert(neighbor);
|
||||
}
|
||||
if node_b_neighbors.len() != 2 {
|
||||
continue;
|
||||
}
|
||||
|
||||
// Check if node_b (relay) is an AP with clients - if so, skip
|
||||
if let GraphMapping::AccessPoint { id, .. } = &graph[node_b]
|
||||
&& aps_with_clients.contains(id)
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
// Find the other neighbor of node_b (the endpoint on the far side)
|
||||
let mut b_neighbors = std::collections::HashSet::new();
|
||||
for neighbor in graph.neighbors_directed(node_b, petgraph::Incoming) {
|
||||
b_neighbors.insert(neighbor);
|
||||
}
|
||||
for neighbor in graph.neighbors_directed(node_b, petgraph::Outgoing) {
|
||||
b_neighbors.insert(neighbor);
|
||||
}
|
||||
b_neighbors.remove(&relay_node); // Remove the current relay node
|
||||
|
||||
if b_neighbors.len() == 1 {
|
||||
let endpoint_b = *b_neighbors.iter().next().unwrap();
|
||||
// Verify endpoint_b is not a relay
|
||||
let mut endpoint_b_neighbors = std::collections::HashSet::new();
|
||||
for neighbor in graph.neighbors_directed(endpoint_b, petgraph::Incoming) {
|
||||
endpoint_b_neighbors.insert(neighbor);
|
||||
}
|
||||
for neighbor in graph.neighbors_directed(endpoint_b, petgraph::Outgoing) {
|
||||
endpoint_b_neighbors.insert(neighbor);
|
||||
}
|
||||
let endpoint_b_degree = graph
|
||||
.neighbors_directed(endpoint_b, petgraph::Incoming)
|
||||
.count()
|
||||
+ graph
|
||||
.neighbors_directed(endpoint_b, petgraph::Outgoing)
|
||||
.count();
|
||||
|
||||
// Endpoint must not be a relay AND must have sufficient connections to be meaningful
|
||||
// Also prefer Sites over AccessPoints as endpoints
|
||||
let is_meaningful_endpoint = !(endpoint_b_degree == 4 && endpoint_b_neighbors.len() == 2)
|
||||
&& endpoint_b_neighbors.len() >= 3 // Must connect to at least 3 other nodes
|
||||
&& !matches!(graph[endpoint_b], GraphMapping::AccessPoint { .. }); // Avoid APs as endpoints
|
||||
if is_meaningful_endpoint {
|
||||
// Check do_not_squash_sites - skip if any node in the chain is in the exclusion list
|
||||
let do_not_squash = &config
|
||||
.uisp_integration
|
||||
.do_not_squash_sites
|
||||
.clone()
|
||||
.unwrap_or_default();
|
||||
let node_names = [
|
||||
&graph[node_a].name(),
|
||||
&graph[relay_node].name(),
|
||||
&graph[node_b].name(),
|
||||
&graph[endpoint_b].name(),
|
||||
];
|
||||
|
||||
if node_names.iter().any(|name| do_not_squash.contains(*name)) {
|
||||
continue;
|
||||
}
|
||||
|
||||
candidates.push(SquashCandidate {
|
||||
endpoint_a: node_a,
|
||||
endpoint_a_name: graph[node_a].name(),
|
||||
relay_a: relay_node,
|
||||
relay_a_name: graph[relay_node].name(),
|
||||
relay_b: node_b,
|
||||
relay_b_name: graph[node_b].name(),
|
||||
endpoint_b,
|
||||
endpoint_b_name: graph[endpoint_b].name(),
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Case 2: node_a is relay, node_b is endpoint
|
||||
if is_node_a_relay && !is_node_b_relay {
|
||||
// First check that node_b is a meaningful endpoint
|
||||
let mut node_b_neighbors = std::collections::HashSet::new();
|
||||
for neighbor in graph.neighbors_directed(node_b, petgraph::Incoming) {
|
||||
node_b_neighbors.insert(neighbor);
|
||||
}
|
||||
for neighbor in graph.neighbors_directed(node_b, petgraph::Outgoing) {
|
||||
node_b_neighbors.insert(neighbor);
|
||||
}
|
||||
let is_node_b_meaningful = node_b_neighbors.len() >= 3
|
||||
&& !matches!(graph[node_b], GraphMapping::AccessPoint { .. });
|
||||
|
||||
if !is_node_b_meaningful {
|
||||
continue;
|
||||
}
|
||||
|
||||
// Also verify that node_a is a pure relay (exactly 2 unique neighbors, no more)
|
||||
let mut node_a_neighbors = std::collections::HashSet::new();
|
||||
for neighbor in graph.neighbors_directed(node_a, petgraph::Incoming) {
|
||||
node_a_neighbors.insert(neighbor);
|
||||
}
|
||||
for neighbor in graph.neighbors_directed(node_a, petgraph::Outgoing) {
|
||||
node_a_neighbors.insert(neighbor);
|
||||
}
|
||||
if node_a_neighbors.len() != 2 {
|
||||
continue;
|
||||
}
|
||||
|
||||
// Check if node_a (relay) is an AP with clients - if so, skip
|
||||
if let GraphMapping::AccessPoint { id, .. } = &graph[node_a]
|
||||
&& aps_with_clients.contains(id)
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
// Find the other neighbor of node_a (the endpoint on the far side)
|
||||
let mut a_neighbors = std::collections::HashSet::new();
|
||||
for neighbor in graph.neighbors_directed(node_a, petgraph::Incoming) {
|
||||
a_neighbors.insert(neighbor);
|
||||
}
|
||||
for neighbor in graph.neighbors_directed(node_a, petgraph::Outgoing) {
|
||||
a_neighbors.insert(neighbor);
|
||||
}
|
||||
a_neighbors.remove(&relay_node); // Remove the current relay node
|
||||
|
||||
if a_neighbors.len() == 1 {
|
||||
let endpoint_a = *a_neighbors.iter().next().unwrap();
|
||||
// Verify endpoint_a is not a relay
|
||||
let mut endpoint_a_neighbors = std::collections::HashSet::new();
|
||||
for neighbor in graph.neighbors_directed(endpoint_a, petgraph::Incoming) {
|
||||
endpoint_a_neighbors.insert(neighbor);
|
||||
}
|
||||
for neighbor in graph.neighbors_directed(endpoint_a, petgraph::Outgoing) {
|
||||
endpoint_a_neighbors.insert(neighbor);
|
||||
}
|
||||
let endpoint_a_degree = graph
|
||||
.neighbors_directed(endpoint_a, petgraph::Incoming)
|
||||
.count()
|
||||
+ graph
|
||||
.neighbors_directed(endpoint_a, petgraph::Outgoing)
|
||||
.count();
|
||||
|
||||
// Endpoint must not be a relay AND must have sufficient connections to be meaningful
|
||||
// Also prefer Sites over AccessPoints as endpoints
|
||||
let is_meaningful_endpoint = !(endpoint_a_degree == 4 && endpoint_a_neighbors.len() == 2)
|
||||
&& endpoint_a_neighbors.len() >= 3 // Must connect to at least 3 other nodes
|
||||
&& !matches!(graph[endpoint_a], GraphMapping::AccessPoint { .. }); // Avoid APs as endpoints
|
||||
if is_meaningful_endpoint {
|
||||
// Check do_not_squash_sites - skip if any node in the chain is in the exclusion list
|
||||
let do_not_squash = &config
|
||||
.uisp_integration
|
||||
.do_not_squash_sites
|
||||
.clone()
|
||||
.unwrap_or_default();
|
||||
let node_names = [
|
||||
&graph[endpoint_a].name(),
|
||||
&graph[node_a].name(),
|
||||
&graph[relay_node].name(),
|
||||
&graph[node_b].name(),
|
||||
];
|
||||
|
||||
if node_names.iter().any(|name| do_not_squash.contains(*name)) {
|
||||
continue;
|
||||
}
|
||||
|
||||
candidates.push(SquashCandidate {
|
||||
endpoint_a,
|
||||
endpoint_a_name: graph[endpoint_a].name(),
|
||||
relay_a: node_a,
|
||||
relay_a_name: graph[node_a].name(),
|
||||
relay_b: relay_node,
|
||||
relay_b_name: graph[relay_node].name(),
|
||||
endpoint_b: node_b,
|
||||
endpoint_b_name: graph[node_b].name(),
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Remove duplicates (same chain detected from both relay nodes)
|
||||
candidates.dedup_by(|a, b| {
|
||||
(a.endpoint_a == b.endpoint_a && a.endpoint_b == b.endpoint_b)
|
||||
|| (a.endpoint_a == b.endpoint_b && a.endpoint_b == b.endpoint_a)
|
||||
});
|
||||
let candidates = collect_point_to_point_squash_candidates(graph, aps_with_clients, config);
|
||||
|
||||
info!(
|
||||
"Found {} point-to-point squash candidates:",
|
||||
@@ -1097,6 +938,117 @@ fn find_point_to_point_squash_candidates(
|
||||
perform_squashing(graph, &candidates);
|
||||
}
|
||||
|
||||
fn collect_point_to_point_squash_candidates(
|
||||
graph: &GraphType,
|
||||
aps_with_clients: &HashSet<String>,
|
||||
config: &Arc<Config>,
|
||||
) -> Vec<SquashCandidate> {
|
||||
let mut candidates = Vec::new();
|
||||
|
||||
// Find all nodes with exactly total degree 4 in bidirectional graph (2 unique neighbors)
|
||||
// This accounts for bidirectional edges: A->B and B->A both exist
|
||||
let mut relay_nodes: Vec<NodeIndex> = graph
|
||||
.node_indices()
|
||||
.filter(|&node| is_relay_node(graph, node))
|
||||
.collect();
|
||||
relay_nodes.sort_unstable_by_key(|node| stable_node_key(graph, *node));
|
||||
|
||||
// For each potential relay node, check if it's part of a 2-relay chain
|
||||
for &relay_node in &relay_nodes {
|
||||
// Skip if this relay node is an AP with clients
|
||||
if let GraphMapping::AccessPoint { id, .. } = &graph[relay_node]
|
||||
&& aps_with_clients.contains(id)
|
||||
{
|
||||
continue;
|
||||
}
|
||||
let neighbors = sorted_unique_neighbors(graph, relay_node);
|
||||
if neighbors.len() != 2 {
|
||||
continue;
|
||||
}
|
||||
let node_a = neighbors[0];
|
||||
let node_b = neighbors[1];
|
||||
|
||||
// Check if each neighbor is a relay or endpoint in bidirectional context
|
||||
let is_node_a_relay = is_relay_node(graph, node_a);
|
||||
let is_node_b_relay = is_relay_node(graph, node_b);
|
||||
|
||||
// We want exactly one of the neighbors to be a relay and one to be an endpoint
|
||||
// Case 1: node_a is endpoint, node_b is relay
|
||||
if !is_node_a_relay && is_node_b_relay {
|
||||
if !is_meaningful_endpoint(graph, node_a) {
|
||||
continue;
|
||||
}
|
||||
|
||||
if sorted_unique_neighbors(graph, node_b).len() != 2 {
|
||||
continue;
|
||||
}
|
||||
|
||||
// Check if node_b (relay) is an AP with clients - if so, skip
|
||||
if let GraphMapping::AccessPoint { id, .. } = &graph[node_b]
|
||||
&& aps_with_clients.contains(id)
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
// Find the other neighbor of node_b (the endpoint on the far side)
|
||||
let candidate_neighbors: Vec<NodeIndex> = sorted_unique_neighbors(graph, node_b)
|
||||
.into_iter()
|
||||
.filter(|neighbor| *neighbor != relay_node)
|
||||
.collect();
|
||||
|
||||
if candidate_neighbors.len() == 1 {
|
||||
let endpoint_b = candidate_neighbors[0];
|
||||
if is_meaningful_endpoint(graph, endpoint_b)
|
||||
&& !should_skip_squash(graph, config, [node_a, relay_node, node_b, endpoint_b])
|
||||
{
|
||||
candidates.push(SquashCandidate::new(
|
||||
graph, node_a, relay_node, node_b, endpoint_b,
|
||||
));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Case 2: node_a is relay, node_b is endpoint
|
||||
if is_node_a_relay && !is_node_b_relay {
|
||||
if !is_meaningful_endpoint(graph, node_b) {
|
||||
continue;
|
||||
}
|
||||
|
||||
if sorted_unique_neighbors(graph, node_a).len() != 2 {
|
||||
continue;
|
||||
}
|
||||
|
||||
// Check if node_a (relay) is an AP with clients - if so, skip
|
||||
if let GraphMapping::AccessPoint { id, .. } = &graph[node_a]
|
||||
&& aps_with_clients.contains(id)
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
// Find the other neighbor of node_a (the endpoint on the far side)
|
||||
let candidate_neighbors: Vec<NodeIndex> = sorted_unique_neighbors(graph, node_a)
|
||||
.into_iter()
|
||||
.filter(|neighbor| *neighbor != relay_node)
|
||||
.collect();
|
||||
|
||||
if candidate_neighbors.len() == 1 {
|
||||
let endpoint_a = candidate_neighbors[0];
|
||||
if is_meaningful_endpoint(graph, endpoint_a)
|
||||
&& !should_skip_squash(graph, config, [endpoint_a, node_a, relay_node, node_b])
|
||||
{
|
||||
candidates.push(SquashCandidate::new(
|
||||
graph, endpoint_a, node_a, relay_node, node_b,
|
||||
));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
candidates.sort_unstable_by_key(SquashCandidate::canonical_key);
|
||||
candidates.dedup_by(|a, b| a.canonical_key() == b.canonical_key());
|
||||
candidates
|
||||
}
|
||||
|
||||
fn perform_squashing(graph: &mut GraphType, candidates: &[SquashCandidate]) {
|
||||
let mut nodes_to_remove = std::collections::HashSet::new();
|
||||
|
||||
@@ -1161,6 +1113,8 @@ fn perform_squashing(graph: &mut GraphType, candidates: &[SquashCandidate]) {
|
||||
|
||||
// Remove all relay nodes that are no longer needed
|
||||
let removed_count = nodes_to_remove.len();
|
||||
let mut nodes_to_remove: Vec<NodeIndex> = nodes_to_remove.into_iter().collect();
|
||||
nodes_to_remove.sort_unstable_by_key(|node| std::cmp::Reverse(node.index()));
|
||||
for node_to_remove in nodes_to_remove {
|
||||
if graph.node_weight(node_to_remove).is_some() {
|
||||
let node_name = graph[node_to_remove].name();
|
||||
@@ -1212,3 +1166,163 @@ fn calculate_chain_capacity(graph: &GraphType, chain_nodes: &[NodeIndex]) -> (u6
|
||||
|
||||
(forward_capacity, reverse_capacity)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
fn add_site(graph: &mut GraphType, name: &str, id: &str) -> NodeIndex {
|
||||
graph.add_node(GraphMapping::Site {
|
||||
name: name.to_string(),
|
||||
id: id.to_string(),
|
||||
latitude: None,
|
||||
longitude: None,
|
||||
})
|
||||
}
|
||||
|
||||
fn add_ap(graph: &mut GraphType, name: &str, id: &str, site_name: &str) -> NodeIndex {
|
||||
graph.add_node(GraphMapping::AccessPoint {
|
||||
name: name.to_string(),
|
||||
id: id.to_string(),
|
||||
site_name: site_name.to_string(),
|
||||
download_mbps: 1000,
|
||||
upload_mbps: 1000,
|
||||
})
|
||||
}
|
||||
|
||||
fn add_bidirectional_edge(graph: &mut GraphType, a: NodeIndex, b: NodeIndex, speed_mbps: u64) {
|
||||
graph.add_edge(a, b, LinkMapping::ethernet(speed_mbps));
|
||||
graph.add_edge(b, a, LinkMapping::ethernet(speed_mbps));
|
||||
}
|
||||
|
||||
fn build_point_to_point_graph(reverse_insertion: bool) -> GraphType {
|
||||
let mut graph = GraphType::new();
|
||||
|
||||
let mut add_node = |label: &str| match label {
|
||||
"left_endpoint" => add_site(&mut graph, "Left Endpoint", "site-left"),
|
||||
"right_endpoint" => add_site(&mut graph, "Right Endpoint", "site-right"),
|
||||
"left_extra_a" => add_site(&mut graph, "Left Extra A", "site-left-extra-a"),
|
||||
"left_extra_b" => add_site(&mut graph, "Left Extra B", "site-left-extra-b"),
|
||||
"right_extra_a" => add_site(&mut graph, "Right Extra A", "site-right-extra-a"),
|
||||
"right_extra_b" => add_site(&mut graph, "Right Extra B", "site-right-extra-b"),
|
||||
"relay_a" => add_ap(&mut graph, "Relay A", "device-relay-a", "Left Endpoint"),
|
||||
"relay_b" => add_ap(&mut graph, "Relay B", "device-relay-b", "Right Endpoint"),
|
||||
_ => unreachable!("unknown node label"),
|
||||
};
|
||||
|
||||
let ordered_labels = if reverse_insertion {
|
||||
vec![
|
||||
"right_extra_b",
|
||||
"right_extra_a",
|
||||
"relay_b",
|
||||
"right_endpoint",
|
||||
"left_extra_b",
|
||||
"left_extra_a",
|
||||
"relay_a",
|
||||
"left_endpoint",
|
||||
]
|
||||
} else {
|
||||
vec![
|
||||
"left_endpoint",
|
||||
"relay_a",
|
||||
"left_extra_a",
|
||||
"left_extra_b",
|
||||
"right_endpoint",
|
||||
"relay_b",
|
||||
"right_extra_a",
|
||||
"right_extra_b",
|
||||
]
|
||||
};
|
||||
|
||||
let mut nodes = HashMap::new();
|
||||
for label in ordered_labels {
|
||||
nodes.insert(label, add_node(label));
|
||||
}
|
||||
|
||||
let mut edges = vec![
|
||||
("left_endpoint", "relay_a", 500),
|
||||
("relay_a", "relay_b", 400),
|
||||
("relay_b", "right_endpoint", 300),
|
||||
("left_endpoint", "left_extra_a", 100),
|
||||
("left_endpoint", "left_extra_b", 100),
|
||||
("right_endpoint", "right_extra_a", 100),
|
||||
("right_endpoint", "right_extra_b", 100),
|
||||
];
|
||||
if reverse_insertion {
|
||||
edges.reverse();
|
||||
}
|
||||
|
||||
for (from, to, speed) in edges {
|
||||
add_bidirectional_edge(&mut graph, nodes[from], nodes[to], speed);
|
||||
}
|
||||
|
||||
graph
|
||||
}
|
||||
|
||||
fn graph_node_ids(graph: &GraphType) -> Vec<String> {
|
||||
let mut ids: Vec<String> = graph
|
||||
.node_indices()
|
||||
.map(|node| graph[node].network_json_id())
|
||||
.collect();
|
||||
ids.sort();
|
||||
ids
|
||||
}
|
||||
|
||||
fn find_node_by_id(graph: &GraphType, id: &str) -> NodeIndex {
|
||||
graph
|
||||
.node_indices()
|
||||
.find(|node| graph[*node].network_json_id() == id)
|
||||
.expect("expected node to exist")
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn squash_candidate_collection_is_deterministic() {
|
||||
let config = Arc::new(Config::default());
|
||||
let aps_with_clients = HashSet::new();
|
||||
|
||||
let graph_a = build_point_to_point_graph(false);
|
||||
let graph_b = build_point_to_point_graph(true);
|
||||
|
||||
let candidates_a =
|
||||
collect_point_to_point_squash_candidates(&graph_a, &aps_with_clients, &config);
|
||||
let candidates_b =
|
||||
collect_point_to_point_squash_candidates(&graph_b, &aps_with_clients, &config);
|
||||
|
||||
assert_eq!(candidates_a.len(), 1);
|
||||
assert_eq!(candidates_b.len(), 1);
|
||||
assert_eq!(
|
||||
candidates_a[0].canonical_key(),
|
||||
candidates_b[0].canonical_key()
|
||||
);
|
||||
assert_eq!(candidates_a[0].endpoint_a_id, "uisp:site:site-left");
|
||||
assert_eq!(candidates_a[0].relay_a_id, "uisp:device:device-relay-a");
|
||||
assert_eq!(candidates_a[0].relay_b_id, "uisp:device:device-relay-b");
|
||||
assert_eq!(candidates_a[0].endpoint_b_id, "uisp:site:site-right");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn squashing_result_is_stable_across_insertion_order() {
|
||||
let config = Arc::new(Config::default());
|
||||
let aps_with_clients = HashSet::new();
|
||||
|
||||
let mut graph_a = build_point_to_point_graph(false);
|
||||
let mut graph_b = build_point_to_point_graph(true);
|
||||
|
||||
find_point_to_point_squash_candidates(&mut graph_a, &aps_with_clients, &config);
|
||||
find_point_to_point_squash_candidates(&mut graph_b, &aps_with_clients, &config);
|
||||
|
||||
assert_eq!(graph_node_ids(&graph_a), graph_node_ids(&graph_b));
|
||||
assert!(!graph_node_ids(&graph_a).contains(&"uisp:device:device-relay-a".to_string()));
|
||||
assert!(!graph_node_ids(&graph_a).contains(&"uisp:device:device-relay-b".to_string()));
|
||||
|
||||
let left_a = find_node_by_id(&graph_a, "uisp:site:site-left");
|
||||
let right_a = find_node_by_id(&graph_a, "uisp:site:site-right");
|
||||
let left_b = find_node_by_id(&graph_b, "uisp:site:site-left");
|
||||
let right_b = find_node_by_id(&graph_b, "uisp:site:site-right");
|
||||
|
||||
assert!(graph_a.find_edge(left_a, right_a).is_some());
|
||||
assert!(graph_a.find_edge(right_a, left_a).is_some());
|
||||
assert!(graph_b.find_edge(left_b, right_b).is_some());
|
||||
assert!(graph_b.find_edge(right_b, left_b).is_some());
|
||||
}
|
||||
}
|
||||
+11
-10
@@ -11,7 +11,7 @@ from io import StringIO
|
||||
from liblqos_python import automatic_import_uisp, automatic_import_splynx, queue_refresh_interval_mins, \
|
||||
automatic_import_powercode, automatic_import_sonar, influx_db_enabled, get_libreqos_directory, \
|
||||
blackboard_finish, blackboard_submit, automatic_import_wispgate, enable_insight_topology, insight_topology_role, \
|
||||
automatic_import_netzur, automatic_import_visp, calculate_hash, efficiency_core_ids, scheduler_alive, scheduler_error, \
|
||||
automatic_import_netzur, automatic_import_visp, calculate_shaping_runtime_hash, efficiency_core_ids, scheduler_alive, scheduler_error, \
|
||||
overrides_persistent_devices_materialized, overrides_circuit_adjustments_materialized, \
|
||||
overrides_network_adjustments_materialized, \
|
||||
scheduler_output, wait_for_bus_ready
|
||||
@@ -22,7 +22,7 @@ import os.path
|
||||
import os
|
||||
|
||||
ads = BlockingScheduler(executors={'default': ThreadPoolExecutor(1)})
|
||||
network_hash = 0
|
||||
shaping_runtime_hash = 0
|
||||
|
||||
|
||||
def clear_scheduler_error():
|
||||
@@ -594,16 +594,17 @@ def importAndShapeFullReload():
|
||||
|
||||
|
||||
def importAndShapePartialReload():
|
||||
global network_hash
|
||||
global shaping_runtime_hash
|
||||
|
||||
importFromCRM()
|
||||
# Calculate if the network.json or ShapedDevices.csv has changed and reload only if it has.
|
||||
new_hash = calculate_hash()
|
||||
if new_hash != network_hash:
|
||||
refreshShapersUpdateOnly()
|
||||
network_hash = new_hash
|
||||
# Rebuild when runtime shaping inputs change, including effective adaptive
|
||||
# circuit overrides that do not belong in source-of-truth files.
|
||||
new_hash = calculate_shaping_runtime_hash()
|
||||
if new_hash != shaping_runtime_hash:
|
||||
refreshShapers()
|
||||
shaping_runtime_hash = calculate_shaping_runtime_hash()
|
||||
else:
|
||||
print("No changes detected in network.json or ShapedDevices.csv, skipping shaper refresh.")
|
||||
print("No runtime shaping changes detected, skipping shaper refresh.")
|
||||
|
||||
|
||||
def not_dead_yet():
|
||||
@@ -619,7 +620,7 @@ if __name__ == '__main__':
|
||||
try:
|
||||
ensure_bus_ready()
|
||||
importAndShapeFullReload()
|
||||
network_hash = calculate_hash()
|
||||
shaping_runtime_hash = calculate_shaping_runtime_hash()
|
||||
|
||||
print("Starting scheduler with jobs:")
|
||||
print(f"- not_dead_yet every 1 minute")
|
||||
|
||||
@@ -27,13 +27,16 @@ def install_scheduler_stubs():
|
||||
lqlib.automatic_import_netzur = lambda: False
|
||||
lqlib.automatic_import_visp = lambda: False
|
||||
lqlib.calculate_hash = lambda: 0
|
||||
lqlib.calculate_shaping_runtime_hash = lambda: 0
|
||||
lqlib.efficiency_core_ids = lambda: []
|
||||
lqlib.scheduler_alive = Mock()
|
||||
lqlib.scheduler_error = Mock()
|
||||
lqlib.scheduler_output = 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: []
|
||||
sys.modules["liblqos_python"] = lqlib
|
||||
@@ -266,10 +269,10 @@ class TestSchedulerOverrideMerge(unittest.TestCase):
|
||||
|
||||
with patch.object(scheduler, "shaped_devices_csv_path", return_value="/tmp/ShapedDevices.csv"):
|
||||
with patch.object(scheduler, "read_shaped_devices_csv", return_value=(header, rows)):
|
||||
with patch.object(scheduler, "overrides_persistent_devices_effective", return_value=[]):
|
||||
with patch.object(scheduler, "overrides_persistent_devices_materialized", return_value=[]):
|
||||
with patch.object(
|
||||
scheduler,
|
||||
"overrides_circuit_adjustments_effective",
|
||||
"overrides_circuit_adjustments_materialized",
|
||||
return_value=[{
|
||||
"type": "device_adjust_sqm",
|
||||
"device_id": "splynx_service_93",
|
||||
|
||||
Reference in new issue
Block a user