Add the protocols top 10 summary

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

@@ -176,6 +176,9 @@ pub enum BusRequest {
 
   /// Ether Protocol Summary
   EtherProtocolSummary,
+
+  /// IP Protocol Summary
+  IpProtocolSummary,
 }
 
 /// Defines the type of "top" flow being requested

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

@@ -148,5 +148,8 @@ pub enum BusResponse {
     v4_rtt: [u64; 2],
     /// Number of IPv6 Flows
     v6_rtt: [u64; 2],
-  }
+  },
+  
+  /// Summary of IP Protocols
+  IpProtocols(Vec<(String, (u64, u64))>),
 }

src/rust/lqos_node_manager/src/flow_monitor.rs

@@ -77,5 +77,17 @@ pub async fn flows_ether_protocol() -> NoCache<Json<BusResponse>> {
     bus_request(vec![BusRequest::EtherProtocolSummary]).await.unwrap();
   let result = responses[0].to_owned();
 
+  NoCache::new(Json(result))
+}
+
+#[get("/api/flows/ip_protocol")]
+pub async fn flows_ip_protocol() -> NoCache<Json<Vec<(String, (u64, u64))>>> {
+  let responses =
+    bus_request(vec![BusRequest::IpProtocolSummary]).await.unwrap();
+  let result = match &responses[0] {
+    BusResponse::IpProtocols(ip_protocols) => ip_protocols.to_owned(),
+    _ => Vec::new(),
+  };
+
   NoCache::new(Json(result))
 }

src/rust/lqos_node_manager/src/main.rs

@@ -118,6 +118,7 @@ fn rocket() -> _ {
         flow_monitor::flows_by_country,
         flow_monitor::flows_lat_lon,
         flow_monitor::flows_ether_protocol,
+        flow_monitor::flows_ip_protocol,
       ],
     );
 

src/rust/lqos_node_manager/static/main.html

@@ -412,7 +412,26 @@
             });
         }
 
-        function updateTop10Protocols() {}
+        function updateTop10Protocols() {
+            $.get("/api/flows/ip_protocol", data => {
+                let html = "<table class='table' style='font-size: 8pt'>";
+                html += "<thead>";
+                html += "<th>Protocol</th>";
+                html += "<th>UL ⬆️</th>";
+                html += "<th>DL ⬇️</th>";
+                html += "</thead></tbody>";
+
+                for (i=0; i<data.length; i++) {
+                    html += "<tr>";
+                    html += "<td>" + data[i][0] + "</td>";
+                    html += "<td>" + scaleNumber(data[i][1][0]) + "</td>";
+                    html += "<td>" + scaleNumber(data[i][1][1]) + "</td>";
+                    html += "</tr>";
+                }
+                html += "</tbody></table>";
+                $("#top10pro").html(html);
+            });
+        }
 
         let top10view = "circuits";
 

src/rust/lqosd/src/main.rs

@@ -242,6 +242,7 @@ fn handle_bus_requests(
       BusRequest::CurrentEndpointsByCountry => throughput_tracker::current_endpoints_by_country(),
       BusRequest::CurrentEndpointLatLon => throughput_tracker::current_lat_lon(),
       BusRequest::EtherProtocolSummary => throughput_tracker::ether_protocol_summary(),
+      BusRequest::IpProtocolSummary => throughput_tracker::ip_protocol_summary(),
     });
   }
 }

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

@@ -1,5 +1,6 @@
 use super::{get_asn_lat_lon, get_asn_name_and_country, FlowAnalysis};
 use crate::throughput_tracker::flow_data::{FlowbeeLocalData, FlowbeeRecipient};
+use fxhash::FxHashMap;
 use lqos_bus::BusResponse;
 use lqos_sys::flowbee_data::FlowbeeKey;
 use once_cell::sync::Lazy;
@@ -212,6 +213,28 @@ impl TimeBuffer {
             v6_rtt,
         }
     }
+
+    pub fn ip_protocol_summary(&self) -> Vec<(String, (u64, u64))> {
+        let buffer = self.buffer.lock().unwrap();
+
+        let mut results = FxHashMap::default();
+
+        buffer
+            .iter()
+            .for_each(|v| {
+                let (_key, data, analysis) = &v.data;
+                let proto = analysis.protocol_analysis.to_string();
+                let entry = results.entry(proto).or_insert((0, 0));
+                entry.0 += data.bytes_sent[0];
+                entry.1 += data.bytes_sent[1];
+            });
+
+        let mut results = results.into_iter().collect::<Vec<(String, (u64, u64))>>();
+        results.sort_by(|a, b| b.2.cmp(&a.2));
+        // Keep only the top 10
+        results.truncate(10);
+        results
+    }
 }
 
 pub static RECENT_FLOWS: Lazy<TimeBuffer> = Lazy::new(|| TimeBuffer::new());

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

@@ -678,3 +678,10 @@ pub fn current_lat_lon() -> BusResponse {
 pub fn ether_protocol_summary() -> BusResponse {
     flow_data::RECENT_FLOWS.ether_protocol_summary()
 }
+
+/// IP Protocol Summary
+pub fn ip_protocol_summary() -> BusResponse {
+    BusResponse::IpProtocols(
+        flow_data::RECENT_FLOWS.ip_protocol_summary()
+    )
+}