diff --git a/docs/v2.0/configuration-advanced.md b/docs/v2.0/configuration-advanced.md index 57f5a68b..a45c8de9 100644 --- a/docs/v2.0/configuration-advanced.md +++ b/docs/v2.0/configuration-advanced.md @@ -175,6 +175,8 @@ LibreQoS currently uses `queue_mode` in the `[queues]` section to control whethe - `queue_mode = "shape"`: normal shaping mode - `queue_mode = "observe"`: remove the subscriber shaping tree for a true baseline while keeping the root MQ in place +Switching from `shape` to `observe` cancels any in-flight Bakery live queue-migration verification so the Observe transition does not surface stale queue-tree cleanup as Bakery errors. IP mapping and per-circuit traffic attribution continue to work in Observe mode. + The older `monitor_only` setting is retained as a compatibility alias in some configs and serialized output, but `queue_mode` is the current operator-facing setting and documentation term. If you plan to use the built-in UISP, Splynx, or Netzur integrations, you do not need to create a network.json file quite yet. diff --git a/src/LibreQoS.py b/src/LibreQoS.py index ee413e49..3b74f14c 100755 --- a/src/LibreQoS.py +++ b/src/LibreQoS.py @@ -224,25 +224,39 @@ def _shaped_devices_row_value(row, layout, field): def _resolve_effective_parent_node(circuit, parent_node_ids, parent_node_aliases): - parent_node_id = str(circuit.get('ParentNodeID', '') or '').strip() - if parent_node_id: + for candidate_id in ( + circuit.get('ParentNodeID', ''), + circuit.get('effectiveAttachmentID', ''), + ): + parent_node_id = str(candidate_id or '').strip() + if not parent_node_id: + continue resolved_parent = parent_node_ids.get(parent_node_id) if resolved_parent: return resolved_parent, parent_node_id - parent_node = str(circuit.get('ParentNode', '') or '').strip() - if not parent_node or parent_node == 'none': - return parent_node, '' - - resolved_parent = parent_node_aliases.get(parent_node) - if resolved_parent: - resolved_parent_id = '' + resolved_parent_id = '' + for candidate_name in ( + circuit.get('ParentNode', ''), + circuit.get('effectiveAttachmentName', ''), + circuit.get('logicalParentNode', ''), + ): + parent_node = str(candidate_name or '').strip() + if not parent_node or parent_node == 'none': + continue + resolved_parent = parent_node_aliases.get(parent_node) + if not resolved_parent: + continue for node_id, node_name in parent_node_ids.items(): if node_name == resolved_parent: resolved_parent_id = node_id break return resolved_parent, resolved_parent_id + parent_node = str(circuit.get('ParentNode', '') or '').strip() + parent_node_id = str(circuit.get('ParentNodeID', '') or '').strip() + if not parent_node or parent_node == 'none': + return parent_node, '' return parent_node, parent_node_id @@ -360,10 +374,10 @@ def load_planner_state(state_path=None, planner_module=None): try: state = planner_module.load_state(state_path) if isinstance(state, dict): - return state + return sanitize_planner_state(state) except Exception: pass - return _load_json_dict(state_path) + return sanitize_planner_state(_load_json_dict(state_path)) def save_planner_state(state, state_path=None, planner_module=None): @@ -382,6 +396,56 @@ def save_planner_state(state, state_path=None, planner_module=None): os.replace(temp_path, state_path) +def sanitize_planner_state(state): + if not isinstance(state, dict): + return {} + + sanitized = False + + def _sanitize_entry(entry): + nonlocal sanitized + if not isinstance(entry, dict): + return + + raw_minor = None + for key in ('class_minor', 'minor', 'classMinor'): + if key in entry: + raw_minor = _parse_int_token(entry.get(key)) + if raw_minor is not None: + break + + # TC minor 0xffff is not safe for leaf classes here. If stale planner + # state reuses it, Bakery later emits qdisc parents like 1:ffff and the + # kernel rejects them. Drop the stored identity and let the planner + # allocate a new minor on this run. + if raw_minor is not None and (raw_minor < 3 or raw_minor >= 0xFFFF): + for key in ('class_minor', 'minor', 'classMinor', 'class_major', 'up_class_major'): + if key in entry: + del entry[key] + sanitized = True + + for value in entry.values(): + if isinstance(value, dict): + _sanitize_entry(value) + elif isinstance(value, list): + for item in value: + if isinstance(item, dict): + _sanitize_entry(item) + + state_copy = json.loads(json.dumps(state)) + for section in ('sites', 'circuits'): + section_value = state_copy.get(section) + if isinstance(section_value, dict): + _sanitize_entry(section_value) + + if sanitized: + logging.warning( + "Sanitized invalid planner state identities; removed reserved or out-of-range minor assignments before planning." + ) + + return state_copy + + def _parse_int_token(value): if value is None: return None @@ -1479,10 +1543,12 @@ def refreshShapers(): parent_node_ids = collect_physical_parent_node_ids(network) parent_node_aliases = collect_physical_parent_node_aliases(network) for circuit in subscriberCircuits: - if circuit.get('parentResolvedByShapingInputs') == True: - continue parent_node = str(circuit.get('ParentNode', '') or '').strip() - if not parent_node or parent_node == 'none': + if ( + (not parent_node or parent_node == 'none') + and not str(circuit.get('effectiveAttachmentName', '') or '').strip() + and not str(circuit.get('effectiveAttachmentID', '') or '').strip() + ): continue resolved_parent, resolved_parent_id = _resolve_effective_parent_node( circuit, @@ -1512,23 +1578,21 @@ def refreshShapers(): elif resolved_parent_id: circuit['ParentNodeID'] = resolved_parent_id - # Group circuits by parent node. Reduces runtime for section below this one. - circuits_by_parent_node = {} - circuit_min_down_combined_by_parent_node = {} - circuit_min_up_combined_by_parent_node = {} + # Group circuits by stable parent identity first, then by names as fallback. + circuits_by_parent_id = {} + circuits_by_parent_name = {} for circuit in subscriberCircuits: - #If a device from ShapedDevices.csv lists this node as its Parent Node, attach it as a leaf to this node HTB - if circuit['ParentNode'] not in circuits_by_parent_node: - circuits_by_parent_node[circuit['ParentNode']] = [] - temp = circuits_by_parent_node[circuit['ParentNode']] - temp.append(circuit) - circuits_by_parent_node[circuit['ParentNode']] = temp - if circuit['ParentNode'] not in circuit_min_down_combined_by_parent_node: - circuit_min_down_combined_by_parent_node[circuit['ParentNode']] = 0 - circuit_min_down_combined_by_parent_node[circuit['ParentNode']] += circuit['minDownload'] - if circuit['ParentNode'] not in circuit_min_up_combined_by_parent_node: - circuit_min_up_combined_by_parent_node[circuit['ParentNode']] = 0 - circuit_min_up_combined_by_parent_node[circuit['ParentNode']] += circuit['minUpload'] + parent_id = str(circuit.get('effectiveParentNodeID', '') or circuit.get('ParentNodeID', '') or '').strip() + if parent_id: + circuits_by_parent_id.setdefault(parent_id, []).append(circuit) + + for parent_name in { + str(circuit.get('ParentNode', '') or '').strip(), + str(circuit.get('effectiveParentNodeName', '') or '').strip(), + str(circuit.get('logicalParentNode', '') or '').strip(), + }: + if parent_name: + circuits_by_parent_name.setdefault(parent_name, []).append(circuit) # Parse network structure and add devices from ShapedDevices.csv print("Parsing network structure and tallying devices") @@ -1552,7 +1616,7 @@ def refreshShapers(): raise ValueError(msg) def ensure_minor_capacity(queue, minor): - if minor > 0xFFFF: + if minor >= 0xFFFF: report_minor_overflow(queue, minor) def next_free_minor(start_minor, reserved): @@ -1792,9 +1856,23 @@ def refreshShapers(): "has_children": has_children, } ) - if node in circuits_by_parent_node: + node_id = str(data[node].get('id', '') or '').strip() + node_name = str(data[node].get('name', '') or '').strip() + selected_circuits = [] + if node_id and node_id in circuits_by_parent_id: + selected_circuits = list(circuits_by_parent_id[node_id]) + else: + seen_circuit_ids = set() + for candidate in (node, node_name): + for circuit in circuits_by_parent_name.get(candidate, []): + circuit_id = planner_circuit_identity_key(circuit) + if circuit_id in seen_circuit_ids: + continue + selected_circuits.append(circuit) + seen_circuit_ids.add(circuit_id) + if selected_circuits: sorted_circuits = sorted( - circuits_by_parent_node[node], + selected_circuits, key=lambda c: c.get('circuitName', c.get('circuitID', '')), ) planner_circuit_groups.append( @@ -1804,7 +1882,6 @@ def refreshShapers(): "circuit_ids": [ planner_circuit_identity_key(circuit) for circuit in sorted_circuits - if node == circuit['ParentNode'] ], } ) @@ -1917,6 +1994,8 @@ def refreshShapers(): parentMinUL=upstream_bandwidth_capacity_upload_mbps(), ) + attached_circuit_ids = set() + def attach_circuits(data, depth, path=()): for node in sorted_node_keys(data, depth): node_data = data[node] @@ -1926,19 +2005,43 @@ def refreshShapers(): continue queue = queue_token + 1 circuitsForThisNetworkNode = [] - if node in circuits_by_parent_node: + + node_id = str(node_data.get('id', '') or '').strip() + node_name = str(node_data.get('name', '') or '').strip() + parent_candidates = [] + for candidate in (node_id, node, node_name): + if candidate and candidate not in parent_candidates: + parent_candidates.append(candidate) + + selected_circuits = [] + if node_id and node_id in circuits_by_parent_id: + selected_circuits = list(circuits_by_parent_id[node_id]) + else: + seen_circuit_ids = set() + for candidate in parent_candidates[1:]: + for circuit in circuits_by_parent_name.get(candidate, []): + circuit_id = str(circuit.get('circuitID', '') or '') + if circuit_id in seen_circuit_ids: + continue + selected_circuits.append(circuit) + seen_circuit_ids.add(circuit_id) + + if selected_circuits: override_min_down = None override_min_up = None - if (circuit_min_down_combined_by_parent_node[node] > node_data['downloadBandwidthMbpsMin']) or (circuit_min_up_combined_by_parent_node[node] > node_data['uploadBandwidthMbpsMin']): + combined_min_down = sum(float(circuit.get('minDownload', 0) or 0) for circuit in selected_circuits) + combined_min_up = sum(float(circuit.get('minUpload', 0) or 0) for circuit in selected_circuits) + if (combined_min_down > node_data['downloadBandwidthMbpsMin']) or (combined_min_up > node_data['uploadBandwidthMbpsMin']): override_min_down = 1 override_min_up = 1 logging.info("The combined minimums of circuits in Parent Node [" + node + "] exceeded that of the parent node. Reducing these circuits' minimums to 1 now.", stacklevel=2) - if ((override_min_down * len(circuits_by_parent_node[node])) > node_data['downloadBandwidthMbpsMin']) or ((override_min_up * len(circuits_by_parent_node[node])) > node_data['uploadBandwidthMbpsMin']): + if ((override_min_down * len(selected_circuits)) > node_data['downloadBandwidthMbpsMin']) or ((override_min_up * len(selected_circuits)) > node_data['uploadBandwidthMbpsMin']): logging.info("Even with this change, minimums will exceed the min rate of the parent node. Using 10 kbps as the minimum for these circuits instead.", stacklevel=2) nodes_requiring_min_squashing[node] = True - sorted_circuits = sorted(circuits_by_parent_node[node], key=lambda c: c.get('circuitName', c.get('circuitID', ''))) + sorted_circuits = sorted(selected_circuits, key=lambda c: c.get('circuitName', c.get('circuitID', ''))) for circuit in sorted_circuits: - if node != circuit['ParentNode']: + circuit_id = str(circuit.get('circuitID', '') or '') + if circuit_id in attached_circuit_ids: continue if circuit['maxDownload'] > node_data['downloadBandwidthMbps']: logging.info("downloadMax of Circuit ID [" + circuit['circuitID'] + "] exceeded that of its parent node. Reducing to that of its parent node now.", stacklevel=2) @@ -1956,6 +2059,7 @@ def refreshShapers(): upFlowIDstring = hex(up_major) + ':' + hex(candidate_minor) circuit['classid'] = flowIDstring circuit['up_classid'] = upFlowIDstring + attached_circuit_ids.add(circuit_id) logging.info("Added up_classid to circuit: " + circuit['up_classid']) maxDownload = min(circuit['maxDownload'], node_data['downloadBandwidthMbps']) maxUpload = min(circuit['maxUpload'], node_data['uploadBandwidthMbps']) @@ -2018,6 +2122,25 @@ def refreshShapers(): attach_circuits(sorted_children, depth+1, path + (node,)) attach_circuits(network, 0) + + unattached_circuits = [] + for circuit in subscriberCircuits: + circuit_id = str(circuit.get('circuitID', '') or '') + if circuit_id and circuit_id not in attached_circuit_ids: + unattached_circuits.append( + f"{circuit_id} ({circuit.get('circuitName', '')}) parent={circuit.get('ParentNode', '')} parent_id={circuit.get('effectiveParentNodeID', circuit.get('ParentNodeID', ''))}" + ) + if unattached_circuits: + warnings.warn( + "Some shaped circuits did not attach to the live queue tree. First examples: " + "; ".join(unattached_circuits[:20]), + stacklevel=2, + ) + logging.warning( + "Unattached shaped circuits after queue build: %s total. Examples: %s", + len(unattached_circuits), + "; ".join(unattached_circuits[:20]), + ) + minorByCPU = { int(queue): int(minor) for queue, minor in (identity_plan.get("last_used_minor_by_queue", {}) or {}).items() diff --git a/src/rust/lqos_bakery/src/lib.rs b/src/rust/lqos_bakery/src/lib.rs index 77421235..ce78157e 100644 --- a/src/rust/lqos_bakery/src/lib.rs +++ b/src/rust/lqos_bakery/src/lib.rs @@ -1457,6 +1457,33 @@ fn mark_reload_required(summary: String) { } } +fn is_live_migration_reload_required_reason(reason: &str) -> bool { + reason.starts_with("Bakery live-move ") || reason.starts_with("Bakery live migration ") +} + +fn cancel_pending_migrations_for_observe_mode( + migrations: &mut HashMap, + reason: &str, +) -> usize { + let canceled = migrations.len(); + if canceled > 0 { + let summary = format!("Bakery canceled {canceled} pending live migration(s): {reason}"); + info!("{summary}"); + push_bakery_event("live_migrations_canceled", "info", summary); + migrations.clear(); + } + + if let Some(reload_reason) = bakery_reload_required_reason() + && is_live_migration_reload_required_reason(&reload_reason) + { + clear_reload_required( + "Observe mode canceled pending Bakery live-migration verification state; incremental live-move reload requirements were cleared.", + ); + } + + canceled +} + fn clear_reload_required(summary: &str) { let mut should_emit = false; { @@ -3342,6 +3369,14 @@ fn bakery_main(rx: Receiver, tx: Sender) { virtualized_sites: &mut HashMap, runtime_node_operations: &mut HashMap, ) { + if config.queues.queue_mode.is_observe() { + cancel_pending_migrations_for_observe_mode( + migrations, + "queue_mode is observe; the shaping tree is not live.", + ); + return; + } + let mut advanced = 0usize; let mut to_remove = Vec::new(); let mut effective_state_changed = false; @@ -4055,11 +4090,21 @@ fn handle_commit_batch( let effective_new_batch = apply_runtime_virtualization_overlay(raw_batch.clone(), virtualized_sites); let resolved_mq_layout = current_mq_layout(&raw_batch, &config, mq_layout); + let has_mq_been_setup = MQ_CREATED.load(std::sync::atomic::Ordering::Relaxed); + let shaping_tree_active = SHAPING_TREE_ACTIVE.load(std::sync::atomic::Ordering::Relaxed); + let desired_tree_active = desired_shaping_tree_active(&config); let mapped_limit = resolve_mapped_circuit_limit(); let effective_limit = mapped_limit.effective_limit; let limit_label = format_mapped_limit(effective_limit); + if shaping_tree_active && !desired_tree_active { + cancel_pending_migrations_for_observe_mode( + migrations, + "queue mode transitioned to observe; a full reload will rebuild the retained root MQ without the shaping tree.", + ); + } + if let Some(reason) = bakery_reload_required_reason() { let summary = format!( "Bakery full reload triggered by reload-required state: {}", @@ -4102,9 +4147,6 @@ fn handle_commit_batch( return; } - let has_mq_been_setup = MQ_CREATED.load(std::sync::atomic::Ordering::Relaxed); - let shaping_tree_active = SHAPING_TREE_ACTIVE.load(std::sync::atomic::Ordering::Relaxed); - let desired_tree_active = desired_shaping_tree_active(&config); if !has_mq_been_setup { push_bakery_event( "baseline_rebuild_required", @@ -13064,6 +13106,105 @@ mod tests { assert!(bakery_reload_required_reason().is_none()); } + fn sample_test_migration(hash: i64) -> Migration { + Migration { + circuit_hash: hash, + circuit_name: Some(format!("circuit-{hash}")), + site_name: Some(format!("site-{hash}")), + old_class_major: 0x1, + old_up_class_major: 0x2, + old_down_qdisc_handle: Some(0x9000), + old_up_qdisc_handle: Some(0x9001), + parent_class_id: TcHandle::from_u32(0x10034), + up_parent_class_id: TcHandle::from_u32(0x20034), + class_major: 0x1, + up_class_major: 0x2, + down_qdisc_handle: Some(0x9000), + up_qdisc_handle: Some(0x9001), + old_down_min: 1.0, + old_down_max: 10.0, + old_up_min: 1.0, + old_up_max: 10.0, + new_down_min: 1.0, + new_down_max: 20.0, + new_up_min: 1.0, + new_up_max: 20.0, + old_minor: 0x21, + shadow_minor: 0x2000, + final_minor: 0x35, + ips: vec![format!("192.0.2.{hash}/32")], + sqm_override: None, + desired_cmd: mk_test_circuit( + hash, + 0x10034, + 0x20034, + 0x35, + 0x1, + 0x2, + &format!("192.0.2.{hash}/32"), + ), + stage: MigrationStage::VerifyFinalReady, + shadow_verify_attempts: 0, + final_verify_attempts: 0, + } + } + + #[test] + fn observe_transition_cancels_pending_live_migrations_and_clears_live_migration_reload_state() { + let _guard = bakery_test_lock().lock().expect("test lock"); + reset_bakery_test_state(); + clear_reload_required( + "reset before observe_transition_cancels_pending_live_migrations_and_clears_live_migration_reload_state", + ); + + let mut migrations = HashMap::from([(9007, sample_test_migration(9007))]); + mark_reload_required( + "Bakery live-move final verification did not find the expected final class/qdisc for circuit 9007: observed missing. A full reload is now required before further incremental topology mutations." + .to_string(), + ); + + let canceled = cancel_pending_migrations_for_observe_mode( + &mut migrations, + "queue mode transitioned to observe; the shaping tree is no longer live.", + ); + + assert_eq!(canceled, 1); + assert!(migrations.is_empty()); + assert!(bakery_reload_required_reason().is_none()); + assert!(bakery_activity_snapshot().iter().any(|entry| { + entry.event == "live_migrations_canceled" + && entry.summary.contains("pending live migration") + })); + } + + #[test] + fn observe_transition_preserves_non_migration_reload_required_state() { + let _guard = bakery_test_lock().lock().expect("test lock"); + reset_bakery_test_state(); + clear_reload_required( + "reset before observe_transition_preserves_non_migration_reload_required_state", + ); + + let unrelated_reason = "Bakery full reload triggered because site speed live-change enqueue failed: synthetic failure."; + let mut migrations = HashMap::from([(9008, sample_test_migration(9008))]); + mark_reload_required(unrelated_reason.to_string()); + + let canceled = cancel_pending_migrations_for_observe_mode( + &mut migrations, + "queue mode transitioned to observe; the shaping tree is no longer live.", + ); + + assert_eq!(canceled, 1); + assert!(migrations.is_empty()); + assert_eq!( + bakery_reload_required_reason().as_deref(), + Some(unrelated_reason) + ); + clear_reload_required( + "reset after observe_transition_preserves_non_migration_reload_required_state", + ); + } + #[test] fn migration_target_label_includes_circuit_and_site_names_when_known() { let migration = Migration {