Support display by protocol

src/rust/lqosd/src/node_manager/js_build/src/asn_explorer.js

@@ -8,6 +8,7 @@ const FLOW_URL = API_URL + "flowTimeline/";
 
 let asnList = [];
 let countryList = [];
+let protocolList = [];
 let asnData = [];
 let graphMinTime = Number.MAX_SAFE_INTEGER;
 let graphMaxTime = Number.MIN_SAFE_INTEGER;
@@ -40,7 +41,7 @@ function asnDropdown() {
         let parentDiv = document.createElement("div");
         parentDiv.classList.add("dropdown");
         let button = document.createElement("button");
-        button.classList.add("btn", "btn-secondary", "dropdown-toggle");
+        button.classList.add("btn", "btn-secondary", "dropdown-toggle", "btn-sm");
         button.type = "button";
         button.innerHTML = "Select ASN";
         button.setAttribute("data-bs-toggle", "dropdown");
@@ -86,7 +87,7 @@ function countryDropdown() {
         let parentDiv = document.createElement("div");
         parentDiv.classList.add("dropdown");
         let button = document.createElement("button");
-        button.classList.add("btn", "btn-secondary", "dropdown-toggle");
+        button.classList.add("btn", "btn-secondary", "dropdown-toggle", "btn-sm");
         button.type = "button";
         button.innerHTML = "Select Country";
         button.setAttribute("data-bs-toggle", "dropdown");
@@ -114,21 +115,71 @@ function countryDropdown() {
     });
 }
 
+function protocolDropdown() {
+    $.get(API_URL + "protocolList", (data) => {
+        protocolList = data;
+
+        // Sort data by row.count, descending
+        data.sort((a, b) => {
+            return b.count - a.count;
+        });
+        //console.log(data);
+
+        // Build the dropdown
+        let parentDiv = document.createElement("div");
+        parentDiv.classList.add("dropdown");
+        let button = document.createElement("button");
+        button.classList.add("btn", "btn-secondary", "dropdown-toggle", "btn-sm");
+        button.type = "button";
+        button.innerHTML = "Select Protocol";
+        button.setAttribute("data-bs-toggle", "dropdown");
+        button.setAttribute("aria-expanded", "false");
+        parentDiv.appendChild(button);
+        let dropdownList = document.createElement("ul");
+        dropdownList.classList.add("dropdown-menu");
+
+        // Add items
+        data.forEach((row) => {
+            let li = document.createElement("li");
+            li.innerHTML = row.protocol + " (" + row.count + ")";
+            li.classList.add("dropdown-item");
+            li.onclick = () => {
+                selectProtocol(row.protocol);
+                renderMode = "protocol";
+            };
+            dropdownList.appendChild(li);
+        });
+
+        parentDiv.appendChild(dropdownList);
+        let target = document.getElementById("protocolList");
+        clearDiv(target);
+        target.appendChild(parentDiv);
+    });
+}
+
 function selectAsn(asn) {
-    $.get(FLOW_URL + asn, (data) => {
+    $.get(FLOW_URL + encodeURI(asn), (data) => {
         page = 0;
         renderAsn(asn, data);
     });
 }
 
 function selectCountry(country) {
-    let url = API_URL + "countryTimeline/" + country;
+    let url = API_URL + "countryTimeline/" + encodeURI(country);
     $.get(url, (data) => {
         page = 0;
         renderAsn(country, data);
     });
 }
 
+function selectProtocol(protocol) {
+    let url = API_URL + "protocolTimeline/" + encodeURI(protocol.replace('/', '_'));
+    $.get(url, (data) => {
+        page = 0;
+        renderAsn(protocol, data);
+    });
+}
+
 function renderAsn(asn, data) {
     let heading = document.createElement("h2");
     if (renderMode === "asn") {
@@ -445,3 +496,4 @@ function drawTimeline() {
 
 asnDropdown();
 countryDropdown();
+protocolDropdown();

src/rust/lqosd/src/node_manager/local_api.rs

@@ -53,7 +53,9 @@ pub fn local_api() -> Router {
         .route("/globalWarnings", get(warnings::get_global_warnings))
         .route("/asnList", get(flow_explorer::asn_list))
         .route("/countryList", get(flow_explorer::country_list))
+        .route("/protocolList", get(flow_explorer::protocol_list))
         .route("/flowTimeline/:asn_id", get(flow_explorer::flow_timeline))
         .route("/countryTimeline/:iso_code", get(flow_explorer::country_timeline))
+        .route("/protocolTimeline/:protocol", get(flow_explorer::protocol_timeline))
         .route_layer(axum::middleware::from_fn(auth_layer))
 }

src/rust/lqosd/src/node_manager/local_api/flow_explorer.rs

@@ -6,7 +6,8 @@ use lqos_sys::flowbee_data::FlowbeeKey;
 use lqos_utils::units::DownUpOrder;
 use lqos_utils::unix_time::{time_since_boot, unix_now};
 use crate::shaped_devices_tracker::SHAPED_DEVICES;
-use crate::throughput_tracker::flow_data::{AsnListEntry, AsnCountryListEntry, RECENT_FLOWS, RttData, FlowbeeLocalData, FlowAnalysis};
+use crate::throughput_tracker::flow_data::{AsnListEntry, AsnCountryListEntry, AsnProtocolListEntry,
+   RECENT_FLOWS, RttData, FlowbeeLocalData, FlowAnalysis};
 
 pub async fn asn_list() -> Json<Vec<AsnListEntry>> {
     Json(RECENT_FLOWS.asn_list())
@@ -16,6 +17,10 @@ pub async fn country_list() -> Json<Vec<AsnCountryListEntry>> {
     Json(RECENT_FLOWS.country_list())
 }
 
+pub async fn protocol_list() -> Json<Vec<AsnProtocolListEntry>> {
+    Json(RECENT_FLOWS.protocol_list())
+}
+
 #[derive(Serialize)]
 pub struct FlowTimeline {
     start: u64,
@@ -95,3 +100,16 @@ pub async fn country_timeline(Path(country_name): Path<String>) -> Json<Vec<Flow
 
     Json(flows)
 }
+
+pub async fn protocol_timeline(Path(protocol_name): Path<String>) -> Json<Vec<FlowTimeline>> {
+    let protocol_name = protocol_name.replace("_", "/");
+    let time_since_boot = time_since_boot().unwrap();
+    let since_boot = Duration::from(time_since_boot);
+    let boot_time = unix_now().unwrap() - since_boot.as_secs();
+
+    let all_flows_for_asn = RECENT_FLOWS.all_flows_for_protocol(&protocol_name);
+
+    let flows = all_flows_to_transport(boot_time, all_flows_for_asn);
+
+    Json(flows)
+}

src/rust/lqosd/src/node_manager/static2/asn_explorer.html

@@ -5,6 +5,9 @@
     <div class="col-1">
         <span id="countryList"><i class="fa fa-spin fa-spinner"></i> Loading, Please Wait (this can take a minute)</span>
     </div>
+    <div class="col-1 ms-4">
+        <span id="protocolList"><i class="fa fa-spin fa-spinner"></i> Loading, Please Wait (this can take a minute)</span>
+    </div>
 </div>
 <div class="row">
     <div class="col-12" id="asnDetails">

src/rust/lqosd/src/throughput_tracker/flow_data/flow_analysis/finished_flows.rs

@@ -40,6 +40,12 @@ pub struct AsnCountryListEntry {
     iso_code: String,
 }
 
+#[derive(Debug, Serialize)]
+pub struct AsnProtocolListEntry {
+    count: usize,
+    protocol: String,
+}
+
 impl TimeBuffer {
     fn new() -> Self {
         Self {
@@ -298,6 +304,17 @@ impl TimeBuffer {
             .collect()
     }
 
+    pub fn all_flows_for_protocol(&self, protocol_name: &str) -> Vec<(FlowbeeKey, FlowbeeLocalData, FlowAnalysis)> {
+        let buffer = self.buffer.lock().unwrap();
+        buffer
+            .iter()
+            .filter(|flow| {
+                flow.data.2.protocol_analysis.to_string() == protocol_name
+            })
+            .map(|flow| flow.data.clone())
+            .collect()
+    }
+
     /// Builds a list of all ASNs with recent data, and how many flows they have.
     pub fn asn_list(&self) -> Vec<AsnListEntry> {
         // 1: Clone: large operation, don't keep the buffer locked longer than we have to
@@ -368,6 +385,41 @@ impl TimeBuffer {
             })
             .collect()
     }
+
+    /// Builds a list of protocols with recent data, and how many flows they have.
+    pub fn protocol_list(&self) -> Vec<AsnProtocolListEntry> {
+        // 1: Clone: large operation, don't keep the buffer locked longer than we have to
+        let buffer = {
+            let buffer = self.buffer.lock().unwrap();
+            buffer.clone()
+        };
+
+        // Filter out the short flows and get the country & flag
+        let mut buffer: Vec<String> = buffer
+            .into_iter()
+            .filter(|flow| {
+                // Total flow time > 3 seconds
+                flow.data.1.last_seen - flow.data.1.start_time > 3_000_000_000
+            })
+            .map(|flow| {
+                flow.data.2.protocol_analysis.to_string()
+            })
+            .collect();
+
+        // Sort the buffer
+        buffer.sort_unstable_by(|a, b| a.cmp(&b));
+
+        // Deduplicate and count, decorate with name
+        buffer
+            .into_iter()
+            .sorted()
+            .dedup_with_count()
+            .map(|(count, protocol)| AsnProtocolListEntry {
+                count,
+                protocol,
+            })
+            .collect()
+    }
 }
 
 pub static RECENT_FLOWS: Lazy<TimeBuffer> = Lazy::new(|| TimeBuffer::new());

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

@@ -14,7 +14,7 @@ mod kernel_ringbuffer;
 pub use kernel_ringbuffer::*;
 mod rtt_types;
 pub use rtt_types::RttData;
-pub use finished_flows::{AsnListEntry, AsnCountryListEntry};
+pub use finished_flows::{AsnListEntry, AsnCountryListEntry, AsnProtocolListEntry};
 use crate::throughput_tracker::flow_data::flow_analysis::asn::AsnNameCountryFlag;
 
 static ANALYSIS: Lazy<FlowAnalysisSystem> = Lazy::new(|| FlowAnalysisSystem::new());

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

@@ -16,7 +16,7 @@ use std::sync::{
 pub(crate) use flow_analysis::{setup_flow_analysis, get_asn_name_and_country, 
     FlowAnalysis, RECENT_FLOWS, flowbee_handle_events, get_flowbee_event_count_and_reset,
     expire_rtt_flows, flowbee_rtt_map, RttData, get_rtt_events_per_second, AsnListEntry,
-    AsnCountryListEntry
+    AsnCountryListEntry, AsnProtocolListEntry,
 };
 
 trait FlowbeeRecipient {