Use the new flows system rather than Heimdall to display the circuits flows tab. Not complete yet - the backend is all there still. Also needs some tweaking on the hyperfocus for packet capture mode.

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

@@ -134,6 +134,7 @@ pub enum BusRequest {
   GetLqosStats,
 
   /// Tell me flow stats for a given IP address
+  #[deprecated(note = "Heimdall flows are being migrated to flows 2")]
   GetFlowStats(String),
 
   /// Tell Heimdall to hyper-focus on an IP address for a bit
@@ -167,6 +168,9 @@ pub enum BusRequest {
     /// The number of flows to return
     n: u32 
   },
+
+  /// Flows by IP Address
+  FlowsByIp(String),
 }
 
 /// Defines the type of "top" flow being requested

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

@@ -92,6 +92,7 @@ pub enum BusResponse {
   },
 
   /// Flow Data
+  #[deprecated(note = "Being replaced by FlowbeeData")]
   FlowData(Vec<(FlowTransport, Option<FlowTransport>)>),
 
   /// The index of the new packet collection session
@@ -125,4 +126,7 @@ pub enum BusResponse {
 
   /// Top Flopws
   TopFlows(Vec<FlowbeeData>),
+
+  /// Flows by IP
+  FlowsByIp(Vec<FlowbeeData>),
 }

src/rust/lqos_node_manager/src/queue_info.rs

@@ -1,7 +1,7 @@
 use crate::auth_guard::AuthGuard;
 use crate::cache_control::NoCache;
 use crate::tracker::{SHAPED_DEVICES, lookup_dns};
-use lqos_bus::{bus_request, BusRequest, BusResponse, FlowTransport, PacketHeader, QueueStoreTransit};
+use lqos_bus::{bus_request, BusRequest, BusResponse, FlowTransport, FlowbeeData, PacketHeader, QueueStoreTransit};
 use rocket::fs::NamedFile;
 use rocket::http::Status;
 use rocket::response::content::RawJson;
@@ -107,6 +107,19 @@ pub async fn raw_queue_by_circuit(
 }
 
 #[get("/api/flows/<ip_list>")]
+pub async fn flow_stats(ip_list: String, _auth: AuthGuard) -> NoCache<Json<Vec<FlowbeeData>>> {
+  let mut result = Vec::new();
+  let request: Vec<BusRequest> = ip_list.split(',').map(|ip| BusRequest::FlowsByIp(ip.to_string())).collect();
+  let responses = bus_request(request).await.unwrap();
+  for r in responses.iter() {
+    if let BusResponse::FlowsByIp(flow) = r {
+      result.extend_from_slice(flow);
+    }
+  }
+  NoCache::new(Json(result))
+}
+
+/*#[get("/api/flows/<ip_list>")]
 pub async fn flow_stats(ip_list: String, _auth: AuthGuard) -> NoCache<MsgPack<Vec<(FlowTransport, Option<FlowTransport>)>>> {
   let mut result = Vec::new();
   let request: Vec<BusRequest> = ip_list.split(',').map(|ip| BusRequest::GetFlowStats(ip.to_string())).collect();
@@ -117,7 +130,7 @@ pub async fn flow_stats(ip_list: String, _auth: AuthGuard) -> NoCache<MsgPack<Ve
     }
   }
   NoCache::new(MsgPack(result))
-}
+}*/
 
 #[derive(Serialize, Clone)]
 #[serde(crate = "rocket::serde")]

src/rust/lqos_node_manager/static/circuit_queue.html

@@ -215,10 +215,6 @@
                         <div class="card bg-light">
                             <div class="card-body">
                                 <h5 class="card-title"><i class="fa fa-bar-chart"></i> Flows (Last 30 Seconds)</h5>
-                                <p class="alert alert-warning" role="alert">
-                                    <i class="fa fa-warning"></i> Gathering packet data can cause high CPU load during
-                                    the capture window.
-                                </p>
                                 <div id="packetButtons"></div>
                                 <div id="flowList"></div>
                             </div>
@@ -779,10 +775,35 @@
             }
             ip_list = ip_list.substring(0, ip_list.length - 1);
             if (ip_list == "") return;
-            msgPackGet("/api/flows/" + ip_list, (data) => {
+            $.get("/api/flows/" + ip_list, (data) => {
+            //msgPackGet("/api/flows/" + ip_list, (data) => {
                 //console.log(data);
                 let html = "<table class='table table-striped'>";
+                
                 html += "<thead>";
+                html += "<th>Connection</th>";
+                html += "<th>Bytes</th>";
+                html += "<th>Packets</th>";
+                html += "<th>TCP Retries</th>";
+                html += "<th>TCP RTT</th>";
+                html += "<th>ASN</th>";
+                html += "<th>ASN Country</th>";
+                html += "</thead>";
+                html += "<tbody>";
+                for (var i=0; i<data.length; i++) {
+                    html += "<tr>";
+                    html += "<td>" + data[i].analysis + "</td>";
+                    html += "<td>" + scaleNumber(data[i].bytes_sent[0]) + " / " + scaleNumber(data[i].bytes_sent[1]) + "</td>";
+                    html += "<td>" + scaleNumber(data[i].packets_sent[0]) + " / " + scaleNumber(data[i].packets_sent[1]) + "</td>";
+                    html += "<td>" + data[i].retries[0] + " / " + data[i].retries[1] + "</td>";
+                    html += "<td>" + (data[i].last_rtt[0] / 1000000).toFixed(2) + " / " + (data[i].last_rtt[1] / 1000000).toFixed(2) + "</td>";
+                    html += "<td>(" + data[i].remote_asn + ") " + data[i].remote_asn_name + "</td>";
+                    html += "<td>" + data[i].remote_asn_country + "</td>";
+                    html += "</tr>";
+                }
+                html += "</tbody>";
+
+                /*html += "<thead>";
                 html += "<th>Protocol</th>";
                 html += "<th>Src</th>";
                 html += "<th>Src Port</th>";
@@ -837,6 +858,7 @@
                     html += "</tr>";
                 }
                 html += "</tbody></table>";
+                */
                 $("#flowList").html(html);
             })
         }

src/rust/lqos_node_manager/static/main.html

@@ -333,8 +333,8 @@
                     html += "<td>" + data[i].local_ip + "</td>";
                     html += "<td>" + data[i].remote_ip + "</td>";
                     // TODO: Check scaling
-                    html += "<td>" + scaleNumber(data[i].rate_estimate_bps[0] * 8) + "</td>";
-                    html += "<td>" + scaleNumber(data[i].rate_estimate_bps[1] * 8) + "</td>";
+                    html += "<td>" + scaleNumber(data[i].rate_estimate_bps[0]) + "</td>";
+                    html += "<td>" + scaleNumber(data[i].rate_estimate_bps[1]) + "</td>";
                     html += "<td>" + (data[i].last_rtt[0] / 1000000).toFixed(2) + "</td>";
                     html += "<td>" + (data[i].last_rtt[1] / 1000000).toFixed(2) + "</td>";
                     html += "<td>" + data[i].retries[0] + "/" + data[i].retries[1] + "</td>";

src/rust/lqos_sys/src/bpf/common/flows.h

@@ -158,7 +158,7 @@ static __always_inline void update_flow_rates(
     if (dissector->now > data->next_count_time[rate_index]) {
         // Calculate the rate estimate
         __u64 bits = (data->bytes_sent[rate_index] - data->next_count_bytes[rate_index])*8;
-        __u64 time = (dissector->now - data->last_count_time[rate_index]) / 10000000; // 1 Second
+        __u64 time = (dissector->now - data->last_count_time[rate_index]) / SECOND_IN_NANOS; // 1 Second
         data->rate_estimate_bps[rate_index] = (bits/time); // bits per second
         data->next_count_time[rate_index] = dissector->now + SECOND_IN_NANOS;
         data->next_count_bytes[rate_index] = data->bytes_sent[rate_index];

src/rust/lqosd/src/main.rs

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

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

@@ -2,6 +2,8 @@ pub mod flow_data;
 mod heimdall_data;
 mod throughput_entry;
 mod tracking_data;
+use std::net::IpAddr;
+
 use crate::{
     long_term_stats::get_network_tree,
     shaped_devices_tracker::{NETWORK_JSON, SHAPED_DEVICES, STATS_NEEDS_NEW_SHAPED_DEVICES},
@@ -614,3 +616,41 @@ pub fn top_flows(n: u32, flow_type: TopFlowType) -> BusResponse {
 
     BusResponse::TopFlows(result)
 }
+
+/// Flows by IP
+pub fn flows_by_ip(ip: &str) -> BusResponse {
+    if let Ok(ip) = ip.parse::<IpAddr>() {
+        let ip = XdpIpAddress::from_ip(ip);
+        let lock = ALL_FLOWS.lock().unwrap();
+        let matching_flows: Vec<_> = lock
+            .iter()
+            .filter(|(key, _)| key.local_ip == ip)
+            .map(|(key, row)| {
+                let (remote_asn_name, remote_asn_country) = get_asn_name_and_country(row.1.asn_id.0);
+    
+                lqos_bus::FlowbeeData {
+                    remote_ip: key.remote_ip.as_ip().to_string(),
+                    local_ip: key.local_ip.as_ip().to_string(),
+                    src_port: key.src_port,
+                    dst_port: key.dst_port,
+                    ip_protocol: FlowbeeProtocol::from(key.ip_protocol),
+                    bytes_sent: row.0.bytes_sent,
+                    packets_sent: row.0.packets_sent,
+                    rate_estimate_bps: row.0.rate_estimate_bps,
+                    retries: row.0.retries,
+                    last_rtt: row.0.last_rtt,
+                    end_status: row.0.end_status,
+                    tos: row.0.tos,
+                    flags: row.0.flags,
+                    remote_asn: row.1.asn_id.0,
+                    remote_asn_name,
+                    remote_asn_country,
+                    analysis: row.1.protocol_analysis.to_string(),
+                }
+            })
+            .collect();
+
+        return BusResponse::FlowsByIp(matching_flows);
+    }
+    BusResponse::Ack
+}