Correctly gather ECN marks from the tins rather than the top-level an…

…d sum them. Remove zero entry items from the Cake stats. This should trim the size of Cake storage a bit, but isn't the full picture yet.

src/rust/lts_client/src/collector/quick_drops/retriever.rs

@@ -93,8 +93,15 @@ impl AsyncQueueReader {
  if let Some(serde_json::Value::Number(drops)) = map.get("drops") {
  stats.drops = drops.as_u64().unwrap_or(0);
  }
- if let Some(serde_json::Value::Number(marks)) = map.get("ecn_mark") {
- stats.marks = marks.as_u64().unwrap_or(0);
+ if let Some(serde_json::Value::Array(tins)) = map.get("tins") {
+ stats.marks = 0;
+ for tin in tins.iter() {
+ if let Some(tin) = tin.as_object() {
+ if let Some(serde_json::Value::Number(marks)) = tin.get("ecn_mark") {
+ stats.marks += marks.as_u64().unwrap_or(0);
+ }
+ }
+ }
  }
  result.push(stats);
  }

src/rust/lts_client/src/collector/quick_drops/stats_diff.rs

@@ -1,7 +1,7 @@
+use super::CakeStats;
 use lqos_config::load_config;
-use tokio::sync::Mutex;
 use once_cell::sync::Lazy;
-use super::CakeStats;
+use tokio::sync::Mutex;
 
 static CAKE_TRACKER: Lazy<Mutex<CakeTracker>> = Lazy::new(|| Mutex::new(CakeTracker::new()));
 
@@ -35,10 +35,7 @@ impl CakeTracker {
  } else {
  let out_reader = super::AsyncQueueReader::new(outbound);
  let in_reader = super::AsyncQueueReader::new(inbound);
- let (up, down) = tokio::join!(
- out_reader.run(),
- in_reader.run(),
- );
+ let (up, down) = tokio::join!(out_reader.run(), in_reader.run(),);
  if let (Ok(Some(up)), Ok(Some(down))) = (up, down) {
  return self.read_up_down(up, down);
  }
@@ -47,21 +44,39 @@ impl CakeTracker {
  None
  }
 
- fn read_up_down(&mut self, up: Vec<CakeStats>, down: Vec<CakeStats>) -> Option<(Vec<CakeStats>, Vec<CakeStats>)> {
+ fn read_up_down(
+ &mut self,
+ mut up: Vec<CakeStats>,
+ mut down: Vec<CakeStats>,
+ ) -> Option<(Vec<CakeStats>, Vec<CakeStats>)> {
  if self.prev.is_none() {
  self.prev = Some((up, down));
  None
  } else {
  // Delta time
  if let Some((down, up)) = &mut self.current {
  down.iter_mut().for_each(|d| {
- if let Some(prev) = self.prev.as_ref().unwrap().0.iter().find(|p| p.circuit_id == d.circuit_id) {
+ if let Some(prev) = self
+ .prev
+ .as_ref()
+ .unwrap()
+ .0
+ .iter()
+ .find(|p| p.circuit_id == d.circuit_id)
+ {
  d.drops = d.drops.saturating_sub(prev.drops);
  d.marks = d.marks.saturating_sub(prev.marks);
  }
  });
  up.iter_mut().for_each(|d| {
- if let Some(prev) = self.prev.as_ref().unwrap().1.iter().find(|p| p.circuit_id == d.circuit_id) {
+ if let Some(prev) = self
+ .prev
+ .as_ref()
+ .unwrap()
+ .1
+ .iter()
+ .find(|p| p.circuit_id == d.circuit_id)
+ {
  d.drops = d.drops.saturating_sub(prev.drops);
  d.marks = d.marks.saturating_sub(prev.marks);
  }
@@ -71,7 +86,11 @@ impl CakeTracker {
  // Advance the previous
  self.prev = self.current.take();
 
+ // Remove all zeroes
+ up.retain(|u| u.drops > 0 || u.marks > 0);
+ down.retain(|d| d.drops > 0 || d.marks > 0);
+
  Some((up, down))
  }
  }
-}
+}