Merge branch 'develop' of https://github.com/LibreQoE/LibreQoS into develop

# Conflicts:
#	src/scheduler.py
This commit is contained in:
Name committed 2026-04-22 12:50:44 -06:00
commit 275ef0dd27
21 files changed
+1319 -126

No files matched your search

+43 -2
View File
@@ -1,7 +1,7 @@
# Optimized version of integrationCommon.py for better performance with large datasets
# Provides common functionality shared between integrations.
from typing import List, Any, Dict, Set
from typing import List, Any, Dict, Set, Tuple
from liblqos_python import allowed_subnets, ignore_subnets, generated_pn_download_mbps, generated_pn_upload_mbps, \
circuit_name_use_address, upstream_bandwidth_capacity_download_mbps, upstream_bandwidth_capacity_upload_mbps, \
find_ipv6_using_mikrotik, exclude_sites, bandwidth_overhead_factor, committed_bandwidth_multiplier, \
@@ -32,6 +32,46 @@ def _state_directory():
def _state_path(category, filename):
return os.path.join(_state_directory(), category, filename)
def _normalize_exception_cpes(raw_exception_cpes) -> Tuple[Dict[str, str], List[str]]:
normalized = {}
errors = []
if raw_exception_cpes is None:
return normalized, errors
if isinstance(raw_exception_cpes, dict):
for cpe, parent in raw_exception_cpes.items():
cpe = str(cpe).strip()
parent = str(parent).strip()
if cpe and parent:
normalized[cpe] = parent
return normalized, errors
if not isinstance(raw_exception_cpes, (list, tuple, set)):
errors.append(
f"exception_cpes ignored: expected dict or list, got {type(raw_exception_cpes).__name__}"
)
return normalized, errors
for entry in raw_exception_cpes:
cpe = None
parent = None
if isinstance(entry, str):
if ":" in entry:
cpe, parent = entry.split(":", 1)
else:
errors.append(f"exception_cpes entry ignored: {entry!r} is missing ':'")
continue
elif isinstance(entry, dict):
cpe = entry.get("cpe")
parent = entry.get("parent")
else:
cpe = getattr(entry, "cpe", None)
parent = getattr(entry, "parent", None)
cpe = str(cpe or "").strip()
parent = str(parent or "").strip()
if not cpe or not parent:
errors.append(f"exception_cpes entry ignored: {entry!r} is missing cpe or parent")
continue
normalized[cpe] = parent
return normalized, errors
def isInAllowedSubnets(inputIP):
# Check whether an IP address occurs inside the allowedSubnets list
isAllowed = False
@@ -203,8 +243,9 @@ class NetworkGraph:
parentId="", displayName="Shaper Root")
]
self.excludeSites = exclude_sites()
self.exceptionCPEs = exception_cpes()
self.errors: List[str] = []
self.exceptionCPEs, exception_cpe_errors = _normalize_exception_cpes(exception_cpes())
self.errors.extend(exception_cpe_errors)
# Initialize optimization structures
self._id_to_index = {"FakeRoot": 0}
+94 -3
View File
@@ -1,5 +1,6 @@
from pythonCheck import checkPythonVersion
checkPythonVersion()
import ipaddress
import requests
import subprocess
import time
@@ -65,6 +66,10 @@ def sonarRadiusAccountDeviceNodeId(radius_account_id):
return f"sonar:radius-account:{radius_account_id}"
def sonarAccountIpAssignmentDeviceNodeId(account_id):
return f"sonar:account-ip-assignments:{account_id}"
def sonarSession():
global _SONAR_SESSION
if _SONAR_SESSION is None:
@@ -223,6 +228,23 @@ def dedupeSubnets(subnets):
return deduped
def splitIpVersions(subnets):
ipv4 = []
ipv6 = []
invalid = []
for subnet in subnets or []:
try:
network = ipaddress.ip_network(subnet, strict=False)
except ValueError:
invalid.append(subnet)
continue
if network.version == 4:
ipv4.append(subnet)
else:
ipv6.append(subnet)
return ipv4, ipv6, invalid
def normalizeServiceName(name):
return (name or "").strip().casefold()
@@ -236,6 +258,15 @@ def findRadiusAccountIPs(radius_account):
return dedupeSubnets(ips)
def findAccountIPs(account):
ips = []
for ip in account.get('ip_assignments', {}).get('entities', []):
subnet = ip.get('subnet')
if subnet:
ips.append(subnet)
return dedupeSubnets(ips)
def recurringServiceRateRules():
rules = {}
for enabled, service_name, download_mbps, upload_mbps in sonar_recurring_service_rates():
@@ -413,7 +444,8 @@ def buildAccountDevices(account):
'raw_id': item['id'],
'name': item['inventory_model']['name'],
'ips': ips,
'mac': findPrimaryMac(item)
'mac': findPrimaryMac(item),
'source': 'inventory_item',
})
for radius_account in account.get('radius_accounts', {}).get('entities', []):
@@ -429,7 +461,23 @@ def buildAccountDevices(account):
'raw_id': radius_account['id'],
'name': f"Radius Account {radius_account['id']}",
'ips': radius_ips,
'mac': ''
'mac': '',
'source': 'radius_account',
})
account_ips = [
subnet for subnet in findAccountIPs(account)
if subnet not in known_ip_subnets
]
if account_ips:
known_ip_subnets.update(account_ips)
devices.append({
'id': sonarAccountIpAssignmentDeviceNodeId(account['id']),
'raw_id': account['id'],
'name': 'Account IP Assignments',
'ips': account_ips,
'mac': '',
'source': 'account_ip_assignments',
})
return devices
@@ -520,6 +568,11 @@ def getAccounts(sonar_active_status_ids):
account_status_id
id
name
ip_assignments {
entities {
subnet
}
}
account_services (reverse_relation_filters: [$data]) {
entities {
service {
@@ -580,6 +633,11 @@ def getAccounts(sonar_active_status_ids):
entities {
id
name
ip_assignments {
entities {
subnet
}
}
account_services (reverse_relation_filters: [$data]) {
entities {
service {
@@ -729,6 +787,34 @@ def buildMacToApMap(aps):
mac_to_ap[cpe_mac] = ap['id']
return mac_to_ap
def sonarImportSummary(sites, aps, accounts):
total_devices = 0
devices_with_ips = 0
ipv4_assignments = 0
ipv6_assignments = 0
account_level_ip_devices = 0
accounts_with_parent = 0
for account in accounts:
if account.get('parent'):
accounts_with_parent += 1
for device in account.get('devices', []):
total_devices += 1
if len(device.get('ips') or []) > 0:
devices_with_ips += 1
ipv4, ipv6, _invalid = splitIpVersions(device.get('ips', []))
ipv4_assignments += len(ipv4)
ipv6_assignments += len(ipv6)
if device.get('source') == 'account_ip_assignments':
account_level_ip_devices += 1
return (
f"Sonar import summary: sites={len(sites)} aps={len(aps)} "
f"accounts={len(accounts)} accounts_with_ap_parent={accounts_with_parent} "
f"devices={total_devices} devices_with_ips={devices_with_ips} "
f"ipv4_assignments={ipv4_assignments} ipv6_assignments={ipv6_assignments} "
f"account_level_ip_devices={account_level_ip_devices}"
)
def createShaper():
net = NetworkGraph()
@@ -757,6 +843,8 @@ def createShaper():
if account['parent']:
break
print(sonarImportSummary(sites, aps, accounts))
for site in sites:
net.addRawNode(NetworkNode(id=site['id'], displayName=site['name'], parentId="", type=NodeType.site, networkJsonId=f"sonar:site:{site['raw_id']}"))
@@ -772,7 +860,10 @@ def createShaper():
net.addRawNode(customer)
for device in account['devices']:
libre_device = NetworkNode(id=device['id'], displayName=device['name'],parentId=account['id'],type=NodeType.device, ipv4=device['ips'],ipv6=[],mac=device['mac'])
ipv4, ipv6, invalid_ips = splitIpVersions(device.get('ips', []))
if invalid_ips:
print(f"Skipping invalid Sonar IP assignment(s) for {device['id']}: {invalid_ips}")
libre_device = NetworkNode(id=device['id'], displayName=device['name'],parentId=account['id'],type=NodeType.device, ipv4=ipv4,ipv6=ipv6,mac=device['mac'])
net.addRawNode(libre_device)
net.prepareTree()
@@ -207,4 +207,50 @@ mod tests {
1
);
}
#[test]
fn config_roundtrips_with_multiple_dynamic_circuit_ranges() {
let mut config = Config::default();
let ipv4_rule = DynamicCircuitRangeRule {
name: "IPv4 default".to_string(),
ip_range: "0.0.0.0/0"
.parse()
.expect("default ipv4 network should parse"),
download_min_mbps: 10.0,
upload_min_mbps: 10.0,
download_max_mbps: 100.0,
upload_max_mbps: 100.0,
attach_to: "Tower-A".to_string(),
};
let ipv6_rule = DynamicCircuitRangeRule {
name: "IPv6 subscribers".to_string(),
ip_range: "2001:db8::/48"
.parse()
.expect("default ipv6 network should parse"),
download_min_mbps: 25.0,
upload_min_mbps: 5.0,
download_max_mbps: 250.0,
upload_max_mbps: 50.0,
attach_to: String::new(),
};
config.dynamic_circuits = Some(DynamicCircuitsConfig {
enabled: true,
ttl_seconds: 300,
enable_unknown_ip_promotion: true,
ranges: vec![ipv4_rule.clone(), ipv6_rule.clone()],
});
let raw = toml::to_string_pretty(&config).expect("config should serialize to TOML");
let parsed = Config::load_from_string(&raw).expect("config should deserialize");
let ranges = &parsed
.dynamic_circuits
.as_ref()
.expect("dynamic circuits should deserialize")
.ranges;
assert_eq!(ranges.len(), 2);
assert_eq!(ranges[0], ipv4_rule);
assert_eq!(ranges[1], ipv6_rule);
}
}
+1 -1
View File
@@ -238,6 +238,6 @@ pub(crate) fn calculate_tree_weights() -> Result<Vec<NetworkNodeWeight>> {
});
});
result.sort_by(|a, b| b.weight.cmp(&a.weight));
result.sort_by_key(|entry| std::cmp::Reverse(entry.weight));
Ok(result)
}
+2
View File
@@ -2240,8 +2240,10 @@ fn committed_bandwidth_multiplier() -> PyResult<f32> {
/// A UISP exception CPE entry paired with its forced parent.
pub struct PyExceptionCpe {
/// Child CPE site or device identifier.
#[pyo3(get)]
pub cpe: String,
/// Parent identifier assigned to the exception CPE.
#[pyo3(get)]
pub parent: String,
}
+10 -8
View File
@@ -616,16 +616,18 @@ fn handle_bus_requests(requests: &[BusRequest], responses: &mut Vec<BusResponse>
lqos_bus::BusResponse::Ack
}
BusRequest::UpdateLqosDTuning(..) => tuning::tune_lqosd_from_bus(req),
BusRequest::UpdateLqosdConfig(config) => {
let result = lqos_config::update_config(config);
if result.is_err() {
error!("Error updating config: {:?}", result);
BusRequest::UpdateLqosdConfig(config) => match lqos_config::update_config(config) {
Ok(()) => {
if let Ok(cfg) = lqos_config::load_config() {
let _ = stick::recompute_stick_offset(&cfg);
}
BusResponse::Ack
}
if let Ok(cfg) = lqos_config::load_config() {
let _ = stick::recompute_stick_offset(&cfg);
Err(err) => {
error!("Error updating config: {err:?}");
BusResponse::Fail(err.to_string())
}
BusResponse::Ack
}
},
BusRequest::CreateDynamicCircuit { shaped_device } => {
crate::dynamic_circuits::create_dynamic_circuit((**shaped_device).clone())
}
@@ -2,27 +2,39 @@ import { get_ws_client } from "../pubsub/ws";
const wsClient = get_ws_client();
const secretBindings = [];
const CONFIG_SAVE_TIMEOUT_MS = 10000;
export function sendWsRequest(responseEvent, request, onComplete, onError) {
export function sendWsRequest(responseEvent, request, onComplete, onError, options = {}) {
let done = false;
const responseHandler = (msg) => {
let timeoutId = null;
const finish = (callback, msg) => {
if (done) return;
done = true;
if (timeoutId !== null) {
clearTimeout(timeoutId);
timeoutId = null;
}
wsClient.off(responseEvent, responseHandler);
wsClient.off("Error", errorHandler);
onComplete(msg);
if (callback) {
callback(msg);
}
};
const responseHandler = (msg) => {
finish(onComplete, msg);
};
const errorHandler = (msg) => {
if (done) return;
done = true;
wsClient.off(responseEvent, responseHandler);
wsClient.off("Error", errorHandler);
if (onError) {
onError(msg);
}
finish(onError, msg);
};
wsClient.on(responseEvent, responseHandler);
wsClient.on("Error", errorHandler);
if (Number.isFinite(options.timeoutMs) && options.timeoutMs > 0) {
timeoutId = setTimeout(() => {
finish(onError, {
message: options.timeoutMessage || `Timed out waiting for ${responseEvent}`,
});
}, options.timeoutMs);
}
wsClient.send(request);
}
@@ -144,6 +156,17 @@ export function loadConfig(onComplete, onError) {
export function saveConfig(onComplete, onError) {
const clearSecrets = window.configSecretClears || {};
const reportError = (msg) => {
const errorMsg = (msg && msg.message) ? msg.message : "Request failed";
const detail = (msg && typeof msg === "object")
? { ...msg, message: errorMsg }
: { message: errorMsg };
if (onError) {
onError(detail);
} else {
alert("Error saving configuration: " + errorMsg);
}
};
sendWsRequest(
"UpdateConfigResult",
{
@@ -153,29 +176,29 @@ export function saveConfig(onComplete, onError) {
},
},
(msg) => {
if (msg && msg.ok) {
secretBindings.forEach((binding) => {
const sectionState = secretFieldMap(window.configSecretState || {}, binding.section);
const inputValue = binding.input.value.trim();
sectionState[binding.field] = clearSecrets?.[binding.section]?.[binding.field]
? false
: (inputValue.length > 0 || sectionState[binding.field]);
if (window.config?.[binding.section] && typeof window.config[binding.section] === "object") {
window.config[binding.section][binding.field] = "";
}
binding.input.value = "";
setSecretClear(binding.section, binding.field, false);
binding.updateStatus();
});
if (!msg || !msg.ok) {
reportError(msg);
return;
}
secretBindings.forEach((binding) => {
const sectionState = secretFieldMap(window.configSecretState || {}, binding.section);
const inputValue = binding.input.value.trim();
sectionState[binding.field] = clearSecrets?.[binding.section]?.[binding.field]
? false
: (inputValue.length > 0 || sectionState[binding.field]);
if (window.config?.[binding.section] && typeof window.config[binding.section] === "object") {
window.config[binding.section][binding.field] = "";
}
binding.input.value = "";
setSecretClear(binding.section, binding.field, false);
binding.updateStatus();
});
if (onComplete) onComplete(msg);
},
(msg) => {
if (onError) {
onError(msg);
} else {
alert("That didn't work");
}
reportError,
{
timeoutMs: CONFIG_SAVE_TIMEOUT_MS,
timeoutMessage: "Timed out waiting for the config save result. The websocket may have disconnected.",
},
);
}
@@ -21,6 +21,8 @@ let totalRows = 0;
let totalCircuits = 0;
let circuitMetricsWatchSignature = null;
const wsClient = get_ws_client();
const DYNAMIC_VIEW_MODE_STORAGE_KEY = "lqosShapedDevicesDynamicViewMode";
let dynamicViewMode = loadDynamicViewMode();
const QOO_TOOLTIP_HTML =
"<h5>Quality of Outcome (QoO)</h5>" +
@@ -75,6 +77,38 @@ function formatPlanValue(value) {
return formatted;
}
function normalizeDynamicViewMode(mode) {
return mode === "list" ? "list" : "cards";
}
function loadDynamicViewMode() {
try {
if (window.localStorage) {
return normalizeDynamicViewMode(window.localStorage.getItem(DYNAMIC_VIEW_MODE_STORAGE_KEY));
}
} catch (_error) {
// Ignore storage failures and fall back to the default cards view.
}
return "cards";
}
function persistDynamicViewMode(mode) {
try {
if (window.localStorage) {
window.localStorage.setItem(
DYNAMIC_VIEW_MODE_STORAGE_KEY,
normalizeDynamicViewMode(mode),
);
}
} catch (_error) {
// Ignore storage failures and keep the current in-memory preference.
}
}
function isDynamicListView() {
return shapedDevicesKind === "Dynamic" && dynamicViewMode === "list";
}
function countCircuits() {
return totalCircuits;
}
@@ -125,6 +159,16 @@ function setShapedDevicesKind(kind) {
requestShapedDevicesPage();
}
function setDynamicViewMode(mode) {
const normalized = normalizeDynamicViewMode(mode);
if (dynamicViewMode === normalized) {
return;
}
dynamicViewMode = normalized;
persistDynamicViewMode(dynamicViewMode);
renderDevices();
}
function filterDevices() {
page = 0;
requestShapedDevicesPage();
@@ -304,6 +348,96 @@ function metricTableRow(labelEl, downEl, upEl) {
return tr;
}
function buildDeviceIdentity(device, options = {}) {
const {
circuitClasses = [],
deviceClasses = [],
} = options;
const wrapper = document.createElement("div");
wrapper.style.minWidth = "0";
if (device.circuit_id) {
const circuitLink = document.createElement("a");
circuitLink.href = "circuit.html?id=" + encodeURI(device.circuit_id);
circuitClasses.forEach((className) => circuitLink.classList.add(className));
circuitLink.innerText = device.circuit_name || "(Unknown circuit)";
wrapper.appendChild(circuitLink);
if (device.device_name) {
const deviceLink = document.createElement("a");
deviceLink.href = "circuit.html?id=" + encodeURI(device.circuit_id);
deviceClasses.forEach((className) => deviceLink.classList.add(className));
deviceLink.innerText = device.device_name;
wrapper.appendChild(deviceLink);
}
} else {
const circuitName = document.createElement("div");
circuitClasses.forEach((className) => circuitName.classList.add(className));
circuitName.innerText = device.circuit_name || "(Unknown circuit)";
wrapper.appendChild(circuitName);
if (device.device_name) {
const deviceName = document.createElement("div");
deviceClasses.forEach((className) => deviceName.classList.add(className));
deviceName.innerText = device.device_name;
wrapper.appendChild(deviceName);
}
}
return wrapper;
}
function buildPlanBadge(device, extraClasses = []) {
const badge = document.createElement("span");
badge.classList.add("badge", "text-bg-secondary", "exec-badge", ...extraClasses);
badge.innerText =
formatPlanValue(device.download_max_mbps) + " / " + formatPlanValue(device.upload_max_mbps) + " Mbps";
return badge;
}
function buildMetricValueEl() {
const span = document.createElement("span");
span.innerHTML = "-";
return span;
}
function buildMetricPairCell(circuitId, downMetricName, upMetricName) {
const wrapper = document.createElement("div");
wrapper.classList.add("lqos-direction-metric");
const addLine = (label, metricName) => {
const line = document.createElement("div");
line.classList.add("lqos-direction-metric-line");
const labelEl = document.createElement("span");
labelEl.classList.add("lqos-direction-metric-label");
labelEl.innerText = label;
line.appendChild(labelEl);
const valueEl = buildMetricValueEl();
line.appendChild(valueEl);
wrapper.appendChild(line);
registerMetricEl(circuitId, metricName, valueEl);
};
addLine("DL", downMetricName);
addLine("UL", upMetricName);
return wrapper;
}
function appendListCell(row, content, classNames = []) {
const td = document.createElement("td");
classNames.forEach((className) => td.classList.add(className));
if (typeof content === "string") {
td.innerText = content;
} else if (content) {
td.appendChild(content);
}
row.appendChild(td);
return td;
}
function buildDeviceCard(device) {
const card = document.createElement("div");
card.classList.add("executive-card", "h-100");
@@ -312,37 +446,12 @@ function buildDeviceCard(device) {
const header = document.createElement("div");
header.classList.add("d-flex", "justify-content-between", "align-items-start", "gap-2");
const titleWrap = document.createElement("div");
titleWrap.style.minWidth = "0";
const titleWrap = buildDeviceIdentity(device, {
circuitClasses: ["redactable", "fw-semibold", "text-decoration-none"],
deviceClasses: ["redactable", "d-block", "small", "text-body-secondary", "text-decoration-none"],
});
if (device.circuit_id) {
const circuitLink = document.createElement("a");
circuitLink.href = "circuit.html?id=" + encodeURI(device.circuit_id);
circuitLink.classList.add("redactable", "fw-semibold", "text-decoration-none");
circuitLink.innerText = device.circuit_name || "(Unknown circuit)";
titleWrap.appendChild(circuitLink);
const deviceLink = document.createElement("a");
deviceLink.href = "circuit.html?id=" + encodeURI(device.circuit_id);
deviceLink.classList.add("redactable", "d-block", "small", "text-body-secondary", "text-decoration-none");
deviceLink.innerText = device.device_name || "";
titleWrap.appendChild(deviceLink);
} else {
const circuitName = document.createElement("div");
circuitName.classList.add("fw-semibold", "redactable");
circuitName.innerText = device.circuit_name || "(Unknown circuit)";
titleWrap.appendChild(circuitName);
const deviceName = document.createElement("div");
deviceName.classList.add("small", "text-body-secondary", "redactable");
deviceName.innerText = device.device_name || "";
titleWrap.appendChild(deviceName);
}
const planBadge = document.createElement("span");
planBadge.classList.add("badge", "text-bg-secondary", "exec-badge", "ms-auto");
planBadge.innerText =
formatPlanValue(device.download_max_mbps) + " / " + formatPlanValue(device.upload_max_mbps) + " Mbps";
const planBadge = buildPlanBadge(device, ["ms-auto"]);
header.appendChild(titleWrap);
header.appendChild(planBadge);
@@ -446,19 +555,57 @@ function buildDeviceCard(device) {
return card;
}
function buildDeviceListRow(device) {
const row = document.createElement("tr");
row.classList.add("align-middle");
const identity = buildDeviceIdentity(device, {
circuitClasses: ["redactable", "fw-semibold", "text-decoration-none"],
deviceClasses: ["redactable", "d-block", "small", "text-body-secondary", "text-decoration-none"],
});
appendListCell(row, identity);
const parent = document.createElement("div");
parent.classList.add("small", "redactable");
parent.innerText = device.parent_node || "-";
appendListCell(row, parent);
appendListCell(row, buildPlanBadge(device));
appendListCell(row, buildMetricPairCell(device.circuit_id, "tpDown", "tpUp"), ["text-nowrap"]);
appendListCell(row, buildMetricPairCell(device.circuit_id, "rttDown", "rttUp"), ["text-nowrap"]);
appendListCell(row, buildMetricPairCell(device.circuit_id, "qooDown", "qooUp"), ["text-nowrap"]);
appendListCell(row, buildMetricPairCell(device.circuit_id, "reXmitDown", "reXmitUp"), ["text-nowrap"]);
const lastSeenValue = document.createElement("span");
lastSeenValue.innerText = "-";
appendListCell(row, lastSeenValue, ["text-nowrap"]);
registerMetricEl(device.circuit_id, "lastSeen", lastSeenValue);
appendListCell(row, buildIpListEl(device), ["text-nowrap"]);
return row;
}
function ensureLayout() {
const target = document.getElementById("deviceTable");
if (!target) return null;
let toolbar = document.getElementById("sdToolbar");
let grid = document.getElementById("sdCardsGrid");
if (toolbar && grid) {
let listWrap = document.getElementById("sdListWrap");
let listBody = document.getElementById("sdListBody");
if (toolbar && grid && listWrap && listBody) {
return {
target,
toolbar,
grid,
listWrap,
listBody,
searchInput: document.getElementById("sdSearch"),
perPageSelect: document.getElementById("sdPerPage"),
viewToggleWrap: document.getElementById("sdViewToggleWrap"),
viewCardsButton: document.getElementById("sdViewCards"),
viewListButton: document.getElementById("sdViewList"),
prevButton: document.getElementById("sdPrevPage"),
nextButton: document.getElementById("sdNextPage"),
pageCounter: document.getElementById("sdPageCounter"),
@@ -529,6 +676,41 @@ function ensureLayout() {
perPageWrap.appendChild(perPageSelect);
toolbar.appendChild(perPageWrap);
const viewToggleWrap = document.createElement("div");
viewToggleWrap.id = "sdViewToggleWrap";
viewToggleWrap.classList.add("d-none", "d-flex", "align-items-center", "gap-1");
const viewLabel = document.createElement("span");
viewLabel.classList.add("small", "text-body-secondary");
viewLabel.innerText = "View";
const viewToggle = document.createElement("div");
viewToggle.classList.add("btn-group", "btn-group-sm", "lqos-shaped-devices-view-toggle");
viewToggle.setAttribute("role", "group");
viewToggle.setAttribute("aria-label", "Dynamic circuits view");
const viewCards = document.createElement("button");
viewCards.id = "sdViewCards";
viewCards.type = "button";
viewCards.classList.add("btn", "btn-outline-secondary");
viewCards.innerHTML = "<i class='fa fa-th-large me-1'></i>Cards";
viewCards.onclick = () => {
setDynamicViewMode("cards");
};
const viewList = document.createElement("button");
viewList.id = "sdViewList";
viewList.type = "button";
viewList.classList.add("btn", "btn-outline-secondary");
viewList.innerHTML = "<i class='fa fa-list me-1'></i>List";
viewList.onclick = () => {
setDynamicViewMode("list");
};
viewToggle.appendChild(viewCards);
viewToggle.appendChild(viewList);
viewToggleWrap.appendChild(viewLabel);
viewToggleWrap.appendChild(viewToggle);
toolbar.appendChild(viewToggleWrap);
const pagerWrap = document.createElement("div");
pagerWrap.classList.add("d-flex", "align-items-center", "gap-2");
const pager = document.createElement("div");
@@ -580,12 +762,50 @@ function ensureLayout() {
grid.classList.add("row", "row-cols-1", "row-cols-md-2", "row-cols-xl-3", "g-3");
target.appendChild(grid);
listWrap = document.createElement("div");
listWrap.id = "sdListWrap";
listWrap.classList.add("table-responsive", "lqos-table-wrap", "d-none");
const listTable = document.createElement("table");
listTable.classList.add("lqos-table", "lqos-table-tight", "mb-0", "lqos-shaped-devices-list-table");
const listHead = document.createElement("thead");
const listHeadRow = document.createElement("tr");
[
"Circuit",
"Parent",
"Plan",
"Throughput",
"RTT",
"QoO",
"Retransmits",
"Last Seen",
"IPs",
].forEach((label) => {
const th = document.createElement("th");
th.innerText = label;
listHeadRow.appendChild(th);
});
listHead.appendChild(listHeadRow);
listTable.appendChild(listHead);
listBody = document.createElement("tbody");
listBody.id = "sdListBody";
listTable.appendChild(listBody);
listWrap.appendChild(listTable);
target.appendChild(listWrap);
return {
target,
toolbar,
grid,
listWrap,
listBody,
searchInput,
perPageSelect,
viewToggleWrap,
viewCardsButton: viewCards,
viewListButton: viewList,
prevButton: prev,
nextButton: next,
pageCounter,
@@ -656,19 +876,36 @@ async function requestShapedDevicesPage() {
current.up = Math.max(toNumber(current.up, 0), toNumber(device.upload_max_mbps, 0));
planByCircuitId.set(device.circuit_id, current);
});
renderCards();
renderDevices();
requestCircuitMetricsWatch(true);
renderCounts();
} catch (_error) {
shapedDevices = [];
totalRows = 0;
totalCircuits = 0;
renderCards();
renderDevices();
renderCounts();
}
}
function renderCards() {
function updateViewToggle(layout) {
const showToggle = shapedDevicesKind === "Dynamic";
if (layout.viewToggleWrap) {
layout.viewToggleWrap.classList.toggle("d-none", !showToggle);
layout.viewToggleWrap.classList.toggle("d-flex", showToggle);
}
const cardsActive = !isDynamicListView();
if (layout.viewCardsButton) {
layout.viewCardsButton.classList.toggle("active", cardsActive);
layout.viewCardsButton.setAttribute("aria-pressed", cardsActive.toString());
}
if (layout.viewListButton) {
layout.viewListButton.classList.toggle("active", !cardsActive);
layout.viewListButton.setAttribute("aria-pressed", (!cardsActive).toString());
}
}
function renderDevices() {
const layout = ensureLayout();
if (!layout) return;
@@ -679,6 +916,7 @@ function renderCards() {
if (layout.perPageSelect && String(devicesPerPage) !== layout.perPageSelect.value) {
layout.perPageSelect.value = String(devicesPerPage);
}
updateViewToggle(layout);
if (layout.prevButton) layout.prevButton.disabled = page <= 0;
if (layout.nextButton) layout.nextButton.disabled = page >= totalPages - 1;
if (layout.pageCounter) layout.pageCounter.innerText = "Page " + (page + 1) + " / " + totalPages;
@@ -688,21 +926,36 @@ function renderCards() {
} else {
const start = page * devicesPerPage + 1;
const end = Math.min((page + 1) * devicesPerPage, totalRows);
layout.summary.innerText = "Showing " + start + "–" + end + " of " + totalRows;
let summary = "Showing " + start + "–" + end + " of " + totalRows;
if (shapedDevicesKind === "Dynamic") {
summary += " (highest throughput first)";
}
layout.summary.innerText = summary;
}
}
metricElsByCircuitId = new Map();
clearDiv(layout.grid);
clearDiv(layout.listBody);
shapedDevices.forEach((device) => {
const col = document.createElement("div");
col.classList.add("col");
col.appendChild(buildDeviceCard(device));
layout.grid.appendChild(col);
});
const showList = isDynamicListView();
layout.grid.classList.toggle("d-none", showList);
layout.listWrap.classList.toggle("d-none", !showList);
// Fill visible cards from cached live data immediately.
if (showList) {
shapedDevices.forEach((device) => {
layout.listBody.appendChild(buildDeviceListRow(device));
});
} else {
shapedDevices.forEach((device) => {
const col = document.createElement("div");
col.classList.add("col");
col.appendChild(buildDeviceCard(device));
layout.grid.appendChild(col);
});
}
// Fill visible metrics from cached live data immediately.
metricElsByCircuitId.forEach((_metrics, circuitId) => {
const latest = latestByCircuitId.get(circuitId);
if (latest) {
@@ -179,12 +179,21 @@ function schedulerHasLiveError(data) {
return !schedulerLooksRecovered(data);
}
function schedulerWaitsForTopologyRuntime(data) {
const progress = data?.progress || null;
if (!progress) {
return false;
}
return String(progress.phase || "").trim() === "waiting_for_topology_runtime";
}
function schedulerStateDescriptor(data) {
const progress = data?.progress || null;
const hasError = schedulerHasLiveError(data);
const available = !!data?.available;
const active = !!progress?.active;
const setupRequired = !!data?.setup_required;
const topologyRuntimeWait = schedulerWaitsForTopologyRuntime(data);
if (hasError) {
return {
@@ -219,6 +228,17 @@ function schedulerStateDescriptor(data) {
icon: "fa-clock",
};
}
if (topologyRuntimeWait) {
return {
tone: "warning",
badgeClass: "text-bg-warning",
label: "Deferred",
title: progress?.phase_label || "Waiting for topology runtime",
subtitle: "Scheduler is waiting for topology runtime outputs",
ringTone: "tone-warning",
icon: "fa-hourglass-half",
};
}
if (active) {
return {
tone: "info",
@@ -316,6 +336,8 @@ function renderSchedulerDetails(data, options = {}) {
? "Setup Required"
: data?.stale
? "Stale"
: schedulerWaitsForTopologyRuntime(data)
? "Deferred"
: (data?.available ? "Healthy" : "Unavailable");
const recentResult = schedulerSetupMessage(data) || summarizeSchedulerOutput(data?.output, historicalError || liveError);
const activity = schedulerActivityItems(data?.output, historicalError || liveError, data);
@@ -324,6 +346,8 @@ function renderSchedulerDetails(data, options = {}) {
progressMeta = "Showing cached scheduler details while the UI reconnects";
} else if (data?.setup_required) {
progressMeta = "Complete runtime setup to enable scheduler work";
} else if (schedulerWaitsForTopologyRuntime(data)) {
progressMeta = "The next shaping update is waiting for topology runtime outputs";
} else if (progress?.active) {
progressMeta = `${percent}% complete`;
} else if (descriptor.label === "Idle") {
@@ -450,6 +474,14 @@ function renderSchedulerStatus(container, state, progress, labelOverride = null)
color = "text-warning";
label = "Scheduler status is stale";
indicator = schedulerRingMarkup(100, "tone-warning", "fa-clock");
} else if (state === "deferred") {
color = "text-warning";
label = labelOverride || "Scheduler is waiting for topology runtime";
indicator = schedulerRingMarkup(
schedulerProgressPercent(progress),
"tone-warning",
"fa-hourglass-half"
);
} else if (state === "setup") {
color = "text-warning";
label = "Scheduler needs topology setup";
@@ -538,6 +570,17 @@ function loadSchedulerStatus(force = false) {
return;
}
if (schedulerWaitsForTopologyRuntime(data)) {
renderSchedulerStatus(
container,
"deferred",
progress,
progress?.phase_label || "Scheduler is waiting for topology runtime"
);
scheduleNextSchedulerStatusPoll();
return;
}
if (progressActive) {
renderSchedulerStatus(container, "progress", progress);
scheduleNextSchedulerStatusPoll();
@@ -7,7 +7,7 @@ use axum::body::Bytes;
use axum::http::StatusCode;
use axum::http::header;
use default_net::get_interfaces;
use lqos_bus::{BusRequest, bus_request};
use lqos_bus::{BusRequest, BusResponse, bus_request};
use lqos_config::{Config, ConfigShapedDevices, ShapedDevice, UserRole, WebUser, WebUsers};
use lqos_utils::hash_to_i64;
use serde::{Deserialize, Serialize};
@@ -793,16 +793,23 @@ pub async fn update_lqosd_config_data(
login: LoginResult,
mut config: Config,
clear_secrets: ConfigSecretClearRequest,
) -> Result<(), StatusCode> {
) -> Result<(), String> {
if login != LoginResult::Admin {
return Err(StatusCode::FORBIDDEN);
return Err("Unauthorized".to_string());
}
let existing = lqos_config::load_config().map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
let existing =
lqos_config::load_config().map_err(|_| "Unable to load the current config".to_string())?;
apply_secret_updates(existing.as_ref(), &mut config, &clear_secrets);
bus_request(vec![BusRequest::UpdateLqosdConfig(Box::new(config))])
let mut responses = bus_request(vec![BusRequest::UpdateLqosdConfig(Box::new(config))])
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
Ok(())
.map_err(|err| format!("Unable to update config: {err}"))?;
match responses.pop() {
Some(BusResponse::Ack) => Ok(()),
Some(BusResponse::Fail(message)) => Err(message),
Some(other) => Err(format!("Unexpected config update response: {other:?}")),
None => Err("No response received for config update".to_string()),
}
}
/// Persists both `network.json` and `ShapedDevices.csv` for administrative
@@ -1,5 +1,9 @@
use crate::shaped_devices_tracker::circuit_live::{
CircuitLiveRollup, CircuitLiveSnapshot, fresh_circuit_live_snapshot,
};
use lqos_config::ShapedDevice;
use serde::{Deserialize, Deserializer, Serialize};
use std::cmp::Ordering;
use std::collections::HashSet;
const DEFAULT_SHAPED_DEVICES_PAGE_SIZE: usize = 24;
@@ -57,7 +61,49 @@ fn normalized_page_size(query: &ShapedDevicesPageQuery) -> usize {
.clamp(1, MAX_SHAPED_DEVICES_PAGE_SIZE)
}
fn alphabetic_inventory_order(left: &ShapedDevice, right: &ShapedDevice) -> Ordering {
left.circuit_name
.cmp(&right.circuit_name)
.then_with(|| left.device_name.cmp(&right.device_name))
.then_with(|| left.device_id.cmp(&right.device_id))
}
fn live_sort_rank(snapshot: &CircuitLiveSnapshot, device: &ShapedDevice) -> (u64, u64, u64, u64) {
let live = snapshot.by_circuit_id.get(&device.circuit_id);
let bytes_per_second = live
.map(|row: &CircuitLiveRollup| row.bytes_per_second)
.unwrap_or_default();
let total_bytes_per_second = bytes_per_second.down.saturating_add(bytes_per_second.up);
let last_seen_nanos = live.map(|row| row.last_seen_nanos).unwrap_or(u64::MAX);
(
total_bytes_per_second,
bytes_per_second.down,
bytes_per_second.up,
last_seen_nanos,
)
}
fn dynamic_inventory_order(
snapshot: &CircuitLiveSnapshot,
left: &ShapedDevice,
right: &ShapedDevice,
) -> Ordering {
let left_rank = live_sort_rank(snapshot, left);
let right_rank = live_sort_rank(snapshot, right);
right_rank
.0
.cmp(&left_rank.0)
.then_with(|| right_rank.1.cmp(&left_rank.1))
.then_with(|| right_rank.2.cmp(&left_rank.2))
.then_with(|| left_rank.3.cmp(&right_rank.3))
.then_with(|| alphabetic_inventory_order(left, right))
}
/// Returns one filtered, sorted page of shaped-device rows.
///
/// Static inventory rows are ordered alphabetically. Dynamic inventory rows are
/// ordered by current observed throughput before pagination so the busiest
/// circuits stay at the top of the page.
pub fn shaped_devices_page(query: ShapedDevicesPageQuery) -> ShapedDevicesPage {
let page = query.page.unwrap_or(0);
let page_size = normalized_page_size(&query);
@@ -103,12 +149,13 @@ pub fn shaped_devices_page(query: ShapedDevicesPageQuery) -> ShapedDevicesPage {
.cloned()
.collect(),
};
filtered.sort_by(|left, right| {
left.circuit_name
.cmp(&right.circuit_name)
.then_with(|| left.device_name.cmp(&right.device_name))
.then_with(|| left.device_id.cmp(&right.device_id))
});
match kind {
ShapedDevicesPageKind::Static => filtered.sort_by(alphabetic_inventory_order),
ShapedDevicesPageKind::Dynamic => {
let snapshot = fresh_circuit_live_snapshot();
filtered.sort_by(|left, right| dynamic_inventory_order(snapshot.as_ref(), left, right));
}
}
let total_rows = filtered.len();
let total_circuits = filtered
@@ -140,3 +187,84 @@ pub fn shaped_devices_page(query: ShapedDevicesPageQuery) -> ShapedDevicesPage {
rows,
}
}
#[cfg(test)]
mod tests {
use super::*;
use fxhash::FxHashMap;
use lqos_utils::units::DownUpOrder;
fn test_device(circuit_id: &str, circuit_name: &str, device_name: &str) -> ShapedDevice {
ShapedDevice {
circuit_id: circuit_id.to_string(),
circuit_name: circuit_name.to_string(),
device_id: format!("{circuit_id}-device"),
device_name: device_name.to_string(),
..Default::default()
}
}
fn test_snapshot(entries: &[(&str, u64, u64, u64)]) -> CircuitLiveSnapshot {
let mut by_circuit_id = FxHashMap::default();
for (circuit_id, down, up, last_seen_nanos) in entries {
by_circuit_id.insert(
(*circuit_id).to_string(),
CircuitLiveRollup {
circuit_id: (*circuit_id).to_string(),
circuit_name: String::new(),
parent_node: String::new(),
device_names: Vec::new(),
ip_addrs: Vec::new(),
plan_mbps: DownUpOrder::default(),
bytes_per_second: DownUpOrder::new(*down, *up),
rtt_current_p50_nanos: DownUpOrder::default(),
qoo: DownUpOrder::default(),
tcp_retransmit_sample: DownUpOrder::default(),
last_seen_nanos: *last_seen_nanos,
},
);
}
CircuitLiveSnapshot { by_circuit_id }
}
#[test]
fn dynamic_inventory_order_prefers_higher_combined_throughput() {
let snapshot = test_snapshot(&[
("slow", 10, 20, 3_000),
("fast", 40, 90, 2_000),
("idle", 0, 0, u64::MAX),
]);
let mut rows = vec![
test_device("slow", "Slow Circuit", "Slow Device"),
test_device("idle", "Idle Circuit", "Idle Device"),
test_device("fast", "Fast Circuit", "Fast Device"),
];
rows.sort_by(|left, right| dynamic_inventory_order(&snapshot, left, right));
assert_eq!(
rows.iter()
.map(|row| row.circuit_id.as_str())
.collect::<Vec<_>>(),
vec!["fast", "slow", "idle"]
);
}
#[test]
fn dynamic_inventory_order_uses_alphabetic_tie_breakers() {
let snapshot = test_snapshot(&[("alpha", 25, 25, 100), ("beta", 25, 25, 100)]);
let mut rows = vec![
test_device("beta", "Beta Circuit", "Second Device"),
test_device("alpha", "Alpha Circuit", "First Device"),
];
rows.sort_by(|left, right| dynamic_inventory_order(&snapshot, left, right));
assert_eq!(
rows.iter()
.map(|row| row.circuit_id.as_str())
.collect::<Vec<_>>(),
vec!["alpha", "beta"]
);
}
}
@@ -2501,6 +2501,41 @@ table tr td a {
background: var(--lqos-surface-solid);
box-shadow: var(--lqos-shadow-sm);
}
.lqos-shaped-devices-view-toggle .btn {
border-color: color-mix(in srgb, var(--lqos-border) 80%, transparent);
}
.lqos-shaped-devices-view-toggle .btn.active,
.lqos-shaped-devices-view-toggle .btn:active {
color: var(--bs-emphasis-color);
border-color: color-mix(in srgb, var(--lqos-border) 72%, transparent);
background: color-mix(in srgb, var(--lqos-surface) 86%, transparent);
box-shadow:
inset 0 1px 0 rgba(255, 255, 255, 0.08),
0 0 0 1px color-mix(in srgb, var(--lqos-border) 30%, transparent);
}
.lqos-shaped-devices-list-table td {
vertical-align: middle;
}
.lqos-shaped-devices-list-table td .badge {
white-space: nowrap;
}
.lqos-direction-metric {
display: flex;
flex-direction: column;
gap: 0.15rem;
min-width: 7rem;
}
.lqos-direction-metric-line {
display: flex;
align-items: center;
justify-content: space-between;
gap: 0.75rem;
}
.lqos-direction-metric-label {
font-size: 0.74rem;
font-weight: 600;
color: var(--bs-secondary-color);
}
.exec-icon {
display: inline-flex;
align-items: center;
+1 -2
View File
@@ -1579,8 +1579,7 @@ async fn receive_channel_message(
config::update_lqosd_config_data(*request_state.login, cfg, clear_secrets).await;
let (ok, message) = match result {
Ok(()) => (true, "Ok".to_string()),
Err(StatusCode::FORBIDDEN) => (false, "Unauthorized".to_string()),
Err(_) => (false, "Error".to_string()),
Err(message) => (false, message),
};
let response = WsResponse::UpdateConfigResult { ok, message };
if send_ws_response(&tx, response).await {
+120 -22
View File
@@ -4,7 +4,7 @@ import csv
import io
import json
import chardet
from LibreQoS import ValidationFailure, refreshShapers, refreshShapersUpdateOnly
from LibreQoS import RefreshFailure, ValidationFailure, refreshShapers, refreshShapersUpdateOnly
import subprocess
import sys
import tempfile
@@ -46,6 +46,7 @@ partial_topology_runtime_pending = False
partial_topology_runtime_generation = None
partial_topology_runtime_started_monotonic = None
partial_topology_runtime_last_report_state = None
last_integration_failure_message = ""
def configure_scheduler_stdio():
@@ -128,6 +129,37 @@ def clear_scheduler_output():
scheduler_output("")
def remember_integration_failure(message: str):
global last_integration_failure_message
last_integration_failure_message = str(message or "").strip()
def clear_integration_failure():
global last_integration_failure_message
last_integration_failure_message = ""
def clear_scheduler_error_unless_integration_failed():
if not last_integration_failure_message:
clear_scheduler_error()
def publish_ready_progress(active: bool, phase: str, phase_label: str, step_index: int, step_count: int, *, percent=None):
publish_scheduler_progress(
active,
phase,
phase_label,
step_index,
step_count,
percent=percent,
)
if last_integration_failure_message:
scheduler_error(
"Scheduler ready using last-known-good topology; latest integration import failed.\n"
+ last_integration_failure_message
)
def _scheduler_progress_percent(step_index: int, step_count: int, *, active: bool) -> int:
if step_count <= 0:
return 0
@@ -152,17 +184,6 @@ def publish_scheduler_progress(active: bool, phase: str, phase_label: str, step_
print(f"Failed to publish scheduler progress: {e}")
def publish_scheduler_ready(step_count: int):
publish_scheduler_progress(
False,
"ready",
"Scheduler ready",
step_count,
step_count,
percent=100,
)
def report_scheduler_runtime_failure(context: str, exc: Exception, *, startup: bool = False):
message = f"{context}: {exc}"
print(message)
@@ -253,6 +274,7 @@ def _write_integration_output_artifact(label, output):
def _publish_integration_result(label, result):
output = ((result.stdout or "") + (result.stderr or "")).replace("\r\n", "\n").strip()
if result.returncode == 0:
clear_integration_failure()
line_count = len(_integration_output_lines(output))
summary = (
f"{label} completed successfully."
@@ -275,6 +297,7 @@ def _publish_integration_result(label, result):
if artifact is not None:
message += f"\nFull output saved to {artifact}"
print(message)
remember_integration_failure(message)
scheduler_error(message)
@@ -396,6 +419,7 @@ def run_python_integration(module_name: str, func_name: str, label: str = ""):
except Exception as e:
err = f"Failed to invoke integration {label or (module_name + '.' + func_name)}: {e}"
print(err)
remember_integration_failure(err)
scheduler_error(err)
def importFromCRM(
@@ -408,6 +432,7 @@ def importFromCRM(
overrides_label="Applying overrides",
overrides_step=3,
):
clear_integration_failure()
clear_scheduler_error()
clear_scheduler_output()
publish_scheduler_progress(
@@ -433,6 +458,7 @@ def importFromCRM(
except Exception as e:
error_msg = f"Failed to run UISP integration: {str(e)}"
print(error_msg)
remember_integration_failure(error_msg)
scheduler_error(error_msg)
elif automatic_import_splynx():
run_python_integration("integrationSplynx", "importFromSplynx", label="Splynx")
@@ -1032,12 +1058,18 @@ def importAndShapeFullReload():
report_topology_runtime_not_ready(
"Scheduler startup shaping refresh deferred",
phase_label="Scheduler waiting for topology runtime",
step_index=4,
step_count=SCHEDULER_STARTUP_STEP_COUNT,
)
return False
publish_scheduler_progress(True, "initial_shaping_reload", "Refreshing shaper state", 5, SCHEDULER_STARTUP_STEP_COUNT)
if not enable_insight_topology():
refreshShapers()
try:
refreshShapers()
except Exception as e:
if _defer_stale_shaping_inputs_failure(e, startup=True):
return False
raise
shaping_runtime_hash = calculate_shaping_runtime_hash()
else:
shaping_runtime_hash = calculate_shaping_runtime_hash()
@@ -1068,6 +1100,7 @@ def importAndShapePartialReload():
report_topology_runtime_not_ready(
"Scheduled shaping refresh deferred",
phase_label="Scheduler waiting for topology runtime",
step_index=3,
step_count=SCHEDULER_REFRESH_STEP_COUNT,
)
return
@@ -1083,12 +1116,14 @@ def importAndShapePartialReload():
report_scheduler_validation_failure("Scheduled shaping refresh blocked by validation", e)
return
except Exception as e:
if _defer_stale_shaping_inputs_failure(e, startup=False):
return
report_scheduler_runtime_failure("Scheduled shaping refresh failed", e)
return
shaping_runtime_hash = calculate_shaping_runtime_hash()
if shaping_runtime_hash != 0:
set_scheduler_status_bus_enabled(True)
publish_scheduler_ready(SCHEDULER_REFRESH_STEP_COUNT)
publish_ready_progress(False, "ready", "Scheduler ready", SCHEDULER_REFRESH_STEP_COUNT, SCHEDULER_REFRESH_STEP_COUNT, percent=100)
def topology_runtime_status_path():
@@ -1184,7 +1219,13 @@ def topology_runtime_readiness_detail():
return (True, "", current_generation)
def report_topology_runtime_not_ready(context: str, *, phase_label: str, step_count: int):
def report_topology_runtime_not_ready(
context: str,
*,
phase_label: str,
step_index: int,
step_count: int,
):
ready, detail, generation = topology_runtime_readiness_detail()
if ready:
return
@@ -1192,7 +1233,19 @@ def report_topology_runtime_not_ready(context: str, *, phase_label: str, step_co
if generation:
message += f" Generation {generation[:12]}."
print(message)
if _topology_runtime_not_ready_is_transient(detail):
clear_scheduler_error_unless_integration_failed()
scheduler_output(message)
publish_scheduler_progress(
False,
"waiting_for_topology_runtime",
phase_label,
step_index,
step_count,
)
return
scheduler_error(message)
scheduler_output(message)
publish_scheduler_progress(
False,
"degraded",
@@ -1234,6 +1287,29 @@ def _topology_runtime_not_ready_is_transient(detail: str) -> bool:
return True
def _is_transient_stale_shaping_inputs_failure(exc: Exception) -> bool:
if not isinstance(exc, RefreshFailure):
return False
detail = str(exc or "").strip()
return detail.startswith("Missing or stale shaping_inputs.json.")
def _defer_stale_shaping_inputs_failure(exc: Exception, *, startup: bool) -> bool:
if not _is_transient_stale_shaping_inputs_failure(exc):
return False
ready, detail, generation = topology_runtime_readiness_detail()
if detail and not _topology_runtime_not_ready_is_transient(detail):
return False
clear_scheduler_error_unless_integration_failed()
if startup:
_begin_startup_topology_runtime_wait(generation if not ready else current_topology_source_generation())
else:
_begin_partial_topology_runtime_wait(generation if not ready else current_topology_source_generation())
return True
def _publish_startup_topology_runtime_wait(generation: str | None):
global startup_topology_runtime_last_report_state
@@ -1360,6 +1436,8 @@ def _continue_startup_topology_runtime_wait():
shaping_runtime_hash = 0
return
except Exception as e:
if _defer_stale_shaping_inputs_failure(e, startup=True):
return
_reset_startup_topology_runtime_wait()
report_scheduler_runtime_failure(
"Scheduler startup shaping refresh failed",
@@ -1371,8 +1449,15 @@ def _continue_startup_topology_runtime_wait():
_reset_startup_topology_runtime_wait()
if shaping_runtime_hash != 0:
clear_scheduler_error()
publish_scheduler_ready(SCHEDULER_STARTUP_STEP_COUNT)
clear_scheduler_error_unless_integration_failed()
publish_ready_progress(
False,
"ready",
"Scheduler ready",
SCHEDULER_STARTUP_STEP_COUNT,
SCHEDULER_STARTUP_STEP_COUNT,
percent=100,
)
return
report_scheduler_runtime_failure(
@@ -1402,6 +1487,7 @@ def _continue_startup_topology_runtime_wait():
report_topology_runtime_not_ready(
"Scheduler startup shaping refresh deferred",
phase_label="Scheduler waiting for topology runtime",
step_index=4,
step_count=SCHEDULER_STARTUP_STEP_COUNT,
)
@@ -1457,6 +1543,8 @@ def _continue_partial_topology_runtime_wait():
)
return
except Exception as e:
if _defer_stale_shaping_inputs_failure(e, startup=False):
return
_reset_partial_topology_runtime_wait()
report_scheduler_runtime_failure("Scheduled shaping refresh failed", e)
return
@@ -1471,8 +1559,15 @@ def _continue_partial_topology_runtime_wait():
)
return
_reset_partial_topology_runtime_wait()
clear_scheduler_error()
publish_scheduler_ready(SCHEDULER_REFRESH_STEP_COUNT)
clear_scheduler_error_unless_integration_failed()
publish_ready_progress(
False,
"ready",
"Scheduler ready",
SCHEDULER_REFRESH_STEP_COUNT,
SCHEDULER_REFRESH_STEP_COUNT,
percent=100,
)
return
started = partial_topology_runtime_started_monotonic
@@ -1495,6 +1590,7 @@ def _continue_partial_topology_runtime_wait():
report_topology_runtime_not_ready(
"Scheduled shaping refresh deferred",
phase_label="Scheduler waiting for topology runtime",
step_index=3,
step_count=SCHEDULER_REFRESH_STEP_COUNT,
)
@@ -1543,6 +1639,8 @@ def topology_runtime_refresh_tick():
report_scheduler_validation_failure("Topology runtime refresh blocked by validation", e)
return
except Exception as e:
if _defer_stale_shaping_inputs_failure(e, startup=False):
return
report_scheduler_runtime_failure("Topology runtime refresh failed", e)
return
shaping_runtime_hash = calculate_shaping_runtime_hash()
@@ -1552,8 +1650,8 @@ def topology_runtime_refresh_tick():
RuntimeError("Shaping runtime hash was not published after topology runtime refresh."),
)
return
clear_scheduler_error()
publish_scheduler_ready(SCHEDULER_REFRESH_STEP_COUNT)
clear_scheduler_error_unless_integration_failed()
publish_ready_progress(False, "ready", "Scheduler ready", SCHEDULER_REFRESH_STEP_COUNT, SCHEDULER_REFRESH_STEP_COUNT, percent=100)
def not_dead_yet():
@@ -1576,7 +1674,7 @@ def run_scheduler_main():
try:
startup_ready = importAndShapeFullReload()
if startup_ready:
publish_scheduler_ready(SCHEDULER_STARTUP_STEP_COUNT)
publish_ready_progress(False, "ready", "Scheduler ready", SCHEDULER_STARTUP_STEP_COUNT, SCHEDULER_STARTUP_STEP_COUNT, percent=100)
except ValidationFailure as e:
report_scheduler_validation_failure(
"Scheduler startup shaping refresh blocked by validation",
+39
View File
@@ -493,6 +493,45 @@ class TestGraph(unittest.TestCase):
net.prepareTree()
self.assertEqual(net.nodes[2].parentIndex, 1)
def test_empty_exception_cpes_list_does_not_crash(self):
import integrationCommon
from integrationCommon import NetworkGraph, NetworkNode, NodeType
original_exception_cpes = integrationCommon.exception_cpes
try:
integrationCommon.exception_cpes = lambda: []
net = NetworkGraph()
net.addRawNode(NetworkNode("Site_1", "Site_1", "", NodeType.site, 1000, 1000))
finally:
integrationCommon.exception_cpes = original_exception_cpes
self.assertEqual(net.nodes[1].displayName, "Site_1")
def test_exception_cpes_list_entries_are_normalized(self):
from integrationCommon import NetworkGraph, NetworkNode, NodeType, _normalize_exception_cpes
class ExceptionCpe:
cpe = "Site_3"
parent = "Site_1"
normalized, errors = _normalize_exception_cpes([
"Site_2:Site_1",
{"cpe": "Site_4", "parent": "Site_1"},
ExceptionCpe(),
])
self.assertEqual(errors, [])
self.assertEqual(normalized["Site_2"], "Site_1")
self.assertEqual(normalized["Site_3"], "Site_1")
self.assertEqual(normalized["Site_4"], "Site_1")
net = NetworkGraph()
net.exceptionCPEs = normalized
net.addRawNode(NetworkNode("Site_1", "Site_1", "", NodeType.site, 1000, 1000))
net.addRawNode(NetworkNode("Site_2", "Site_2", "", NodeType.site, 500, 500))
self.assertEqual(net.nodes[2].parentId, "Site_1")
def test_native_topology_editor_uses_nearest_real_infrastructure_parent(self):
from integrationCommon import NetworkGraph, NetworkNode, NodeType
+125 -1
View File
@@ -29,6 +29,9 @@ def install_sonar_stubs():
lqlib.sonar_active_status_ids = lambda: []
lqlib.sonar_recurring_service_rates = lambda: []
lqlib.sonar_recurring_excluded_service_names = lambda: []
lqlib.sonar_strategy = lambda: "full"
lqlib.write_compiled_topology_from_python_graph_payload = lambda *_args, **_kwargs: None
lqlib.get_libreqos_directory = lambda: ""
sys.modules["liblqos_python"] = lqlib
@@ -59,8 +62,13 @@ def inventory_item(item_id, name, ips=None, mac=""):
}
def sonar_account(address_items=None, radius_accounts=None):
def sonar_account(address_items=None, radius_accounts=None, account_ips=None):
return {
"id": 1,
"name": "Account 1",
"ip_assignments": {
"entities": [{"subnet": subnet} for subnet in (account_ips or [])]
},
"addresses": {
"entities": [
{
@@ -147,6 +155,64 @@ class TestIntegrationSonarDevices(unittest.TestCase):
self.assertEqual(devices[1]["id"], "sonar:radius-account:88")
self.assertEqual(devices[1]["ips"], ["100.64.3.31/32"])
def test_account_level_ips_emit_stable_account_ip_device(self):
devices = integrationSonar.buildAccountDevices(
sonar_account(
address_items=[
inventory_item(
42,
"ONU",
ips=[],
mac="AA:BB:CC:DD:EE:03",
)
],
account_ips=["100.64.8.80", "100.64.8.81"],
)
)
self.assertEqual(len(devices), 2)
self.assertEqual(devices[0]["id"], "sonar:device:42")
self.assertEqual(devices[0]["mac"], "AA:BB:CC:DD:EE:03")
self.assertEqual(devices[0]["ips"], [])
self.assertEqual(devices[1]["id"], "sonar:account-ip-assignments:1")
self.assertEqual(devices[1]["name"], "Account IP Assignments")
self.assertEqual(devices[1]["ips"], ["100.64.8.80", "100.64.8.81"])
self.assertEqual(devices[1]["mac"], "")
self.assertEqual(devices[1]["source"], "account_ip_assignments")
def test_split_ip_versions_preserves_ipv6_prefixes(self):
ipv4, ipv6, invalid = integrationSonar.splitIpVersions([
"100.64.8.80",
"2605:40c1:300::/56",
"not an ip",
])
self.assertEqual(ipv4, ["100.64.8.80"])
self.assertEqual(ipv6, ["2605:40c1:300::/56"])
self.assertEqual(invalid, ["not an ip"])
def test_account_level_ips_are_deduped_against_inventory_and_radius_ips(self):
devices = integrationSonar.buildAccountDevices(
sonar_account(
address_items=[
inventory_item(
42,
"ONU",
ips=["100.64.9.90"],
mac="AA:BB:CC:DD:EE:09",
)
],
radius_accounts=[
radius_account(99, ["100.64.9.91"])
],
account_ips=["100.64.9.90", "100.64.9.91", "100.64.9.92"],
)
)
self.assertEqual(len(devices), 3)
self.assertEqual(devices[2]["id"], "sonar:account-ip-assignments:1")
self.assertEqual(devices[2]["ips"], ["100.64.9.92"])
def test_inventory_mac_devices_are_preserved_when_radius_supplies_ips(self):
devices = integrationSonar.buildAccountDevices(
sonar_account(
@@ -195,6 +261,31 @@ class TestIntegrationSonarDevices(unittest.TestCase):
self.assertEqual(record["address"], "Parent Address")
self.assertEqual(record["id"], "sonar:account:500")
def test_child_account_imports_account_level_ips(self):
child = sonar_account(
address_items=[],
account_ips=["100.64.10.10"],
)
child["addresses"] = {"entities": []}
child["id"] = 501
child["name"] = "Suite 501"
child["account_services"] = {
"entities": [{
"service": {
"data_service_detail": {
"download_speed_kilobits_per_second": 100000,
"upload_speed_kilobits_per_second": 20000,
}
}
}]
}
record = integrationSonar.buildAccountRecord(child, fallback_address="Parent Address")
self.assertIsNotNone(record)
self.assertEqual(record["devices"][0]["id"], "sonar:account-ip-assignments:501")
self.assertEqual(record["devices"][0]["ips"], ["100.64.10.10"])
def test_child_accounts_are_added_without_duplicate_top_level_records(self):
top_level = sonar_account(
address_items=[
@@ -338,6 +429,39 @@ class TestIntegrationSonarDevices(unittest.TestCase):
self.assertIsNone(record)
def test_import_summary_counts_account_level_ip_devices(self):
account = {
"parent": "ap_1",
"devices": [
{
"id": "sonar:device:42",
"ips": [],
"source": "inventory_item",
},
{
"id": "sonar:account-ip-assignments:1",
"ips": ["100.64.11.11", "2605:40c1:300::/56"],
"source": "account_ip_assignments",
},
],
}
summary = integrationSonar.sonarImportSummary(
sites=[{"id": "site_1"}],
aps=[{"id": "ap_1"}],
accounts=[account],
)
self.assertIn("sites=1", summary)
self.assertIn("aps=1", summary)
self.assertIn("accounts=1", summary)
self.assertIn("accounts_with_ap_parent=1", summary)
self.assertIn("devices=2", summary)
self.assertIn("devices_with_ips=1", summary)
self.assertIn("ipv4_assignments=1", summary)
self.assertIn("ipv6_assignments=1", summary)
self.assertIn("account_level_ip_devices=1", summary)
class FakeResponse:
def __init__(self, payload):
+258
View File
@@ -22,8 +22,11 @@ scheduler = None
def install_scheduler_stubs():
libre = types.ModuleType("LibreQoS")
class RefreshFailure(Exception):
pass
class ValidationFailure(Exception):
pass
libre.RefreshFailure = RefreshFailure
libre.ValidationFailure = ValidationFailure
libre.refreshShapers = Mock()
libre.refreshShapersUpdateOnly = Mock()
@@ -196,6 +199,7 @@ class TestSchedulerErrorReporting(unittest.TestCase):
scheduler.set_scheduler_status_bus_enabled(True)
scheduler._reset_startup_topology_runtime_wait()
scheduler._reset_partial_topology_runtime_wait()
scheduler.clear_integration_failure()
scheduler.shaping_runtime_hash = 0
def test_python_integration_output_does_not_set_scheduler_error(self):
@@ -293,12 +297,38 @@ class TestSchedulerErrorReporting(unittest.TestCase):
)
self.assertEqual(mock_scheduler_output.call_args_list, [(( "",),)])
def test_ready_progress_keeps_latest_integration_failure_visible(self):
scheduler.remember_integration_failure("Sonar exited with code 1. Continuing.")
with patch.object(scheduler, "publish_scheduler_progress") as mock_progress:
with patch.object(scheduler, "scheduler_error") as mock_scheduler_error:
scheduler.publish_ready_progress(
False,
"ready",
"Scheduler ready",
scheduler.SCHEDULER_REFRESH_STEP_COUNT,
scheduler.SCHEDULER_REFRESH_STEP_COUNT,
percent=100,
)
mock_progress.assert_called_once()
mock_scheduler_error.assert_called_once()
self.assertIn(
"Scheduler ready using last-known-good topology; latest integration import failed.",
mock_scheduler_error.call_args.args[0],
)
self.assertIn(
"Sonar exited with code 1. Continuing.",
mock_scheduler_error.call_args.args[0],
)
class TestSchedulerLogging(unittest.TestCase):
def setUp(self):
scheduler.set_scheduler_status_bus_enabled(True)
scheduler._reset_startup_topology_runtime_wait()
scheduler._reset_partial_topology_runtime_wait()
scheduler.clear_integration_failure()
scheduler.shaping_runtime_hash = 0
def test_configure_scheduler_stdio_enables_line_buffering_when_supported(self):
@@ -377,6 +407,55 @@ class TestSchedulerLogging(unittest.TestCase):
"Topology runtime refresh failed: bad runtime"
)
def test_topology_runtime_refresh_tick_waits_for_stale_shaping_inputs(self):
stale_exc = scheduler.RefreshFailure(
"Missing or stale shaping_inputs.json. Run topology runtime before shaping."
)
with patch.object(scheduler, "ensure_topology_runtime_process"):
with patch.object(
scheduler,
"topology_runtime_readiness_detail",
return_value=(True, "", "generation-1"),
):
with patch.object(
scheduler,
"current_topology_source_generation",
return_value="generation-1",
):
with patch.object(scheduler, "calculate_shaping_runtime_hash", return_value=5):
with patch.object(
scheduler,
"refreshShapers",
side_effect=stale_exc,
):
with patch.object(scheduler, "clear_scheduler_error") as mock_clear_error:
with patch.object(
scheduler,
"report_scheduler_runtime_failure",
) as mock_report:
with patch.object(
scheduler,
"publish_scheduler_progress",
) as mock_progress:
with patch("builtins.print"):
scheduler.shaping_runtime_hash = 1
scheduler.topology_runtime_refresh_tick()
mock_clear_error.assert_called_once()
mock_report.assert_not_called()
self.assertTrue(scheduler.partial_topology_runtime_pending)
self.assertTrue(
any(
call.args[:3] == (
True,
"waiting_for_topology_runtime",
"Waiting for topology runtime",
)
for call in mock_progress.call_args_list
)
)
def test_topology_runtime_refresh_tick_reports_validation_failure(self):
validation_exc = scheduler.ValidationFailure("Validation failed. Will now exit.")
@@ -514,6 +593,83 @@ class TestSchedulerLogging(unittest.TestCase):
)
)
def test_transient_topology_runtime_deferred_wait_does_not_raise_scheduler_error(self):
with patch.object(
scheduler,
"topology_runtime_readiness_detail",
return_value=(
False,
"Topology runtime is still building outputs for the current source generation.",
"generation-1",
),
):
with patch.object(scheduler, "clear_scheduler_error") as mock_clear_error:
with patch.object(scheduler, "scheduler_error") as mock_scheduler_error:
with patch.object(scheduler, "scheduler_output") as mock_scheduler_output:
with patch.object(
scheduler,
"publish_scheduler_progress",
) as mock_progress:
with patch("builtins.print"):
scheduler.report_topology_runtime_not_ready(
"Scheduled shaping refresh deferred",
phase_label="Scheduler waiting for topology runtime",
step_index=3,
step_count=scheduler.SCHEDULER_REFRESH_STEP_COUNT,
)
mock_clear_error.assert_called_once()
mock_scheduler_error.assert_not_called()
mock_scheduler_output.assert_called_once()
mock_progress.assert_called_once_with(
False,
"waiting_for_topology_runtime",
"Scheduler waiting for topology runtime",
3,
scheduler.SCHEDULER_REFRESH_STEP_COUNT,
)
def test_failed_topology_runtime_deferred_wait_reports_scheduler_error(self):
with patch.object(
scheduler,
"topology_runtime_readiness_detail",
return_value=(
False,
"Topology runtime failed for the current source generation: publish failed",
"generation-1",
),
):
with patch.object(scheduler, "clear_scheduler_error") as mock_clear_error:
with patch.object(scheduler, "scheduler_error") as mock_scheduler_error:
with patch.object(scheduler, "scheduler_output") as mock_scheduler_output:
with patch.object(
scheduler,
"publish_scheduler_progress",
) as mock_progress:
with patch("builtins.print"):
scheduler.report_topology_runtime_not_ready(
"Scheduled shaping refresh deferred",
phase_label="Scheduler waiting for topology runtime",
step_index=3,
step_count=scheduler.SCHEDULER_REFRESH_STEP_COUNT,
)
mock_clear_error.assert_not_called()
mock_scheduler_error.assert_called_once_with(
"Scheduled shaping refresh deferred: Topology runtime failed for the current source generation: publish failed Generation generation-1."
)
mock_scheduler_output.assert_called_once_with(
"Scheduled shaping refresh deferred: Topology runtime failed for the current source generation: publish failed Generation generation-1."
)
mock_progress.assert_called_once_with(
False,
"degraded",
"Scheduler waiting for topology runtime",
scheduler.SCHEDULER_REFRESH_STEP_COUNT,
scheduler.SCHEDULER_REFRESH_STEP_COUNT,
percent=100,
)
def test_topology_runtime_refresh_tick_waits_for_partial_runtime_outputs(self):
scheduler.partial_topology_runtime_pending = True
scheduler.partial_topology_runtime_generation = "generation-1"
@@ -744,6 +900,59 @@ class TestSchedulerLogging(unittest.TestCase):
)
)
def test_partial_reload_waits_for_stale_shaping_inputs(self):
stale_exc = scheduler.RefreshFailure(
"Missing or stale shaping_inputs.json. Run topology runtime before shaping."
)
with patch.object(scheduler, "importFromCRM"):
with patch.object(scheduler, "ensure_topology_runtime_process", return_value=True):
with patch.object(scheduler, "calculate_shaping_runtime_hash", return_value=2):
with patch.object(scheduler, "shaping_runtime_hash", 1):
with patch.object(
scheduler,
"topology_runtime_readiness_detail",
return_value=(True, "", "generation-1"),
):
with patch.object(
scheduler,
"current_topology_source_generation",
return_value="generation-1",
):
with patch.object(
scheduler,
"refreshShapers",
side_effect=stale_exc,
):
with patch.object(
scheduler,
"clear_scheduler_error",
) as mock_clear_error:
with patch.object(
scheduler,
"report_scheduler_runtime_failure",
) as mock_report:
with patch.object(
scheduler,
"publish_scheduler_progress",
) as mock_progress:
with patch("builtins.print"):
scheduler.importAndShapePartialReload()
mock_clear_error.assert_called_once()
mock_report.assert_not_called()
self.assertTrue(scheduler.partial_topology_runtime_pending)
self.assertTrue(
any(
call.args[:3] == (
True,
"waiting_for_topology_runtime",
"Waiting for topology runtime",
)
for call in mock_progress.call_args_list
)
)
class TestTopologyRuntimeReadiness(unittest.TestCase):
def test_missing_status_is_not_ready(self):
@@ -898,6 +1107,7 @@ class TestTopologyRuntimeGating(unittest.TestCase):
def setUp(self):
scheduler._reset_startup_topology_runtime_wait()
scheduler._reset_partial_topology_runtime_wait()
scheduler.clear_integration_failure()
scheduler.shaping_runtime_hash = 0
def test_full_reload_skips_refresh_when_topology_runtime_not_ready(self):
@@ -953,6 +1163,54 @@ class TestTopologyRuntimeGating(unittest.TestCase):
mock_refresh.assert_not_called()
mock_report.assert_called_once()
def test_full_reload_waits_for_stale_shaping_inputs(self):
stale_exc = scheduler.RefreshFailure(
"Missing or stale shaping_inputs.json. Run topology runtime before shaping."
)
with patch.object(scheduler, "importFromCRM"):
with patch.object(scheduler, "ensure_topology_runtime_process", return_value=True):
with patch.object(scheduler, "enable_insight_topology", return_value=False):
with patch.object(
scheduler,
"topology_runtime_readiness_detail",
return_value=(True, "", "generation-1"),
):
with patch.object(
scheduler,
"current_topology_source_generation",
return_value="generation-1",
):
with patch.object(
scheduler,
"refreshShapers",
side_effect=stale_exc,
):
with patch.object(
scheduler,
"clear_scheduler_error",
) as mock_clear_error:
with patch.object(
scheduler,
"publish_scheduler_progress",
) as mock_progress:
with patch("builtins.print"):
ready = scheduler.importAndShapeFullReload()
self.assertFalse(ready)
mock_clear_error.assert_called_once()
self.assertTrue(scheduler.startup_topology_runtime_pending)
self.assertTrue(
any(
call.args[:3] == (
True,
"waiting_for_topology_runtime",
"Waiting for topology runtime",
)
for call in mock_progress.call_args_list
)
)
class TestSchedulerOverrideMerge(unittest.TestCase):
def test_merge_rows_replaces_matching_device_id(self):