Top 10 Downloaders and infrastructure to support it.

src/rust/lqosd/src/node_manager/js_build/src/dashlets/dashlet_index.js

@@ -4,6 +4,7 @@ import {ShapedUnshapedDash} from "./shaped_unshaped_dash";
 import {TrackedFlowsCount} from "./tracked_flow_count_dash";
 import {ThroughputRingDash} from "./throughput_ring_dash";
 import {RttHistoDash} from "./rtt_histo_dash";
+import {Top10Downloaders} from "./top10_downloaders";
 
 export const DashletMenu = [
     { name: "Throughput Bits/Second", tag: "throughputBps", size: 3 },
@@ -12,6 +13,7 @@ export const DashletMenu = [
     { name: "Tracked Flows Counter", tag: "trackedFlowsCount", size: 3 },
     { name: "Last 5 Minutes Throughput", tag: "throughputRing", size: 6 },
     { name: "Round-Trip Time Histogram", tag: "rttHistogram", size: 6 },
+    { name: "Top 10 Downloaders", tag: "top10downloaders", size: 6 },
 ];
 
 export function widgetFactory(widgetName, count) {
@@ -23,6 +25,7 @@ export function widgetFactory(widgetName, count) {
         case "trackedFlowsCount": widget = new TrackedFlowsCount(count); break;
         case "throughputRing":  widget = new ThroughputRingDash(count); break;
         case "rttHistogram":    widget = new RttHistoDash(count); break;
+        case "top10downloaders":widget = new Top10Downloaders(count); break;
         default: {
             console.log("I don't know how to construct a widget of type [" + widgetName + "]");
             return null;

src/rust/lqosd/src/node_manager/js_build/src/dashlets/top10_downloaders.js

@@ -0,0 +1,90 @@
+import {BaseDashlet} from "./base_dashlet";
+import {RttHistogram} from "../graphs/rtt_histo";
+import {theading} from "../helpers/builders";
+import {scaleNumber, rttCircleSpan} from "../helpers/scaling";
+
+export class Top10Downloaders extends BaseDashlet {
+    constructor(slot) {
+        super(slot);
+    }
+
+    title() {
+        return "Top 10 Downloaders";
+    }
+
+    subscribeTo() {
+        return [ "top10downloaders" ];
+    }
+
+    buildContainer() {
+        let base = super.buildContainer();
+        base.style.height = "250px";
+        base.style.overflow = "auto";
+        return base;
+    }
+
+    setup() {
+        super.setup();
+    }
+
+    onMessage(msg) {
+        if (msg.event === "top10downloaders") {
+            let target = document.getElementById(this.id);
+
+            let t = document.createElement("table");
+            t.classList.add("table", "table-striped", "tiny");
+
+            let th = document.createElement("thead");
+            th.appendChild(theading(""));
+            th.appendChild(theading("IP Address/Circuit"));
+            th.appendChild(theading("DL ⬇️"));
+            th.appendChild(theading("UL ⬆️"));
+            th.appendChild(theading("RTT (ms)"));
+            th.appendChild(theading("TCP Retransmits"));
+            th.appendChild(theading("Shaped"));
+            t.appendChild(th);
+
+            let tbody = document.createElement("tbody");
+            msg.data.forEach((r) => {
+                let row = document.createElement("tr");
+
+                let circle = document.createElement("td");
+                circle.appendChild(rttCircleSpan(r.median_tcp_rtt));
+                row.appendChild(circle);
+
+                let ip = document.createElement("td");
+                ip.innerText = r.ip_address;
+                row.append(ip);
+
+                let dl = document.createElement("td");
+                dl.innerText = scaleNumber(r.bits_per_second[0]);
+                row.append(dl);
+
+                let ul = document.createElement("td");
+                ul.innerText = scaleNumber(r.bits_per_second[1]);
+                row.append(ul);
+
+                let rtt = document.createElement("td");
+                rtt.innerText = r.median_tcp_rtt.toFixed(2);
+                row.append(rtt);
+
+                let tcp_xmit = document.createElement("td");
+                tcp_xmit.innerText = r.tcp_retransmits[0] + " / " + r.tcp_retransmits[1];
+                row.append(tcp_xmit);
+
+                let shaped = document.createElement("td");
+                shaped.innerText = r.plan[0] + " / " + r.plan[1];
+                row.append(shaped);
+
+                t.appendChild(row);
+            });
+            t.appendChild(tbody);
+
+            // Display it
+            while (target.children.length > 1) {
+                target.removeChild(target.lastChild);
+            }
+            target.appendChild(t);
+        }
+    }
+}

src/rust/lqosd/src/node_manager/js_build/src/graphs/bits_gauge.js

@@ -1,5 +1,5 @@
 import {DashboardGraph} from "./dashboard_graph";
-import {scaleNumber} from "../scaling";
+import {scaleNumber} from "../helpers/scaling";
 
 export class BitsPerSecondGauge extends DashboardGraph {
     constructor(id) {

src/rust/lqosd/src/node_manager/js_build/src/graphs/flows_graph.js

@@ -1,5 +1,5 @@
 import {DashboardGraph} from "./dashboard_graph";
-import {scaleNumber} from "../scaling";
+import {scaleNumber} from "../helpers/scaling";
 
 const RING_SIZE = 60 * 5; // 5 Minutes
 

src/rust/lqosd/src/node_manager/js_build/src/graphs/packets_bar.js

@@ -1,5 +1,5 @@
 import {DashboardGraph} from "./dashboard_graph";
-import {scaleNumber} from "../scaling";
+import {scaleNumber} from "../helpers/scaling";
 
 export class PacketsPerSecondBar extends DashboardGraph {
     constructor(id) {

src/rust/lqosd/src/node_manager/js_build/src/graphs/shaped_unshaped_pie.js

@@ -1,5 +1,5 @@
 import {DashboardGraph} from "./dashboard_graph";
-import {scaleNumber} from "../scaling";
+import {scaleNumber} from "../helpers/scaling";
 
 export class ShapedUnshapedPie extends DashboardGraph {
     constructor(id) {

src/rust/lqosd/src/node_manager/js_build/src/graphs/throughput_ring_graph.js

@@ -1,5 +1,5 @@
 import {DashboardGraph} from "./dashboard_graph";
-import {scaleNumber} from "../scaling";
+import {scaleNumber} from "../helpers/scaling";
 
 const RING_SIZE = 60 * 5; // 5 Minutes
 

src/rust/lqosd/src/node_manager/js_build/src/helpers/scaling.js

@@ -9,4 +9,21 @@ export function scaleNumber(n, fixed=2) {
         return (n / 1000).toFixed(fixed) + "K";
     }
     return n;
+}
+
+export function colorRamp(n) {
+    if (n <= 100) {
+        return "#aaffaa";
+    } else if (n <= 150) {
+        return "goldenrod";
+    } else {
+        return "#ffaaaa";
+    }
+}
+
+export function rttCircleSpan(rtt) {
+    let span = document.createElement("span");
+    span.style.color = colorRamp(rtt);
+    span.innerText = "⬤";
+    return span;
 }

src/rust/lqosd/src/node_manager/static2/node_manager.css

@@ -50,4 +50,7 @@ body.dark-mode {
 .dashgraph {
     width: 100%;
     height: 250px;
+}
+.tiny {
+    font-size: 8pt;
 }

src/rust/lqosd/src/node_manager/ws/publish_subscribe.rs

@@ -30,6 +30,7 @@ impl PubSub {
             PublisherChannel::new(PublishedChannels::RttHistogram),
             PublisherChannel::new(PublishedChannels::FlowCount),
             PublisherChannel::new(PublishedChannels::Cadence),
+            PublisherChannel::new(PublishedChannels::Top10Downloaders),
         ];
 
         let result = Self {

src/rust/lqosd/src/node_manager/ws/published_channels.rs

@@ -5,15 +5,17 @@ pub enum PublishedChannels {
     Throughput,
     RttHistogram,
     FlowCount,
+    Top10Downloaders,
 }
 
 impl PublishedChannels {
     pub(super) fn as_str(&self) -> &'static str {
         match self {
-            PublishedChannels::Throughput => "throughput",
+            Self::Throughput => "throughput",
-            PublishedChannels::RttHistogram => "rttHistogram",
+            Self::RttHistogram => "rttHistogram",
-            PublishedChannels::FlowCount => "flowCount",
+            Self::FlowCount => "flowCount",
-            PublishedChannels::Cadence => "cadence",
+            Self::Cadence => "cadence",
+            Self::Top10Downloaders => "top10downloaders",
         }
     }
 
@@ -23,6 +25,7 @@ impl PublishedChannels {
             "rttHistogram" => Some(Self::RttHistogram),
             "flowCount" => Some(Self::FlowCount),
             "cadence" => Some(Self::Cadence),
+            "top10downloaders" => Some(Self::Top10Downloaders),
             _ => None,
         }
     }

src/rust/lqosd/src/node_manager/ws/ticker.rs

@@ -2,6 +2,8 @@ mod cadence;
 mod throughput;
 mod rtt_histogram;
 mod flow_counter;
+mod top_10;
+mod ipstats_conversion;
 
 use std::sync::Arc;
 use crate::node_manager::ws::publish_subscribe::PubSub;
@@ -17,6 +19,7 @@ pub(super) async fn channel_ticker(channels: Arc<PubSub>) {
             throughput::throughput(channels.clone()),
             rtt_histogram::rtt_histo(channels.clone()),
             flow_counter::flow_count(channels.clone()),
+            top_10::top_10_downloaders(channels.clone()),
         );
     }
 }

src/rust/lqosd/src/node_manager/ws/ticker/ipstats_conversion.rs

@@ -0,0 +1,50 @@
+use serde::{Deserialize, Serialize};
+use lqos_bus::{IpStats, TcHandle};
+use crate::shaped_devices_tracker::SHAPED_DEVICES;
+
+#[derive(Serialize, Deserialize, Clone, Debug)]
+pub struct IpStatsWithPlan {
+    pub ip_address: String,
+    pub bits_per_second: (u64, u64),
+    pub packets_per_second: (u64, u64),
+    pub median_tcp_rtt: f32,
+    pub tc_handle: TcHandle,
+    pub circuit_id: String,
+    pub plan: (u32, u32),
+    pub tcp_retransmits: (u64, u64),
+}
+
+impl From<&IpStats> for IpStatsWithPlan {
+    fn from(i: &IpStats) -> Self {
+        let mut result = Self {
+            ip_address: i.ip_address.clone(),
+            bits_per_second: i.bits_per_second,
+            packets_per_second: i.packets_per_second,
+            median_tcp_rtt: i.median_tcp_rtt,
+            tc_handle: i.tc_handle,
+            circuit_id: i.circuit_id.clone(),
+            plan: (0, 0),
+            tcp_retransmits: i.tcp_retransmits,
+        };
+
+        if !result.circuit_id.is_empty() {
+            if let Some(circuit) = SHAPED_DEVICES
+                .read()
+                .unwrap()
+                .devices
+                .iter()
+                .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!("{} ({})", name, result.ip_address);
+                result.plan = (circuit.download_max_mbps, circuit.upload_max_mbps);
+            }
+        }
+
+        result
+    }
+}

src/rust/lqosd/src/node_manager/ws/ticker/top_10.rs

@@ -0,0 +1,28 @@
+use std::sync::Arc;
+use serde_json::json;
+use lqos_bus::BusResponse;
+use crate::node_manager::ws::publish_subscribe::PubSub;
+use crate::node_manager::ws::published_channels::PublishedChannels;
+use crate::node_manager::ws::ticker::ipstats_conversion::IpStatsWithPlan;
+use crate::throughput_tracker::top_n;
+
+pub async fn top_10_downloaders(channels: Arc<PubSub>) {
+    if !channels.is_channel_alive(PublishedChannels::Top10Downloaders).await {
+        return;
+    }
+    
+    if let BusResponse::TopDownloaders(top) = top_n(0, 10) {
+        let result: Vec<IpStatsWithPlan> = top
+            .iter()
+            .map(|stat| stat.into())
+            .collect();
+
+        let message = json!(
+        {
+            "event": "top10downloaders",
+            "data": result
+        }
+    ).to_string();
+        channels.send(PublishedChannels::Top10Downloaders, message).await;
+    }
+}