Top 10 Flows by Rate

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

@@ -7,6 +7,7 @@ import {RttHistoDash} from "./rtt_histo_dash";
 import {Top10Downloaders} from "./top10_downloaders";
 import {Worst10Downloaders} from "./worst10_downloaders";
 import {Top10FlowsBytes} from "./top10flows_bytes";
+import {Top10FlowsRate} from "./top10flows_rate";
 
 export const DashletMenu = [
     { name: "Throughput Bits/Second", tag: "throughputBps", size: 3 },
@@ -18,6 +19,7 @@ export const DashletMenu = [
     { name: "Top 10 Downloaders", tag: "top10downloaders", size: 6 },
     { name: "Worst 10 Round-Trip Time", tag: "worst10downloaders", size: 6 },
     { name: "Top 10 Flows (total bytes)", tag: "top10flowsBytes", size: 6 },
+    { name: "Top 10 Flows (rate)", tag: "top10flowsRate", size: 6 },
 ];
 
 export function widgetFactory(widgetName, count) {
@@ -32,6 +34,7 @@ export function widgetFactory(widgetName, count) {
         case "top10downloaders":widget = new Top10Downloaders(count); break;
         case "worst10downloaders":widget = new Worst10Downloaders(count); break;
         case "top10flowsBytes"  : widget = new Top10FlowsBytes(count); break;
+        case "top10flowsRate"   : widget = new Top10FlowsRate(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/top10flows_rate.js

@@ -0,0 +1,109 @@
+import {BaseDashlet} from "./base_dashlet";
+import {theading} from "../helpers/builders";
+import {scaleNumber, scaleNanos} from "../helpers/scaling";
+
+export class Top10FlowsRate extends BaseDashlet {
+    constructor(slot) {
+        super(slot);
+    }
+
+    title() {
+        return "Top 10 Flows (by rate)";
+    }
+
+    subscribeTo() {
+        return [ "topFlowsRate" ];
+    }
+
+    buildContainer() {
+        let base = super.buildContainer();
+        base.style.height = "250px";
+        base.style.overflow = "auto";
+        return base;
+    }
+
+    setup() {
+        super.setup();
+    }
+
+    onMessage(msg) {
+        if (msg.event === "topFlowsRate") {
+            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("Protocol"));
+            th.appendChild(theading("Local IP"));
+            th.appendChild(theading("Remote IP"));
+            th.appendChild(theading("DL ⬇️"));
+            th.appendChild(theading("UL ⬆️"));
+            th.appendChild(theading("Total"));
+            th.appendChild(theading("⬇ RTT"));
+            th.appendChild(theading("️️⬆ RTT"));
+            th.appendChild(theading("TCP Retransmits"));
+            th.appendChild(theading("Remote ASN"));
+            th.appendChild(theading("Country"));
+            t.appendChild(th);
+
+            let tbody = document.createElement("tbody");
+            msg.data.forEach((r) => {
+                let row = document.createElement("tr");
+
+                let proto = document.createElement("td");
+                proto.innerText = r.analysis;
+                row.appendChild(proto);
+
+                let localIp = document.createElement("td");
+                localIp.innerText = r.local_ip;
+                row.appendChild(localIp);
+
+                let remoteIp = document.createElement("td");
+                remoteIp.innerText = r.remote_ip;
+                row.appendChild(remoteIp);
+
+                let dl = document.createElement("td");
+                dl.innerText = scaleNumber(r.rate_estimate_bps[0]);
+                row.appendChild(dl);
+
+                let ul = document.createElement("td");
+                ul.innerText = scaleNumber(r.rate_estimate_bps[1]);
+                row.appendChild(ul);
+
+                let total = document.createElement("td");
+                total.innerText = scaleNumber(r.bytes_sent[0]) + " / " + scaleNumber(r.bytes_sent[1]);
+                row.appendChild(total);
+
+                let rttD = document.createElement("td");
+                rttD.innerText = scaleNanos(r.rtt_nanos[0]);
+                row.appendChild(rttD);
+
+                let rttU = document.createElement("td");
+                rttU.innerText = scaleNanos(r.rtt_nanos[1]);
+                row.appendChild(rttU);
+
+                let tcp = document.createElement("td");
+                tcp.innerText = r.tcp_retransmits[0] + " / " + r.tcp_retransmits[1];
+                row.appendChild(tcp);
+
+                let asn = document.createElement("td");
+                asn.innerText = r.remote_asn_name;
+                row.appendChild(asn);
+
+                let country = document.createElement("td");
+                country.innerText = r.remote_asn_country;
+                row.appendChild(country);
+
+                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/ws/published_channels.rs

@@ -10,6 +10,7 @@ pub enum PublishedChannels {
     Top10Downloaders,
     Worst10Downloaders,
     TopFlowsBytes,
+    TopFlowsRate,
 }
 
 impl PublishedChannels {
@@ -22,6 +23,7 @@ impl PublishedChannels {
             Self::Top10Downloaders => "top10downloaders",
             Self::Worst10Downloaders => "worst10downloaders",
             Self::TopFlowsBytes => "topFlowsBytes",
+            Self::TopFlowsRate => "topFlowsRate",
         }
     }
 
@@ -34,6 +36,7 @@ impl PublishedChannels {
             "top10downloaders" => Some(Self::Top10Downloaders),
             "worst10downloaders" => Some(Self::Worst10Downloaders),
             "topFlowsBytes" => Some(Self::TopFlowsBytes),
+            "topFlowsRate" => Some(Self::TopFlowsRate),
             _ => None,
         }
     }

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

@@ -23,6 +23,7 @@ pub(super) async fn channel_ticker(channels: Arc<PubSub>) {
             top_10::top_10_downloaders(channels.clone()),
             top_10::worst_10_downloaders(channels.clone()),
             top_flows::top_flows_bytes(channels.clone()),
+            top_flows::top_flows_rate(channels.clone()),
         );
     }
 }

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

@@ -23,3 +23,19 @@ pub async fn top_flows_bytes(channels: Arc<PubSub>) {
         channels.send(PublishedChannels::TopFlowsBytes, message).await;
     }
 }
+
+pub async fn top_flows_rate(channels: Arc<PubSub>) {
+    if !channels.is_channel_alive(PublishedChannels::TopFlowsBytes).await {
+        return;
+    }
+
+    if let BusResponse::TopFlows(flows) = throughput_tracker::top_flows(10, TopFlowType::RateEstimate) {
+        let message = json!(
+            {
+                "event": "topFlowsRate",
+                "data": flows,
+            }
+        ).to_string();
+        channels.send(PublishedChannels::TopFlowsBytes, message).await;
+    }
+}