Merge branch 'main' of github.com:LibreQoE/LibreQoS

README.md

@@ -19,7 +19,7 @@ Please support the continued development of LibreQoS by sponsoring us via [GitHu
 Our Matrix chat channel is available at [https://matrix.to/#/#libreqos:matrix.org](https://matrix.to/#/#libreqos:matrix.org).
 
 
-<img alt="LibreQoS" src="https://user-images.githubusercontent.com/22501920/216796542-e3a2d4c3-a735-4cd1-adf8-5dae160fd7e9.png"></a>
+<img alt="LibreQoS" src="https://user-images.githubusercontent.com/22501920/223866474-603e1112-e2e6-4c67-93e4-44c17b1b7c43.png"></a>
 
 ## Features
 ### Flexible Hierarchical Shaping / Back-Haul Congestion Mitigation

src/rust/Cargo.lock

@@ -117,9 +117,9 @@ dependencies = [
 
 [[package]]
 name = "async-trait"
-version = "0.1.64"
+version = "0.1.66"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1cd7fce9ba8c3c042128ce72d8b2ddbf3a05747efb67ea0313c635e10bda47a2"
+checksum = "b84f9ebcc6c1f5b8cb160f6990096a5c127f423fcb6e1ccc46c370cbdfb75dfc"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -1169,9 +1169,9 @@ dependencies = [
 
 [[package]]
 name = "itoa"
-version = "1.0.5"
+version = "1.0.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fad582f4b9e86b6caa621cabeb0963332d92eea04729ab12892c2533951e6440"
+checksum = "453ad9f582a441959e5f0d088b02ce04cfe8d51a8eaf077f12ac6d3e94164ca6"
 
 [[package]]
 name = "jemalloc-sys"
@@ -1334,6 +1334,7 @@ dependencies = [
  "ip_network_table",
  "log",
  "serde",
+ "serde_json",
  "sha2",
  "thiserror",
  "toml 0.7.2",
@@ -1436,7 +1437,6 @@ dependencies = [
  "anyhow",
  "env_logger",
  "jemallocator",
- "lazy_static",
  "log",
  "lqos_bus",
  "lqos_config",
@@ -1444,6 +1444,7 @@ dependencies = [
  "lqos_sys",
  "lqos_utils",
  "nix",
+ "once_cell",
  "parking_lot",
  "rayon",
  "serde",
@@ -1699,9 +1700,9 @@ dependencies = [
 
 [[package]]
 name = "paste"
-version = "1.0.11"
+version = "1.0.12"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d01a5bd0424d00070b0098dd17ebca6f961a959dead1dbcbbbc1d1cd8d3deeba"
+checksum = "9f746c4065a8fa3fe23974dd82f15431cc8d40779821001404d10d2e79ca7d79"
 
 [[package]]
 name = "pear"
@@ -1949,9 +1950,9 @@ dependencies = [
 
 [[package]]
 name = "rayon"
-version = "1.6.1"
+version = "1.7.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6db3a213adf02b3bcfd2d3846bb41cb22857d131789e01df434fb7e7bc0759b7"
+checksum = "1d2df5196e37bcc87abebc0053e20787d73847bb33134a69841207dd0a47f03b"
 dependencies = [
  "either",
  "rayon-core",
@@ -1959,9 +1960,9 @@ dependencies = [
 
 [[package]]
 name = "rayon-core"
-version = "1.10.2"
+version = "1.11.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "356a0625f1954f730c0201cdab48611198dc6ce21f4acff55089b5a78e6e835b"
+checksum = "4b8f95bd6966f5c87776639160a66bd8ab9895d9d4ab01ddba9fc60661aebe8d"
 dependencies = [
  "crossbeam-channel",
  "crossbeam-deque",
@@ -1980,18 +1981,18 @@ dependencies = [
 
 [[package]]
 name = "ref-cast"
-version = "1.0.14"
+version = "1.0.15"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8c78fb8c9293bcd48ef6fce7b4ca950ceaf21210de6e105a883ee280c0f7b9ed"
+checksum = "a9af2cf09ef80e610097515e80095b7f76660a92743c4185aff5406cd5ce3dd5"
 dependencies = [
  "ref-cast-impl",
 ]
 
 [[package]]
 name = "ref-cast-impl"
-version = "1.0.14"
+version = "1.0.15"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9f9c0c92af03644e4806106281fe2e068ac5bc0ae74a707266d06ea27bccee5f"
+checksum = "9c501201393982e275433bc55de7d6ae6f00e7699cd5572c5b57581cd69c881b"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -2152,9 +2153,9 @@ checksum = "08d43f7aa6b08d49f382cde6a7982047c3426db949b1424bc4b7ec9ae12c6ce2"
 
 [[package]]
 name = "rustix"
-version = "0.36.8"
+version = "0.36.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f43abb88211988493c1abb44a70efa56ff0ce98f233b7b276146f1f3f7ba9644"
+checksum = "fd5c6ff11fecd55b40746d1995a02f2eb375bf8c00d192d521ee09f42bef37bc"
 dependencies = [
  "bitflags",
  "errno",
@@ -2166,15 +2167,15 @@ dependencies = [
 
 [[package]]
 name = "rustversion"
-version = "1.0.11"
+version = "1.0.12"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5583e89e108996506031660fe09baa5011b9dd0341b89029313006d1fb508d70"
+checksum = "4f3208ce4d8448b3f3e7d168a73f5e0c43a61e32930de3bceeccedb388b6bf06"
 
 [[package]]
 name = "ryu"
-version = "1.0.12"
+version = "1.0.13"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7b4b9743ed687d4b4bcedf9ff5eaa7398495ae14e61cba0a295704edbc7decde"
+checksum = "f91339c0467de62360649f8d3e185ca8de4224ff281f66000de5eb2a77a79041"
 
 [[package]]
 name = "same-file"
@@ -2219,9 +2220,9 @@ dependencies = [
 
 [[package]]
 name = "serde_json"
-version = "1.0.93"
+version = "1.0.94"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "cad406b69c91885b5107daf2c29572f6c8cdb3c66826821e286c533490c0bc76"
+checksum = "1c533a59c9d8a93a09c6ab31f0fd5e5f4dd1b8fc9434804029839884765d04ea"
 dependencies = [
  "itoa",
  "ryu",
@@ -2310,9 +2311,9 @@ checksum = "a507befe795404456341dfab10cef66ead4c041f62b8b11bbb92bffe5d0953e0"
 
 [[package]]
 name = "socket2"
-version = "0.4.7"
+version = "0.4.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "02e2d2db9033d13a1567121ddd7a095ee144db4e1ca1b1bda3419bc0da294ebd"
+checksum = "64a4a911eed85daf18834cfaa86a79b7d266ff93ff5ba14005426219480ed662"
 dependencies = [
  "libc",
  "winapi",
@@ -2373,9 +2374,9 @@ dependencies = [
 
 [[package]]
 name = "sysinfo"
-version = "0.28.1"
+version = "0.28.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "38a81bbc26c485910df47772df6bbcdb417036132caa9e51e29d2e39c4636d4e"
+checksum = "d3e847e2de7a137c8c2cede5095872dbb00f4f9bf34d061347e36b43322acd56"
 dependencies = [
  "cfg-if",
  "core-foundation-sys",
@@ -2443,18 +2444,18 @@ checksum = "222a222a5bfe1bba4a77b45ec488a741b3cb8872e5e499451fd7d0129c9c7c3d"
 
 [[package]]
 name = "thiserror"
-version = "1.0.38"
+version = "1.0.39"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6a9cd18aa97d5c45c6603caea1da6628790b37f7a34b6ca89522331c5180fed0"
+checksum = "a5ab016db510546d856297882807df8da66a16fb8c4101cb8b30054b0d5b2d9c"
 dependencies = [
  "thiserror-impl",
 ]
 
 [[package]]
 name = "thiserror-impl"
-version = "1.0.38"
+version = "1.0.39"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1fb327af4685e4d03fa8cbcf1716380da910eeb2bb8be417e7f9fd3fb164f36f"
+checksum = "5420d42e90af0c38c3290abcca25b9b3bdf379fc9f55c528f53a269d9c9a267e"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -2721,9 +2722,9 @@ dependencies = [
 
 [[package]]
 name = "unicode-ident"
-version = "1.0.6"
+version = "1.0.8"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "84a22b9f218b40614adcb3f4ff08b703773ad44fa9423e4e0d346d5db86e4ebc"
+checksum = "e5464a87b239f13a63a501f2701565754bae92d243d4bb7eb12f6d57d2269bf4"
 
 [[package]]
 name = "unicode-segmentation"
@@ -3050,9 +3051,9 @@ checksum = "447660ad36a13288b1db4d4248e857b510e8c3a225c822ba4fb748c0aafecffd"
 
 [[package]]
 name = "winnow"
-version = "0.3.3"
+version = "0.3.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "faf09497b8f8b5ac5d3bb4d05c0a99be20f26fd3d5f2db7b0716e946d5103658"
+checksum = "ee7b2c67f962bf5042bfd8b6a916178df33a26eec343ae064cb8e069f638fa6f"
 dependencies = [
  "memchr",
 ]

src/rust/lqos_bus/src/bus/request.rs

@@ -111,6 +111,12 @@ pub enum BusRequest {
   /// Request that the Rust side of things validate the CSV
   ValidateShapedDevicesCsv,
 
+  /// Request details of part of the network tree
+  GetNetworkMap{
+    /// The parent of the map to retrieve
+    parent: usize
+  },
+
   /// If running on Equinix (the `equinix_test` feature is enabled),
   /// display a "run bandwidht test" link.
   #[cfg(feature = "equinix_tests")]

src/rust/lqos_bus/src/bus/response.rs

@@ -68,4 +68,7 @@ pub enum BusResponse {
   /// A string containing a JSON dump of a queue stats. Analagos to
   /// the response from `tc show qdisc`.
   RawQueueData(String),
+
+  /// Results from network map queries
+  NetworkMap(Vec<(usize, lqos_config::NetworkJsonNode)>),
 }

src/rust/lqos_bus/src/bus/unix_socket_server.rs

@@ -12,7 +12,7 @@ use tokio::{
 
 use super::BUS_SOCKET_DIRECTORY;
 
-const READ_BUFFER_SIZE: usize = 2048000;
+const READ_BUFFER_SIZE: usize = 20_480;
 
 /// Implements a Tokio-friendly server using Unix Sockets and the bus protocol.
 /// Requests are handled and then forwarded to the handler.

src/rust/lqos_bus/src/ip_stats.rs

@@ -8,6 +8,9 @@ pub struct IpStats {
   /// The host's IP address, as detected by the XDP program.
   pub ip_address: String,
 
+  /// The host's mapped circuit ID
+  pub circuit_id: String,
+
   /// The current bits-per-second passing through this host. Tuple
   /// 0 is download, tuple 1 is upload.
   pub bits_per_second: (u64, u64),

src/rust/lqos_config/Cargo.toml

@@ -7,6 +7,7 @@ edition = "2021"
 thiserror = "1"
 toml = "0"
 serde = { version = "1.0", features = [ "derive" ] }
+serde_json = "1"
 csv = "1"
 ip_network_table = "0"
 ip_network = "0"

src/rust/lqos_config/src/lib.rs

@@ -11,12 +11,14 @@ mod etc;
 mod libre_qos_config;
 mod program_control;
 mod shaped_devices;
+mod network_json;
 
 pub use authentication::{UserRole, WebUsers};
 pub use etc::{BridgeConfig, BridgeInterface, BridgeVlan, EtcLqos, Tunables};
 pub use libre_qos_config::LibreQoSConfig;
 pub use program_control::load_libreqos;
 pub use shaped_devices::{ConfigShapedDevices, ShapedDevice};
+pub use network_json::{NetworkJson, NetworkJsonNode};
 
 /// Used as a constant in determining buffer preallocation
 pub const SUPPORTED_CUSTOMERS: usize = 16_000_000;

src/rust/lqos_config/src/network_json/mod.rs

@@ -0,0 +1,230 @@
+use crate::etc;
+use log::{error, info, warn};
+use serde::{Deserialize, Serialize};
+use serde_json::{Map, Value};
+use std::{
+  fs,
+  path::{Path, PathBuf},
+};
+use thiserror::Error;
+
+/// Describes a node in the network map tree.
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+pub struct NetworkJsonNode {
+  /// The node name, as it appears in `network.json`
+  pub name: String,
+
+  /// The maximum throughput allowed per `network.json` for this node
+  pub max_throughput: (u32, u32), // In mbps
+
+  /// Current throughput (in bytes/second) at this node
+  pub current_throughput: (u64, u64), // In bytes
+
+  /// Approximate RTTs reported for this level of the tree.
+  /// It's never going to be as statistically accurate as the actual
+  /// numbers, being based on medians.
+  pub rtts: Vec<f32>,
+
+  /// A list of indices in the `NetworkJson` vector of nodes
+  /// linking to parent nodes
+  pub parents: Vec<usize>,
+
+  /// The immediate parent node
+  pub immediate_parent: Option<usize>,
+}
+
+/// Holder for the network.json representation.
+/// This is condensed into a single level vector with index-based referencing
+/// for easy use in funnel calculations.
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+pub struct NetworkJson {
+  nodes: Vec<NetworkJsonNode>,
+}
+
+impl Default for NetworkJson {
+  fn default() -> Self {
+    Self::new()
+  }
+}
+
+impl NetworkJson {
+  /// Generates an empty network.json
+  pub fn new() -> Self {
+    Self { nodes: Vec::new() }
+  }
+
+  /// The path to the current `network.json` file, determined
+  /// by acquiring the prefix from the `/etc/lqos.conf` configuration
+  /// file.
+  pub fn path() -> Result<PathBuf, NetworkJsonError> {
+    let cfg =
+      etc::EtcLqos::load().map_err(|_| NetworkJsonError::ConfigLoadError)?;
+    let base_path = Path::new(&cfg.lqos_directory);
+    Ok(base_path.join("network.json"))
+  }
+
+  /// Does network.json exist?
+  pub fn exists() -> bool {
+    if let Ok(path) = Self::path() {
+      path.exists()
+    } else {
+      false
+    }
+  }
+
+  /// Attempt to load network.json from disk
+  pub fn load() -> Result<Self, NetworkJsonError> {
+    let mut nodes = vec![NetworkJsonNode {
+      name: "Root".to_string(),
+      max_throughput: (0, 0),
+      current_throughput: (0, 0),
+      parents: Vec::new(),
+      immediate_parent: None,
+      rtts: Vec::new(),
+    }];
+    if !Self::exists() {
+      return Err(NetworkJsonError::FileNotFound);
+    }
+    let path = Self::path()?;
+    let raw = fs::read_to_string(path)
+      .map_err(|_| NetworkJsonError::ConfigLoadError)?;
+    let json: Value = serde_json::from_str(&raw)
+      .map_err(|_| NetworkJsonError::ConfigLoadError)?;
+
+    // Start reading from the top. We are at the root node.
+    let parents = vec![0];
+    if let Value::Object(map) = &json {
+      for (key, value) in map.iter() {
+        if let Value::Object(inner_map) = value {
+          recurse_node(&mut nodes, key, inner_map, &parents, 0);
+        }
+      }
+    }
+
+    Ok(Self { nodes })
+  }
+
+  /// Find the index of a circuit_id
+  pub fn get_index_for_name(&self, name: &str) -> Option<usize> {
+    self.nodes.iter().position(|n| n.name == name)
+  }
+
+  /// Retrieve a cloned copy of a NetworkJsonNode entry, or None if there isn't
+  /// an entry at that index.
+  pub fn get_cloned_entry_by_index(
+    &self,
+    index: usize,
+  ) -> Option<NetworkJsonNode> {
+    self.nodes.get(index).cloned()
+  }
+
+  /// Retrieve a cloned copy of all children with a parent containing a specific
+  /// node index.
+  pub fn get_cloned_children(&self, index: usize) -> Vec<(usize, NetworkJsonNode)> {
+    self
+      .nodes
+      .iter()
+      .enumerate()
+      .filter(|(_i,n)| n.immediate_parent == Some(index))
+      .map(|(i, n)| (i, n.clone()))
+      .collect()
+  }
+
+  /// Find a circuit_id, and if it exists return its list of parent nodes
+  /// as indices within the network_json layout.
+  pub fn get_parents_for_circuit_id(
+    &self,
+    circuit_id: &str,
+  ) -> Option<Vec<usize>> {
+    self
+      .nodes
+      .iter()
+      .find(|n| n.name == circuit_id)
+      .map(|node| node.parents.clone())
+  }
+
+  /// Sets all current throughput values to zero
+  pub fn zero_throughput_and_rtt(&mut self) {
+    self.nodes.iter_mut().for_each(|n| {
+      n.current_throughput = (0, 0);
+      n.rtts.clear();
+    });
+  }
+
+  /// Add throughput numbers to node entries
+  pub fn add_throughput_cycle(
+    &mut self,
+    targets: &[usize],
+    bytes: (u64, u64),
+    median_rtt: f32,
+  ) {
+    for idx in targets {
+      // Safety first: use "get" to ensure that the node exists
+      if let Some(node) = self.nodes.get_mut(*idx) {
+        node.current_throughput.0 += bytes.0;
+        node.current_throughput.1 += bytes.1;
+        if median_rtt > 0.0 {
+          node.rtts.push(median_rtt);
+        }
+      } else {
+        warn!("No network tree entry for index {idx}");
+      }
+    }
+  }
+}
+
+fn json_to_u32(val: Option<&Value>) -> u32 {
+  if let Some(val) = val {
+    if let Some(n) = val.as_u64() {
+      n as u32
+    } else {
+      0
+    }
+  } else {
+    0
+  }
+}
+
+fn recurse_node(
+  nodes: &mut Vec<NetworkJsonNode>,
+  name: &str,
+  json: &Map<String, Value>,
+  parents: &[usize],
+  immediate_parent: usize,
+) {
+  info!("Mapping {name} from network.json");
+  let node = NetworkJsonNode {
+    parents: parents.to_vec(),
+    max_throughput: (
+      json_to_u32(json.get("downloadBandwidthMbps")),
+      json_to_u32(json.get("uploadBandwidthMbps")),
+    ),
+    current_throughput: (0, 0),
+    name: name.to_string(),
+    immediate_parent: Some(immediate_parent),
+    rtts: Vec::new(),
+  };
+
+  let my_id = nodes.len();
+  nodes.push(node);
+  let mut parents = parents.to_vec();
+  parents.push(my_id);
+
+  // Recurse children
+  for (key, value) in json.iter() {
+    let key_str = key.as_str();
+    if key_str != "uploadBandwidthMbps" && key_str != "downloadBandwidthMbps" {
+      if let Value::Object(value) = value {
+        recurse_node(nodes, key, value, &parents, my_id);
+      }
+    }
+  }
+}
+
+#[derive(Error, Debug)]
+pub enum NetworkJsonError {
+  #[error("Unable to find or load network.json")]
+  ConfigLoadError,
+  #[error("network.json not found or does not exist")]
+  FileNotFound,
+}

src/rust/lqos_node_manager/src/main.rs

@@ -10,6 +10,7 @@ use rocket_async_compression::Compression;
 mod auth_guard;
 mod config_control;
 mod queue_info;
+mod network_tree;
 
 // Use JemAllocator only on supported platforms
 #[cfg(any(target_arch = "x86", target_arch = "x86_64"))]
@@ -38,6 +39,7 @@ fn rocket() -> _ {
         static_pages::unknown_devices_page,
         static_pages::circuit_queue,
         config_control::config_page,
+        network_tree::tree_page,
         // Our JS library
         static_pages::lqos_js,
         static_pages::lqos_css,
@@ -77,6 +79,7 @@ fn rocket() -> _ {
         auth_guard::admin_check,
         static_pages::login_page,
         auth_guard::username,
+        network_tree::tree_entry,
         // Supporting files
         static_pages::bootsrap_css,
         static_pages::plotly_js,

src/rust/lqos_node_manager/src/network_tree.rs

@@ -0,0 +1,26 @@
+use lqos_bus::{bus_request, BusRequest, BusResponse};
+use lqos_config::NetworkJsonNode;
+use rocket::{fs::NamedFile, serde::json::Json};
+
+use crate::cache_control::NoCache;
+
+// Note that NoCache can be replaced with a cache option
+// once the design work is complete.
+#[get("/tree")]
+pub async fn tree_page<'a>() -> NoCache<Option<NamedFile>> {
+  NoCache::new(NamedFile::open("static/tree.html").await.ok())
+}
+
+#[get("/api/network_tree/<parent>")]
+pub async fn tree_entry(
+  parent: usize,
+) -> NoCache<Json<Vec<(usize, NetworkJsonNode)>>> {
+  let responses =
+    bus_request(vec![BusRequest::GetNetworkMap { parent }]).await.unwrap();
+  let result = match &responses[0] {
+    BusResponse::NetworkMap(nodes) => nodes.to_owned(),
+    _ => Vec::new(),
+  };
+
+  NoCache::new(Json(result))
+}

src/rust/lqos_node_manager/src/tracker/mod.rs

@@ -13,7 +13,6 @@ use lqos_bus::{IpStats, TcHandle};
 use lqos_config::LibreQoSConfig;
 use parking_lot::Mutex;
 use rocket::serde::{json::Json, Deserialize, Serialize};
-use std::net::IpAddr;
 
 #[derive(Serialize, Deserialize, Clone, Debug)]
 #[serde(crate = "rocket::serde")]
@@ -35,23 +34,28 @@ impl From<&IpStats> for IpStatsWithPlan {
       packets_per_second: i.packets_per_second,
       median_tcp_rtt: i.median_tcp_rtt,
       tc_handle: i.tc_handle,
-      circuit_id: String::new(),
+      circuit_id: i.circuit_id.clone(),
       plan: (0, 0),
     };
-    if let Ok(ip) = result.ip_address.parse::<IpAddr>() {
+
-      let lookup = match ip {
+    if !result.circuit_id.is_empty() {
-        IpAddr::V4(ip) => ip.to_ipv6_mapped(),
+      if let Some(circuit) = SHAPED_DEVICES
-        IpAddr::V6(ip) => ip,
+        .read()
-      };
+        .devices
-      let cfg = SHAPED_DEVICES.read();
+        .iter()
-      if let Some((_, id)) = cfg.trie.longest_match(lookup) {
+        .find(|sd| sd.circuit_id == result.circuit_id)
+      {
+        let name = if circuit.circuit_name.len() > 20 {
+          &circuit.circuit_name[0..20]
+        } else {
+          &circuit.circuit_name
+        };
         result.ip_address =
-          format!("{} ({})", cfg.devices[*id].circuit_name, result.ip_address);
+          format!("{} ({})", name, result.ip_address);
-        result.plan.0 = cfg.devices[*id].download_max_mbps;
+        result.plan = (circuit.download_max_mbps, circuit.download_min_mbps);
-        result.plan.1 = cfg.devices[*id].upload_max_mbps;
-        result.circuit_id = cfg.devices[*id].circuit_id.clone();
       }
     }
+
     result
   }
 }

src/rust/lqos_node_manager/static/main.html

@@ -26,9 +26,9 @@
                         <a class="nav-link active" aria-current="page" href="/"><i class="fa fa-home"></i> Dashboard</a>
                     </li>
                     <li class="nav-item" id="currentLogin"></li>
-                    <!--<li class="nav-item">
+                    <li class="nav-item">
-                        <a class="nav-link" href="#"><i class="fa fa-globe"></i> Network Layout</a>
+                        <a class="nav-link" href="/tree?parent=0"><i class="fa fa-globe"></i> Network Layout</a>
-                    </li>-->
+                    </li>
                     <li class="nav-item">
                         <a class="nav-link" href="/shaped"><i class="fa fa-users"></i> Shaped Devices <span id="shapedCount" class="badge badge-pill badge-success green-badge">?</span></a>
                     </li>

src/rust/lqos_node_manager/static/tree.html

@@ -0,0 +1,129 @@
+<!doctype html>
+<html lang="en">
+<head>
+    <meta charset="utf-8">
+    <meta name="viewport" content="width=device-width, initial-scale=1">
+    <link href="/vendor/bootstrap.min.css" rel="stylesheet">
+    <link rel="stylesheet" href="/vendor/solid.min.css">
+    <link rel="stylesheet" href="/lqos.css">
+    <title>LibreQoS - Local Node Manager</title>
+    <script src="/lqos.js"></script>
+    <script src="/vendor/plotly-2.16.1.min.js"></script>
+    <script src="/vendor/jquery.min.js"></script>
+    <script defer src="/vendor/bootstrap.bundle.min.js"></script>
+</head>
+<body class="bg-secondary">
+    <!-- Navigation -->
+    <nav class="navbar navbar-expand-lg navbar-dark bg-dark">
+        <div class="container-fluid">
+            <a class="navbar-brand" href="/"><img src="/vendor/tinylogo.svg" alt="LibreQoS SVG Logo" width="25" height="25" />&nbsp;LibreQoS</a>
+            <button class="navbar-toggler" type="button" data-bs-toggle="collapse" data-bs-target="#navbarSupportedContent" aria-controls="navbarSupportedContent" aria-expanded="false" aria-label="Toggle navigation">
+                <span class="navbar-toggler-icon"></span>
+            </button>
+            <div class="collapse navbar-collapse" id="navbarSupportedContent">
+                <ul class="navbar-nav me-auto mb-2 mb-lg-0">
+                    <li class="nav-item">
+                        <a class="nav-link active" aria-current="page" href="/"><i class="fa fa-home"></i> Dashboard</a>
+                    </li>
+                    <li class="nav-item" id="currentLogin"></li>
+                    <li class="nav-item">
+                        <a class="nav-link" href="/tree?parent=0"><i class="fa fa-globe"></i> Network Layout</a>
+                    </li>
+                    <li class="nav-item">
+                        <a class="nav-link" href="/shaped"><i class="fa fa-users"></i> Shaped Devices <span id="shapedCount" class="badge badge-pill badge-success green-badge">?</span></a>
+                    </li>
+                    <li class="nav-item">
+                        <a class="nav-link" href="/unknown"><i class="fa fa-address-card"></i> Unknown IPs <span id="unshapedCount" class="badge badge-warning orange-badge">?</span></a>
+                    </li>
+                </ul>
+            </div>
+
+            <ul class="navbar-nav ms-auto">
+                <li class="nav-item">
+                    <a class="nav-link" href="#" id="startTest"><i class="fa fa-flag-checkered"></i> Run Bandwidth Test</a>
+                </li>
+                <li class="nav-item ms-auto">
+                    <a class="nav-link" href="/config"><i class="fa fa-gear"></i> Configuration</a>
+                </li>
+                <li>
+                    <a class="nav-link btn btn-small" href="#" id="btnReload"><i class="fa fa-refresh"></i> Reload LibreQoS</a>
+                </li>
+            </ul>
+        </div>
+    </nav>
+
+    <div id="container" class="pad4">
+
+        <div class="row top-shunt">
+            <div class="col-sm-12 bg-light center-txt">
+                THIS NODE
+            </div>
+        </div>
+
+        <div class="row">
+            <div class="col-sm-12 bg-light center-txt">
+                <div id="treeList"></div>
+            </div>
+        </div>
+
+    </div>
+
+    <footer>&copy; 2022-2023, LibreQoE LLC</footer>
+
+    <script>
+        let node = 0;
+
+        function getTree() {
+            $.get("/api/network_tree/" + node, (data) => {
+                //console.log(data);
+
+                let tbl = "<table class='table'>";
+                tbl += "<thead><th>Circuit</th><th>Limit</th><th>Download</th><th>Upload</th><th>RTT Latency</th></thead>";
+                for (let i=1; i<data.length; ++i) {
+                    tbl += "<tr>";
+                    tbl += "<td style='width: 20%'><a href='/tree?parent=" + encodeURI(data[i][0]) + "'>" + data[i][1].name + "</a></td>";
+                    if (data[i][1].max_throughput[0]==0 && data[i][1].max_throughput[1] == 0) {
+                        tbl += "<td>No Limit</td>";
+                    } else {
+                        let down = scaleNumber(data[i][1].max_throughput[0] * 1000000);
+                        let up = scaleNumber(data[i][1].max_throughput[1] * 1000000);
+                        tbl += "<td>" + down + " / " + up + "</td>";
+                    }
+                    let down = scaleNumber(data[i][1].current_throughput[0] * 8);
+                    let up = scaleNumber(data[i][1].current_throughput[1] * 8);
+                    tbl += "<td>" + down + "</td>";
+                    tbl += "<td>" + up + "</td>";
+                    let rtt = "-";
+                    if (data[i][1].rtts.length > 0) {
+                        let sum = 0;
+                        for (let j=0; j<data[i][1].rtts.length; ++j) {
+                            sum += data[i][1].rtts[j];
+                        }
+                        sum /= data[i][1].rtts.length;
+                        rtt = sum.toFixed(2) + " ms";
+                    }
+                    tbl += "<td>" + rtt + "</td>";
+                    tbl += "</tr>";
+                }
+                tbl += "</table>";
+                $("#treeList").html(tbl);
+            });
+            setTimeout(getTree, 1000);
+        }
+
+        function start() {
+            colorReloadButton();
+            updateHostCounts();
+            getTree();
+        }
+
+        const params = new Proxy(new URLSearchParams(window.location.search), {
+            get: (searchParams, prop) => searchParams.get(prop),
+        });
+        node = params.parent;
+
+        $(document).ready(start);
+    </script>
+
+</body>
+</html>

src/rust/lqos_python/src/lib.rs

@@ -178,11 +178,15 @@ impl BatchedCommands {
   }
 
   pub fn submit(&mut self) -> PyResult<usize> {
+    const MAX_BATH_SIZE: usize = 512;
     // We're draining the request list out, which is a move that
     // *should* be elided by the optimizing compiler.
     let len = self.batch.len();
-    let batch: Vec<BusRequest> = self.batch.drain(0..).collect();
+    while !self.batch.is_empty() {
-    run_query(batch).unwrap();
+      let batch_size = usize::min(MAX_BATH_SIZE, self.batch.len());
+      let batch: Vec<BusRequest> = self.batch.drain(0..batch_size).collect();
+      run_query(batch).unwrap();
+    }
     Ok(len)
   }
 }

src/rust/lqos_sys/src/xdp_ip_address.rs

@@ -41,9 +41,8 @@ impl XdpIpAddress {
     result
   }
 
-  /// Converts an `XdpIpAddress` type to a Rust `IpAddr` type
+  fn is_v4(&self) -> bool {
-  pub fn as_ip(&self) -> IpAddr {
+    self.0[0] == 0xFF
-    if self.0[0] == 0xFF
       && self.0[1] == 0xFF
       && self.0[2] == 0xFF
       && self.0[3] == 0xFF
@@ -55,6 +54,31 @@ impl XdpIpAddress {
       && self.0[9] == 0xFF
       && self.0[10] == 0xFF
       && self.0[11] == 0xFF
+  }
+
+  /// Convers an `XdpIpAddress` type to a Rust `IpAddr` type, using
+  /// the in-build mapped function for squishing IPv4 into IPv6
+  pub fn as_ipv6(&self) -> Ipv6Addr {
+    if self.is_v4()
+    {
+      Ipv4Addr::new(self.0[12], self.0[13], self.0[14], self.0[15]).to_ipv6_mapped()
+    } else {
+      Ipv6Addr::new(
+        BigEndian::read_u16(&self.0[0..2]),
+        BigEndian::read_u16(&self.0[2..4]),
+        BigEndian::read_u16(&self.0[4..6]),
+        BigEndian::read_u16(&self.0[6..8]),
+        BigEndian::read_u16(&self.0[8..10]),
+        BigEndian::read_u16(&self.0[10..12]),
+        BigEndian::read_u16(&self.0[12..14]),
+        BigEndian::read_u16(&self.0[14..]),
+      )
+    }
+  }
+
+  /// Converts an `XdpIpAddress` type to a Rust `IpAddr` type
+  pub fn as_ip(&self) -> IpAddr {
+    if self.is_v4()
     {
       // It's an IPv4 Address
       IpAddr::V4(Ipv4Addr::new(self.0[12], self.0[13], self.0[14], self.0[15]))

src/rust/lqosd/Cargo.toml

@@ -14,7 +14,7 @@ lqos_sys = { path = "../lqos_sys" }
 lqos_queue_tracker = { path = "../lqos_queue_tracker" }
 lqos_utils = { path = "../lqos_utils" }
 tokio = { version = "1", features = [ "full", "parking_lot" ] }
-lazy_static = "1.4"
+once_cell = "1.17.1"
 parking_lot = "0.12"
 lqos_bus = { path = "../lqos_bus" }
 signal-hook = "0.3"

src/rust/lqosd/src/main.rs

@@ -3,6 +3,7 @@ mod ip_mapping;
 #[cfg(feature = "equinix_tests")]
 mod lqos_daht_test;
 mod program_control;
+mod shaped_devices_tracker;
 mod throughput_tracker;
 mod tuning;
 mod validation;
@@ -63,7 +64,11 @@ async fn main() -> Result<()> {
   };
 
   // Spawn tracking sub-systems
-  join!(spawn_queue_structure_monitor(),);
+  join!(
+    spawn_queue_structure_monitor(),
+    shaped_devices_tracker::shaped_devices_watcher(),
+    shaped_devices_tracker::network_json_watcher()
+  );
   throughput_tracker::spawn_throughput_monitor();
   spawn_queue_monitor();
 
@@ -155,7 +160,10 @@ fn handle_bus_requests(
       BusRequest::RequestLqosEquinixTest => lqos_daht_test::lqos_daht_test(),
       BusRequest::ValidateShapedDevicesCsv => {
         validation::validate_shaped_devices_csv()
-      }
+      },
+      BusRequest::GetNetworkMap { parent } => {
+        shaped_devices_tracker::get_one_network_map_layer(*parent)
+      },
     });
   }
 }

src/rust/lqosd/src/shaped_devices_tracker/mod.rs

@@ -0,0 +1,66 @@
+use anyhow::Result;
+use log::{error, info, warn};
+use lqos_bus::BusResponse;
+use lqos_config::ConfigShapedDevices;
+use lqos_utils::file_watcher::FileWatcher;
+use once_cell::sync::Lazy;
+use parking_lot::RwLock;
+use tokio::task::spawn_blocking;
+mod netjson;
+pub use netjson::*;
+
+pub static SHAPED_DEVICES: Lazy<RwLock<ConfigShapedDevices>> =
+  Lazy::new(|| RwLock::new(ConfigShapedDevices::default()));
+
+fn load_shaped_devices() {
+  info!("ShapedDevices.csv has changed. Attempting to load it.");
+  let shaped_devices = ConfigShapedDevices::load();
+  if let Ok(new_file) = shaped_devices {
+    info!("ShapedDevices.csv loaded");
+    *SHAPED_DEVICES.write() = new_file;
+    crate::throughput_tracker::THROUGHPUT_TRACKER.write().refresh_circuit_ids();
+  } else {
+    warn!("ShapedDevices.csv failed to load, see previous error messages. Reverting to empty set.");
+    *SHAPED_DEVICES.write() = ConfigShapedDevices::default();
+  }
+}
+
+pub async fn shaped_devices_watcher() {
+  spawn_blocking(|| {
+    info!("Watching for ShapedDevices.csv changes");
+    let _ = watch_for_shaped_devices_changing();
+  });
+}
+
+/// Fires up a Linux file system watcher than notifies
+/// when `ShapedDevices.csv` changes, and triggers a reload.
+fn watch_for_shaped_devices_changing() -> Result<()> {
+  let watch_path = ConfigShapedDevices::path();
+  if watch_path.is_err() {
+    error!("Unable to generate path for ShapedDevices.csv");
+    return Err(anyhow::Error::msg(
+      "Unable to create path for ShapedDevices.csv",
+    ));
+  }
+  let watch_path = watch_path.unwrap();
+
+  let mut watcher = FileWatcher::new("ShapedDevices.csv", watch_path);
+  watcher.set_file_exists_callback(load_shaped_devices);
+  watcher.set_file_created_callback(load_shaped_devices);
+  watcher.set_file_changed_callback(load_shaped_devices);
+  loop {
+    let result = watcher.watch();
+    info!("ShapedDevices watcher returned: {result:?}");
+  }
+}
+
+pub fn get_one_network_map_layer(parent_idx: usize) -> BusResponse {
+  let net_json = NETWORK_JSON.read();
+  if let Some(parent) = net_json.get_cloned_entry_by_index(parent_idx) {
+    let mut nodes = vec![(parent_idx, parent)];
+    nodes.extend_from_slice(&net_json.get_cloned_children(parent_idx));
+    BusResponse::NetworkMap(nodes)
+  } else {
+    BusResponse::Fail("No such node".to_string())
+  }
+}

src/rust/lqosd/src/shaped_devices_tracker/netjson.rs

@@ -0,0 +1,47 @@
+use log::{info, error, warn};
+use lqos_config::NetworkJson;
+use lqos_utils::file_watcher::FileWatcher;
+use once_cell::sync::Lazy;
+use parking_lot::RwLock;
+use tokio::task::spawn_blocking;
+use anyhow::Result;
+
+pub static NETWORK_JSON: Lazy<RwLock<NetworkJson>> = Lazy::new(|| RwLock::new(NetworkJson::default()));
+
+pub async fn network_json_watcher() {
+  spawn_blocking(|| {
+    info!("Watching for network.kson changes");
+    let _ = watch_for_network_json_changing();
+  });
+}
+
+/// Fires up a Linux file system watcher than notifies
+/// when `network.json` changes, and triggers a reload.
+fn watch_for_network_json_changing() -> Result<()> {
+    let watch_path = NetworkJson::path();
+    if watch_path.is_err() {
+      error!("Unable to generate path for network.json");
+      return Err(anyhow::Error::msg(
+        "Unable to create path for network.json",
+      ));
+    }
+    let watch_path = watch_path.unwrap();
+  
+    let mut watcher = FileWatcher::new("network.json", watch_path);
+    watcher.set_file_exists_callback(load_network_json);
+    watcher.set_file_created_callback(load_network_json);
+    watcher.set_file_changed_callback(load_network_json);
+    loop {
+      let result = watcher.watch();
+      info!("network.json watcher returned: {result:?}");
+    }
+  }
+  
+  fn load_network_json() {
+    let njs = NetworkJson::load();
+    if let Ok(njs) = njs {
+        *NETWORK_JSON.write() = njs;
+    } else {
+        warn!("Unable to load network.json");
+    }
+  }

src/rust/lqosd/src/throughput_tracker/mod.rs

@@ -1,20 +1,17 @@
 mod throughput_entry;
 mod tracking_data;
-use crate::throughput_tracker::tracking_data::ThroughputTracker;
+use crate::{throughput_tracker::tracking_data::ThroughputTracker, shaped_devices_tracker::NETWORK_JSON};
-use lazy_static::*;
 use log::{info, warn};
 use lqos_bus::{BusResponse, IpStats, TcHandle, XdpPpingResult};
 use lqos_sys::XdpIpAddress;
 use lqos_utils::{fdtimer::periodic, unix_time::time_since_boot};
+use once_cell::sync::Lazy;
 use parking_lot::RwLock;
 use std::time::Duration;
 
 const RETIRE_AFTER_SECONDS: u64 = 30;
 
-lazy_static! {
+pub static THROUGHPUT_TRACKER: Lazy<RwLock<ThroughputTracker>> = Lazy::new(|| RwLock::new(ThroughputTracker::new()));
-  static ref THROUGHPUT_TRACKER: RwLock<ThroughputTracker> =
-    RwLock::new(ThroughputTracker::new());
-}
 
 pub fn spawn_throughput_monitor() {
   info!("Starting the bandwidth monitor thread.");
@@ -24,10 +21,11 @@ pub fn spawn_throughput_monitor() {
   std::thread::spawn(move || {
     periodic(interval_ms, "Throughput Monitor", &mut || {
       let mut throughput = THROUGHPUT_TRACKER.write();
-      throughput.copy_previous_and_reset_rtt();
+      let mut net_json = NETWORK_JSON.write();
+      throughput.copy_previous_and_reset_rtt(&mut net_json);
       throughput.apply_new_throughput_counters();
       throughput.apply_rtt_data();
-      throughput.update_totals();
+      throughput.update_totals(&mut net_json);
       throughput.next_cycle();
     });
   });
@@ -65,7 +63,7 @@ fn retire_check(cycle: u64, recent_cycle: u64) -> bool {
   cycle < recent_cycle + RETIRE_AFTER_SECONDS
 }
 
-type TopList = (XdpIpAddress, (u64, u64), (u64, u64), f32, TcHandle);
+type TopList = (XdpIpAddress, (u64, u64), (u64, u64), f32, TcHandle, String);
 
 pub fn top_n(start: u32, end: u32) -> BusResponse {
   let mut full_list: Vec<TopList> = {
@@ -81,6 +79,7 @@ pub fn top_n(start: u32, end: u32) -> BusResponse {
           te.packets_per_second,
           te.median_latency(),
           te.tc_handle,
+          te.circuit_id.as_ref().unwrap_or(&String::new()).clone(),
         )
       })
       .collect()
@@ -97,8 +96,10 @@ pub fn top_n(start: u32, end: u32) -> BusResponse {
         (packets_dn, packets_up),
         median_rtt,
         tc_handle,
+        circuit_id,
       )| IpStats {
         ip_address: ip.as_ip().to_string(),
+        circuit_id: circuit_id.clone(),
         bits_per_second: (bytes_dn * 8, bytes_up * 8),
         packets_per_second: (*packets_dn, *packets_up),
         median_tcp_rtt: *median_rtt,
@@ -124,6 +125,7 @@ pub fn worst_n(start: u32, end: u32) -> BusResponse {
           te.packets_per_second,
           te.median_latency(),
           te.tc_handle,
+          te.circuit_id.as_ref().unwrap_or(&String::new()).clone(),
         )
       })
       .collect()
@@ -140,8 +142,10 @@ pub fn worst_n(start: u32, end: u32) -> BusResponse {
         (packets_dn, packets_up),
         median_rtt,
         tc_handle,
+        circuit_id,
       )| IpStats {
         ip_address: ip.as_ip().to_string(),
+        circuit_id: circuit_id.clone(),
         bits_per_second: (bytes_dn * 8, bytes_up * 8),
         packets_per_second: (*packets_dn, *packets_up),
         median_tcp_rtt: *median_rtt,
@@ -166,6 +170,7 @@ pub fn best_n(start: u32, end: u32) -> BusResponse {
           te.packets_per_second,
           te.median_latency(),
           te.tc_handle,
+          te.circuit_id.as_ref().unwrap_or(&String::new()).clone(),
         )
       })
       .collect()
@@ -183,8 +188,10 @@ pub fn best_n(start: u32, end: u32) -> BusResponse {
         (packets_dn, packets_up),
         median_rtt,
         tc_handle,
+        circuit_id,
       )| IpStats {
         ip_address: ip.as_ip().to_string(),
+        circuit_id: circuit_id.clone(),
         bits_per_second: (bytes_dn * 8, bytes_up * 8),
         packets_per_second: (*packets_dn, *packets_up),
         median_tcp_rtt: *median_rtt,
@@ -318,6 +325,7 @@ pub fn all_unknown_ips() -> BusResponse {
         _last_seen,
       )| IpStats {
         ip_address: ip.as_ip().to_string(),
+        circuit_id: String::new(),
         bits_per_second: (bytes_dn * 8, bytes_up * 8),
         packets_per_second: (*packets_dn, *packets_up),
         median_tcp_rtt: *median_rtt,

src/rust/lqosd/src/throughput_tracker/throughput_entry.rs

@@ -2,6 +2,8 @@ use lqos_bus::TcHandle;
 
 #[derive(Debug)]
 pub(crate) struct ThroughputEntry {
+  pub(crate) circuit_id: Option<String>,
+  pub(crate) network_json_parents: Option<Vec<usize>>,
   pub(crate) first_cycle: u64,
   pub(crate) most_recent_cycle: u64,
   pub(crate) bytes: (u64, u64),

src/rust/lqosd/src/throughput_tracker/tracking_data.rs

@@ -1,5 +1,8 @@
+use crate::shaped_devices_tracker::SHAPED_DEVICES;
+
 use super::{throughput_entry::ThroughputEntry, RETIRE_AFTER_SECONDS};
 use lqos_bus::TcHandle;
+use lqos_config::NetworkJson;
 use lqos_sys::{rtt_for_each, throughput_for_each, XdpIpAddress};
 use rayon::prelude::{IntoParallelRefMutIterator, ParallelIterator};
 use std::collections::HashMap;
@@ -26,7 +29,13 @@ impl ThroughputTracker {
     }
   }
 
-  pub(crate) fn copy_previous_and_reset_rtt(&mut self) {
+  pub(crate) fn copy_previous_and_reset_rtt(
+    &mut self,
+    netjson: &mut NetworkJson,
+  ) {
+    // Zero the previous funnel hierarchy current numbers
+    netjson.zero_throughput_and_rtt();
+
     // Copy previous byte/packet numbers and reset RTT data
     // We're using Rayon's "par_iter_mut" to spread the operation across
     // all CPU cores.
@@ -52,6 +61,51 @@ impl ThroughputTracker {
     });
   }
 
+  fn lookup_circuit_id(xdp_ip: &XdpIpAddress) -> Option<String> {
+    let mut circuit_id = None;
+    let lookup = xdp_ip.as_ipv6();
+    let cfg = SHAPED_DEVICES.read();
+    if let Some((_, id)) = cfg.trie.longest_match(lookup) {
+      circuit_id = Some(cfg.devices[*id].circuit_id.clone());
+    }
+    //println!("{lookup:?} Found circuit_id: {circuit_id:?}");
+    circuit_id
+  }
+
+  pub(crate) fn get_node_name_for_circuit_id(
+    circuit_id: Option<String>,
+  ) -> Option<String> {
+    if let Some(circuit_id) = circuit_id {
+      let shaped = SHAPED_DEVICES.read();
+      shaped
+        .devices
+        .iter()
+        .find(|d| d.circuit_id == circuit_id)
+        .map(|device| device.parent_node.clone())
+    } else {
+      None
+    }
+  }
+
+  pub(crate) fn lookup_network_parents(
+    circuit_id: Option<String>,
+  ) -> Option<Vec<usize>> {
+    if let Some(parent) = Self::get_node_name_for_circuit_id(circuit_id) {
+      let lock = crate::shaped_devices_tracker::NETWORK_JSON.read();
+      lock.get_parents_for_circuit_id(&parent)
+    } else {
+      None
+    }
+  }
+
+  pub(crate) fn refresh_circuit_ids(&mut self) {
+    self.raw_data.par_iter_mut().for_each(|(ip, data)| {
+      data.circuit_id = Self::lookup_circuit_id(ip);
+      data.network_json_parents =
+        Self::lookup_network_parents(data.circuit_id.clone());
+    });
+  }
+
   pub(crate) fn apply_new_throughput_counters(&mut self) {
     let cycle = self.cycle;
     let raw_data = &mut self.raw_data;
@@ -75,7 +129,10 @@ impl ThroughputTracker {
           entry.most_recent_cycle = cycle;
         }
       } else {
+        let circuit_id = Self::lookup_circuit_id(xdp_ip);
         let mut entry = ThroughputEntry {
+          circuit_id: circuit_id.clone(),
+          network_json_parents: Self::lookup_network_parents(circuit_id),
           first_cycle: self.cycle,
           most_recent_cycle: 0,
           bytes: (0, 0),
@@ -115,7 +172,7 @@ impl ThroughputTracker {
     });
   }
 
-  pub(crate) fn update_totals(&mut self) {
+  pub(crate) fn update_totals(&mut self, net_json: &mut NetworkJson) {
     self.bytes_per_second = (0, 0);
     self.packets_per_second = (0, 0);
     self.shaped_bytes_per_second = (0, 0);
@@ -129,27 +186,43 @@ impl ThroughputTracker {
           v.packets.0.saturating_sub(v.prev_packets.0),
           v.packets.1.saturating_sub(v.prev_packets.1),
           v.tc_handle.as_u32() > 0,
+          &v.network_json_parents,
+          v.median_latency(),
         )
       })
-      .for_each(|(bytes_down, bytes_up, packets_down, packets_up, shaped)| {
+      .for_each(
-        self.bytes_per_second.0 =
+        |(bytes_down, bytes_up, packets_down, packets_up, shaped, parents, median_rtt)| {
-          self.bytes_per_second.0.checked_add(bytes_down).unwrap_or(0);
+          self.bytes_per_second.0 =
-        self.bytes_per_second.1 =
+            self.bytes_per_second.0.checked_add(bytes_down).unwrap_or(0);
-          self.bytes_per_second.1.checked_add(bytes_up).unwrap_or(0);
+          self.bytes_per_second.1 =
-        self.packets_per_second.0 =
+            self.bytes_per_second.1.checked_add(bytes_up).unwrap_or(0);
-          self.packets_per_second.0.checked_add(packets_down).unwrap_or(0);
+          self.packets_per_second.0 =
-        self.packets_per_second.1 =
+            self.packets_per_second.0.checked_add(packets_down).unwrap_or(0);
-          self.packets_per_second.1.checked_add(packets_up).unwrap_or(0);
+          self.packets_per_second.1 =
-        if shaped {
+            self.packets_per_second.1.checked_add(packets_up).unwrap_or(0);
-          self.shaped_bytes_per_second.0 = self
+          if shaped {
-            .shaped_bytes_per_second
+            self.shaped_bytes_per_second.0 = self
-            .0
+              .shaped_bytes_per_second
-            .checked_add(bytes_down)
+              .0
-            .unwrap_or(0);
+              .checked_add(bytes_down)
-          self.shaped_bytes_per_second.1 =
+              .unwrap_or(0);
-            self.shaped_bytes_per_second.1.checked_add(bytes_up).unwrap_or(0);
+            self.shaped_bytes_per_second.1 = self
-        }
+              .shaped_bytes_per_second
-      });
+              .1
+              .checked_add(bytes_up)
+              .unwrap_or(0);
+          }
+
+          // If we have parent node data, we apply it now
+          if let Some(parents) = parents {
+            net_json.add_throughput_cycle(
+              parents,
+              (self.bytes_per_second.0, self.bytes_per_second.1),
+              median_rtt,
+            )
+          }
+        },
+      );
   }
 
   pub(crate) fn next_cycle(&mut self) {