Experimental: list ASN timelines by country

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

@@ -7,12 +7,14 @@ const LIST_URL = API_URL + "asnList";
 const FLOW_URL = API_URL + "flowTimeline/";
 
 let asnList = [];
+let countryList = [];
 let asnData = [];
 let graphMinTime = Number.MAX_SAFE_INTEGER;
 let graphMaxTime = Number.MIN_SAFE_INTEGER;
 
 const itemsPerPage = 20;
 let page = 0;
+let renderMode = "asn";
 
 let sortBy = "start";
 let sortOptionsList = [
@@ -58,6 +60,7 @@ function asnDropdown() {
             li.classList.add("dropdown-item");
             li.onclick = () => {
                 selectAsn(row.asn);
+                renderMode = "asn";
             };
             dropdownList.appendChild(li);
         });
@@ -66,10 +69,48 @@ function asnDropdown() {
         let target = document.getElementById("asnList");
         clearDiv(target);
         target.appendChild(parentDiv);
+    });
+}
 
-        /*if (data.length > 0) {
+function countryDropdown() {
-            selectAsn(data[0].asn);
+    $.get(API_URL + "countryList", (data) => {
-        }*/
+        countryList = 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");
+        button.type = "button";
+        button.innerHTML = "Select Country";
+        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 = "<img alt='" + row.iso_code + "' src='flags/" + row.iso_code + ".svg' height=12 width=12 />" + row.name + " (" + row.count + ")";
+            li.classList.add("dropdown-item");
+            li.onclick = () => {
+                selectCountry(row.iso_code);
+                renderMode = "country";
+            };
+            dropdownList.appendChild(li);
+        });
+
+        parentDiv.appendChild(dropdownList);
+        let target = document.getElementById("countryList");
+        clearDiv(target);
+        target.appendChild(parentDiv);
     });
 }
 
@@ -80,19 +121,38 @@ function selectAsn(asn) {
     });
 }
 
+function selectCountry(country) {
+    let url = API_URL + "countryTimeline/" + country;
+    $.get(url, (data) => {
+        page = 0;
+        renderAsn(country, data);
+    });
+}
+
 function renderAsn(asn, data) {
-    let targetAsn = asnList.find((row) => row.asn === asn);
+    let heading = document.createElement("h2");
-    if (targetAsn === undefined || targetAsn === null) {
+    if (renderMode === "asn") {
-        console.error("Could not find ASN: " + asn);
+        let targetAsn = asnList.find((row) => row.asn === asn);
-        return;
+        if (targetAsn === undefined || targetAsn === null) {
+            console.error("Could not find ASN: " + asn);
+            return;
+        }
+
+        // Build the heading
+        heading.innerText = "ASN #" + asn.toFixed(0) + " (" + targetAsn.name + ")";
+    } else if (renderMode === "country") {
+        let targetCountry = countryList.find((row) => row.iso_code === asn);
+        if (targetCountry === undefined || targetCountry === null) {
+            console.error("Could not find country: " + asn);
+            return;
+        }
+
+        // Build the heading
+        heading.innerHTML = "<img alt='" + targetCountry.iso_code + "' src='flags/" + targetCountry.iso_code + ".svg' height=32 width=32 />" + targetCountry.name;
     }
 
     let target = document.getElementById("asnDetails");
 
-    // Build the heading
-    let heading = document.createElement("h2");
-    heading.innerText = "ASN #" + asn.toFixed(0) + " (" + targetAsn.name + ")";
-
     // Get the flow data
     asnData = data;
 
@@ -384,3 +444,4 @@ function drawTimeline() {
 }
 
 asnDropdown();
+countryDropdown();

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

@@ -52,6 +52,8 @@ pub fn local_api() -> Router {
         .route("/flowMap", get(flow_map::flow_lat_lon))
         .route("/globalWarnings", get(warnings::get_global_warnings))
         .route("/asnList", get(flow_explorer::asn_list))
+        .route("/countryList", get(flow_explorer::country_list))
         .route("/flowTimeline/:asn_id", get(flow_explorer::flow_timeline))
+        .route("/countryTimeline/:iso_code", get(flow_explorer::country_timeline))
         .route_layer(axum::middleware::from_fn(auth_layer))
 }

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

@@ -2,15 +2,20 @@ use std::time::Duration;
 use axum::extract::Path;
 use axum::Json;
 use serde::Serialize;
+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, RECENT_FLOWS, RttData};
+use crate::throughput_tracker::flow_data::{AsnListEntry, AsnCountryListEntry, RECENT_FLOWS, RttData, FlowbeeLocalData, FlowAnalysis};
 
 pub async fn asn_list() -> Json<Vec<AsnListEntry>> {
     Json(RECENT_FLOWS.asn_list())
 }
 
+pub async fn country_list() -> Json<Vec<AsnCountryListEntry>> {
+    Json(RECENT_FLOWS.country_list())
+}
+
 #[derive(Serialize)]
 pub struct FlowTimeline {
     start: u64,
@@ -34,14 +39,19 @@ pub async fn flow_timeline(Path(asn_id): Path<u32>) -> Json<Vec<FlowTimeline>> {
 
     let all_flows_for_asn = RECENT_FLOWS.all_flows_for_asn(asn_id);
 
-    let flows = all_flows_for_asn
+    let flows = all_flows_to_transport(boot_time, all_flows_for_asn);
+
+    Json(flows)
+}
+
+fn all_flows_to_transport(boot_time: u64, all_flows_for_asn: Vec<(FlowbeeKey, FlowbeeLocalData, FlowAnalysis)>) -> Vec<FlowTimeline> {
+    all_flows_for_asn
         .iter()
         .filter(|flow| {
             // Total flow time > 2 seconds
             flow.1.last_seen - flow.1.start_time > 2_000_000_000
         })
         .map(|flow| {
-
             let (circuit_id, mut circuit_name) = {
                 let sd = SHAPED_DEVICES.read().unwrap();
                 sd.get_circuit_id_and_name_from_ip(&flow.0.local_ip).unwrap_or((String::new(), String::new()))
@@ -71,7 +81,17 @@ pub async fn flow_timeline(Path(asn_id): Path<u32>) -> Json<Vec<FlowTimeline>> {
                 circuit_name,
             }
         })
-        .collect::<Vec<_>>();
+        .collect::<Vec<_>>()
+}
+
+pub async fn country_timeline(Path(iso_code): Path<String>) -> Json<Vec<FlowTimeline>> {
+    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_country(&iso_code);
+
+    let flows = all_flows_to_transport(boot_time, all_flows_for_asn);
 
     Json(flows)
 }

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

@@ -1,7 +1,10 @@
 <div class="row">
-    <div class="col-12">
+    <div class="col-1">
         <span id="asnList"><i class="fa fa-spin fa-spinner"></i> Loading, Please Wait (this can take a minute)</span>
     </div>
+    <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>
 <div class="row">
     <div class="col-12" id="asnDetails">

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

@@ -139,9 +139,9 @@ impl GeoTable {
             IpAddr::V4(ip) => ip.to_ipv6_mapped(),
             IpAddr::V6(ip) => ip,
         };
-        let mut owners = String::new();
+        let mut owners = "Unknown".to_string();
-        let mut country = String::new();
+        let mut country = "Unknown".to_string();
-        let mut flag = String::new();
+        let mut flag = "Unknown".to_string();
 
         if let Some(matched) = self.asn_trie.longest_match(ip) {
             log::debug!("Matched ASN: {:?}", matched.1.asn);

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

@@ -33,6 +33,13 @@ pub struct AsnListEntry {
     name: String,
 }
 
+#[derive(Debug, Serialize)]
+pub struct AsnCountryListEntry {
+    count: usize,
+    name: String,
+    iso_code: String,
+}
+
 impl TimeBuffer {
     fn new() -> Self {
         Self {
@@ -279,6 +286,18 @@ impl TimeBuffer {
             .collect()
     }
 
+    pub fn all_flows_for_country(&self, iso_code: &str) -> Vec<(FlowbeeKey, FlowbeeLocalData, FlowAnalysis)> {
+        let buffer = self.buffer.lock().unwrap();
+        buffer
+            .iter()
+            .filter(|flow| {
+                let country = get_asn_name_and_country(flow.data.0.remote_ip.as_ip());
+                country.flag == iso_code
+            })
+            .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
@@ -312,6 +331,43 @@ impl TimeBuffer {
             })
             .collect()
     }
+
+    /// Builds a list of ASNs by country with recent data, and how many flows they have.
+    pub fn country_list(&self) -> Vec<AsnCountryListEntry> {
+        // 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, 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| {
+                let country = get_asn_name_and_country(flow.data.0.remote_ip.as_ip());
+                (country.country, country.flag)
+            })
+            .collect();
+
+        // Sort the buffer
+        buffer.sort_unstable_by(|a, b| a.0.cmp(&b.0));
+
+        // Deduplicate and count, decorate with name
+        buffer
+            .into_iter()
+            .sorted()
+            .dedup_with_count()
+            .map(|(count, asn)| AsnCountryListEntry {
+                count,
+                name: asn.0,
+                iso_code: asn.1,
+            })
+            .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;
+pub use finished_flows::{AsnListEntry, AsnCountryListEntry};
 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

@@ -15,7 +15,8 @@ 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
+    expire_rtt_flows, flowbee_rtt_map, RttData, get_rtt_events_per_second, AsnListEntry,
+    AsnCountryListEntry
 };
 
 trait FlowbeeRecipient {