Merge remote-tracking branch 'origin/feat/1094_cake_extended' into codex/integrate-api-cake

# Conflicts:
#	src/rust/lqos_bus/src/bus/protocol.rs
#	src/rust/lqos_bus/src/bus/queue_data.rs
#	src/rust/lqos_bus/src/lib.rs
#	src/rust/lqos_queue_tracker/src/queue_diff.rs
#	src/rust/lqos_queue_tracker/src/queue_types/tc_cake.rs
#	src/rust/lqosd/src/throughput_tracker/mod.rs
This commit is contained in:
Herbert Wolverson committed 2026-07-17 14:59:46 -05:00
commit 174b4cf3ef
13 files changed
+1281 -42

No files matched your search

+119 -2
View File
@@ -159,8 +159,12 @@ mod tests {
encode_session_cbor, read_frame, write_frame,
};
use crate::{
BusReply, BusRequest, BusResponse, BusSession, OverrideLayerSelection, OverrideMutation,
bus::BusClientError,
BusReply, BusRequest, BusResponse, BusSession, CircuitRollup, OverrideLayerSelection,
OverrideMutation, QooData, bus::BusClientError,
};
use lqos_utils::{
qoq_heatmap::QoqHeatmapBlocks,
units::{DownUpOrder, TcpRetransmitSample},
};
use tokio::io::{AsyncWriteExt, duplex};
@@ -202,6 +206,39 @@ mod tests {
assert_eq!(decoded.requests, session.requests);
}
#[test]
fn cbor_round_trip_qoo_requests() {
let session = BusSession {
requests: vec![
BusRequest::GetQoo,
BusRequest::GetSiteQoo {
site_name: "North".to_string(),
},
BusRequest::GetCircuitQoo {
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_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 {
@@ -212,6 +249,86 @@ mod tests {
assert_eq!(decoded.responses, reply.responses);
}
#[test]
fn cbor_round_trip_qoo_reply() {
let mut download_total = [None; 15];
let mut upload_total = [None; 15];
download_total[14] = Some(91.0);
upload_total[14] = Some(89.0);
let reply = BusReply {
responses: vec![BusResponse::Qoo(Some(QooData {
key: "site:North".to_string(),
entity_kind: "site".to_string(),
label: "North".to_string(),
site_name: Some("North".to_string()),
circuit_id: None,
blocks: QoqHeatmapBlocks {
download_total,
upload_total,
},
latest: DownUpOrder {
down: Some(91.0),
up: Some(89.0),
},
}))],
};
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);
}
#[test]
fn cbor_round_trip_empty_qoo_reply() {
let reply = BusReply {
responses: vec![BusResponse::Qoo(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);
}
#[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);
+79
View File
@@ -38,6 +38,15 @@ pub struct CakeDiffTinTransit {
pub drops: u64,
pub marks: u64,
pub base_delay_us: u64,
pub sent_packets: Option<u64>,
pub peak_delay_us: Option<u64>,
pub avg_delay_us: Option<u64>,
pub way_indirect_hits: Option<u64>,
pub way_misses: Option<u64>,
pub way_collisions: Option<u64>,
pub sparse_flows: Option<u64>,
pub bulk_flows: Option<u64>,
pub unresponsive_flows: Option<u64>,
}
#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Default, Allocative)]
@@ -108,3 +117,73 @@ pub struct CakeTinTransit {
//pub flow_quantum: u64,
}
*/
#[cfg(test)]
mod tests {
use super::*;
use serde::Serialize;
#[derive(Serialize)]
struct LegacyCakeDiffTinTransit {
sent_bytes: u64,
backlog_bytes: u32,
drops: u32,
marks: u32,
base_delay_us: u32,
}
#[test]
fn cake_diff_tin_transit_accepts_legacy_payloads() {
let legacy = LegacyCakeDiffTinTransit {
sent_bytes: 100,
backlog_bytes: 10,
drops: 2,
marks: 3,
base_delay_us: 42,
};
let bytes = serde_cbor::to_vec(&legacy).expect("legacy tin should serialize");
let decoded: CakeDiffTinTransit =
serde_cbor::from_slice(&bytes).expect("legacy tin should deserialize");
assert_eq!(decoded.sent_bytes, 100);
assert_eq!(decoded.backlog_bytes, 10);
assert_eq!(decoded.drops, 2);
assert_eq!(decoded.marks, 3);
assert_eq!(decoded.base_delay_us, 42);
assert_eq!(decoded.sent_packets, None);
assert_eq!(decoded.peak_delay_us, None);
assert_eq!(decoded.avg_delay_us, None);
assert_eq!(decoded.way_indirect_hits, None);
assert_eq!(decoded.way_misses, None);
assert_eq!(decoded.way_collisions, None);
assert_eq!(decoded.sparse_flows, None);
assert_eq!(decoded.bulk_flows, None);
assert_eq!(decoded.unresponsive_flows, None);
}
#[test]
fn cake_diff_tin_transit_round_trips_extended_payloads() {
let tin = CakeDiffTinTransit {
sent_bytes: 100,
backlog_bytes: 10,
drops: 2,
marks: 3,
base_delay_us: 42,
sent_packets: Some(5_000_000_000),
peak_delay_us: Some(101),
avg_delay_us: Some(51),
way_indirect_hits: Some(70_000),
way_misses: Some(80_000),
way_collisions: Some(90_000),
sparse_flows: Some(11),
bulk_flows: Some(12),
unresponsive_flows: Some(13),
};
let bytes = serde_cbor::to_vec(&tin).expect("extended tin should serialize");
let decoded: CakeDiffTinTransit =
serde_cbor::from_slice(&bytes).expect("extended tin should deserialize");
assert_eq!(decoded, tin);
}
}
+24
View File
@@ -299,6 +299,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
@@ -621,6 +630,21 @@ pub enum BusRequest {
/// Retrieve queue marks/drops totals
GetQueueStatsTotal,
/// Retrieve current top-level QoO history.
GetQoo,
/// Retrieve current QoO history for a site by site name.
GetSiteQoo {
/// Site name to query.
site_name: String,
},
/// Retrieve current QoO history for a circuit by circuit ID.
GetCircuitQoo {
/// Circuit ID to query.
circuit_id: String,
},
/// Retrieve per-circuit capacity utilization
GetCircuitCapacity,
+61 -1
View File
@@ -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;
@@ -87,6 +91,53 @@ pub struct LtsCapabilitiesSummary {
/// Effective mapped-circuit limit. `None` means unlimited.
pub mapped_circuit_limit: Option<u64>,
}
/// QoO history and latest values for a dashboard entity.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Allocative)]
pub struct QooData {
/// Stable key for this QoO row.
pub key: String,
/// Entity kind: "global", "site", or "circuit".
pub entity_kind: String,
/// Display label for this QoO row.
pub label: String,
/// Site name for site rows.
pub site_name: Option<String>,
/// Circuit ID for circuit rows.
pub circuit_id: Option<String>,
/// Fifteen QoO history blocks, oldest to newest.
pub blocks: QoqHeatmapBlocks,
/// Latest download and upload QoO score.
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 {
@@ -588,6 +639,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
@@ -714,6 +771,9 @@ pub enum BusResponse {
/// Queue stats totals (marks/drops)
QueueStatsTotal(QueueStatsTotal),
/// Current QoO data.
Qoo(Option<QooData>),
/// Circuit capacity utilization
CircuitCapacity(Vec<CircuitCapacityRow>),
+3 -3
View File
@@ -26,9 +26,9 @@ 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,
OverrideMutationResult, ProtocolListEntry, QueueStatsTotal, RetransmitSummary,
CircuitHeatmapData, CircuitRollup, CountryListEntry, DeviceCounts, ExecutiveSummaryHeader,
FlowMapPoint, FlowTimelineEntry, InsightLicenseSummary, LtsCapabilitiesSummary, NodeCapacity,
OverrideMutationResult, ProtocolListEntry, QooData, QueueStatsTotal, RetransmitSummary,
SchedulerDetails, SearchResultEntry, SiteHeatmapData, StormguardDebugDirection,
StormguardDebugEntry, TreeGuardRuntimeNodeBranchSnapshot,
TreeGuardRuntimeNodeOperationSnapshot, UrgentIssue, WarningLevel,
+228 -1
View File
@@ -1,7 +1,7 @@
use crate::queue_types::QueueType;
use serde::Serialize;
use thiserror::Error;
use tracing::warn;
use tracing::{debug, warn};
#[derive(Debug, Clone, Serialize)]
pub enum QueueDiff {
@@ -76,6 +76,15 @@ pub struct CakeDiffTin {
pub drops: u64,
pub marks: u64,
pub base_delay_us: u64,
pub sent_packets: u64,
pub peak_delay_us: u64,
pub avg_delay_us: u64,
pub way_indirect_hits: u64,
pub way_misses: u64,
pub way_collisions: u64,
pub sparse_flows: u64,
pub bulk_flows: u64,
pub unresponsive_flows: u64,
}
fn cake_diff(previous: &QueueType, current: &QueueType) -> Result<QueueDiff, QueueDiffError> {
@@ -83,6 +92,15 @@ fn cake_diff(previous: &QueueType, current: &QueueType) -> Result<QueueDiff, Que
if let QueueType::Cake(prev) = previous
&& let QueueType::Cake(new) = current
{
if prev.tins.len() != new.tins.len() {
debug!(
previous_tins = prev.tins.len(),
current_tins = new.tins.len(),
"CAKE tin count changed; resetting queue diff sample"
);
return Ok(QueueDiff::None);
}
let tins = new
.tins
.iter()
@@ -93,6 +111,15 @@ fn cake_diff(previous: &QueueType, current: &QueueType) -> Result<QueueDiff, Que
drops: new.drops.saturating_sub(prev.drops),
marks: new.ecn_marks.saturating_sub(prev.ecn_marks),
base_delay_us: new.base_delay_us,
sent_packets: new.sent_packets.saturating_sub(prev.sent_packets),
peak_delay_us: new.peak_delay_us,
avg_delay_us: new.avg_delay_us,
way_indirect_hits: new.way_indirect_hits.saturating_sub(prev.way_indirect_hits),
way_misses: new.way_misses.saturating_sub(prev.way_misses),
way_collisions: new.way_collisions.saturating_sub(prev.way_collisions),
sparse_flows: new.sparse_flows,
bulk_flows: new.bulk_flows,
unresponsive_flows: new.unresponsive_flows,
})
.collect();
return Ok(QueueDiff::Cake(CakeDiff {
@@ -173,4 +200,204 @@ mod tests {
assert!(matches!(diff, QueueDiff::None));
}
#[test]
fn cake_tin_count_change_resets_diff_sample() {
let previous = deserialize_tc_tree(
r#"[{
"kind":"cake",
"handle":"9001:",
"parent":"1:2",
"options":{},
"bytes":1000,
"packets":10,
"tins":[{
"sent_bytes":100,
"backlog_bytes":3,
"base_delay_us":1
}]
}]"#,
)
.expect("previous cake qdisc should parse")
.remove(0);
let current = deserialize_tc_tree(
r#"[{
"kind":"cake",
"handle":"9001:",
"parent":"1:2",
"options":{},
"bytes":2000,
"packets":20,
"tins":[]
}]"#,
)
.expect("current cake qdisc should parse")
.remove(0);
let diff = make_queue_diff(&previous, &current)
.expect("tin count changes should reset without error");
assert!(matches!(diff, QueueDiff::None));
}
#[test]
fn cake_diff_includes_extended_delta_and_gauge_fields() {
let previous = deserialize_tc_tree(
r#"[{
"kind":"cake",
"handle":"9001:",
"parent":"1:2",
"options":{},
"bytes":1000,
"packets":10,
"qlen":1,
"tins":[{
"sent_bytes":100,
"backlog_bytes":3,
"peak_delay_us":10,
"avg_delay_us":5,
"base_delay_us":1,
"sent_packets":5000000000,
"way_indirect_hits":70000,
"way_misses":80000,
"way_collisions":90000,
"drops":7,
"ecn_mark":9,
"sparse_flows":1,
"bulk_flows":2,
"unresponsive_flows":3
}]
}]"#,
)
.expect("previous cake qdisc should parse")
.remove(0);
let current = deserialize_tc_tree(
r#"[{
"kind":"cake",
"handle":"9001:",
"parent":"1:2",
"options":{},
"bytes":1600,
"packets":18,
"qlen":2,
"tins":[{
"sent_bytes":175,
"backlog_bytes":4,
"peak_delay_us":25,
"avg_delay_us":8,
"base_delay_us":2,
"sent_packets":5000000010,
"way_indirect_hits":70011,
"way_misses":80010,
"way_collisions":90003,
"drops":8,
"ecn_mark":13,
"sparse_flows":4,
"bulk_flows":5,
"unresponsive_flows":6
}]
}]"#,
)
.expect("current cake qdisc should parse")
.remove(0);
let diff = make_queue_diff(&previous, &current).expect("cake diff should be available");
let QueueDiff::Cake(diff) = diff else {
panic!("expected cake diff");
};
let tin = diff.tins.first().expect("cake diff should include one tin");
assert_eq!(tin.sent_bytes, 75);
assert_eq!(tin.drops, 1);
assert_eq!(tin.marks, 4);
assert_eq!(tin.sent_packets, 10);
assert_eq!(tin.way_indirect_hits, 11);
assert_eq!(tin.way_misses, 10);
assert_eq!(tin.way_collisions, 3);
assert_eq!(tin.backlog_bytes, 4);
assert_eq!(tin.peak_delay_us, 25);
assert_eq!(tin.avg_delay_us, 8);
assert_eq!(tin.base_delay_us, 2);
assert_eq!(tin.sparse_flows, 4);
assert_eq!(tin.bulk_flows, 5);
assert_eq!(tin.unresponsive_flows, 6);
}
#[test]
fn cake_diff_saturates_extended_counter_resets() {
let previous = deserialize_tc_tree(
r#"[{
"kind":"cake",
"handle":"9001:",
"parent":"1:2",
"options":{},
"bytes":5000,
"packets":100,
"qlen":1,
"tins":[{
"sent_bytes":5000,
"backlog_bytes":3,
"peak_delay_us":10,
"avg_delay_us":5,
"base_delay_us":1,
"sent_packets":5000000000,
"way_indirect_hits":70000,
"way_misses":80000,
"way_collisions":90000,
"drops":7,
"ecn_mark":9,
"sparse_flows":1,
"bulk_flows":2,
"unresponsive_flows":3
}]
}]"#,
)
.expect("previous cake qdisc should parse")
.remove(0);
let current = deserialize_tc_tree(
r#"[{
"kind":"cake",
"handle":"9001:",
"parent":"1:2",
"options":{},
"bytes":1000,
"packets":10,
"qlen":2,
"tins":[{
"sent_bytes":1000,
"backlog_bytes":4,
"peak_delay_us":25,
"avg_delay_us":8,
"base_delay_us":2,
"sent_packets":100,
"way_indirect_hits":10,
"way_misses":20,
"way_collisions":30,
"drops":1,
"ecn_mark":2,
"sparse_flows":4,
"bulk_flows":5,
"unresponsive_flows":6
}]
}]"#,
)
.expect("current cake qdisc should parse")
.remove(0);
let diff = make_queue_diff(&previous, &current).expect("cake diff should be available");
let QueueDiff::Cake(diff) = diff else {
panic!("expected cake diff");
};
let tin = diff.tins.first().expect("cake diff should include one tin");
assert_eq!(tin.sent_packets, 0);
assert_eq!(tin.way_indirect_hits, 0);
assert_eq!(tin.way_misses, 0);
assert_eq!(tin.way_collisions, 0);
assert_eq!(tin.peak_delay_us, 25);
assert_eq!(tin.avg_delay_us, 8);
assert_eq!(tin.sparse_flows, 4);
assert_eq!(tin.bulk_flows, 5);
assert_eq!(tin.unresponsive_flows, 6);
}
}
+80 -2
View File
@@ -106,8 +106,7 @@ impl Into<CakeDiffTransit> for QueueDiff {
sent_bytes: c.bytes,
backlog_bytes: c.backlog,
drops: c.ddrops,
marks: 0,
base_delay_us: 0,
..CakeDiffTinTransit::default()
});
// Add three zeroed tins for UI expectations
for _ in 0..3 {
@@ -134,6 +133,15 @@ impl Into<CakeDiffTinTransit> for CakeDiffTin {
drops: self.drops,
marks: self.marks,
base_delay_us: self.base_delay_us,
sent_packets: Some(self.sent_packets),
peak_delay_us: Some(self.peak_delay_us),
avg_delay_us: Some(self.avg_delay_us),
way_indirect_hits: Some(self.way_indirect_hits),
way_misses: Some(self.way_misses),
way_collisions: Some(self.way_collisions),
sparse_flows: Some(self.sparse_flows),
bulk_flows: Some(self.bulk_flows),
unresponsive_flows: Some(self.unresponsive_flows),
}
}
}
@@ -169,6 +177,76 @@ impl Into<CakeTransit> for QueueType {
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::queue_diff::{CakeDiff, FqCodelDiff};
#[test]
fn cake_diff_conversion_populates_extended_tin_fields() {
let diff = QueueDiff::Cake(CakeDiff {
bytes: 1000,
packets: 20,
qlen: 3,
tins: vec![CakeDiffTin {
sent_bytes: 500,
backlog_bytes: 30,
drops: 2,
marks: 1,
base_delay_us: 10,
sent_packets: 5_000_000_000,
peak_delay_us: 50,
avg_delay_us: 25,
way_indirect_hits: 70_000,
way_misses: 80_000,
way_collisions: 90_000,
sparse_flows: 4,
bulk_flows: 5,
unresponsive_flows: 6,
}],
});
let transit: CakeDiffTransit = diff.into();
let tin = transit.tins.first().expect("tin should convert");
assert_eq!(tin.sent_packets, Some(5_000_000_000));
assert_eq!(tin.peak_delay_us, Some(50));
assert_eq!(tin.avg_delay_us, Some(25));
assert_eq!(tin.way_indirect_hits, Some(70_000));
assert_eq!(tin.way_misses, Some(80_000));
assert_eq!(tin.way_collisions, Some(90_000));
assert_eq!(tin.sparse_flows, Some(4));
assert_eq!(tin.bulk_flows, Some(5));
assert_eq!(tin.unresponsive_flows, Some(6));
}
#[test]
fn fq_codel_conversion_leaves_cake_only_fields_empty() {
let diff = QueueDiff::FqCodel(FqCodelDiff {
bytes: 1000,
packets: 20,
backlog: 3,
flows: 4,
ddrops: 2,
});
let transit: CakeDiffTransit = diff.into();
assert_eq!(transit.tins.len(), 4);
for tin in transit.tins {
assert_eq!(tin.sent_packets, None);
assert_eq!(tin.peak_delay_us, None);
assert_eq!(tin.avg_delay_us, None);
assert_eq!(tin.way_indirect_hits, None);
assert_eq!(tin.way_misses, None);
assert_eq!(tin.way_collisions, None);
assert_eq!(tin.sparse_flows, None);
assert_eq!(tin.bulk_flows, None);
assert_eq!(tin.unresponsive_flows, None);
}
}
}
/*
#[allow(clippy::from_over_into)]
impl Into<CakeOptionsTransit> for TcCakeOptions {
+56 -2
View File
@@ -1,6 +1,6 @@
//! Rolling 15-minute heatmap storage for QoQ (0..100) scores.
//!
//! This mirrors the structure of `temporal_heatmap`, but stores four QoQ series:
//! This mirrors the structure of `temporal_heatmap`, but stores two QoQ series:
//! - download_total, upload_total
const RAW_SAMPLES: usize = 60;
@@ -10,6 +10,8 @@ const TOTAL_BLOCKS: usize = SUMMARY_BLOCKS + 1;
use allocative::Allocative;
use serde::{Deserialize, Serialize};
use crate::units::DownUpOrder;
/// Heatmap block medians for QoQ (0..100) scores.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, Allocative)]
pub struct QoqHeatmapBlocks {
@@ -31,6 +33,24 @@ pub struct QoqHeatmapBlocks {
pub upload_total: [Option<f32>; TOTAL_BLOCKS],
}
impl QoqHeatmapBlocks {
/// Returns the current in-progress download and upload QoO/QoQ scores.
pub fn current_values(&self) -> DownUpOrder<Option<f32>> {
DownUpOrder {
down: self.download_total.last().copied().flatten(),
up: self.upload_total.last().copied().flatten(),
}
}
/// Returns the latest non-empty download and upload QoO/QoQ scores.
pub fn latest_values(&self) -> DownUpOrder<Option<f32>> {
DownUpOrder {
down: self.download_total.iter().rev().flatten().copied().next(),
up: self.upload_total.iter().rev().flatten().copied().next(),
}
}
}
/// Fixed-size rolling QoQ heatmap storage for 15 minutes of data.
#[derive(Clone, Debug, Allocative)]
pub struct TemporalQoqHeatmap {
@@ -151,7 +171,7 @@ impl Default for TemporalQoqHeatmap {
#[cfg(test)]
mod tests {
use super::{TOTAL_BLOCKS, TemporalQoqHeatmap};
use super::{QoqHeatmapBlocks, TOTAL_BLOCKS, TemporalQoqHeatmap};
#[test]
fn new_is_empty() {
@@ -169,4 +189,38 @@ mod tests {
assert_eq!(blocks.download_total[TOTAL_BLOCKS - 1], Some(10.0));
assert_eq!(blocks.upload_total[TOTAL_BLOCKS - 1], Some(20.0));
}
#[test]
fn latest_values_read_newest_non_empty_blocks() {
let mut download_total = [None; TOTAL_BLOCKS];
let mut upload_total = [None; TOTAL_BLOCKS];
download_total[TOTAL_BLOCKS - 3] = Some(91.0);
upload_total[TOTAL_BLOCKS - 2] = Some(82.0);
let latest = QoqHeatmapBlocks {
download_total,
upload_total,
}
.latest_values();
assert_eq!(latest.down, Some(91.0));
assert_eq!(latest.up, Some(82.0));
}
#[test]
fn current_values_read_current_block_only() {
let mut download_total = [None; TOTAL_BLOCKS];
let mut upload_total = [None; TOTAL_BLOCKS];
download_total[TOTAL_BLOCKS - 2] = Some(91.0);
upload_total[TOTAL_BLOCKS - 2] = Some(82.0);
let current = QoqHeatmapBlocks {
download_total,
upload_total,
}
.current_values();
assert_eq!(current.down, None);
assert_eq!(current.up, None);
}
}
+40
View File
@@ -649,6 +649,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),
@@ -1163,6 +1169,18 @@ fn handle_bus_requests(requests: &[BusRequest], responses: &mut Vec<BusResponse>
let totals = queue_stats_total_data();
BusResponse::QueueStatsTotal(totals)
}
BusRequest::GetQoo => {
let data = node_manager::local_api::executive::qoo_global();
BusResponse::Qoo(Some(data))
}
BusRequest::GetSiteQoo { site_name } => {
let data = node_manager::local_api::executive::qoo_site(site_name);
BusResponse::Qoo(data)
}
BusRequest::GetCircuitQoo { circuit_id } => {
let data = node_manager::local_api::executive::qoo_circuit(circuit_id);
BusResponse::Qoo(data)
}
BusRequest::GetCircuitCapacity => {
let data = circuit_capacity_data();
BusResponse::CircuitCapacity(data)
@@ -1432,4 +1450,26 @@ mod tests {
assert!(matches!(responses.as_slice(), [BusResponse::Ack]));
assert!(!urgent::list().iter().any(|issue| issue.code == code));
}
#[test]
fn bus_qoo_requests_dispatch_to_qoo_responses() {
let requests = [
BusRequest::GetQoo,
BusRequest::GetSiteQoo {
site_name: "missing-site".to_string(),
},
BusRequest::GetCircuitQoo {
circuit_id: "missing-circuit".to_string(),
},
];
let mut responses = Vec::new();
handle_bus_requests(&requests, &mut responses);
assert!(matches!(responses.as_slice(), [
BusResponse::Qoo(Some(global)),
BusResponse::Qoo(None),
BusResponse::Qoo(None),
] if global.key == "global"));
}
}
@@ -1,4 +1,6 @@
use crate::node_manager::local_api::executive_cache::fresh_executive_cache_snapshot;
use crate::node_manager::local_api::executive_cache::{
ExecutiveCacheSnapshot, ExecutiveEntitySnapshot, fresh_executive_cache_snapshot,
};
use lqos_bus::ExecutiveSummaryHeader;
use lqos_utils::{HeatmapBlocks, qoq_heatmap::QoqHeatmapBlocks};
use serde::{Deserialize, Serialize};
@@ -189,6 +191,64 @@ pub struct ExecutiveLeaderboardPage {
pub rows: Vec<ExecutiveLeaderboardRow>,
}
fn qoo_site_from_snapshot(
snapshot: &ExecutiveCacheSnapshot,
site_name: &str,
) -> Option<lqos_bus::QooData> {
snapshot.site_entity(site_name).and_then(entity_qoo_data)
}
fn qoo_circuit_from_snapshot(
snapshot: &ExecutiveCacheSnapshot,
circuit_id: &str,
) -> Option<lqos_bus::QooData> {
snapshot
.circuit_entity(circuit_id)
.and_then(entity_qoo_data)
}
fn qoo_data(
key: String,
entity_kind: &'static str,
label: String,
site_name: Option<String>,
circuit_id: Option<String>,
blocks: QoqHeatmapBlocks,
) -> lqos_bus::QooData {
lqos_bus::QooData {
key,
entity_kind: entity_kind.to_string(),
label,
site_name,
circuit_id,
latest: blocks.latest_values(),
blocks,
}
}
fn entity_qoo_data(entity: &ExecutiveEntitySnapshot) -> Option<lqos_bus::QooData> {
let blocks = entity.qoq_blocks.clone()?;
match entity.entity_kind {
ExecutiveEntityKind::Site => Some(qoo_data(
entity.row_key.clone(),
"site",
entity.label.clone(),
Some(entity.label.clone()),
None,
blocks,
)),
ExecutiveEntityKind::Circuit => Some(qoo_data(
entity.row_key.clone(),
"circuit",
entity.label.clone(),
None,
entity.circuit_id.clone(),
blocks,
)),
ExecutiveEntityKind::Asn => None,
}
}
/// Rows for the standalone executive leaderboard pages.
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
#[serde(tag = "kind")]
@@ -275,15 +335,13 @@ fn latest_value(row: &ExecutiveHeatmapPageRow, metric: &ExecutiveMetric) -> Opti
.as_ref()
.and_then(|blocks| blocks.values.iter().rev().flatten().copied().next()),
ExecutiveMetric::Qoo => row.split_blocks.as_ref().and_then(|blocks| {
let values = [
blocks.download.iter().rev().flatten().copied().next(),
blocks.upload.iter().rev().flatten().copied().next(),
];
let present: Vec<f32> = values.into_iter().flatten().collect();
if present.is_empty() {
None
} else {
Some(present.iter().sum::<f32>() / present.len() as f32)
let download = blocks.download.iter().rev().flatten().copied().next();
let upload = blocks.upload.iter().rev().flatten().copied().next();
match (download, upload) {
(Some(download), Some(upload)) => Some((download + upload) / 2.0),
(Some(download), None) => Some(download),
(None, Some(upload)) => Some(upload),
(None, None) => None,
}
}),
ExecutiveMetric::Rtt => row
@@ -307,9 +365,9 @@ fn sample_count(row: &ExecutiveHeatmapPageRow, metric: &ExecutiveMetric) -> usiz
blocks
.download
.iter()
.chain(blocks.upload.iter())
.filter(|value| value.is_some())
.count()
.max(blocks.upload.iter().filter(|value| value.is_some()).count())
})
.unwrap_or(0),
ExecutiveMetric::Rtt => row
@@ -353,6 +411,32 @@ pub fn executive_dashboard_summary() -> ExecutiveDashboardSummary {
fresh_executive_cache_snapshot().dashboard.clone()
}
/// Returns the current top-level QoO history from the executive dashboard cache.
pub(crate) fn qoo_global() -> lqos_bus::QooData {
let snapshot = fresh_executive_cache_snapshot();
let blocks = snapshot.dashboard.global_qoq.clone();
qoo_data(
"global".to_string(),
"global",
"Global QoO".to_string(),
None,
None,
blocks,
)
}
/// Returns the current QoO history for a site name.
pub(crate) fn qoo_site(site_name: &str) -> Option<lqos_bus::QooData> {
let snapshot = fresh_executive_cache_snapshot();
qoo_site_from_snapshot(&snapshot, site_name)
}
/// Returns the current QoO history for a circuit ID.
pub(crate) fn qoo_circuit(circuit_id: &str) -> Option<lqos_bus::QooData> {
let snapshot = fresh_executive_cache_snapshot();
qoo_circuit_from_snapshot(&snapshot, circuit_id)
}
/// Returns one filtered, sorted executive heatmap detail page.
pub fn executive_heatmap_page(query: ExecutiveHeatmapPageQuery) -> ExecutiveHeatmapPage {
let snapshot = fresh_executive_cache_snapshot();
@@ -469,12 +553,39 @@ pub fn executive_leaderboard_page(
#[cfg(test)]
mod tests {
use crate::node_manager::local_api::executive_cache::{
ExecutiveCacheSnapshot, ExecutiveEntitySnapshot, build_entity_indexes,
};
use lqos_utils::{qoq_heatmap::QoqHeatmapBlocks, temporal_heatmap::TemporalHeatmap};
use super::{
ExecutiveEntityKind, ExecutiveHeatmapPageQuery, ExecutiveHeatmapSort,
ExecutiveLeaderboardKind, ExecutiveLeaderboardPageQuery, ExecutiveMetric,
executive_heatmap_page, executive_leaderboard_page,
ExecutiveLeaderboardKind, ExecutiveLeaderboardPageQuery, ExecutiveMetric, entity_qoo_data,
executive_heatmap_page, executive_leaderboard_page, qoo_circuit_from_snapshot,
qoo_site_from_snapshot,
};
fn qoo_blocks(download_latest: Option<f32>, upload_latest: Option<f32>) -> QoqHeatmapBlocks {
let mut download_total = [None; 15];
let mut upload_total = [None; 15];
download_total[14] = download_latest;
upload_total[14] = upload_latest;
QoqHeatmapBlocks {
download_total,
upload_total,
}
}
fn snapshot_with_entities(entities: Vec<ExecutiveEntitySnapshot>) -> ExecutiveCacheSnapshot {
let (site_entity_index, circuit_entity_index) = build_entity_indexes(&entities);
ExecutiveCacheSnapshot {
entities,
site_entity_index,
circuit_entity_index,
..ExecutiveCacheSnapshot::default()
}
}
#[test]
fn executive_heatmap_page_preserves_client_request_id() {
let page = executive_heatmap_page(ExecutiveHeatmapPageQuery {
@@ -509,4 +620,209 @@ mod tests {
Some("leaderboard-req-1")
);
}
#[test]
fn latest_qoo_reads_newest_heatmap_blocks() {
let latest = qoo_blocks(Some(98.0), Some(87.5)).latest_values();
assert_eq!(latest.down, Some(98.0));
assert_eq!(latest.up, Some(87.5));
}
#[test]
fn latest_qoo_reads_latest_non_empty_heatmap_blocks() {
let mut download_total = [None; 15];
let mut upload_total = [None; 15];
download_total[12] = Some(96.0);
upload_total[13] = Some(88.0);
let latest = QoqHeatmapBlocks {
download_total,
upload_total,
}
.latest_values();
assert_eq!(latest.down, Some(96.0));
assert_eq!(latest.up, Some(88.0));
}
#[test]
fn entity_qoo_data_preserves_site_keying() {
let entity = ExecutiveEntitySnapshot {
row_key: "site:North".to_string(),
entity_kind: ExecutiveEntityKind::Site,
label: "North".to_string(),
circuit_id: None,
asn: None,
tree: None,
heatmap: TemporalHeatmap::new().blocks(),
qoq_blocks: Some(qoo_blocks(Some(91.0), Some(89.0))),
};
let qoo = entity_qoo_data(&entity).expect("site QoO should be returned");
assert_eq!(qoo.key, "site:North");
assert_eq!(qoo.entity_kind, "site");
assert_eq!(qoo.site_name.as_deref(), Some("North"));
assert_eq!(qoo.circuit_id, None);
assert_eq!(qoo.latest.down, Some(91.0));
assert_eq!(qoo.latest.up, Some(89.0));
}
#[test]
fn entity_qoo_data_preserves_circuit_keying() {
let entity = ExecutiveEntitySnapshot {
row_key: "circuit:Circuit-1".to_string(),
entity_kind: ExecutiveEntityKind::Circuit,
label: "Circuit One".to_string(),
circuit_id: Some("Circuit-1".to_string()),
asn: None,
tree: None,
heatmap: TemporalHeatmap::new().blocks(),
qoq_blocks: Some(qoo_blocks(Some(81.0), Some(79.0))),
};
let qoo = entity_qoo_data(&entity).expect("circuit QoO should be returned");
assert_eq!(qoo.key, "circuit:Circuit-1");
assert_eq!(qoo.entity_kind, "circuit");
assert_eq!(qoo.site_name, None);
assert_eq!(qoo.circuit_id.as_deref(), Some("Circuit-1"));
}
#[test]
fn qoo_site_lookup_trims_and_matches_unicode_case_insensitive_name() {
let snapshot = snapshot_with_entities(vec![ExecutiveEntitySnapshot {
row_key: "site:École".to_string(),
entity_kind: ExecutiveEntityKind::Site,
label: "École".to_string(),
circuit_id: None,
asn: None,
tree: None,
heatmap: TemporalHeatmap::new().blocks(),
qoq_blocks: Some(qoo_blocks(Some(91.0), Some(89.0))),
}]);
let qoo = qoo_site_from_snapshot(&snapshot, " école ").expect("site QoO should match");
assert_eq!(qoo.key, "site:École");
assert_eq!(qoo.site_name.as_deref(), Some("École"));
assert_eq!(qoo.latest.down, Some(91.0));
}
#[test]
fn qoo_circuit_lookup_trims_and_requires_exact_id() {
let snapshot = snapshot_with_entities(vec![ExecutiveEntitySnapshot {
row_key: "circuit:Circuit-1".to_string(),
entity_kind: ExecutiveEntityKind::Circuit,
label: "Circuit One".to_string(),
circuit_id: Some("Circuit-1".to_string()),
asn: None,
tree: None,
heatmap: TemporalHeatmap::new().blocks(),
qoq_blocks: Some(qoo_blocks(Some(81.0), Some(79.0))),
}]);
let qoo =
qoo_circuit_from_snapshot(&snapshot, " Circuit-1 ").expect("circuit QoO should match");
assert_eq!(qoo.key, "circuit:Circuit-1");
assert_eq!(qoo.circuit_id.as_deref(), Some("Circuit-1"));
assert!(qoo_circuit_from_snapshot(&snapshot, "circuit-1").is_none());
}
#[test]
fn qoo_lookup_returns_none_for_missing_or_empty_keys() {
let snapshot = snapshot_with_entities(Vec::new());
assert!(qoo_site_from_snapshot(&snapshot, " ").is_none());
assert!(qoo_site_from_snapshot(&snapshot, "missing").is_none());
assert!(qoo_circuit_from_snapshot(&snapshot, " ").is_none());
assert!(qoo_circuit_from_snapshot(&snapshot, "missing").is_none());
}
#[test]
fn qoo_lookup_preserves_first_match_for_duplicate_keys() {
let snapshot = snapshot_with_entities(vec![
ExecutiveEntitySnapshot {
row_key: "site:first".to_string(),
entity_kind: ExecutiveEntityKind::Site,
label: "North".to_string(),
circuit_id: None,
asn: None,
tree: None,
heatmap: TemporalHeatmap::new().blocks(),
qoq_blocks: Some(qoo_blocks(Some(91.0), Some(89.0))),
},
ExecutiveEntitySnapshot {
row_key: "site:second".to_string(),
entity_kind: ExecutiveEntityKind::Site,
label: "NORTH".to_string(),
circuit_id: None,
asn: None,
tree: None,
heatmap: TemporalHeatmap::new().blocks(),
qoq_blocks: Some(qoo_blocks(Some(81.0), Some(79.0))),
},
ExecutiveEntitySnapshot {
row_key: "circuit:first".to_string(),
entity_kind: ExecutiveEntityKind::Circuit,
label: "Circuit One".to_string(),
circuit_id: Some("Circuit-1".to_string()),
asn: None,
tree: None,
heatmap: TemporalHeatmap::new().blocks(),
qoq_blocks: Some(qoo_blocks(Some(71.0), Some(69.0))),
},
ExecutiveEntitySnapshot {
row_key: "circuit:second".to_string(),
entity_kind: ExecutiveEntityKind::Circuit,
label: "Circuit One Duplicate".to_string(),
circuit_id: Some("Circuit-1".to_string()),
asn: None,
tree: None,
heatmap: TemporalHeatmap::new().blocks(),
qoq_blocks: Some(qoo_blocks(Some(61.0), Some(59.0))),
},
]);
let site = qoo_site_from_snapshot(&snapshot, "north").expect("site QoO should match");
let circuit = qoo_circuit_from_snapshot(&snapshot, "Circuit-1")
.expect("circuit QoO should match");
assert_eq!(site.key, "site:first");
assert_eq!(circuit.key, "circuit:first");
}
#[test]
fn qoo_lookup_trims_source_keys() {
let snapshot = snapshot_with_entities(vec![
ExecutiveEntitySnapshot {
row_key: "site:padded".to_string(),
entity_kind: ExecutiveEntityKind::Site,
label: " North ".to_string(),
circuit_id: None,
asn: None,
tree: None,
heatmap: TemporalHeatmap::new().blocks(),
qoq_blocks: Some(qoo_blocks(Some(91.0), Some(89.0))),
},
ExecutiveEntitySnapshot {
row_key: "circuit:padded".to_string(),
entity_kind: ExecutiveEntityKind::Circuit,
label: "Circuit One".to_string(),
circuit_id: Some(" Circuit-1 ".to_string()),
asn: None,
tree: None,
heatmap: TemporalHeatmap::new().blocks(),
qoq_blocks: Some(qoo_blocks(Some(71.0), Some(69.0))),
},
]);
let site = qoo_site_from_snapshot(&snapshot, "north").expect("site QoO should match");
let circuit = qoo_circuit_from_snapshot(&snapshot, "Circuit-1")
.expect("circuit QoO should match");
assert_eq!(site.key, "site:padded");
assert_eq!(circuit.key, "circuit:padded");
}
}
@@ -51,6 +51,8 @@ pub(crate) struct ExecutiveCacheSnapshot {
pub generated_at_unix_ms: u64,
pub dashboard: ExecutiveDashboardSummary,
pub entities: Vec<ExecutiveEntitySnapshot>,
pub site_entity_index: FxHashMap<String, usize>,
pub circuit_entity_index: FxHashMap<String, usize>,
pub leaderboard_rows: FxHashMap<ExecutiveLeaderboardKind, Vec<ExecutiveLeaderboardRow>>,
}
@@ -72,12 +74,36 @@ impl Default for ExecutiveCacheSnapshot {
top_asns: Vec::new(),
},
entities: Vec::new(),
site_entity_index: FxHashMap::default(),
circuit_entity_index: FxHashMap::default(),
leaderboard_rows: FxHashMap::default(),
}
}
}
impl ExecutiveCacheSnapshot {
/// Returns the cached entity row for a site name, using case-insensitive exact matching.
pub(crate) fn site_entity(&self, site_name: &str) -> Option<&ExecutiveEntitySnapshot> {
let normalized_site_name = site_name.trim().to_lowercase();
if normalized_site_name.is_empty() {
return None;
}
self.site_entity_index
.get(&normalized_site_name)
.and_then(|idx| self.entities.get(*idx))
}
/// Returns the cached entity row for an exact circuit ID.
pub(crate) fn circuit_entity(&self, circuit_id: &str) -> Option<&ExecutiveEntitySnapshot> {
let circuit_id = circuit_id.trim();
if circuit_id.is_empty() {
return None;
}
self.circuit_entity_index
.get(circuit_id)
.and_then(|idx| self.entities.get(*idx))
}
/// Builds transport rows for one requested metric from cached raw executive entities.
pub(crate) fn heatmap_rows_for_metric(
&self,
@@ -116,6 +142,31 @@ impl ExecutiveCacheSnapshot {
}
}
pub(crate) fn build_entity_indexes(
entities: &[ExecutiveEntitySnapshot],
) -> (FxHashMap<String, usize>, FxHashMap<String, usize>) {
let mut site_entity_index = FxHashMap::default();
let mut circuit_entity_index = FxHashMap::default();
for (idx, entity) in entities.iter().enumerate() {
match entity.entity_kind {
ExecutiveEntityKind::Site => {
site_entity_index
.entry(entity.label.trim().to_lowercase())
.or_insert(idx);
}
ExecutiveEntityKind::Circuit => {
if let Some(circuit_id) = entity.circuit_id.as_ref() {
circuit_entity_index
.entry(circuit_id.trim().to_string())
.or_insert(idx);
}
}
ExecutiveEntityKind::Asn => {}
}
}
(site_entity_index, circuit_entity_index)
}
static EXECUTIVE_CACHE_SNAPSHOT: Lazy<ArcSwap<ExecutiveCacheSnapshot>> =
Lazy::new(|| ArcSwap::new(Arc::new(ExecutiveCacheSnapshot::default())));
static EXECUTIVE_CACHE_LAST_REFRESH_SECS: AtomicU64 = AtomicU64::new(0);
@@ -179,18 +230,12 @@ fn non_null_count(values: &[Option<f32>]) -> usize {
}
fn latest_qoo(blocks: Option<&QoqHeatmapBlocks>) -> Option<f32> {
let blocks = blocks?;
let mut present = Vec::new();
if let Some(download) = latest_value(&blocks.download_total) {
present.push(download);
}
if let Some(upload) = latest_value(&blocks.upload_total) {
present.push(upload);
}
if present.is_empty() {
None
} else {
Some(present.iter().sum::<f32>() / present.len() as f32)
let latest = blocks?.latest_values();
match (latest.down, latest.up) {
(Some(download), Some(upload)) => Some((download + upload) / 2.0),
(Some(download), None) => Some(download),
(None, Some(upload)) => Some(upload),
(None, None) => None,
}
}
@@ -818,6 +863,7 @@ pub(crate) fn rebuild_executive_cache_snapshot() -> Arc<ExecutiveCacheSnapshot>
site_heatmap_rows(),
asn_heatmap_rows(),
);
let (site_entity_index, circuit_entity_index) = build_entity_indexes(&entities);
let oversubscribed_sites = build_oversubscribed_sites(&entities);
let top_download = top_metric_rows(&entities, ExecutiveMetric::Download, EXECUTIVE_TOP_LIMIT);
let top_upload = top_metric_rows(&entities, ExecutiveMetric::Upload, EXECUTIVE_TOP_LIMIT);
@@ -844,6 +890,8 @@ pub(crate) fn rebuild_executive_cache_snapshot() -> Arc<ExecutiveCacheSnapshot>
top_asns,
},
entities,
site_entity_index,
circuit_entity_index,
leaderboard_rows,
});
EXECUTIVE_CACHE_SNAPSHOT.store(snapshot.clone());
@@ -863,3 +911,78 @@ pub(crate) fn fresh_executive_cache_snapshot() -> Arc<ExecutiveCacheSnapshot> {
}
rebuild_executive_cache_snapshot()
}
#[cfg(test)]
mod tests {
use super::{ExecutiveEntitySnapshot, latest_qoo, top_metric_rows};
use crate::node_manager::local_api::executive::{ExecutiveEntityKind, ExecutiveMetric};
use lqos_utils::{qoq_heatmap::QoqHeatmapBlocks, temporal_heatmap::TemporalHeatmap};
fn qoo_blocks(
download_values: &[(usize, f32)],
upload_values: &[(usize, f32)],
) -> QoqHeatmapBlocks {
let mut download_total = [None; 15];
let mut upload_total = [None; 15];
for &(idx, value) in download_values {
download_total[idx] = Some(value);
}
for &(idx, value) in upload_values {
upload_total[idx] = Some(value);
}
QoqHeatmapBlocks {
download_total,
upload_total,
}
}
fn qoo_entity(label: &str, blocks: QoqHeatmapBlocks) -> ExecutiveEntitySnapshot {
ExecutiveEntitySnapshot {
row_key: format!("site:{label}"),
entity_kind: ExecutiveEntityKind::Site,
label: label.to_string(),
circuit_id: None,
asn: None,
tree: None,
heatmap: TemporalHeatmap::new().blocks(),
qoq_blocks: Some(blocks),
}
}
#[test]
fn latest_qoo_averages_or_falls_back_to_available_direction() {
let both = qoo_blocks(&[(14, 90.0)], &[(14, 70.0)]);
let download_only = qoo_blocks(&[(14, 88.0)], &[]);
let upload_only = qoo_blocks(&[], &[(14, 66.0)]);
let empty = qoo_blocks(&[], &[]);
assert_eq!(latest_qoo(Some(&both)), Some(80.0));
assert_eq!(latest_qoo(Some(&download_only)), Some(88.0));
assert_eq!(latest_qoo(Some(&upload_only)), Some(66.0));
assert_eq!(latest_qoo(Some(&empty)), None);
assert_eq!(latest_qoo(None), None);
}
#[test]
fn top_qoo_rows_rank_by_latest_qoo_with_sample_penalty() {
let sparse_low_qoo = qoo_entity("Sparse Low", qoo_blocks(&[(14, 10.0)], &[(14, 10.0)]));
let sampled_high_qoo = qoo_entity(
"Sampled High",
qoo_blocks(&[(12, 80.0), (13, 80.0), (14, 80.0)], &[]),
);
let sampled_low_qoo = qoo_entity(
"Sampled Low",
qoo_blocks(&[(12, 20.0), (13, 20.0), (14, 20.0)], &[]),
);
let rows = top_metric_rows(
&[sparse_low_qoo, sampled_high_qoo, sampled_low_qoo],
ExecutiveMetric::Qoo,
3,
);
assert_eq!(rows[0].label, "Sampled Low");
assert_eq!(rows[1].label, "Sampled High");
assert_eq!(rows[2].label, "Sparse Low");
}
}
@@ -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());
}
}
+1 -4
View File
@@ -74,10 +74,7 @@ pub(crate) fn circuit_current_qoo(circuit_hash: i64) -> DownUpOrder<Option<f32>>
return DownUpOrder::default();
};
let blocks = heatmap.blocks();
DownUpOrder {
down: blocks.download_total.last().copied().flatten(),
up: blocks.upload_total.last().copied().flatten(),
}
blocks.current_values()
}
fn finish_expired_flows(