Add plumbing for the top 10 displays. Display ethertypes from expired flows.

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

@@ -173,6 +173,9 @@ pub enum BusRequest {
 
   /// Lat/Lon of Endpoints
   CurrentEndpointLatLon,
+
+  /// Ether Protocol Summary
+  EtherProtocolSummary,
 }
 
 /// Defines the type of "top" flow being requested

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

@@ -133,4 +133,20 @@ pub enum BusResponse {
 
   /// Current Lat/Lon of endpoints
   CurrentLatLon(Vec<(f64, f64)>),
+
+  /// Summary of Ether Protocol
+  EtherProtocols{
+    /// Number of IPv4 Bytes
+    v4_bytes: [u64; 2],
+    /// Number of IPv6 Bytes
+    v6_bytes: [u64; 2],
+    /// Number of IPv4 Packets
+    v4_packets: [u64; 2],
+    /// Number of IPv6 Packets
+    v6_packets: [u64; 2],
+    /// Number of IPv4 Flows
+    v4_rtt: [u64; 2],
+    /// Number of IPv6 Flows
+    v6_rtt: [u64; 2],
+  }
 }

src/rust/lqos_node_manager/src/flow_monitor.rs

@@ -68,5 +68,14 @@ pub async fn flows_lat_lon() -> NoCache<Json<Vec<(f64, f64)>>> {
     _ => Vec::new(),
   };
 
+  NoCache::new(Json(result))
+}
+
+#[get("/api/flows/ether_protocol")]
+pub async fn flows_ether_protocol() -> NoCache<Json<BusResponse>> {
+  let responses =
+    bus_request(vec![BusRequest::EtherProtocolSummary]).await.unwrap();
+  let result = responses[0].to_owned();
+
   NoCache::new(Json(result))
 }

src/rust/lqos_node_manager/src/main.rs

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

src/rust/lqos_node_manager/static/main.html

@@ -148,10 +148,14 @@
                             <button class="btn btn-small btn-success" href="/top10" onclick="showCircuits()">Circuits</button>
                             <button class="btn btn-small btn-success" href="/top10" onclick="showFlows()">Flows</button>
                             <button class="btn btn-small btn-success" href="/top10" onclick="showEndpoints()">Geo Endpoints</button>
+                            <button class="btn btn-small btn-success" href="/top10" onclick="showProtocols()">Protocols</button>
+                            <button class="btn btn-small btn-success" href="/top10" onclick="showEthertypes()">Ethertypes</button>
                         </h5>
                         <div id="top10dl" style="display:block;"></div>
                         <div id="top10flows" style="display: none;"></div>
                         <div id="top10ep" style="display: none;"></div>
+                        <div id="top10eth" style="display: none;"></div>
+                        <div id="top10pro" style="display: none;"></div>
                     </div>
                 </div>
             </div>
@@ -329,7 +333,7 @@
                 html += "<th>Country</th>";
                 html += "</thead><tbody>";
                 for (var i = 0; i<data.length; i++) {
-                    console.log(data[i]);
+                    //console.log(data[i]);
                     html += "<tr>";
                     html += "<td>" + data[i].analysis + "</td>";
                     html += "<td>" + data[i].local_ip + "</td>";
@@ -376,12 +380,48 @@
             });
         }
 
+        function updateTop10Ethertypes() {
+            $.get("/api/flows/ether_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 += "<th>UL RTT</th>";
+                html += "<th>DL RTT</th>";
+                html += "</thead></tbody>";
+                let row = data.EtherProtocols;
+                html += "<tr>";
+                html += "<td>IPv4</td>";
+                html += "<td>" + scaleNumber(row.v4_bytes[0]) + "</td>";
+                html += "<td>" + scaleNumber(row.v4_bytes[1]) + "</td>";
+                html += "<td>" + scaleNanos(row.v4_rtt[0]) + "</td>";
+                html += "<td>" + scaleNanos(row.v4_rtt[1]) + "</td>";
+                html += "</tr>";
+
+                html += "<tr>";
+                html += "<td>IPv6</td>";
+                html += "<td>" + scaleNumber(row.v6_bytes[0]) + "</td>";
+                html += "<td>" + scaleNumber(row.v6_bytes[1]) + "</td>";
+                html += "<td>" + scaleNanos(row.v6_rtt[0]) + "</td>";
+                html += "<td>" + scaleNanos(row.v6_rtt[1]) + "</td>";
+                html += "</tr>";
+
+                html += "</tbody></table>";
+                $("#top10eth").html(html);
+            });
+        }
+
+        function updateTop10Protocols() {}
+
         let top10view = "circuits";
 
         function showCircuits() {
             $("#top10dl").show();
             $("#top10flows").hide();
             $("#top10ep").hide();
+            $("#top10eth").hide();
+            $("#top10pro").hide();
             top10view = "circuits";
         }
 
@@ -389,6 +429,8 @@
             $("#top10dl").hide();
             $("#top10flows").show();
             $("#top10ep").hide();
+            $("#top10eth").hide();
+            $("#top10pro").hide();
             top10view = "flows";
         }
 
@@ -396,9 +438,29 @@
             $("#top10dl").hide();
             $("#top10flows").hide();
             $("#top10ep").show();
+            $("#top10eth").hide();
+            $("#top10pro").hide();
             top10view = "endpoints";
         }
 
+        function showProtocols() {
+            $("#top10dl").hide();
+            $("#top10flows").hide();
+            $("#top10ep").hide();
+            $("#top10eth").hide();
+            $("#top10pro").show();
+            top10view = "protocols";
+        }
+
+        function showEthertypes() {
+            $("#top10dl").hide();
+            $("#top10flows").hide();
+            $("#top10ep").hide();
+            $("#top10eth").show();
+            $("#top10pro").hide();
+            top10view = "ethertypes";
+        }
+
         var rttGraph = new RttHistogram();
 
         function updateHistogram() {
@@ -425,8 +487,12 @@
                     updateTop10();
                 } else if (top10view == "flows") {
                     updateTop10Flows();
-                } else {
+                } else if (top10view == "endpoints") {
                     updateTop10Endpoints();
+                } else if (top10view == "protocols") {
+                    updateTop10Protocols();
+                } else if (top10view == "ethertypes") {
+                    updateTop10Ethertypes();
                 }
             }
 

src/rust/lqosd/src/main.rs

@@ -241,6 +241,7 @@ fn handle_bus_requests(
       BusRequest::FlowsByIp(ip) => throughput_tracker::flows_by_ip(ip),
       BusRequest::CurrentEndpointsByCountry => throughput_tracker::current_endpoints_by_country(),
       BusRequest::CurrentEndpointLatLon => throughput_tracker::current_lat_lon(),
+      BusRequest::EtherProtocolSummary => throughput_tracker::ether_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 lqos_bus::BusResponse;
 use lqos_sys::flowbee_data::FlowbeeKey;
 use once_cell::sync::Lazy;
 use std::sync::{Arc, Mutex};
@@ -136,6 +137,81 @@ impl TimeBuffer {
 
         country_summary
     }
+
+    fn median(slice: &[u64]) -> u64 {
+        if slice.is_empty() {
+            return 0;
+        }
+        let mut slice = slice.to_vec();
+        slice.sort_by(|a, b| a.partial_cmp(b).unwrap());
+        let mid = slice.len() / 2;
+        if slice.len() % 2 == 0 {
+            (slice[mid] + slice[mid - 1]) / 2
+        } else {
+            slice[mid]
+        }
+    }
+
+    pub fn ether_protocol_summary(&self) -> BusResponse {
+        let buffer = self.buffer.lock().unwrap();
+
+        let mut v4_bytes_sent = [0,0];
+        let mut v4_packets_sent = [0,0];
+        let mut v6_bytes_sent = [0,0];
+        let mut v6_packets_sent = [0,0];
+        let mut v4_rtt = [Vec::new(), Vec::new()];
+        let mut v6_rtt = [Vec::new(), Vec::new()];
+
+        buffer
+            .iter()
+            .for_each(|v| {
+                let (key, data, _analysis) = &v.data;
+                if key.local_ip.is_v4() {
+                    // It's V4
+                    v4_bytes_sent[0] += data.bytes_sent[0];
+                    v4_bytes_sent[1] += data.bytes_sent[1];
+                    v4_packets_sent[0] += data.packets_sent[0];
+                    v4_packets_sent[1] += data.packets_sent[1];
+                    if data.rtt[0].as_nanos() > 0 {
+                        v4_rtt[0].push(data.rtt[0].as_nanos());
+                    }
+                    if data.rtt[1].as_nanos() > 0 {
+                        v4_rtt[1].push(data.rtt[1].as_nanos());
+                    }
+                } else {
+                    // It's V6
+                    v6_bytes_sent[0] += data.bytes_sent[0];
+                    v6_bytes_sent[1] += data.bytes_sent[1];
+                    v6_packets_sent[0] += data.packets_sent[0];
+                    v6_packets_sent[1] += data.packets_sent[1];
+                    if data.rtt[0].as_nanos() > 0 {
+                        v6_rtt[0].push(data.rtt[0].as_nanos());
+                    }
+                    if data.rtt[1].as_nanos() > 0 {
+                        v6_rtt[1].push(data.rtt[1].as_nanos());
+                    }
+
+                }
+            });
+        
+        let v4_rtt = [
+            Self::median(&v4_rtt[0]),
+            Self::median(&v4_rtt[1]),
+        ];
+        let v6_rtt = [
+            Self::median(&v6_rtt[0]),
+            Self::median(&v6_rtt[1]),
+        ];
+
+        BusResponse::EtherProtocols {
+            v4_bytes: v4_bytes_sent,
+            v6_bytes: v6_bytes_sent,
+            v4_packets: v4_packets_sent,
+            v6_packets: v6_packets_sent,
+            v4_rtt,
+            v6_rtt,
+        }
+    }
 }
 
 pub static RECENT_FLOWS: Lazy<TimeBuffer> = Lazy::new(|| TimeBuffer::new());

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

@@ -673,3 +673,8 @@ pub fn current_lat_lon() -> BusResponse {
     let summary = flow_data::RECENT_FLOWS.lat_lon_endpoints();
     BusResponse::CurrentLatLon(summary)
 }
+
+/// Ether Protocol Summary
+pub fn ether_protocol_summary() -> BusResponse {
+    flow_data::RECENT_FLOWS.ether_protocol_summary()
+}