Feat: Add combined circuit rollup bus data

Refs #1094

src/rust/lqos_bus/src/bus/protocol.rs

@@ -158,8 +158,13 @@ mod tests {
         BUS_CHUNK_SIZE, MAX_FRAME_BYTES, decode_reply_cbor, decode_session_cbor, encode_reply_cbor,
         encode_session_cbor, read_frame, write_frame,
     };
-    use crate::{BusReply, BusRequest, BusResponse, BusSession, QooData, bus::BusClientError};
-    use lqos_utils::{qoq_heatmap::QoqHeatmapBlocks, units::DownUpOrder};
+    use crate::{
+        BusReply, BusRequest, BusResponse, BusSession, CircuitRollup, QooData, bus::BusClientError,
+    };
+    use lqos_utils::{
+        qoq_heatmap::QoqHeatmapBlocks,
+        units::{DownUpOrder, TcpRetransmitSample},
+    };
     use tokio::io::{AsyncWriteExt, duplex};
 
     #[test]
@@ -203,6 +208,21 @@ mod tests {
         assert_eq!(decoded.requests, session.requests);
     }
 
+    #[test]
+    fn cbor_round_trip_circuit_rollup_requests() {
+        let session = BusSession {
+            requests: vec![
+                BusRequest::GetAllCircuitRollups,
+                BusRequest::GetCircuitRollupById {
+                    circuit_id: "Circuit-1".to_string(),
+                },
+            ],
+        };
+        let bytes = encode_session_cbor(&session).expect("encode_session_cbor");
+        let decoded = decode_session_cbor(&bytes).expect("decode_session_cbor");
+        assert_eq!(decoded.requests, session.requests);
+    }
+
     #[test]
     fn cbor_round_trip_reply() {
         let reply = BusReply {
@@ -251,6 +271,48 @@ mod tests {
         assert_eq!(decoded.responses, reply.responses);
     }
 
+    #[test]
+    fn cbor_round_trip_circuit_rollup_reply() {
+        let rollup = CircuitRollup {
+            circuit_id: "Circuit-1".to_string(),
+            circuit_name: "Crew".to_string(),
+            parent_node: "Ship-10001-Crew".to_string(),
+            device_names: vec!["Crew Devices".to_string()],
+            ip_addrs: vec!["10.224.11.10".to_string()],
+            plan_mbps: DownUpOrder {
+                down: 10.0,
+                up: 2.0,
+            },
+            bytes_per_second: DownUpOrder {
+                down: 1234,
+                up: 567,
+            },
+            rtt_current_p50_nanos: DownUpOrder {
+                down: Some(1000),
+                up: Some(2000),
+            },
+            qoo: DownUpOrder {
+                down: Some(95.0),
+                up: Some(93.0),
+            },
+            tcp_retransmit_sample: DownUpOrder {
+                down: TcpRetransmitSample::new(2, 100),
+                up: TcpRetransmitSample::new(1, 50),
+            },
+            last_seen_nanos: 42,
+        };
+        let reply = BusReply {
+            responses: vec![
+                BusResponse::CircuitRollups(vec![rollup.clone()]),
+                BusResponse::CircuitRollup(Some(rollup)),
+                BusResponse::CircuitRollup(None),
+            ],
+        };
+        let bytes = encode_reply_cbor(&reply).expect("encode_reply_cbor");
+        let decoded = decode_reply_cbor(&bytes).expect("decode_reply_cbor");
+        assert_eq!(decoded.responses, reply.responses);
+    }
+
     #[tokio::test]
     async fn frame_round_trip_small_payload() {
         let (mut client, mut server) = duplex(128 * 1024);

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

@@ -278,6 +278,15 @@ pub enum BusRequest {
         circuit_id: String,
     },
 
+    /// Get live usage rollups aggregated by circuit ID.
+    GetAllCircuitRollups,
+
+    /// Get one live usage rollup aggregated by circuit ID.
+    GetCircuitRollupById {
+        /// Circuit ID to query.
+        circuit_id: String,
+    },
+
     /// Retrieve stats for all queues above a named circuit id
     GetFunnel {
         /// Circuit being analyzed, as the named circuit id

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

@@ -7,7 +7,11 @@ use crate::{
     ip_stats::{FlowbeeSummaryData, PacketHeader},
 };
 use allocative::Allocative;
-use lqos_utils::{HeatmapBlocks, qoq_heatmap::QoqHeatmapBlocks, units::DownUpOrder};
+use lqos_utils::{
+    HeatmapBlocks,
+    qoq_heatmap::QoqHeatmapBlocks,
+    units::{DownUpOrder, TcpRetransmitSample},
+};
 use serde::{Deserialize, Serialize};
 use std::net::IpAddr;
 
@@ -98,6 +102,33 @@ pub struct QooData {
     pub latest: DownUpOrder<Option<f32>>,
 }
 
+/// Live traffic and quality rollup for one logical circuit ID.
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Allocative)]
+pub struct CircuitRollup {
+    /// Circuit ID from ShapedDevices.csv.
+    pub circuit_id: String,
+    /// Circuit name from ShapedDevices.csv.
+    pub circuit_name: String,
+    /// Effective parent node for this circuit.
+    pub parent_node: String,
+    /// Device names contributing to this circuit rollup.
+    pub device_names: Vec<String>,
+    /// Active IP addresses contributing to this circuit rollup.
+    pub ip_addrs: Vec<String>,
+    /// Circuit plan in Mbps.
+    pub plan_mbps: DownUpOrder<f32>,
+    /// Current bytes-per-second passing through this circuit.
+    pub bytes_per_second: DownUpOrder<u64>,
+    /// Current RTT p50 in nanoseconds, per direction.
+    pub rtt_current_p50_nanos: DownUpOrder<Option<u64>>,
+    /// Current QoO score, per direction.
+    pub qoo: DownUpOrder<Option<f32>>,
+    /// TCP retransmit samples for this circuit at the current time.
+    pub tcp_retransmit_sample: DownUpOrder<TcpRetransmitSample>,
+    /// Most recent activity age for this circuit, in nanoseconds since boot.
+    pub last_seen_nanos: u64,
+}
+
 /// Serializable snapshot of BakeryStats for bus transmission
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Allocative)]
 pub struct BakeryStatsSnapshot {
@@ -599,6 +630,12 @@ pub enum BusResponse {
     /// Circuit data
     CircuitData(Vec<Circuit>),
 
+    /// Live circuit data aggregated by circuit ID.
+    CircuitRollups(Vec<CircuitRollup>),
+
+    /// Live circuit data for one circuit ID.
+    CircuitRollup(Option<CircuitRollup>),
+
     /// Statistics from lqosd
     LqosdStats {
         /// Number of bus requests handled

src/rust/lqos_bus/src/lib.rs

@@ -26,8 +26,8 @@ pub use ip_stats::{
 mod tc_handle;
 pub use bus::response::{
     AsnHeatmapData, AsnListEntry, BakeryStatsSnapshot, CircuitCapacityRow, CircuitCount,
-    CircuitHeatmapData, CountryListEntry, DeviceCounts, ExecutiveSummaryHeader, FlowMapPoint,
-    FlowTimelineEntry, InsightLicenseSummary, LtsCapabilitiesSummary, NodeCapacity,
+    CircuitHeatmapData, CircuitRollup, CountryListEntry, DeviceCounts, ExecutiveSummaryHeader,
+    FlowMapPoint, FlowTimelineEntry, InsightLicenseSummary, LtsCapabilitiesSummary, NodeCapacity,
     ProtocolListEntry, QooData, QueueStatsTotal, RetransmitSummary, SchedulerDetails,
     SearchResultEntry, SiteHeatmapData, StormguardDebugDirection, StormguardDebugEntry,
     TreeGuardRuntimeNodeBranchSnapshot, TreeGuardRuntimeNodeOperationSnapshot, UrgentIssue,

src/rust/lqosd/src/main.rs

@@ -655,6 +655,12 @@ fn handle_bus_requests(requests: &[BusRequest], responses: &mut Vec<BusResponse>
             BusRequest::GetCircuitById { circuit_id } => {
                 shaped_devices_tracker::get_circuit_by_id(circuit_id.clone())
             }
+            BusRequest::GetAllCircuitRollups => BusResponse::CircuitRollups(
+                shaped_devices_tracker::circuit_live::all_circuit_rollups(),
+            ),
+            BusRequest::GetCircuitRollupById { circuit_id } => BusResponse::CircuitRollup(
+                shaped_devices_tracker::circuit_live::circuit_rollup_by_id(circuit_id),
+            ),
             BusRequest::GetFunnel { target: parent } => shaped_devices_tracker::get_funnel(parent),
             BusRequest::GetLqosStats => BusResponse::LqosdStats {
                 bus_requests: BUS_REQUESTS.load(std::sync::atomic::Ordering::Relaxed),

src/rust/lqosd/src/shaped_devices_tracker/circuit_live.rs

@@ -33,6 +33,43 @@ pub struct CircuitLiveSnapshot {
     pub by_circuit_id: FxHashMap<String, CircuitLiveRollup>,
 }
 
+fn rollup_to_bus(rollup: &CircuitLiveRollup) -> lqos_bus::CircuitRollup {
+    lqos_bus::CircuitRollup {
+        circuit_id: rollup.circuit_id.clone(),
+        circuit_name: rollup.circuit_name.clone(),
+        parent_node: rollup.parent_node.clone(),
+        device_names: rollup.device_names.clone(),
+        ip_addrs: rollup.ip_addrs.clone(),
+        plan_mbps: rollup.plan_mbps,
+        bytes_per_second: rollup.bytes_per_second,
+        rtt_current_p50_nanos: rollup.rtt_current_p50_nanos,
+        qoo: rollup.qoo,
+        tcp_retransmit_sample: rollup.tcp_retransmit_sample,
+        last_seen_nanos: rollup.last_seen_nanos,
+    }
+}
+
+fn rollups_from_snapshot(snapshot: &CircuitLiveSnapshot) -> Vec<lqos_bus::CircuitRollup> {
+    let mut rollups = snapshot
+        .by_circuit_id
+        .values()
+        .map(rollup_to_bus)
+        .collect::<Vec<_>>();
+    rollups.sort_by(|left, right| left.circuit_id.cmp(&right.circuit_id));
+    rollups
+}
+
+fn rollup_from_snapshot(
+    snapshot: &CircuitLiveSnapshot,
+    circuit_id: &str,
+) -> Option<lqos_bus::CircuitRollup> {
+    let circuit_id = circuit_id.trim();
+    if circuit_id.is_empty() {
+        return None;
+    }
+    snapshot.by_circuit_id.get(circuit_id).map(rollup_to_bus)
+}
+
 #[derive(Default)]
 struct CircuitAccumulator {
     circuit_hash: Option<i64>,
@@ -116,8 +153,8 @@ pub fn rebuild_circuit_live_snapshot() -> Arc<CircuitLiveSnapshot> {
             entry.device_names.insert(device.device_name.clone());
         }
         entry.ip_addrs.insert(ip_to_string(ip_key.as_ip()));
-        entry.plan_mbps.down = entry.plan_mbps.down.max(device.download_max_mbps.round());
-        entry.plan_mbps.up = entry.plan_mbps.up.max(device.upload_max_mbps.round());
+        entry.plan_mbps.down = entry.plan_mbps.down.max(device.download_max_mbps);
+        entry.plan_mbps.up = entry.plan_mbps.up.max(device.upload_max_mbps);
         entry.bytes_per_second.down += data.bytes_per_second.down;
         entry.bytes_per_second.up += data.bytes_per_second.up;
         entry.tcp_packets.down += data.tcp_retransmit_packets.down;
@@ -188,3 +225,90 @@ pub fn fresh_circuit_live_snapshot() -> Arc<CircuitLiveSnapshot> {
     }
     rebuild_circuit_live_snapshot()
 }
+
+/// Returns live circuit rollups aggregated by circuit ID.
+///
+/// Side effects: may refresh the shared circuit-live snapshot if it is stale.
+pub fn all_circuit_rollups() -> Vec<lqos_bus::CircuitRollup> {
+    rollups_from_snapshot(&fresh_circuit_live_snapshot())
+}
+
+/// Returns the live circuit rollup for one circuit ID.
+///
+/// Side effects: may refresh the shared circuit-live snapshot if it is stale.
+pub fn circuit_rollup_by_id(circuit_id: &str) -> Option<lqos_bus::CircuitRollup> {
+    rollup_from_snapshot(&fresh_circuit_live_snapshot(), circuit_id)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{
+        CircuitLiveRollup, CircuitLiveSnapshot, rollup_from_snapshot, rollups_from_snapshot,
+    };
+    use fxhash::FxHashMap;
+    use lqos_utils::units::{DownUpOrder, TcpRetransmitSample};
+
+    fn rollup(circuit_id: &str, bytes_down: u64) -> CircuitLiveRollup {
+        CircuitLiveRollup {
+            circuit_id: circuit_id.to_string(),
+            circuit_name: format!("{circuit_id} name"),
+            parent_node: "parent".to_string(),
+            device_names: vec![format!("{circuit_id} device")],
+            ip_addrs: vec!["192.0.2.10".to_string()],
+            plan_mbps: DownUpOrder {
+                down: 100.0,
+                up: 50.0,
+            },
+            bytes_per_second: DownUpOrder {
+                down: bytes_down,
+                up: 123,
+            },
+            rtt_current_p50_nanos: DownUpOrder {
+                down: Some(1000),
+                up: Some(2000),
+            },
+            qoo: DownUpOrder {
+                down: Some(95.0),
+                up: Some(93.0),
+            },
+            tcp_retransmit_sample: DownUpOrder {
+                down: TcpRetransmitSample::new(2, 100),
+                up: TcpRetransmitSample::new(1, 50),
+            },
+            last_seen_nanos: 10,
+        }
+    }
+
+    fn snapshot(rollups: Vec<CircuitLiveRollup>) -> CircuitLiveSnapshot {
+        let mut by_circuit_id = FxHashMap::default();
+        for rollup in rollups {
+            by_circuit_id.insert(rollup.circuit_id.clone(), rollup);
+        }
+        CircuitLiveSnapshot { by_circuit_id }
+    }
+
+    #[test]
+    fn rollups_from_snapshot_are_sorted_and_preserve_totals() {
+        let snapshot = snapshot(vec![rollup("circuit-b", 200), rollup("circuit-a", 100)]);
+
+        let rollups = rollups_from_snapshot(&snapshot);
+
+        assert_eq!(rollups[0].circuit_id, "circuit-a");
+        assert_eq!(rollups[0].bytes_per_second.down, 100);
+        assert_eq!(rollups[1].circuit_id, "circuit-b");
+        assert_eq!(rollups[1].bytes_per_second.down, 200);
+    }
+
+    #[test]
+    fn rollup_from_snapshot_matches_exact_trimmed_circuit_id() {
+        let snapshot = snapshot(vec![rollup("Circuit-1", 300)]);
+
+        let rollup = rollup_from_snapshot(&snapshot, " Circuit-1 ")
+            .expect("trimmed circuit ID should match");
+
+        assert_eq!(rollup.circuit_id, "Circuit-1");
+        assert_eq!(rollup.bytes_per_second.down, 300);
+        assert!(rollup_from_snapshot(&snapshot, "circuit-1").is_none());
+        assert!(rollup_from_snapshot(&snapshot, " ").is_none());
+    }
+}