Feat: Extend CAKE queue stats transit

Refs #1094

src/rust/lqos_bus/src/bus/queue_data.rs

@@ -38,6 +38,15 @@ pub struct CakeDiffTinTransit {
     pub drops: u32,
     pub marks: u32,
     pub base_delay_us: u32,
+    pub sent_packets: Option<u64>,
+    pub peak_delay_us: Option<u32>,
+    pub avg_delay_us: Option<u32>,
+    pub way_indirect_hits: Option<u64>,
+    pub way_misses: Option<u64>,
+    pub way_collisions: Option<u64>,
+    pub sparse_flows: Option<u32>,
+    pub bulk_flows: Option<u32>,
+    pub unresponsive_flows: Option<u32>,
 }
 
 #[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Default, Allocative)]
@@ -108,3 +117,73 @@ pub struct CakeTinTransit {
     //pub flow_quantum: u16,
 }
 */
+
+#[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);
+    }
+}

src/rust/lqos_queue_tracker/src/queue_diff.rs

@@ -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: u32,
     pub marks: u32,
     pub base_delay_us: u32,
+    pub sent_packets: u64,
+    pub peak_delay_us: u32,
+    pub avg_delay_us: u32,
+    pub way_indirect_hits: u64,
+    pub way_misses: u64,
+    pub way_collisions: u64,
+    pub sparse_flows: u32,
+    pub bulk_flows: u32,
+    pub unresponsive_flows: u32,
 }
 
 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);
+    }
 }

src/rust/lqos_queue_tracker/src/queue_store.rs

@@ -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 {

src/rust/lqos_queue_tracker/src/queue_types/tc_cake.rs

@@ -71,16 +71,16 @@ pub(crate) struct TcCakeTin {
     pub(crate) peak_delay_us: u32,
     pub(crate) avg_delay_us: u32,
     pub(crate) base_delay_us: u32,
-    pub(crate) sent_packets: u32,
-    pub(crate) way_indirect_hits: u16,
-    pub(crate) way_misses: u16,
-    pub(crate) way_collisions: u16,
+    pub(crate) sent_packets: u64,
+    pub(crate) way_indirect_hits: u64,
+    pub(crate) way_misses: u64,
+    pub(crate) way_collisions: u64,
     pub(crate) drops: u32,
     pub(crate) ecn_marks: u32,
     pub(crate) ack_drops: u32,
-    pub(crate) sparse_flows: u16,
-    pub(crate) bulk_flows: u16,
-    pub(crate) unresponsive_flows: u16,
+    pub(crate) sparse_flows: u32,
+    pub(crate) bulk_flows: u32,
+    pub(crate) unresponsive_flows: u32,
     pub(crate) max_pkt_len: u16,
     pub(crate) flow_quantum: u16,
 }
@@ -195,21 +195,19 @@ impl TcCakeTin {
                         "base_delay_us" => {
                             result.base_delay_us = value.as_u64().unwrap_or(0) as u32
                         }
-                        "sent_packets" => result.sent_packets = value.as_u64().unwrap_or(0) as u32,
+                        "sent_packets" => result.sent_packets = value.as_u64().unwrap_or(0),
                         "way_indirect_hits" => {
-                            result.way_indirect_hits = value.as_u64().unwrap_or(0) as u16
-                        }
-                        "way_misses" => result.way_misses = value.as_u64().unwrap_or(0) as u16,
-                        "way_collisions" => {
-                            result.way_collisions = value.as_u64().unwrap_or(0) as u16
+                            result.way_indirect_hits = value.as_u64().unwrap_or(0)
                         }
+                        "way_misses" => result.way_misses = value.as_u64().unwrap_or(0),
+                        "way_collisions" => result.way_collisions = value.as_u64().unwrap_or(0),
                         "drops" => result.drops = value.as_u64().unwrap_or(0) as u32,
                         "ecn_mark" => result.ecn_marks = value.as_u64().unwrap_or(0) as u32,
                         "ack_drops" => result.ack_drops = value.as_u64().unwrap_or(0) as u32,
-                        "sparse_flows" => result.sparse_flows = value.as_u64().unwrap_or(0) as u16,
-                        "bulk_flows" => result.bulk_flows = value.as_u64().unwrap_or(0) as u16,
+                        "sparse_flows" => result.sparse_flows = value.as_u64().unwrap_or(0) as u32,
+                        "bulk_flows" => result.bulk_flows = value.as_u64().unwrap_or(0) as u32,
                         "unresponsive_flows" => {
-                            result.unresponsive_flows = value.as_u64().unwrap_or(0) as u16
+                            result.unresponsive_flows = value.as_u64().unwrap_or(0) as u32
                         }
                         "max_pkt_len" => result.max_pkt_len = value.as_u64().unwrap_or(0) as u16,
                         "flow_quantum" => result.flow_quantum = value.as_u64().unwrap_or(0) as u16,