diff --git a/src/rust/lqos_bus/src/bus/request.rs b/src/rust/lqos_bus/src/bus/request.rs index 97199222..e796fa51 100644 --- a/src/rust/lqos_bus/src/bus/request.rs +++ b/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 diff --git a/src/rust/lqos_bus/src/bus/response.rs b/src/rust/lqos_bus/src/bus/response.rs index 80c26bc1..48d716e8 100644 --- a/src/rust/lqos_bus/src/bus/response.rs +++ b/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)>), /// The index of the new packet collection session @@ -125,4 +126,7 @@ pub enum BusResponse { /// Top Flopws TopFlows(Vec), + + /// Flows by IP + FlowsByIp(Vec), } diff --git a/src/rust/lqos_node_manager/src/queue_info.rs b/src/rust/lqos_node_manager/src/queue_info.rs index 3a0a34ab..d34a7668 100644 --- a/src/rust/lqos_node_manager/src/queue_info.rs +++ b/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/")] +pub async fn flow_stats(ip_list: String, _auth: AuthGuard) -> NoCache>> { + let mut result = Vec::new(); + let request: Vec = 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/")] pub async fn flow_stats(ip_list: String, _auth: AuthGuard) -> NoCache)>>> { let mut result = Vec::new(); let request: Vec = 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
Flows (Last 30 Seconds)
-
@@ -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 = ""; + html += ""; + html += ""; + html += ""; + html += ""; + html += ""; + html += ""; + html += ""; + html += ""; + html += ""; + html += ""; + for (var i=0; i"; + html += ""; + html += ""; + html += ""; + html += ""; + html += ""; + html += ""; + html += ""; + } + html += ""; + + /*html += ""; html += ""; html += ""; html += ""; @@ -837,6 +858,7 @@ html += ""; } html += "
ConnectionBytesPacketsTCP RetriesTCP RTTASNASN Country
" + scaleNumber(data[i].bytes_sent[0]) + " / " + scaleNumber(data[i].bytes_sent[1]) + "" + scaleNumber(data[i].packets_sent[0]) + " / " + scaleNumber(data[i].packets_sent[1]) + "" + data[i].retries[0] + " / " + data[i].retries[1] + "" + (data[i].last_rtt[0] / 1000000).toFixed(2) + " / " + (data[i].last_rtt[1] / 1000000).toFixed(2) + "(" + data[i].remote_asn + ") " + data[i].remote_asn_name + "" + data[i].remote_asn_country + "
ProtocolSrcSrc Port
"; + */ $("#flowList").html(html); }) } diff --git a/src/rust/lqos_node_manager/static/main.html b/src/rust/lqos_node_manager/static/main.html index ebde62f2..1c14c9a9 100644 --- a/src/rust/lqos_node_manager/static/main.html +++ b/src/rust/lqos_node_manager/static/main.html @@ -333,8 +333,8 @@ html += "" + data[i].local_ip + ""; html += "" + data[i].remote_ip + ""; // TODO: Check scaling - html += "" + scaleNumber(data[i].rate_estimate_bps[0] * 8) + ""; - html += "" + scaleNumber(data[i].rate_estimate_bps[1] * 8) + ""; + html += "" + scaleNumber(data[i].rate_estimate_bps[0]) + ""; + html += "" + scaleNumber(data[i].rate_estimate_bps[1]) + ""; html += "" + (data[i].last_rtt[0] / 1000000).toFixed(2) + ""; html += "" + (data[i].last_rtt[1] / 1000000).toFixed(2) + ""; html += "" + data[i].retries[0] + "/" + data[i].retries[1] + ""; diff --git a/src/rust/lqos_sys/src/bpf/common/flows.h b/src/rust/lqos_sys/src/bpf/common/flows.h index 9c2cd02a..719c4db1 100644 --- a/src/rust/lqos_sys/src/bpf/common/flows.h +++ b/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]; diff --git a/src/rust/lqosd/src/main.rs b/src/rust/lqosd/src/main.rs index 48f11c24..b18cb6d2 100644 --- a/src/rust/lqosd/src/main.rs +++ b/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), }); } } diff --git a/src/rust/lqosd/src/throughput_tracker/mod.rs b/src/rust/lqosd/src/throughput_tracker/mod.rs index 8457f4c4..8c8f16c0 100644 --- a/src/rust/lqosd/src/throughput_tracker/mod.rs +++ b/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::() { + 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 +} \ No newline at end of file