Very silly commit - adds a /showoff page to the node manager firing particles from all endpoints at my ISP in Missouri. Will turn into something useful in the future.

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

@@ -170,6 +170,9 @@ pub enum BusRequest {
 
   /// Current Endpoints by Country
   CurrentEndpointsByCountry,
+
+  /// Lat/Lon of Endpoints
+  CurrentEndpointLatLon,
 }
 
 /// Defines the type of "top" flow being requested

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

@@ -128,4 +128,7 @@ pub enum BusResponse {
 
   /// Current endpoints by country
   CurrentEndpointsByCountry(Vec<(String, [u64; 2], [f32; 2])>),
+
+  /// Current Lat/Lon of endpoints
+  CurrentLatLon(Vec<(f64, f64)>),
 }

src/rust/lqos_node_manager/src/flow_monitor.rs

@@ -56,5 +56,17 @@ pub async fn flows_by_country() -> NoCache<Json<Vec<(String, [u64; 2], [f32; 2])
     _ => Vec::new(),
   };
 
+  NoCache::new(Json(result))
+}
+
+#[get("/api/flows/lat_lon")]
+pub async fn flows_lat_lon() -> NoCache<Json<Vec<(f64, f64)>>> {
+  let responses =
+    bus_request(vec![BusRequest::CurrentEndpointLatLon]).await.unwrap();
+  let result = match &responses[0] {
+    BusResponse::CurrentLatLon(lat_lon) => lat_lon.to_owned(),
+    _ => Vec::new(),
+  };
+
   NoCache::new(Json(result))
 }

src/rust/lqos_node_manager/src/main.rs

@@ -44,6 +44,7 @@ fn rocket() -> _ {
         static_pages::shaped_devices_add_page,
         static_pages::unknown_devices_page,
         static_pages::circuit_queue,
+        static_pages::pretty_map_graph,
         config_control::config_page,
         network_tree::tree_page,
         static_pages::ip_dump,
@@ -115,6 +116,7 @@ fn rocket() -> _ {
         flow_monitor::count_flows,
         flow_monitor::top_5_flows,
         flow_monitor::flows_by_country,
+        flow_monitor::flows_lat_lon,
       ],
     );
 

src/rust/lqos_node_manager/src/static_pages.rs

@@ -75,6 +75,14 @@ pub async fn shaped_devices_add_page<'a>(
   NoCache::new(NamedFile::open("static/shaped-add.html").await.ok())
 }
 
+// Temporary for funsies
+#[get("/showoff")]
+pub async fn pretty_map_graph<'a>(
+  _auth: AuthGuard,
+) -> NoCache<Option<NamedFile>> {
+  NoCache::new(NamedFile::open("static/showoff.html").await.ok())
+}
+
 #[get("/vendor/bootstrap.min.css")]
 pub async fn bootsrap_css<'a>() -> LongCache<Option<NamedFile>> {
   LongCache::new(NamedFile::open("static/vendor/bootstrap.min.css").await.ok())

src/rust/lqosd/src/main.rs

@@ -232,6 +232,7 @@ fn handle_bus_requests(
       BusRequest::TopFlows { n, flow_type } => throughput_tracker::top_flows(*n, *flow_type),
       BusRequest::FlowsByIp(ip) => throughput_tracker::flows_by_ip(ip),
       BusRequest::CurrentEndpointsByCountry => throughput_tracker::current_endpoints_by_country(),
+      BusRequest::CurrentEndpointLatLon => throughput_tracker::current_lat_lon(),
     });
   }
 }

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

@@ -136,6 +136,21 @@ impl GeoTable {
 
         (owners, country)
     }
+
+    pub fn find_lat_lon_by_ip(&self, ip: IpAddr) -> (f64, f64) {
+        log::debug!("Looking up ASN for IP: {:?}", ip);
+        let ip = match ip {
+            IpAddr::V4(ip) => ip.to_ipv6_mapped(),
+            IpAddr::V6(ip) => ip,
+        };
+
+        if let Some(matched) = self.geo_trie.longest_match(ip) {
+            log::debug!("Matched Geo: {:?}", matched.1.city_and_country);
+            return (matched.1.latitude, matched.1.longitude);
+        }
+
+        (0.0, 0.0)
+    }
 }
 
 ///////////////////////////////////////////////////////////////////////

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

@@ -1,8 +1,8 @@
-use super::{get_asn_name_and_country, FlowAnalysis};
+use super::{get_asn_lat_lon, get_asn_name_and_country, FlowAnalysis};
 use crate::throughput_tracker::flow_data::FlowbeeRecipient;
 use lqos_sys::flowbee_data::{FlowbeeData, FlowbeeKey};
 use once_cell::sync::Lazy;
-use std::sync::{Arc, Mutex};
+use std::{alloc::LayoutError, sync::{Arc, Mutex}};
 
 pub struct TimeBuffer {
     buffer: Mutex<Vec<TimeEntry>>,
@@ -34,6 +34,27 @@ impl TimeBuffer {
         buffer.push(entry);
     }
 
+    pub fn lat_lon_endpoints(&self) -> Vec<(f64, f64)> {
+        let buffer = self.buffer.lock().unwrap();
+        let mut my_buffer = buffer
+            .iter()
+            .map(|v| {
+                let (key, _data, _analysis) = &v.data;
+                let (lat, lon) = get_asn_lat_lon(key.remote_ip.as_ip());
+                (lat, lon)
+            })
+            .filter(|(lat, lon)| *lat != 0.0 && *lon != 0.0)
+            .collect::<Vec<(f64, f64)>>();
+
+        // Sort by lat/lon
+        my_buffer.sort_by(|a, b| a.0.partial_cmp(&b.0).unwrap());
+
+        // Depuplicate
+        my_buffer.dedup();
+
+        my_buffer
+    }
+
     pub fn country_summary(&self) -> Vec<(String, [u64; 2], [f32; 2])> {
         let buffer = self.buffer.lock().unwrap();
         let mut my_buffer = buffer

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

@@ -82,3 +82,12 @@ pub fn get_asn_name_and_country(ip: IpAddr) -> (String, String) {
     }
     (String::new(), String::new())
 }
+
+pub fn get_asn_lat_lon(ip: IpAddr) -> (f64, f64) {
+    if let Ok(table_lock) = ANALYSIS.asn_table.lock() {
+        if let Some(table) = table_lock.as_ref() {
+            return table.find_lat_lon_by_ip(ip);
+        }
+    }
+    (0.0, 0.0)
+}

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

@@ -657,4 +657,10 @@ pub fn flows_by_ip(ip: &str) -> BusResponse {
 pub fn current_endpoints_by_country() -> BusResponse {
     let summary = flow_data::RECENT_FLOWS.country_summary();
     BusResponse::CurrentEndpointsByCountry(summary)
-}
+}
+
+/// Current endpoint lat/lon
+pub fn current_lat_lon() -> BusResponse {
+    let summary = flow_data::RECENT_FLOWS.lat_lon_endpoints();
+    BusResponse::CurrentLatLon(summary)
+}