diff --git a/.gitignore b/.gitignore index 0c20c865..2b3d9acb 100644 --- a/.gitignore +++ b/.gitignore @@ -44,6 +44,8 @@ src/statsByParentNode.json src/lastGoodConfig.json src/lastGoodConfig.csv src/mikrotikDHCPRouterList.csv +src/integrationUISPbandwidths.csv +src/integrationUISProutes.csv src/longTermStats.json src/queuingStructure.json src/tinsStats.json @@ -68,7 +70,7 @@ src/bin/lqos_node_manager src/bin/webusers src/bin/lqusers src/bin/Rocket.toml - +src/bin/lqos_map_perf # Ignore project folders for some IDEs .idea/ diff --git a/requirements.txt b/requirements.txt index 71d8b35f..ba497ca3 100644 --- a/requirements.txt +++ b/requirements.txt @@ -11,3 +11,4 @@ schedule waitress setuptools >= 65.5.1 apscheduler +deepdiff diff --git a/src/LibreQoS.py b/src/LibreQoS.py index ffd458d0..efa4039d 100755 --- a/src/LibreQoS.py +++ b/src/LibreQoS.py @@ -18,6 +18,7 @@ import argparse import logging import shutil import binpacking +from deepdiff import DeepDiff from ispConfig import sqm, upstreamBandwidthCapacityDownloadMbps, upstreamBandwidthCapacityUploadMbps, \ interfaceA, interfaceB, enableActualShellCommands, useBinPackingToBalanceCPU, monitorOnlyMode, \ @@ -917,14 +918,7 @@ def refreshShapersUpdateOnly(): else: print("ERROR: lqosd is not running. Aborting") os._exit(-1) - - # Check that we aren't running LibreQoS.py more than once at a time - if is_libre_already_running(): - print("LibreQoS.py is already running in another process. Aborting.") - os._exit(-1) - - # We've got this far, so create a lock file - create_lock_file() + # Starting print("refreshShapers starting at " + datetime.now().strftime("%d/%m/%Y %H:%M:%S")) @@ -943,35 +937,27 @@ def refreshShapersUpdateOnly(): # Check validation safeToRunRefresh = False if (validateNetworkAndDevices() == True): - shutil.copyfile('ShapedDevices.csv', 'lastGoodConfig.csv') - shutil.copyfile('network.json', 'lastGoodConfig.json') - print("Backed up good config as lastGoodConfig.csv and lastGoodConfig.json") safeToRunRefresh = True else: warnings.warn("Validation failed. Will now exit.", stacklevel=2) if safeToRunRefresh == True: + networkChanged = False + devicesChanged = False + # Check for changes to network.json + if os.path.isfile('lastGoodConfig.json'): + with open('lastGoodConfig.json', 'r') as j: + originalNetwork = json.loads(j.read()) + with open('network.json', 'r') as j: + newestNetwork = json.loads(j.read()) + ddiff = DeepDiff(originalNetwork, newestNetwork, ignore_order=True) + if ddiff != {}: + networkChanged = True - # Load queuingStructure - with open('queuingStructure.json', 'r') as infile: - queuingStructure = json.loads(infile.read()) - - - network = queuingStructure['Network'] - lastUsedClassIDCounterByCPU = queuingStructure['lastUsedClassIDCounterByCPU'] - generatedPNs = queuingStructure['generatedPNs'] - + # Check for changes to ShapedDevices.csv newlyUpdatedSubscriberCircuits, newlyUpdatedDictForCircuitsWithoutParentNodes = loadSubscriberCircuits('ShapedDevices.csv') lastLoadedSubscriberCircuits, lastLoadedDictForCircuitsWithoutParentNodes = loadSubscriberCircuits('ShapedDevices.lastLoaded.csv') - - # Load stats files - with open('statsByParentNode.json', 'r') as j: - parentNodes = json.loads(j.read()) - - with open('statsByCircuit.json', 'r') as j: - subscriberCircuits = json.loads(j.read()) - newlyUpdatedSubscriberCircuitsByID = {} for circuit in newlyUpdatedSubscriberCircuits: circuitid = circuit['circuitID'] @@ -982,327 +968,35 @@ def refreshShapersUpdateOnly(): circuitid = circuit['circuitID'] lastLoadedSubscriberCircuitsByID[circuitid] = circuit - - def removeDeviceIPsFromFilter(circuit): - for device in circuit['devices']: - for ipv4 in device['ipv4s']: - #shell('./bin/xdp_iphash_to_cpu_cmdline del ip ' + str(ipv4)) - delete_ip_mapping(str(ipv4)) - for ipv6 in device['ipv6s']: - #shell('./bin/xdp_iphash_to_cpu_cmdline del ip ' + str(ipv6)) - delete_ip_mapping(str(ipv6)) - - - def addDeviceIPsToFilter(circuit, cpuNumHex, classId, upload = False): - # `lqosd` expects the CPU and ClassIDs to be in Hex. - # `ClassId` wasn't present in the circuit data, so it's now a required parameter. - for device in circuit['devices']: - for ipv4 in device['ipv4s']: - #shell('./bin/xdp_iphash_to_cpu_cmdline add --ip ' + str(ipv4) + ' --cpu ' + cpuNumHex + ' --classid ' + circuit['classid']) - add_ip_mapping(str(ipv4), classId, str(cpuNumHex), upload) - for ipv6 in device['ipv6s']: - #shell('./bin/xdp_iphash_to_cpu_cmdline add --ip ' + str(ipv6) + ' --cpu ' + cpuNumHex + ' --classid ' + circuit['classid']) - add_ip_mapping(str(ipv6), classId, str(cpuNumHex), upload) - - - def getAllParentNodes(data, allParentNodes): - for node in data: - if isinstance(node, str): - if (isinstance(data[node], dict)) and (node != 'children'): - thisParentNode = node - if thisParentNode not in allParentNodes: - allParentNodes.append(thisParentNode) - if 'children' in data[node]: - result = getAllParentNodes(data[node]['children'], allParentNodes) - for entry in result: - if entry not in allParentNodes: - allParentNodes.append(entry) - return allParentNodes - allParentNodes = getAllParentNodes(network, []) - - - def getClassIDofParentNodes(data, classIDOfParentNodes): - revisedClassIDOfParentNodes = classIDOfParentNodes - for node in data: - if isinstance(node, str): - if (isinstance(data[node], dict)) and (node != 'children'): - thisParentNode = node - classIDOfParentNodes[thisParentNode] = data[node]['classid'] - if 'children' in data[node]: - result = getClassIDofParentNodes(data[node]['children'], revisedClassIDOfParentNodes) - revisedClassIDOfParentNodes.update(result) - return revisedClassIDOfParentNodes - classIDOfParentNodes = getClassIDofParentNodes(network, {}) - - - def getAllCircuitIDs(data, allCircuitIDs): - for node in data: - if isinstance(node, str): - if (isinstance(data[node], dict)) and (node != 'children'): - thisParentNode = node - if 'circuits' in data[node]: - for circuit in data[node]['circuits']: - if circuit['circuitID'] not in allCircuitIDs: - allCircuitIDs.append(circuit['circuitID']) - if 'children' in data[node]: - result = getAllCircuitIDs(data[node]['children'], allCircuitIDs) - for entry in result: - if entry not in allCircuitIDs: - allCircuitIDs.append(result) - return allCircuitIDs - allCircuitIDs = getAllCircuitIDs(network, []) - - - def getClassIDofExistingCircuitID(data, classIDofExistingCircuitID): - revisedClassIDofExistingCircuitID = classIDofExistingCircuitID - for node in data: - if isinstance(node, str): - if (isinstance(data[node], dict)) and (node != 'children'): - thisParentNode = node - if 'circuits' in data[node]: - for circuit in data[node]['circuits']: - revisedClassIDofExistingCircuitID[circuit['circuitID']] = circuit['classid'] - if 'children' in data[node]: - result = getClassIDofExistingCircuitID(data[node]['children'], revisedClassIDofExistingCircuitID) - revisedClassIDofExistingCircuitID.update(result) - return revisedClassIDofExistingCircuitID - classIDofExistingCircuitID = getClassIDofExistingCircuitID(network, {}) - - - def getParentNodeOfCircuitID(data, parentNodeOfCircuitID, allCircuitIDs): - for node in data: - if isinstance(node, str): - if (isinstance(data[node], dict)) and (node != 'children'): - thisParentNode = node - if 'circuits' in data[node]: - for circuit in data[node]['circuits']: - parentNodeOfCircuitID[circuit['circuitID']] = thisParentNode - if 'children' in data[node]: - result = getParentNodeOfCircuitID(data[node]['children'], parentNodeOfCircuitID, allCircuitIDs) - parentNodeOfCircuitID.update(result) - return parentNodeOfCircuitID - parentNodeOfCircuitID = getParentNodeOfCircuitID(network, {}, allCircuitIDs) - - - def getCPUnumOfParentNodes(data, cpuNumOfParentNode): - revisedCpuNumOfParentNode = cpuNumOfParentNode - for node in data: - if isinstance(node, str): - if (isinstance(data[node], dict)) and (node != 'children'): - thisParentNode = node - revisedCpuNumOfParentNode[thisParentNode] = data[node]['cpuNum'] - if 'children' in data[node]: - result = getCPUnumOfParentNodes(data[node]['children'], revisedCpuNumOfParentNode) - revisedCpuNumOfParentNode.update(result) - return revisedCpuNumOfParentNode - - cpuNumOfParentNodeHex = getCPUnumOfParentNodes(network, {}) - cpuNumOfParentNodeInt = {} - for key, value in cpuNumOfParentNodeHex.items(): - cpuNumOfParentNodeInt[key] = int(value,16) + 1 - - - def addCircuitHTBandQdisc(circuit, parentNodeClassID): - minor = circuit['classid'].split(':')[1] - for interface in [interfaceA, interfaceB]: - if interface == interfaceA: - rate = str(circuit['minDownload']) - ceil = str(circuit['maxDownload']) - else: - rate = str(circuit['minUpload']) - ceil = str(circuit['maxUpload']) - command = 'tc class add dev ' + interface + ' parent ' + parentNodeClassID + ' classid ' + minor + ' htb rate ' + rate + 'Mbit ceil ' + ceil + 'Mbit' - print(command) - shell(command) - # Only add CAKE / fq_codel qdisc if monitorOnlyMode is Off - if monitorOnlyMode == False: - command = 'tc qdisc add dev ' + interface + ' parent ' + classID + ' ' + sqm - print(command) - shell(command) - - - def delHTBclass(classid): - for interface in [interfaceA, interfaceB]: - command = 'tc class del dev ' + interface + ' classid ' + classid - print(command) - shell(command) - - - generatedPNcounter = 1 - circuitsIDsToRemove = [] - circuitsToUpdateByID = {} - circuitsToAddByParentNode = {} for circuitID, circuit in lastLoadedSubscriberCircuitsByID.items(): - # Same circuit, update parameters (bandwidth, devices) - bandwidthChanged = False - devicesChanged = False if (circuitID in newlyUpdatedSubscriberCircuitsByID) and (circuitID in lastLoadedSubscriberCircuitsByID): if newlyUpdatedSubscriberCircuitsByID[circuitID]['maxDownload'] != lastLoadedSubscriberCircuitsByID[circuitID]['maxDownload']: - bandwidthChanged = True + devicesChanged = True if newlyUpdatedSubscriberCircuitsByID[circuitID]['maxUpload'] != lastLoadedSubscriberCircuitsByID[circuitID]['maxUpload']: - bandwidthChanged = True + devicesChanged = True if newlyUpdatedSubscriberCircuitsByID[circuitID]['minDownload'] != lastLoadedSubscriberCircuitsByID[circuitID]['minDownload']: - bandwidthChanged = True + devicesChanged = True if newlyUpdatedSubscriberCircuitsByID[circuitID]['minUpload'] != lastLoadedSubscriberCircuitsByID[circuitID]['minUpload']: - bandwidthChanged = True + devicesChanged = True if newlyUpdatedSubscriberCircuitsByID[circuitID]['devices'] != lastLoadedSubscriberCircuitsByID[circuitID]['devices']: devicesChanged = True - if bandwidthChanged == True: - if newlyUpdatedSubscriberCircuitsByID[circuitID]['ParentNode'] == lastLoadedSubscriberCircuitsByID[circuitID]['ParentNode']: - parentNodeActual = circuit['ParentNode'] - if parentNodeActual == 'none': - # In flat network, there's still a Parent Node (GENERATED_PN_X) so we need to use that here - parentNodeActual = parentNodeOfCircuitID[circuitID] - parentNodeClassID = classIDOfParentNodes[parentNodeActual] - classid = classIDofExistingCircuitID[circuitID] - minor = classid.split(':')[1] - for interface in [interfaceA, interfaceB]: - if interface == interfaceA: - rate = str(newlyUpdatedSubscriberCircuitsByID[circuitID]['minDownload']) - ceil = str(newlyUpdatedSubscriberCircuitsByID[circuitID]['maxDownload']) - else: - rate = str(newlyUpdatedSubscriberCircuitsByID[circuitID]['minUpload']) - ceil = str(newlyUpdatedSubscriberCircuitsByID[circuitID]['maxUpload']) - command = 'tc class change dev ' + interface + ' parent ' + parentNodeClassID + ' classid ' + minor + ' htb rate ' + rate + 'Mbit ceil ' + ceil + 'Mbit' - shell(command) - else: - removeDeviceIPsFromFilter(lastLoadedSubscriberCircuitsByID[circuitID]) - # Delete HTB class, qdisc. Then recreat it. - classID = classIDofExistingCircuitID[circuitID] - circuit['classid'] = classID - delHTBclass(classID) - parentNodeClassID = classIDOfParentNodes[parentNodeOfCircuitID[circuitID]] - addCircuitHTBandQdisc(circuit, parentNodeClassID) - cpuNumHex = cpuNumOfParentNodeHex[parentNodeActual] - addDeviceIPsToFilter(newlyUpdatedSubscriberCircuitsByID[circuitID], cpuNumHex, classID) - if devicesChanged: - parentNodeActual = circuit['ParentNode'] - if parentNodeActual == 'none': - # In flat network, there's still a Parent Node (GENERATED_PN_X) so we need to use that here - parentNodeActual = parentNodeOfCircuitID[circuitID] - cpuNumHex = cpuNumOfParentNodeHex[parentNodeActual] - parentNodeClassID = classIDOfParentNodes[parentNodeActual] - addDeviceIPsToFilter(newlyUpdatedSubscriberCircuitsByID[circuitID], cpuNumHex, parentNodeClassID) - - newlyUpdatedSubscriberCircuitsByID[circuitID]['classid'] = lastLoadedSubscriberCircuitsByID[circuitID]['classid'] - if (bandwidthChanged) or (devicesChanged): - circuitsToUpdateByID[circuitID] = newlyUpdatedSubscriberCircuitsByID[circuitID] - - - # Removed circuit - if (circuitID in lastLoadedSubscriberCircuitsByID) and (circuitID not in newlyUpdatedSubscriberCircuitsByID): - circuitsIDsToRemove.append(circuitID) - removeDeviceIPsFromFilter(lastLoadedSubscriberCircuitsByID[circuitID]) - classid = classIDofExistingCircuitID[circuitID] - delHTBclass(classid) - - - # New circuit + if newlyUpdatedSubscriberCircuitsByID[circuitID]['ParentNode'] != lastLoadedSubscriberCircuitsByID[circuitID]['ParentNode']: + devicesChanged = True + else: + devicesChanged = True for circuitID, circuit in newlyUpdatedSubscriberCircuitsByID.items(): - if (circuitID in newlyUpdatedSubscriberCircuitsByID) and (circuitID not in lastLoadedSubscriberCircuitsByID): - if newlyUpdatedSubscriberCircuitsByID[circuitID]['ParentNode'] == 'none': - newlyUpdatedSubscriberCircuitsByID[circuitID]['ParentNode'] = 'Generated_PN_' + str(generatedPNcounter) - generatedPNcounter += 1 - if generatedPNcounter > len(generatedPNs): - generatedPNcounter = 1 - cpuNumHex = cpuNumOfParentNodeHex[circuit['ParentNode']] - - cpuNumInt = cpuNumOfParentNodeInt[circuit['ParentNode']] - parentNodeClassID = classIDOfParentNodes[circuit['ParentNode']] - classID = parentNodeClassID.split(':')[0] + ':' + hex(lastUsedClassIDCounterByCPU[str(cpuNumInt)]) - lastUsedClassIDCounterByCPU[str(cpuNumInt)] = lastUsedClassIDCounterByCPU[str(cpuNumInt)] + 1 - circuit['classid'] = classID - # Add HTB class, qdisc - addCircuitHTBandQdisc(circuit, parentNodeClassID) - addDeviceIPsToFilter(circuit, cpuNumHex, parentNodeClassID) - if circuit['ParentNode'] in circuitsToAddByParentNode: - temp = circuitsToAddByParentNode[circuit['ParentNode']] - temp.append(circuit) - circuitsToAddByParentNode[circuit['ParentNode']] = temp - else: - temp = [] - temp.append(circuit) - circuitsToAddByParentNode[circuit['ParentNode']] = temp + if (circuitID not in lastLoadedSubscriberCircuitsByID): + devicesChanged = True - # Update network structure reflecting new circuits, removals, and changes - itemsToChange = (circuitsIDsToRemove, circuitsToUpdateByID, circuitsToAddByParentNode) - def updateNetworkStructure(data, depth, itemsToChange): - circuitsIDsToRemove, circuitsToUpdateByID, circuitsToAddByParentNode = itemsToChange - parentNodesToAddCircuitsTo = [] - for thing in circuitsToAddByParentNode: - parentNodesToAddCircuitsTo.append(thing) - revisedData = data - for node in revisedData: - if isinstance(node, str): - if (isinstance(revisedData[node], dict)) and (node != 'children'): - thisParentNode = node - if 'circuits' in revisedData[node]: - for circuit in revisedData[node]['circuits']: - if circuit['circuitID'] in circuitsToUpdateByID: - circuit = circuitsToUpdateByID[circuit['circuitID']] - print('updated') - if circuit['circuitID'] in circuitsIDsToRemove: - revisedData[node]['circuits'].remove(circuit) - if thisParentNode in parentNodesToAddCircuitsTo: - if 'circuits' in revisedData[node]: - temp = revisedData[node]['circuits'] - for circuit in circuitsToAddByParentNode[thisParentNode]: - temp.append(circuit) - print('added ') - print(circuit) - revisedData[node]['circuits'] = temp - else: - temp = [] - for circuit in circuitsToAddByParentNode[thisParentNode]: - temp.append(circuit) - revisedData[node]['circuits'] = temp - if 'children' in revisedData[node]: - result = updateNetworkStructure(revisedData[node]['children'], depth+1, itemsToChange) - revisedData[node]['children'] = result - return revisedData - network = updateNetworkStructure(network, 0 , itemsToChange) - - - # Record start time of actual filter reload - reloadStartTime = datetime.now() - - - # Record end time of all reload commands - reloadEndTime = datetime.now() - - - queuingStructure = {} - queuingStructure['Network'] = network - queuingStructure['lastUsedClassIDCounterByCPU'] = lastUsedClassIDCounterByCPU - queuingStructure['generatedPNs'] = generatedPNs - # Save queuingStructure - with open('queuingStructure.json', 'w') as infile: - json.dump(queuingStructure, infile, indent=4) - - - # copy ShapedDevices.csv and save as ShapedDevices.lastLoaded.csv and lastGoodConfig.csv - shutil.copyfile('ShapedDevices.csv', 'ShapedDevices.lastLoaded.csv') - shutil.copyfile('ShapedDevices.csv', 'lastGoodConfig.csv') - - - # Save for stats - with open('statsByCircuit.json', 'w') as f: - f.write(json.dumps(subscriberCircuits, indent=4)) - with open('statsByParentNode.json', 'w') as f: - f.write(json.dumps(parentNodes, indent=4)) - - - # Report reload time - reloadTimeSeconds = ((reloadEndTime - reloadStartTime).seconds) + (((reloadEndTime - reloadStartTime).microseconds) / 1000000) - print("Queue and IP filter partial reload completed in " + "{:g}".format(round(reloadTimeSeconds,1)) + " seconds") - + if devicesChanged or networkChanged: + print('Observed changes to ShapedDevices.csv or network.json. Applying full reload now') + refreshShapers() + else: + print('Observed no changes to ShapedDevices.csv or network.json. Leaving queues as is.') # Done print("refreshShapersUpdateOnly completed on " + datetime.now().strftime("%d/%m/%Y %H:%M:%S")) - - # Free the lock file - free_lock_file() if __name__ == '__main__': # Check that the host lqosd is running diff --git a/src/build_dpkg.sh b/src/build_dpkg.sh index 8f7b7640..cc562d6d 100755 --- a/src/build_dpkg.sh +++ b/src/build_dpkg.sh @@ -22,7 +22,7 @@ ETC_DIR=$DPKG_DIR/etc MOTD_DIR=$DPKG_DIR/etc/update-motd.d LQOS_FILES="graphInfluxDB.py influxDBdashboardTemplate.json integrationCommon.py integrationRestHttp.py integrationSplynx.py integrationUISP.py ispConfig.example.py LibreQoS.py lqos.example lqTools.py mikrotikFindIPv6.py network.example.json pythonCheck.py README.md scheduler.py ShapedDevices.example.csv" LQOS_BIN_FILES="lqos_scheduler.service.example lqosd.service.example lqos_node_manager.service.example" -RUSTPROGS="lqosd lqtop xdp_iphash_to_cpu_cmdline xdp_pping lqos_node_manager lqusers lqos_setup" +RUSTPROGS="lqosd lqtop xdp_iphash_to_cpu_cmdline xdp_pping lqos_node_manager lqusers lqos_setup lqos_map_perf" #################################################### # Clean any previous dist build diff --git a/src/build_rust.sh b/src/build_rust.sh index cc5547eb..bebc1c58 100755 --- a/src/build_rust.sh +++ b/src/build_rust.sh @@ -52,7 +52,7 @@ rustup update # Start building echo "Please wait while the system is compiled. Service will not be interrupted during this stage." -PROGS="lqosd lqtop xdp_iphash_to_cpu_cmdline xdp_pping lqos_node_manager lqusers" +PROGS="lqosd lqtop xdp_iphash_to_cpu_cmdline xdp_pping lqos_node_manager lqusers lqos_map_perf" mkdir -p bin/static pushd rust > /dev/null #cargo clean diff --git a/src/csvToNetworkJSON.py b/src/csvToNetworkJSON.py new file mode 100644 index 00000000..b5e2de76 --- /dev/null +++ b/src/csvToNetworkJSON.py @@ -0,0 +1,58 @@ +from pythonCheck import checkPythonVersion +checkPythonVersion() +import os +import csv +import json +from ispConfig import uispSite, uispStrategy, overwriteNetworkJSONalways +from ispConfig import generatedPNUploadMbps, generatedPNDownloadMbps, upstreamBandwidthCapacityDownloadMbps, upstreamBandwidthCapacityUploadMbps +from integrationCommon import NetworkGraph, NetworkNode, NodeType + +def csvToNetworkJSONfile(): + sites = [] + with open('manualNetwork.csv') as csv_file: + csv_reader = csv.reader(csv_file, delimiter=',') + for row in csv_reader: + if 'Site Name' in row[0]: + header = row + else: + name, down, up, parent = row + site = {'name': name, + 'download': down, + 'upload': up, + 'parent': parent} + sites.append(site) + + net = NetworkGraph() + idCounter = 1000 + nameToID = {} + for site in sites: + site['id'] = idCounter + idCounter = idCounter + 1 + nameToID[site['name']] = site['id'] + for site in sites: + id = site['id'] + if site['parent'] == '': + parentID = None + else: + parentID = nameToID[site['parent']] + name = site['name'] + parent = site['parent'] + download = site['download'] + upload = site['upload'] + nodeType = NodeType.site + node = NetworkNode(id=id, displayName=name, type=nodeType, + parentId=parentID, download=download, upload=upload, address=None, customerName=None) + net.addRawNode(node) + net.prepareTree() + net.plotNetworkGraph(False) + if net.doesNetworkJsonExist(): + if overwriteNetworkJSONalways: + net.createNetworkJson() + else: + print("network.json already exists and overwriteNetworkJSONalways set to False. Leaving in-place.") + else: + net.createNetworkJson() + net.createShapedDevices() + +if __name__ == '__main__': + csvToNetworkJSONfile() diff --git a/src/integrationCommon.py b/src/integrationCommon.py index 6e89c3c9..6a771253 100644 --- a/src/integrationCommon.py +++ b/src/integrationCommon.py @@ -63,6 +63,22 @@ class NodeType(enum.IntEnum): clientWithChildren = 5 device = 6 +def nodeTypeToString(integer): + string = '' + match integer: + case 1: + string = 'root' + case 2: + string = 'site' + case 3: + string = 'ap' + case 4: + string = 'client' + case 5: + string = 'clientWithChildren' + case 6: + string = 'device' + return(string) class NetworkNode: # Defines a node on a LibreQoS network graph. @@ -311,6 +327,7 @@ class NetworkGraph: node = { "downloadBandwidthMbps": self.nodes[idx].downloadMbps, "uploadBandwidthMbps": self.nodes[idx].uploadMbps, + 'type': nodeTypeToString(self.nodes[idx].type), } children = {} hasChildren = False @@ -418,7 +435,7 @@ class NetworkGraph: import graphviz dot = graphviz.Digraph( - 'network', comment="Network Graph", engine="dot") + 'network', comment="Network Graph", engine="dot", graph_attr={'rankdir':'LR'}) for (i, node) in enumerate(self.nodes): if ((node.type != NodeType.client and node.type != NodeType.device) or showClients): @@ -436,6 +453,6 @@ class NetworkGraph: if child != i: if (self.nodes[child].type != NodeType.client and self.nodes[child].type != NodeType.device) or showClients: dot.edge("N" + str(i), "N" + str(child)) - - dot.render("network.pdf") + dot = dot.unflatten(stagger=3)#, fanout=True) + dot.render("network") diff --git a/src/integrationUISP.py b/src/integrationUISP.py index 079c3b4d..15870cfe 100644 --- a/src/integrationUISP.py +++ b/src/integrationUISP.py @@ -131,11 +131,20 @@ def findApCapacities(devices, siteBandwidth): if device['identification']['role'] == "ap": name = device['identification']['name'] if not name in siteBandwidth and device['overview']['downlinkCapacity'] and device['overview']['uplinkCapacity']: + safeToUse = True download = int(device['overview'] ['downlinkCapacity'] / 1000000) upload = int(device['overview']['uplinkCapacity'] / 1000000) - siteBandwidth[device['identification']['name']] = { - "download": download, "upload": upload} + if (download < 15) or (upload < 15): + print("WARNING: Device '" + device['identification']['hostname'] + "' has unusually low capacity (" + str(download) + '/' + str(upload) + " Mbps). Discarding in favor of parent site rates.") + safeToUse = False + if device['identification']['model'] == 'WaveAP': + if (download < 500) or (upload < 500): + download = 2450 + upload = 2450 + if safeToUse: + siteBandwidth[device['identification']['name']] = { + "download": download, "upload": upload} def findAirfibers(devices, generatedPNDownloadMbps, generatedPNUploadMbps): foundAirFibersBySite = {} @@ -147,19 +156,17 @@ def findAirfibers(devices, generatedPNDownloadMbps, generatedPNUploadMbps): if device['overview']['downlinkCapacity'] is not None and device['overview']['uplinkCapacity'] is not None: download = int(device['overview']['downlinkCapacity']/ 1000000) upload = int(device['overview']['uplinkCapacity']/ 1000000) - else: - download = generatedPNDownloadMbps - upload = generatedPNUploadMbps - # Make sure to use half of reported bandwidth for AF60-LRs - if device['identification']['model'] == "AF60-LR": - download = int(download / 2) - upload = int(download / 2) - if device['identification']['site']['id'] in foundAirFibersBySite: - if (download > foundAirFibersBySite[device['identification']['site']['id']]['download']) or (upload > foundAirFibersBySite[device['identification']['site']['id']]['upload']): - foundAirFibersBySite[device['identification']['site']['id']]['download'] = download - foundAirFibersBySite[device['identification']['site']['id']]['upload'] = upload - else: - foundAirFibersBySite[device['identification']['site']['id']] = {'download': download, 'upload': upload} + # Make sure to factor in gigabit port for AF60/AF60-LRs + if (device['identification']['model'] == "AF60-LR") or (device['identification']['model'] == "AF60"): + download = min(download,950) + upload = min(upload,950) + if device['identification']['site']['id'] in foundAirFibersBySite: + if (download > foundAirFibersBySite[device['identification']['site']['id']]['download']) or (upload > foundAirFibersBySite[device['identification']['site']['id']]['upload']): + foundAirFibersBySite[device['identification']['site']['id']]['download'] = download + foundAirFibersBySite[device['identification']['site']['id']]['upload'] = upload + #print(device['identification']['name'] + ' will override bandwidth for site ' + device['identification']['site']['name']) + else: + foundAirFibersBySite[device['identification']['site']['id']] = {'download': download, 'upload': upload} return foundAirFibersBySite def buildSiteList(sites, dataLinks): @@ -240,11 +247,66 @@ def loadRoutingOverrides(): #print(overrides) return overrides +def findNodesBranchedOffPtMP(siteList, dataLinks, sites, rootSite, foundAirFibersBySite): + nodeOffPtMP = {} + for site in siteList: + id = site['id'] + name = site['name'] + if id != rootSite['id']: + if id not in foundAirFibersBySite: + trueParent = findInSiteListById(siteList, id)['parent'] + #parent = findInSiteListById(siteList, id)['parent'] + if site['parent'] is not None: + parent = site['parent'] + for link in dataLinks: + if (link['to']['site'] is not None) and (link['to']['site']['identification'] is not None): + if ('identification' in link['to']['site']) and (link['to']['site']['identification'] is not None): + if ('identification' in link['from']['site']) and (link['from']['site']['identification'] is not None): + # Respect parent defined by topology and overrides + if link['from']['site']['identification']['id'] == trueParent: + if link['to']['site']['identification']['id'] == id: + if link['from']['device']['overview']['wirelessMode'] == 'ap-ptmp': + if 'overview' in link['to']['device']: + if ('downlinkCapacity' in link['to']['device']['overview']) and ('uplinkCapacity' in link['to']['device']['overview']): + if (link['to']['device']['overview']['downlinkCapacity'] is not None) and (link['to']['device']['overview']['uplinkCapacity'] is not None): + # Capacity of the PtMP client radio feeding the PoP will be used as the site bandwidth limit + download = int(round(link['to']['device']['overview']['downlinkCapacity']/1000000)) + upload = int(round(link['to']['device']['overview']['uplinkCapacity']/1000000)) + nodeOffPtMP[id] = {'download': download, + 'upload': upload + } + site['parent'] = parent + if site['type'] == 'site': + print('Site ' + name + ' will use PtMP AP as parent.') + return siteList, nodeOffPtMP + +def handleMultipleInternetNodes(sites, dataLinks, uispSite): + internetConnectedSites = [] + for link in dataLinks: + if link['canDelete'] == False: + if link['from']['device']['identification']['id'] == link['to']['device']['identification']['id']: + siteID = link['from']['site']['identification']['id'] + # Found internet link + internetConnectedSites.append(siteID) + if len(internetConnectedSites) > 1: + internetSite = {'id': '001', 'identification': {'id': '001', 'status': 'active', 'name': 'Internet', 'parent': None, 'type': 'site'}} + sites.append(internetSite) + uispSite = 'Internet' + for link in dataLinks: + if link['canDelete'] == False: + if link['from']['device']['identification']['id'] == link['to']['device']['identification']['id']: + link['from']['site']['identification']['id'] = '001' + link['from']['site']['identification']['name'] = 'Internet' + # Found internet link + internetConnectedSites.append(siteID) + print("Multiple Internet links detected. Will create 'Internet' root node.") + return(sites, dataLinks, uispSite) + def buildFullGraph(): # Attempts to build a full network graph, incorporating as much of the UISP # hierarchy as possible. from integrationCommon import NetworkGraph, NetworkNode, NodeType - from ispConfig import generatedPNUploadMbps, generatedPNDownloadMbps + from ispConfig import uispSite, generatedPNUploadMbps, generatedPNDownloadMbps # Load network sites print("Loading Data from UISP") @@ -252,19 +314,30 @@ def buildFullGraph(): devices = uispRequest("devices?withInterfaces=true&authorized=true") dataLinks = uispRequest("data-links?siteLinksOnly=true") - # Build Site Capacities - siteBandwidth = buildSiteBandwidths() - findApCapacities(devices, siteBandwidth) - foundAirFibersBySite = findAirfibers(devices, generatedPNDownloadMbps, generatedPNUploadMbps) + # If multiple Internet-connected sites, create Internet root node: + sites, dataLinks, uispSite = handleMultipleInternetNodes(sites, dataLinks, uispSite) + # Build Site Capacities + print("Compiling Site Bandwidths") + siteBandwidth = buildSiteBandwidths() + print("Finding AP Capacities") + findApCapacities(devices, siteBandwidth) # Create a list of just network sites + print('Creating list of sites') siteList = buildSiteList(sites, dataLinks) rootSite = findInSiteList(siteList, uispSite) + print("Finding PtP Capacities") + foundAirFibersBySite = findAirfibers(devices, generatedPNDownloadMbps, generatedPNUploadMbps) + print('Creating list of route overrides') routeOverrides = loadRoutingOverrides() if rootSite is None: print("ERROR: Unable to find root site in UISP") return + print('Creating graph') walkGraphOutwards(siteList, rootSite, routeOverrides) + print("Finding PtMP Capacities") + siteList, nodeOffPtMP = findNodesBranchedOffPtMP(siteList, dataLinks, sites, rootSite, foundAirFibersBySite) + # Debug code: dump the list of site parents # for s in siteList: # if s['parent'] == "": @@ -273,25 +346,8 @@ def buildFullGraph(): # p = findInSiteListById(siteList, s['parent'])['name'] # print(s['name'] + " (" + str(s['cost']) + ") <-- " + p) - # Find Nodes Connected By PtMP - nodeOffPtMP = {} - for site in sites: - id = site['identification']['id'] - name = site['identification']['name'] - type = site['identification']['type'] - parent = findInSiteListById(siteList, id)['parent'] - if type == 'site': - for link in dataLinks: - if link['from']['site'] is not None and link['from']['site']['identification']['id'] == parent: - if link['to']['site'] is not None and link['to']['site']['identification']['id'] == id: - if link['from']['device'] is not None and link['to']['device'] is not None and link['to']['device']['overview'] is not None and link['to']['device']['overview']['downlinkCapacity'] is not None and link['from']['device']['overview']['wirelessMode'] == 'ap-ptmp': - # Capacity of the PtMP client radio feeding the PoP will be used as the site bandwidth limit - download = int(round(link['to']['device']['overview']['downlinkCapacity']/1000000)) - upload = int(round(link['to']['device']['overview']['uplinkCapacity']/1000000)) - nodeOffPtMP[id] = { 'parent': link['from']['device']['identification']['id'], - 'download': download, - 'upload': upload - } + + print("Building Topology") net = NetworkGraph() # Add all sites and client sites @@ -312,21 +368,23 @@ def buildFullGraph(): match type: case "site": nodeType = NodeType.site - if id in nodeOffPtMP: - parent = nodeOffPtMP[id]['parent'] if name in siteBandwidth: # Use the CSV bandwidth values download = siteBandwidth[name]["download"] upload = siteBandwidth[name]["upload"] - elif id in nodeOffPtMP: - download = nodeOffPtMP[id]['download'] - upload = nodeOffPtMP[id]['upload'] - elif id in foundAirFibersBySite: - download = foundAirFibersBySite[id]['download'] - upload = foundAirFibersBySite[id]['upload'] - else: - # Add them just in case - siteBandwidth[name] = { + else: + # Use limits from foundAirFibersBySite + if id in foundAirFibersBySite: + download = foundAirFibersBySite[id]['download'] + upload = foundAirFibersBySite[id]['upload'] + # If no airFibers were found, and node originates off PtMP, treat as child node of that PtMP AP + else: + if id in nodeOffPtMP: + if (nodeOffPtMP[id]['download'] >= download) or (nodeOffPtMP[id]['upload'] >= upload): + download = nodeOffPtMP[id]['download'] + upload = nodeOffPtMP[id]['upload'] + + siteBandwidth[name] = { "download": download, "upload": upload} case default: nodeType = NodeType.client @@ -364,7 +422,7 @@ def buildFullGraph(): # TODO: Figure out Mikrotik IPv6? mac = device['identification']['mac'] - + net.addRawNode(NetworkNode(id=device['identification']['id'], displayName=device['identification'] ['name'], parentId=id, type=NodeType.device, ipv4=ipv4, ipv6=ipv6, mac=mac)) diff --git a/src/integrationUISProutes.csv b/src/integrationUISProutes.template.csv similarity index 100% rename from src/integrationUISProutes.csv rename to src/integrationUISProutes.template.csv diff --git a/src/ispConfig.example.py b/src/ispConfig.example.py index 9da3813c..53a010b8 100644 --- a/src/ispConfig.example.py +++ b/src/ispConfig.example.py @@ -23,6 +23,9 @@ interfaceA = 'eth1' # Interface connected to edge router interfaceB = 'eth2' +# Queue refresh scheduler (lqos_scheduler). Minutes between reloads. +queueRefreshIntervalMins = 30 + # WORK IN PROGRESS. Note that interfaceA determines the "stick" interface # I could only get scanning to work if I issued ethtool -K enp1s0f1 rxvlan off OnAStick = False diff --git a/src/manualNetwork.template.csv b/src/manualNetwork.template.csv new file mode 100644 index 00000000..79619eee --- /dev/null +++ b/src/manualNetwork.template.csv @@ -0,0 +1,4 @@ +Site Name,Download Mbps,Upload Mbps,Parent Site Name +SiteA,1000,1000, +SiteB,975,975,SiteA +SiteC,950,950,SiteB diff --git a/src/rust/Cargo.lock b/src/rust/Cargo.lock index 4066d6f0..d9995b2d 100644 --- a/src/rust/Cargo.lock +++ b/src/rust/Cargo.lock @@ -1942,6 +1942,13 @@ dependencies = [ "zerocopy", ] +[[package]] +name = "lqos_map_perf" +version = "0.1.0" +dependencies = [ + "lqos_sys", +] + [[package]] name = "lqos_node_manager" version = "0.1.0" @@ -2022,6 +2029,7 @@ dependencies = [ "nix", "once_cell", "thiserror", + "zerocopy", ] [[package]] diff --git a/src/rust/Cargo.toml b/src/rust/Cargo.toml index 54dab3f4..bb6e6b34 100644 --- a/src/rust/Cargo.toml +++ b/src/rust/Cargo.toml @@ -30,4 +30,5 @@ members = [ "long_term_stats/license_server", # Licensing Server for LibreQoS Long-term stats "long_term_stats/lts_node", # Long-term stats cluster node "long_term_stats/pgdb", # PostgreSQL interface for the LTS system + "lqos_map_perf", # A CLI tool for testing eBPF map performance ] diff --git a/src/rust/lqos_heimdall/src/flows.rs b/src/rust/lqos_heimdall/src/flows.rs index c88313e2..215e093d 100644 --- a/src/rust/lqos_heimdall/src/flows.rs +++ b/src/rust/lqos_heimdall/src/flows.rs @@ -1,43 +1,11 @@ use crate::{timeline::expire_timeline, FLOW_EXPIRE_SECS}; use dashmap::DashMap; use lqos_bus::{tos_parser, BusResponse, FlowTransport}; -use lqos_sys::bpf_per_cpu_map::BpfPerCpuMap; +use lqos_sys::heimdall_data::{HeimdallKey, HeimdallData}; use lqos_utils::{unix_time::time_since_boot, XdpIpAddress}; use once_cell::sync::Lazy; use std::{collections::HashSet, time::Duration}; -/// Representation of the eBPF `heimdall_key` type. -#[derive(Debug, Clone, Default, PartialEq, Eq, Hash)] -#[repr(C)] -pub struct HeimdallKey { - /// Mapped `XdpIpAddress` source for the flow. - pub src_ip: XdpIpAddress, - /// Mapped `XdpIpAddress` destination for the flow - pub dst_ip: XdpIpAddress, - /// IP protocol (see the Linux kernel!) - pub ip_protocol: u8, - /// Source port number, or ICMP type. - pub src_port: u16, - /// Destination port number. - pub dst_port: u16, -} - -/// Mapped representation of the eBPF `heimdall_data` type. -#[derive(Debug, Clone, Default)] -#[repr(C)] -pub struct HeimdallData { - /// Last seen, in nanoseconds (since boot time). - pub last_seen: u64, - /// Number of bytes since the flow started being tracked - pub bytes: u64, - /// Number of packets since the flow started being tracked - pub packets: u64, - /// IP header TOS value - pub tos: u8, - /// Reserved to pad the structure - pub reserved: [u8; 3], -} - #[derive(Clone, Debug, PartialEq, Eq, Hash)] struct FlowKey { src: XdpIpAddress, @@ -95,11 +63,12 @@ static FLOW_DATA: Lazy> = Lazy::new(DashMap::new); fn heimdall_for_each( callback: &mut dyn FnMut(&HeimdallKey, &[HeimdallData]), ) { - if let Ok(heimdall) = BpfPerCpuMap::::from_path( + /*if let Ok(heimdall) = BpfPerCpuMap::::from_path( "/sys/fs/bpf/heimdall", ) { heimdall.for_each(callback); - } + }*/ + lqos_sys::iterate_heimdall(callback); } diff --git a/src/rust/lqos_map_perf/Cargo.toml b/src/rust/lqos_map_perf/Cargo.toml new file mode 100644 index 00000000..34ec96e1 --- /dev/null +++ b/src/rust/lqos_map_perf/Cargo.toml @@ -0,0 +1,7 @@ +[package] +name = "lqos_map_perf" +version = "0.1.0" +edition = "2021" + +[dependencies] +lqos_sys = { path = "../lqos_sys" } diff --git a/src/rust/lqos_map_perf/src/main.rs b/src/rust/lqos_map_perf/src/main.rs new file mode 100644 index 00000000..c31c1433 --- /dev/null +++ b/src/rust/lqos_map_perf/src/main.rs @@ -0,0 +1,24 @@ +use std::time::Instant; + +use lqos_sys::{rtt_for_each, throughput_for_each}; + +fn main() { + println!("LibreQoS Map Performance Tool"); + + // Test the RTT map + let mut rtt_count = 0; + let now = Instant::now(); + rtt_for_each(&mut |_rtt, _tracker| { + rtt_count += 1; + }); + let elapsed = now.elapsed(); + println!("RTT map: {} entries in {} µs", rtt_count, elapsed.as_micros()); + + let mut tp_count = 0; + let now = Instant::now(); + throughput_for_each(&mut |_ip, _hosts| { + tp_count += 1; + }); + let elapsed = now.elapsed(); + println!("TP map: {} entries in {} µs", tp_count, elapsed.as_micros()); +} diff --git a/src/rust/lqos_sys/Cargo.toml b/src/rust/lqos_sys/Cargo.toml index a743824f..dcffa3a9 100644 --- a/src/rust/lqos_sys/Cargo.toml +++ b/src/rust/lqos_sys/Cargo.toml @@ -16,6 +16,7 @@ lqos_utils = { path = "../lqos_utils" } once_cell = "1" dashmap = "5" thiserror = "1" +zerocopy = { version = "0.6.1", features = ["simd"] } [build-dependencies] bindgen = "0" diff --git a/src/rust/lqos_sys/src/bpf/common/heimdall.h b/src/rust/lqos_sys/src/bpf/common/heimdall.h index 9657c8fb..9d5b8342 100644 --- a/src/rust/lqos_sys/src/bpf/common/heimdall.h +++ b/src/rust/lqos_sys/src/bpf/common/heimdall.h @@ -67,6 +67,7 @@ struct heimdall_key __u8 ip_protocol; __u16 src_port; __u16 dst_port; + __u8 pad; }; struct heimdall_data { @@ -155,6 +156,7 @@ static __always_inline void update_heimdall(struct dissector_t *dissector, __u32 { bpf_debug("Failed to insert tracking"); } + //bpf_debug("Inserted tracking"); } } else if (mode == 2) { struct heimdall_event event = {0}; diff --git a/src/rust/lqos_sys/src/bpf/common/maximums.h b/src/rust/lqos_sys/src/bpf/common/maximums.h index 4c20f219..7520cbc9 100644 --- a/src/rust/lqos_sys/src/bpf/common/maximums.h +++ b/src/rust/lqos_sys/src/bpf/common/maximums.h @@ -1,10 +1,10 @@ #pragma once // Maximum number of client IPs we are tracking -#define MAX_TRACKED_IPS 128000 +#define MAX_TRACKED_IPS 64000 // Maximum number of TC class mappings to support -#define IP_HASH_ENTRIES_MAX 128000 +#define IP_HASH_ENTRIES_MAX 64000 // Maximum number of supported CPUs #define MAX_CPUS 1024 diff --git a/src/rust/lqos_sys/src/bpf/lqos_kern.c b/src/rust/lqos_sys/src/bpf/lqos_kern.c index 9e4e5295..c4980a69 100644 --- a/src/rust/lqos_sys/src/bpf/lqos_kern.c +++ b/src/rust/lqos_sys/src/bpf/lqos_kern.c @@ -319,4 +319,99 @@ int bifrost(struct __sk_buff *skb) return TC_ACT_UNSPEC; } +/* + * Structs for map iteration programs + * See https://github.com/xdp-project/bpf-examples + */ +struct bpf_iter_meta { + struct seq_file *seq; + __u64 session_id; + __u64 seq_num; +} __attribute__((preserve_access_index)); + +struct bpf_iter__bpf_map_elem { + struct bpf_iter_meta *meta; + struct bpf_map *map; + void *key; + void *value; +}; + +volatile const int NUM_CPUS = 0; + +SEC("iter/bpf_map_elem") +int throughput_reader(struct bpf_iter__bpf_map_elem *ctx) +{ + // The sequence file + struct seq_file *seq = ctx->meta->seq; + void *counter = ctx->value; + struct in6_addr *ip = ctx->key; + __u32 num_cpus = NUM_CPUS; + + // Bail on end + if (counter == NULL || ip == NULL) { + return 0; + } + + if (ctx->meta->seq_num == 0) { + bpf_seq_write(seq, &num_cpus, sizeof(__u32)); + bpf_seq_write(seq, &num_cpus, sizeof(__u32)); // Repeat for padding + } + + bpf_seq_write(seq, ip, sizeof(struct in6_addr)); + for (__u32 i=0; idownload_bytes, counter->upload_bytes); + return 0; +} + +SEC("iter/bpf_map_elem") +int rtt_reader(struct bpf_iter__bpf_map_elem *ctx) +{ + // The sequence file + struct seq_file *seq = ctx->meta->seq; + struct rotating_performance *counter = ctx->value; + struct in6_addr *ip = ctx->key; + + // Bail on end + if (counter == NULL || ip == NULL) { + return 0; + } + + //BPF_SEQ_PRINTF(seq, "%d %d\n", counter->next_entry, counter->rtt[0]); + bpf_seq_write(seq, ip, sizeof(struct in6_addr)); + bpf_seq_write(seq, counter, sizeof(struct rotating_performance)); + return 0; +} + +SEC("iter/bpf_map_elem") +int heimdall_reader(struct bpf_iter__bpf_map_elem *ctx) { + // The sequence file + struct seq_file *seq = ctx->meta->seq; + void *counter = ctx->value; + struct heimdall_key *ip = ctx->key; + __u32 num_cpus = NUM_CPUS; + + if (ctx->meta->seq_num == 0) { + bpf_seq_write(seq, &num_cpus, sizeof(__u32)); + bpf_seq_write(seq, &num_cpus, sizeof(__u32)); // Repeat for padding + } + + // Bail on end + if (counter == NULL || ip == NULL) { + return 0; + } + + bpf_seq_write(seq, ip, sizeof(struct heimdall_key)); + for (__u32 i=0; idownload_bytes, counter->upload_bytes); + return 0; +} + char _license[] SEC("license") = "GPL"; diff --git a/src/rust/lqos_sys/src/bpf/wrapper.c b/src/rust/lqos_sys/src/bpf/wrapper.c index 8ca2f74d..94ae7914 100644 --- a/src/rust/lqos_sys/src/bpf/wrapper.c +++ b/src/rust/lqos_sys/src/bpf/wrapper.c @@ -246,3 +246,77 @@ int tc_attach_ingress(int ifindex, bool verbose, struct lqos_kern *obj) out: return err; } + +// Iterator code +#include +#include + +struct bpf_link *setup_iterator_link( + struct bpf_program *prog, + struct bpf_map *map +) { + int map_fd; // File descriptor for the map itself + struct bpf_link *link; // Value to return with the link + union bpf_iter_link_info linfo = { 0 }; + DECLARE_LIBBPF_OPTS(bpf_iter_attach_opts, iter_opts, + .link_info = &linfo, + .link_info_len = sizeof(linfo)); + + map_fd = bpf_map__fd(map); + if (map_fd < 0) { + fprintf(stderr, "bpf_map__fd() fails\n"); + return NULL; + } + linfo.map.map_fd = map_fd; + + link = bpf_program__attach_iter(prog, &iter_opts); + if (!link) { + fprintf(stderr, "bpf_program__attach_iter() fails\n"); + return NULL; + } + return link; +} + +int read_tp_buffer(struct bpf_program *prog, struct bpf_map *map) +{ + struct bpf_link *link; + char buf[16] = {}; + int iter_fd = -1, len; + int ret = 0; + int map_fd; + + union bpf_iter_link_info linfo = { 0 }; + DECLARE_LIBBPF_OPTS(bpf_iter_attach_opts, iter_opts, + .link_info = &linfo, + .link_info_len = sizeof(linfo)); + + map_fd = bpf_map__fd(map); + if (map_fd < 0) { + fprintf(stderr, "bpf_map__fd() fails\n"); + return map_fd; + } + linfo.map.map_fd = map_fd; + + link = bpf_program__attach_iter(prog, &iter_opts); + if (!link) { + fprintf(stderr, "bpf_program__attach_iter() fails\n"); + return -1; + } + iter_fd = bpf_iter_create(bpf_link__fd(link)); + if (iter_fd < 0) { + fprintf(stderr, "bpf_iter_create() fails\n"); + ret = -1; + goto free_link; + } + /* not check contents, but ensure read() ends without error */ + while ((len = read(iter_fd, buf, sizeof(buf) - 1)) > 0) { + buf[len] = 0; + printf("%s", buf); + } + printf("\n"); +free_link: + if (iter_fd >= 0) + close(iter_fd); + bpf_link__destroy(link); + return 0; +} \ No newline at end of file diff --git a/src/rust/lqos_sys/src/bpf/wrapper.h b/src/rust/lqos_sys/src/bpf/wrapper.h index a6c8e074..e24a3dc0 100644 --- a/src/rust/lqos_sys/src/bpf/wrapper.h +++ b/src/rust/lqos_sys/src/bpf/wrapper.h @@ -1,5 +1,8 @@ #include "lqos_kern_skel.h" #include +#include +#include +#include extern struct lqos_kern * lqos_kern_open(); extern int lqos_kern_load(struct lqos_kern * skel); @@ -8,4 +11,6 @@ extern int tc_detach_egress(int ifindex, bool verbose, bool flush_hook, const ch extern int tc_attach_ingress(int ifindex, bool verbose, struct lqos_kern *obj); extern int tc_detach_ingress(int ifindex, bool verbose, bool flush_hook, const char * ifname); extern __u64 max_tracker_ips(); -extern void do_not_print(); \ No newline at end of file +extern void do_not_print(); +int read_tp_buffer(struct bpf_program *prog, struct bpf_map *map); +struct bpf_link * setup_iterator_link(struct bpf_program *prog, struct bpf_map *map); diff --git a/src/rust/lqos_sys/src/bpf_iterator.rs b/src/rust/lqos_sys/src/bpf_iterator.rs new file mode 100644 index 00000000..f3c77c0e --- /dev/null +++ b/src/rust/lqos_sys/src/bpf_iterator.rs @@ -0,0 +1,264 @@ +use crate::{ + kernel_wrapper::BPF_SKELETON, lqos_kernel::bpf, HostCounter, + RttTrackingEntry, heimdall_data::{HeimdallKey, HeimdallData}, +}; +use lqos_utils::XdpIpAddress; +use once_cell::sync::Lazy; +use std::{ + fmt::Debug, fs::File, io::Read, marker::PhantomData, os::fd::FromRawFd, +}; +use thiserror::Error; +use zerocopy::FromBytes; + +/// Represents a link to an eBPF defined iterator. The iterators +/// must be available in the BPF skeleton, and the skeleton must +/// be loaded. These are designed to be lazy-initialized on a +/// per-map basis. The `MAP_TRAFFIC` and `RTT_ITERATOR` types +/// implement this type. +/// +/// Normal usage is to initialize the iterator and keep it around. +/// When you need to query the iterator, execute the `iter` method +/// and treat it as a normal Rust iterator. +struct BpfMapIterator { + link: *mut bpf::bpf_link, + _phantom: PhantomData<(KEY, VALUE)>, +} + +// The BPF map is re-entrant and thread safe. There's no clean +// way to represent this in Rust, so we just mark it as such. +unsafe impl Sync for BpfMapIterator {} +unsafe impl Send for BpfMapIterator {} + +impl BpfMapIterator +where + KEY: FromBytes + Debug + Clone, + VALUE: FromBytes + Debug + Clone + Default, +{ + /// Create a new link to an eBPF map, that *must* have an iterator + /// function defined in the eBPF program - and exposed in the + /// skeleton. + /// + /// # Safety + /// + /// * This is unsafe, it relies on the skeleton having been properly + /// initialized prior to using this type. + /// + /// # Arguments + /// + /// * `program` - The eBPF program that points to the iterator function. + /// * `map` - The eBPF map that the iterator function will iterate over. + fn new( + program: *mut bpf::bpf_program, + map: *mut bpf::bpf_map, + ) -> Result { + let link = unsafe { bpf::setup_iterator_link(program, map) }; + if !link.is_null() { + Ok(Self { link, _phantom: PhantomData }) + } else { + Err(BpfIteratorError::FailedToLink) + } + } + + /// Create a "link file descriptor", connecting the eBPF iterator + /// to a Linux file descriptor. This instantiates the iterator + /// in the kernel and allows us to read from it. + fn as_file(&self) -> Result { + let link_fd = unsafe { bpf::bpf_link__fd(self.link) }; + let iter_fd = unsafe { bpf::bpf_iter_create(link_fd) }; + if iter_fd < 0 { + log::error!("Unable to create map file descriptor"); + Err(BpfIteratorError::FailedToCreateFd) + } else { + unsafe { Ok(File::from_raw_fd(iter_fd)) } + } + } + + const KEY_SIZE: usize = std::mem::size_of::(); + const VALUE_SIZE: usize = std::mem::size_of::(); + const TOTAL_SIZE: usize = Self::KEY_SIZE + Self::VALUE_SIZE; + + fn for_each_per_cpu( + &self, + callback: &mut dyn FnMut(&KEY, &[VALUE]), + ) -> Result<(), BpfIteratorError> { + let mut file = self.as_file()?; + let mut buf = Vec::new(); + let bytes_read = file.read_to_end(&mut buf); + match bytes_read { + Err(e) => { + log::error!("Unable to read from kernel map iterator file"); + log::error!("{e:?}"); + Err(BpfIteratorError::UnableToCreateIterator) + } + Ok(bytes) => { + if bytes == 0 { + // Not having any data is not an error + return Ok(()); + } + let first_four_bytes: [u8; 4] = [buf[0], buf[1], buf[2], buf[3]]; + let num_cpus = u32::from_ne_bytes(first_four_bytes) as usize; + let mut index = 8; + while index + Self::TOTAL_SIZE <= buf.len() { + let key_start = index; + let key_end = key_start + Self::KEY_SIZE; + let key_slice = &buf[key_start..key_end]; + //println!("{:?}", unsafe { &key_slice.align_to::() }); + let (_head, key, _tail) = unsafe { &key_slice.align_to::() }; + + let value_start = key_end; + let value_end = value_start + (num_cpus * Self::VALUE_SIZE); + let value_slice = &buf[value_start..value_end]; + //println!("{:?}", unsafe { &value_slice.align_to::() }); + let (_head, values, _tail) = + unsafe { &value_slice.align_to::() }; + debug_assert_eq!(values.len(), num_cpus); + + callback(&key[0], values); + + index += Self::KEY_SIZE + (num_cpus * Self::VALUE_SIZE); + } + Ok(()) + } + } + } + + fn for_each( + &self, + callback: &mut dyn FnMut(&KEY, &VALUE), + ) -> Result<(), BpfIteratorError> { + let mut file = self.as_file()?; + let mut buf = Vec::new(); + let bytes_read = file.read_to_end(&mut buf); + match bytes_read { + Err(e) => { + log::error!("Unable to read from kernel map iterator file"); + log::error!("{e:?}"); + Err(BpfIteratorError::UnableToCreateIterator) + } + Ok(_) => { + let mut index = 0; + while index + Self::TOTAL_SIZE <= buf.len() { + let key_start = index; + let key_end = key_start + Self::KEY_SIZE; + let key_slice = &buf[key_start..key_end]; + let (_head, key, _tail) = unsafe { &key_slice.align_to::() }; + + let value_start = key_end; + let value_end = value_start + Self::VALUE_SIZE; + let value_slice = &buf[value_start..value_end]; + let (_head, values, _tail) = + unsafe { &value_slice.align_to::() }; + + callback(&key[0], &values[0]); + + index += Self::KEY_SIZE + Self::VALUE_SIZE; + } + Ok(()) + } + } + } +} + +/// When the iterator is dropped, we need to destroy the link. +/// This is handled by the kernel when the program is unloaded. +impl Drop for BpfMapIterator { + fn drop(&mut self) { + unsafe { + bpf::bpf_link__destroy(self.link); + } + } +} + +#[derive(Debug, Error)] +enum BpfIteratorError { + #[error("Failed to create iterator link")] + FailedToLink, + #[error("Failed to create file descriptor")] + FailedToCreateFd, + #[error("Iterator error")] + UnableToCreateIterator, +} + +static mut MAP_TRAFFIC: Lazy< + Option>, +> = Lazy::new(|| None); + +static mut RTT_TRACKER: Lazy< + Option>, +> = Lazy::new(|| None); + +static mut HEIMDALL_TRACKER: Lazy< + Option>, +> = Lazy::new(|| None); + +pub unsafe fn iterate_throughput( + callback: &mut dyn FnMut(&XdpIpAddress, &[HostCounter]), +) { + if MAP_TRAFFIC.is_none() { + let lock = BPF_SKELETON.lock().unwrap(); + if let Some(skeleton) = lock.as_ref() { + let skeleton = skeleton.get_ptr(); + if let Ok(iter) = unsafe { + BpfMapIterator::new( + (*skeleton).progs.throughput_reader, + (*skeleton).maps.map_traffic, + ) + } { + *MAP_TRAFFIC = Some(iter); + } + } + } + + if let Some(iter) = MAP_TRAFFIC.as_mut() { + let _ = iter.for_each_per_cpu(callback); + } +} + +pub unsafe fn iterate_rtt( + callback: &mut dyn FnMut(&XdpIpAddress, &RttTrackingEntry), +) { + if RTT_TRACKER.is_none() { + let lock = BPF_SKELETON.lock().unwrap(); + if let Some(skeleton) = lock.as_ref() { + let skeleton = skeleton.get_ptr(); + if let Ok(iter) = unsafe { + BpfMapIterator::new( + (*skeleton).progs.rtt_reader, + (*skeleton).maps.rtt_tracker, + ) + } { + *RTT_TRACKER = Some(iter); + } + } + } + + if let Some(iter) = RTT_TRACKER.as_mut() { + let _ = iter.for_each(callback); + } +} + +/// Iterate through the heimdall map and call the callback for each entry. +pub fn iterate_heimdall( + callback: &mut dyn FnMut(&HeimdallKey, &[HeimdallData]), +) { + unsafe { + if HEIMDALL_TRACKER.is_none() { + let lock = BPF_SKELETON.lock().unwrap(); + if let Some(skeleton) = lock.as_ref() { + let skeleton = skeleton.get_ptr(); + if let Ok(iter) = { + BpfMapIterator::new( + (*skeleton).progs.heimdall_reader, + (*skeleton).maps.heimdall, + ) + } { + *HEIMDALL_TRACKER = Some(iter); + } + } + } + + if let Some(iter) = HEIMDALL_TRACKER.as_mut() { + let _ = iter.for_each_per_cpu(callback); + } + } +} \ No newline at end of file diff --git a/src/rust/lqos_sys/src/bpf_map.rs b/src/rust/lqos_sys/src/bpf_map.rs index 6f6f3043..29bb8c24 100644 --- a/src/rust/lqos_sys/src/bpf_map.rs +++ b/src/rust/lqos_sys/src/bpf_map.rs @@ -72,34 +72,6 @@ where result } - /// Iterates the undlering BPF map, and sends references to the - /// results directly to a callback - pub fn for_each(&self, callback: &mut dyn FnMut(&K, &V)) { - let mut prev_key: *mut K = null_mut(); - let mut key: K = K::default(); - let key_ptr: *mut K = &mut key; - let mut value = V::default(); - let value_ptr: *mut V = &mut value; - - unsafe { - while bpf_map_get_next_key( - self.fd, - prev_key as *mut c_void, - key_ptr as *mut c_void, - ) == 0 - { - bpf_map_lookup_elem( - self.fd, - key_ptr as *mut c_void, - value_ptr as *mut c_void, - ); - //result.push((key.clone(), value.clone())); - callback(&key, &value); - prev_key = key_ptr; - } - } - } - /// Inserts an entry into a BPF map. /// Use this sparingly, because it briefly pauses XDP access to the /// underlying map (through internal locking we can't reach from diff --git a/src/rust/lqos_sys/src/bpf_per_cpu_map.rs b/src/rust/lqos_sys/src/bpf_per_cpu_map.rs deleted file mode 100644 index 5fdc31a2..00000000 --- a/src/rust/lqos_sys/src/bpf_per_cpu_map.rs +++ /dev/null @@ -1,77 +0,0 @@ -use anyhow::{Error, Result}; -use libbpf_sys::{ - bpf_map_get_next_key, bpf_map_lookup_elem, bpf_obj_get, -}; -use std::fmt::Debug; -use std::{ - ffi::{c_void, CString}, - marker::PhantomData, - ptr::null_mut, -}; - -use crate::num_possible_cpus; - -/// Represents an underlying BPF map, accessed via the filesystem. -/// `BpfMap` *only* talks to PER-CPU variants of maps. -/// -/// `K` is the *key* type, indexing the map. -/// `V` is the *value* type, and must exactly match the underlying C data type. -pub struct BpfPerCpuMap { - fd: i32, - _key_phantom: PhantomData, - _val_phantom: PhantomData, -} - -impl BpfPerCpuMap -where - K: Default + Clone, - V: Default + Clone + Debug, -{ - /// Connect to a PER-CPU BPF map via a filename. Connects the internal - /// file descriptor, which is held until the structure is - /// dropped. The index of the CPU is *not* specified. - pub fn from_path(filename: &str) -> Result { - let filename_c = CString::new(filename)?; - let fd = unsafe { bpf_obj_get(filename_c.as_ptr()) }; - if fd < 0 { - Err(Error::msg("Unable to open BPF map")) - } else { - Ok(Self { fd, _key_phantom: PhantomData, _val_phantom: PhantomData }) - } - } - - /// Instead of clonining into a vector - /// and allocating, calls `callback` for each key/value slice - /// with references to the data returned from C. - pub fn for_each(&self, callback: &mut dyn FnMut(&K, &[V])) { - let num_cpus = num_possible_cpus().unwrap(); - let mut prev_key: *mut K = null_mut(); - let mut key: K = K::default(); - let key_ptr: *mut K = &mut key; - let mut value = vec![V::default(); num_cpus as usize]; - let value_ptr = value.as_mut_ptr(); - - unsafe { - while bpf_map_get_next_key( - self.fd, - prev_key as *mut c_void, - key_ptr as *mut c_void, - ) == 0 - { - bpf_map_lookup_elem( - self.fd, - key_ptr as *mut c_void, - value_ptr as *mut c_void, - ); - callback(&key, &value); - prev_key = key_ptr; - } - } - } -} - -impl Drop for BpfPerCpuMap { - fn drop(&mut self) { - let _ = nix::unistd::close(self.fd); - } -} diff --git a/src/rust/lqos_sys/src/heimdall_data.rs b/src/rust/lqos_sys/src/heimdall_data.rs new file mode 100644 index 00000000..31be4bca --- /dev/null +++ b/src/rust/lqos_sys/src/heimdall_data.rs @@ -0,0 +1,33 @@ +use lqos_utils::XdpIpAddress; +use zerocopy::FromBytes; + +/// Representation of the eBPF `heimdall_key` type. +#[derive(Debug, Clone, Default, PartialEq, Eq, Hash, FromBytes)] +#[repr(C)] +pub struct HeimdallKey { + /// Mapped `XdpIpAddress` source for the flow. + pub src_ip: XdpIpAddress, + /// Mapped `XdpIpAddress` destination for the flow + pub dst_ip: XdpIpAddress, + /// IP protocol (see the Linux kernel!) + pub ip_protocol: u8, + /// Source port number, or ICMP type. + pub src_port: u16, + /// Destination port number. + pub dst_port: u16, + _padding: u8, +} + +/// Mapped representation of the eBPF `heimdall_data` type. +#[derive(Debug, Clone, Default, FromBytes)] +#[repr(C)] +pub struct HeimdallData { + /// Last seen, in nanoseconds (since boot time). + pub last_seen: u64, + /// Number of bytes since the flow started being tracked + pub bytes: u64, + /// Number of packets since the flow started being tracked + pub packets: u64, + /// IP header TOS value + pub tos: u8, +} \ No newline at end of file diff --git a/src/rust/lqos_sys/src/kernel_wrapper.rs b/src/rust/lqos_sys/src/kernel_wrapper.rs index 6375a83f..1c6cb133 100644 --- a/src/rust/lqos_sys/src/kernel_wrapper.rs +++ b/src/rust/lqos_sys/src/kernel_wrapper.rs @@ -1,8 +1,28 @@ +use std::sync::Mutex; +use once_cell::sync::Lazy; use crate::lqos_kernel::{ attach_xdp_and_tc_to_interface, unload_xdp_from_interface, - InterfaceDirection, bpf::ring_buffer_sample_fn, + InterfaceDirection, bpf::{ring_buffer_sample_fn, self}, }; +/// Safer wrapper around pointers to `bpf::lqos_kern`. It really isn't +/// a great idea to be passing mutable pointers around like this, but the C +/// world insists on it. +pub(crate) struct LqosKernBpfWrapper { + ptr: *mut bpf::lqos_kern, +} + +impl LqosKernBpfWrapper { + pub(crate) fn get_ptr(&self) -> *mut bpf::lqos_kern { + self.ptr + } +} + +unsafe impl Sync for LqosKernBpfWrapper {} +unsafe impl Send for LqosKernBpfWrapper {} + +pub(crate) static BPF_SKELETON: Lazy>> = Lazy::new(|| Mutex::new(None)); + /// A wrapper-type that stores the interfaces to which the XDP and TC programs should /// be attached. Performs the attachment process, and hooks "drop" to unattach the /// programs when the structure falls out of scope. @@ -31,7 +51,7 @@ impl LibreQoSKernels { to_isp: to_isp.to_string(), on_a_stick: false, }; - attach_xdp_and_tc_to_interface( + let skeleton = attach_xdp_and_tc_to_interface( &kernel.to_internet, InterfaceDirection::Internet, heimdall_event_handler, @@ -41,6 +61,7 @@ impl LibreQoSKernels { InterfaceDirection::IspNetwork, heimdall_event_handler, )?; + BPF_SKELETON.lock().unwrap().replace(LqosKernBpfWrapper { ptr: skeleton }); Ok(kernel) } @@ -66,12 +87,12 @@ impl LibreQoSKernels { to_isp: String::new(), on_a_stick: true, }; - attach_xdp_and_tc_to_interface( + let skeleton = attach_xdp_and_tc_to_interface( &kernel.to_internet, InterfaceDirection::OnAStick(internet_vlan, isp_vlan), heimdall_event_handler, )?; - + BPF_SKELETON.lock().unwrap().replace(LqosKernBpfWrapper { ptr: skeleton }); Ok(kernel) } } diff --git a/src/rust/lqos_sys/src/lib.rs b/src/rust/lqos_sys/src/lib.rs index 40b56dae..d8ac2c3e 100644 --- a/src/rust/lqos_sys/src/lib.rs +++ b/src/rust/lqos_sys/src/lib.rs @@ -11,10 +11,6 @@ mod bifrost_maps; /// built-in, compiled eBPF programs. This is very-low level and should /// be handled with caution. pub mod bpf_map; -/// Provides direct access to LibBPF functionality, as exposed by the -/// built-in, compiled eBPF programs. This is very-low level and should -/// be handled with caution. -pub mod bpf_per_cpu_map; mod cpu_map; mod ip_mapping; mod kernel_wrapper; @@ -22,6 +18,10 @@ mod lqos_kernel; mod tcp_rtt; mod throughput; mod linux; +mod bpf_iterator; +/// Data shared between eBPF and Heimdall that needs local access +/// for map control. +pub mod heimdall_data; pub use ip_mapping::{ add_ip_to_tc, clear_ips_from_tc, del_ip_from_tc, list_mapped_ips, @@ -31,3 +31,4 @@ pub use linux::num_possible_cpus; pub use lqos_kernel::max_tracked_ips; pub use tcp_rtt::{rtt_for_each, RttTrackingEntry}; pub use throughput::{throughput_for_each, HostCounter}; +pub use bpf_iterator::iterate_heimdall; \ No newline at end of file diff --git a/src/rust/lqos_sys/src/lqos_kernel.rs b/src/rust/lqos_sys/src/lqos_kernel.rs index 89a132d3..d3db199a 100644 --- a/src/rust/lqos_sys/src/lqos_kernel.rs +++ b/src/rust/lqos_sys/src/lqos_kernel.rs @@ -14,6 +14,8 @@ use log::{info, warn}; use nix::libc::{geteuid, if_nametoindex}; use std::{ffi::{CString, c_void}, process::Command}; +use self::bpf::{lqos_kern, libbpf_num_possible_cpus}; + pub(crate) mod bpf { #![allow(warnings, unused)] include!(concat!(env!("OUT_DIR"), "/bindings.rs")); @@ -114,13 +116,14 @@ pub fn attach_xdp_and_tc_to_interface( interface_name: &str, direction: InterfaceDirection, heimdall_event_handler: bpf::ring_buffer_sample_fn, -) -> Result<()> { +) -> Result<*mut lqos_kern> { check_root()?; // Check the interface is valid let interface_index = interface_name_to_index(interface_name)?; set_strict_mode()?; let skeleton = unsafe { let skeleton = open_kernel()?; + (*(*skeleton).rodata).NUM_CPUS = libbpf_num_possible_cpus(); (*(*skeleton).data).direction = match direction { InterfaceDirection::Internet => 1, InterfaceDirection::IspNetwork => 2, @@ -229,7 +232,8 @@ pub fn attach_xdp_and_tc_to_interface( } } - Ok(()) + + Ok(skeleton) } unsafe fn attach_xdp_best_available( diff --git a/src/rust/lqos_sys/src/tcp_rtt.rs b/src/rust/lqos_sys/src/tcp_rtt.rs index 5742f17b..467440f1 100644 --- a/src/rust/lqos_sys/src/tcp_rtt.rs +++ b/src/rust/lqos_sys/src/tcp_rtt.rs @@ -1,8 +1,10 @@ -use crate::bpf_map::BpfMap; +use lqos_utils::XdpIpAddress; +use zerocopy::FromBytes; +use crate::bpf_iterator::iterate_rtt; /// Entry from the XDP rtt_tracker map. #[repr(C)] -#[derive(Clone, Copy, Debug)] +#[derive(Clone, Copy, Debug, FromBytes)] pub struct RttTrackingEntry { /// Array containing TCP round-trip times. Convert to an `f32` and divide by `100.0` for actual numbers. pub rtt: [u32; 60], @@ -29,10 +31,8 @@ impl Default for RttTrackingEntry { /// Only IP addresses facing the ISP Network side are tracked. /// /// Executes `callback` for each entry. -pub fn rtt_for_each(callback: &mut dyn FnMut(&[u8; 16], &RttTrackingEntry)) { - if let Ok(rtt_tracker) = - BpfMap::<[u8; 16], RttTrackingEntry>::from_path("/sys/fs/bpf/rtt_tracker") - { - rtt_tracker.for_each(callback); +pub fn rtt_for_each(callback: &mut dyn FnMut(&XdpIpAddress, &RttTrackingEntry)) { + unsafe { + iterate_rtt(callback); } } diff --git a/src/rust/lqos_sys/src/throughput.rs b/src/rust/lqos_sys/src/throughput.rs index 75c458b7..f6115ee5 100644 --- a/src/rust/lqos_sys/src/throughput.rs +++ b/src/rust/lqos_sys/src/throughput.rs @@ -1,10 +1,9 @@ use lqos_utils::XdpIpAddress; - -use crate::{bpf_per_cpu_map::BpfPerCpuMap}; +use zerocopy::FromBytes; /// Representation of the XDP map from map_traffic #[repr(C)] -#[derive(Debug, Clone, Default)] +#[derive(Debug, Clone, Default, FromBytes)] pub struct HostCounter { /// Download bytes counter (keeps incrementing) pub download_bytes: u64, @@ -30,9 +29,7 @@ pub struct HostCounter { pub fn throughput_for_each( callback: &mut dyn FnMut(&XdpIpAddress, &[HostCounter]), ) { - if let Ok(throughput) = BpfPerCpuMap::::from_path( - "/sys/fs/bpf/map_traffic", - ) { - throughput.for_each(callback); + unsafe { + crate::bpf_iterator::iterate_throughput(callback); } } diff --git a/src/rust/lqosd/src/throughput_tracker/tracking_data.rs b/src/rust/lqosd/src/throughput_tracker/tracking_data.rs index 73af07fa..f204d8df 100644 --- a/src/rust/lqosd/src/throughput_tracker/tracking_data.rs +++ b/src/rust/lqosd/src/throughput_tracker/tracking_data.rs @@ -131,8 +131,8 @@ impl ThroughputTracker { net_json.add_throughput_cycle( parents, ( - entry.bytes.0 - entry.prev_bytes.0, - entry.bytes.1 - entry.prev_bytes.1, + entry.bytes.0.saturating_sub(entry.prev_bytes.0), + entry.bytes.1.saturating_sub(entry.prev_bytes.1), ), ); } @@ -171,10 +171,9 @@ impl ThroughputTracker { pub(crate) fn apply_rtt_data(&self) { let self_cycle = self.cycle.load(std::sync::atomic::Ordering::Relaxed); - rtt_for_each(&mut |raw_ip, rtt| { + rtt_for_each(&mut |ip, rtt| { if rtt.has_fresh_data != 0 { - let ip = XdpIpAddress(*raw_ip); - if let Some(mut tracker) = self.raw_data.get_mut(&ip) { + if let Some(mut tracker) = self.raw_data.get_mut(ip) { tracker.recent_rtt_data = rtt.rtt; tracker.last_fresh_rtt_data_cycle = self_cycle; if let Some(parents) = &tracker.network_json_parents { diff --git a/src/scheduler.py b/src/scheduler.py index c1235812..0c1de7ba 100644 --- a/src/scheduler.py +++ b/src/scheduler.py @@ -3,6 +3,10 @@ import datetime from LibreQoS import refreshShapers, refreshShapersUpdateOnly from graphInfluxDB import refreshBandwidthGraphs, refreshLatencyGraphs from ispConfig import influxDBEnabled, automaticImportUISP, automaticImportSplynx +try: + from ispConfig import queueRefreshIntervalMins +except: + queueRefreshIntervalMins = 30 if automaticImportUISP: from integrationUISP import importFromUISP if automaticImportSplynx: @@ -44,7 +48,7 @@ def importAndShapePartialReload(): if __name__ == '__main__': importAndShapeFullReload() - ads.add_job(importAndShapePartialReload, 'interval', minutes=30) + ads.add_job(importAndShapePartialReload, 'interval', minutes=queueRefreshIntervalMins) if influxDBEnabled: ads.add_job(graphHandler, 'interval', seconds=10)