Initial work on displaying queue data. Very early days.

Signed-off-by: Herbert Wolverson <herberticus@gmail.com>

src/rust/lqos_node_manager/static/circuit_queue.html

@@ -50,29 +50,149 @@
     <div id="container" style="padding: 4px;">
 
         <div class="row">
-            <div class="col-sm-12">
+            <div class="col-sm-2">
                 <div class="card bg-light">
                     <div class="card-body">
-                        <h5 class="card-title"><i class="fa fa-users"></i> Queue Info Placeholder</h5>
                         <div id="raw"></div>
                 </div>
             </div>
         </div>
+        <div class="row" style="margin-top: 4px">
+            <div class="col-sm-6">
+                <div class="card bg-light">
+                    <div class="card-body">
+                        <h5 class="card-title"><i class="fa fa-bar-chart"></i> Tin 1 (Bulk)</h5>
+                        <div id="tinTp_0" style="height: 150px"></div>
+                        <div id="tinMd_0" style="height: 150px"></div>
+                    </div>                        
+                </div>
+            </div>
+            <div class="col-sm-6">
+                <div class="card bg-light">
+                    <div class="card-body">
+                        <h5 class="card-title"><i class="fa fa-bar-chart"></i> Tin 2 (Best Effort)</h5>
+                        <div id="tinTp_1" style="height: 150px"></div>
+                        <div id="tinMd_1" style="height: 150px"></div>
+                    </div>
+                </div>
+            </div>
+        </div>
+        <div class="row" style="margin-top: 4px">
+            <div class="col-sm-6">
+                <div class="card bg-light">
+                    <div class="card-body">
+                        <h5 class="card-title"><i class="fa fa-bar-chart"></i> Tin 3 (Video)</h5>
+                        <div id="tinTp_2" style="height: 150px"></div>
+                        <div id="tinMd_2" style="height: 150px"></div>
+                    </div>                        
+                </div>
+            </div>
+            <div class="col-sm-6">
+                <div class="card bg-light">
+                    <div class="card-body">
+                        <h5 class="card-title"><i class="fa fa-bar-chart"></i> Tin 4 (Voice)</h5>
+                        <div id="tinTp_3" style="height: 150px"></div>
+                        <div id="tinMd_3" style="height: 150px"></div>
+                    </div>
+                </div>
+            </div>
+        </div>
 
     </div>
 
     <footer>Copyright (c) 2022, LibreQoE LLC</footer>
 
     <script>
-        function start() {
+        function pollQueue() {
-            colorReloadButton();
-            updateHostCounts();
             const params = new Proxy(new URLSearchParams(window.location.search), {
                 get: (searchParams, prop) => searchParams.get(prop),
             });
             if (params.id != null) {
-                $("#raw").html("<a class='btn btn-info' href='/api/raw_queue_by_circuit/" + encodeURI(params.id) + "'><i class='fa fa-search'></i> Raw Data</a>");
+                $("#raw").html("<a class='btn btn-sm btn-info' href='/api/raw_queue_by_circuit/" + encodeURI(params.id) + "'><i class='fa fa-search'></i> Raw Data</a>");
+                $.get("/api/raw_queue_by_circuit/" + encodeURI(params.id), (data) => {
+                    for (var i=0; i<4; i++) {
+                        // Build Throughput per Tin
+                        {
+                            let x = [];
+                            let y = [];
+                            let x2 = [];
+                            let y2 = [];
+                            for (var j=0; j<600; j++) {
+                                x.push(j);
+                                x2.push(j);
+
+                                // Download
+                                if (data.history[j][0] != "None") {
+                                    let sb = data.history[j][0].Cake.tins[i].sent_bytes;
+                                    y.push(sb * 8);
+                                } else {
+                                    y.push(0);
+                                }
+
+                                // Upload
+                                if (data.history[j][1] != "None") {
+                                    let sb = data.history[j][1].Cake.tins[i].sent_bytes;
+                                    y2.push(0.0 - (sb * 8));
+                                } else {
+                                    y2.push(0);
+                                }
+                            }
+                            let graph_data = [
+                                {x: x, y:y, name: 'Download', type: 'scatter', fill: 'tozeroy'},
+                                {x: x2, y:y2, name: 'Upload', type: 'scatter', fill: 'tozeroy'},
+                            ];
+                            let graph = document.getElementById("tinTp_" + i);
+                            Plotly.newPlot(graph, graph_data, { margin: { l:0,r:0,b:0,t:0,pad:4 }, yaxis: { automargin: true }, xaxis: {automargin: true} });
+                        } // End scope
+                        {
+                            let x = [];
+                            let x2 = [];
+                            let y = [];
+                            let y2 = [];
+                            let x3 = [];
+                            let y3 = [];
+                            let x4 = [];
+                            let y4 = [];
+                            for (var j=0; j<600; j++) {
+                                x.push(j);
+                                x2.push(j);
+                                x3.push(j);
+                                x4.push(j);
+                                if (data.history[j][0] != "None") {
+                                    y.push(data.history[j][0].Cake.tins[i].drops);
+                                    y2.push(data.history[j][0].Cake.tins[i].marks);
+                                } else {
+                                    y.push(0);
+                                    y2.push(0);
+                                }
+                                if (data.history[j][1] != "None") {
+                                    y3.push(data.history[j][1].Cake.tins[i].drops);
+                                    y4.push(data.history[j][1].Cake.tins[i].marks);
+                                } else {
+                                    y3.push(0);
+                                    y4.push(0);
+                                }
+                            }
+                            let graph_data = [
+                                {x: x, y:y, name: 'Down Drops', type: 'scatter', fill: 'tozeroy'},
+                                {x: x2, y:y2, name: 'Down Marks', type: 'scatter', fill: 'tozeroy'},
+                                {x: x3, y:y3, name: 'Up Drops', type: 'scatter', fill: 'tozeroy'},
+                                {x: x4, y:y4, name: 'Up Marks', type: 'scatter', fill: 'tozeroy'},
+                            ];
+                            let graph = document.getElementById("tinMd_" + i);
+                            Plotly.newPlot(graph, graph_data, { margin: { l:0,r:0,b:0,t:0,pad:4 }, yaxis: { automargin: true }, xaxis: {automargin: true} });
+                        } // End scope
+                    }
+                });
             }
+
+            setTimeout(pollQueue, 1000);
+        }
+
+        function start() {
+            colorReloadButton();
+            updateHostCounts();
+            pollQueue();
         }
 
         $(document).ready(start);

src/rust/lqosd/src/queue_tracker/mod.rs

@@ -1,15 +1,57 @@
 use std::{time::{Duration, Instant}, collections::HashMap};
 use lqos_bus::BusResponse;
 use lqos_config::LibreQoSConfig;
+use serde::Serialize;
 use tokio::{task, time};
 use crate::libreqos_tracker::QUEUE_STRUCTURE;
-use self::queue_reader::QueueType;
+use self::queue_reader::{QueueType, QueueDiff, make_queue_diff};
 mod queue_reader;
 use lazy_static::*;
 use parking_lot::RwLock;
 
+const NUM_QUEUE_HISTORY: usize = 600;
+
+#[derive(Debug, Serialize)]
+pub struct QueueStore {
+    history: Vec<(QueueDiff, QueueDiff)>,
+    history_head: usize,
+    prev_download: Option<QueueType>,
+    prev_upload: Option<QueueType>,
+    current_download: QueueType,
+    current_upload: QueueType,
+}
+
+impl QueueStore {
+    fn new(download: QueueType, upload: QueueType) -> Self {
+        Self {
+            history: vec![(QueueDiff::None, QueueDiff::None); NUM_QUEUE_HISTORY],
+            history_head: 0,
+            prev_upload: None,
+            prev_download: None,
+            current_download: download,
+            current_upload: upload,
+        }
+    }
+
+    fn update(&mut self, download: &QueueType, upload: &QueueType) {
+        self.prev_upload = Some(self.current_upload.clone());
+        self.prev_download = Some(self.current_download.clone());
+        self.current_download = download.clone();
+        self.current_upload = upload.clone();
+        let new_diff_up = make_queue_diff(self.prev_upload.as_ref().unwrap(), &self.current_upload);
+        let new_diff_dn = make_queue_diff(self.prev_download.as_ref().unwrap(), &self.current_download);
+        if new_diff_dn.is_ok() && new_diff_up.is_ok() {
+            self.history[self.history_head] = (new_diff_dn.unwrap(), new_diff_up.unwrap());
+            self.history_head += 1;
+            if self.history_head >= NUM_QUEUE_HISTORY {
+                self.history_head = 0;
+            }
+        }
+    }
+}
+
 lazy_static! {
-    pub(crate) static ref CIRCUIT_TO_QUEUE : RwLock<HashMap<String, (QueueType, QueueType)>> = RwLock::new(HashMap::new());
+    pub(crate) static ref CIRCUIT_TO_QUEUE : RwLock<HashMap<String, QueueStore>> = RwLock::new(HashMap::new());
 }
 
 fn track_queues() {
@@ -25,8 +67,9 @@ fn track_queues() {
     };
 
     // Time to associate queues with circuits
-    let mut mapping: HashMap<String, (QueueType, QueueType)> = HashMap::new();
+    let mut mapping = CIRCUIT_TO_QUEUE.write();
     let structure_lock = QUEUE_STRUCTURE.read();
+    
     // Do a quick check that we have a queue association
     if let Ok(structure) = &*structure_lock {
         for circuit in structure.iter().filter(|c| c.circuit_id.is_some()) {
@@ -57,10 +100,21 @@ fn track_queues() {
                         _ => false,
                     }
                 });
-                mapping.insert(
+                if let Some(download) = download {
-                    circuit.circuit_id.as_ref().unwrap().clone(),
+                    if let Some(upload) = upload {
-                    (download.unwrap().clone(), upload.unwrap().clone())
+                        if let Some(circuit_id) = &circuit.circuit_id {
-                );
+                            if let Some(circuit) = mapping.get_mut(circuit_id) {
+                                circuit.update(download, upload);
+                            } else {
+                                // It's new: insert it                        
+                                mapping.insert(
+                                    circuit_id.clone(),
+                                    QueueStore::new(download.clone(), upload.clone())
+                                );
+                            }
+                        }
+                    }
+                }
             } else {
                 let download = queues[0].iter().find(|q| {
                     match q {
@@ -88,19 +142,29 @@ fn track_queues() {
                         _ => false,
                     }
                 });
-                mapping.insert(
+                if let Some(download) = download {
-                    circuit.circuit_id.as_ref().unwrap().clone(),
+                    if let Some(upload) = upload {
-                    (download.unwrap().clone(), upload.unwrap().clone())
+                        if let Some(circuit_id) = &circuit.circuit_id {
-                );
+                            if let Some(circuit) = mapping.get_mut(circuit_id) {
+                                circuit.update(download, upload);
+                            } else {
+                                // It's new: insert it                        
+                                mapping.insert(
+                                    circuit_id.clone(),
+                                    QueueStore::new(download.clone(), upload.clone())
+                                );
+                            }
+                        }
+                    }
+                }
             }
         }
-        *CIRCUIT_TO_QUEUE.write() = mapping;
     }
 }
 
 pub async fn spawn_queue_monitor() {
     let _ = task::spawn(async {
-        let mut interval = time::interval(Duration::from_secs(10));
+        let mut interval = time::interval(Duration::from_secs(1));
 
         loop {
             let now = Instant::now();
@@ -109,9 +173,9 @@ pub async fn spawn_queue_monitor() {
             })
             .await;
             let elapsed = now.elapsed();
-            //println!("TC Reader tick with mapping consumed {:.4} seconds.", elapsed.as_secs_f32());
+            println!("TC Reader tick with mapping consumed {:.4} seconds.", elapsed.as_secs_f32());
             if elapsed.as_secs_f32() < 10.0 {
-                let duration = Duration::from_secs(10) - elapsed;
+                let duration = Duration::from_secs(1) - elapsed;
                 //println!("Sleeping for {:.2} seconds", duration.as_secs_f32());
                 tokio::time::sleep(duration).await;
             } else {

src/rust/lqosd/src/queue_tracker/queue_reader/mod.rs

@@ -6,6 +6,9 @@ use anyhow::{Result, Error};
 use serde::Serialize;
 use serde_json::Value;
 use std::process::Command;
+mod queue_diff;
+pub use queue_diff::QueueDiff;
+pub(crate) use queue_diff::make_queue_diff;
 
 #[derive(Debug, Clone, Serialize)]
 pub(crate) enum QueueType {

src/rust/lqosd/src/queue_tracker/queue_reader/queue_diff/mod.rs

@@ -0,0 +1,63 @@
+use serde::Serialize;
+use anyhow::Result;
+use super::QueueType;
+
+#[derive(Debug, Clone, Serialize)]
+pub enum QueueDiff {
+    None,
+    Mq,
+    Htb,
+    FqCodel,
+    Cake(CakeDiff),
+    ClsAct,
+}
+
+pub(crate) fn make_queue_diff(previous: &QueueType, current: &QueueType) -> Result<QueueDiff> {
+    match previous {
+        QueueType::Cake(..) => {
+            match current {
+                QueueType::Cake(..) => Ok(cake_diff(previous, current)?),
+                _ => Err(anyhow::Error::msg("Not implemented"))
+            }
+        }
+        _ => Err(anyhow::Error::msg("Not implemented"))
+    }
+}
+
+#[derive(Serialize, Clone, Debug)]
+pub struct CakeDiff {
+    pub bytes: u64,
+    pub packets: u64,
+    pub tins: Vec<CakeDiffTin>,
+}
+
+#[derive(Serialize, Clone, Debug)]
+pub struct CakeDiffTin {
+    pub sent_bytes: u64,
+    pub backlog_bytes: u64,
+    pub drops: u64,
+    pub marks: u64,
+}
+
+fn cake_diff(previous: &QueueType, current: &QueueType) -> Result<QueueDiff> {
+    // TODO: Wrapping Handler
+    if let QueueType::Cake(prev) = previous {
+        if let QueueType::Cake(new) = current {
+            let tins = new.tins.iter().zip(prev.tins.iter()).map(|(new, prev)| {
+                //println!("{} - {} = {}", new.sent_bytes, prev.sent_bytes, new.sent_bytes -prev.sent_bytes);
+                CakeDiffTin {
+                    sent_bytes: new.sent_bytes - prev.sent_bytes,
+                    backlog_bytes: new.backlog_bytes - prev.backlog_bytes,
+                    drops: new.drops - prev.drops,
+                    marks: new.ecn_marks - prev.ecn_marks,
+                }
+            }).collect();
+            return Ok(QueueDiff::Cake(CakeDiff{
+                bytes: new.bytes - prev.bytes,
+                packets: new.packets - prev.packets,
+                tins,
+            }));
+        }
+    }
+    Err(anyhow::Error::msg("Not implemented"))
+}

src/rust/lqosd/src/queue_tracker/queue_reader/tc_cake.rs

@@ -137,12 +137,12 @@ pub(crate) struct TcCake {
     pub(crate) handle: TcHandle,
     pub(crate) parent: TcHandle,
     options: TcCakeOptions,
-    bytes: u64,
+    pub(crate) bytes: u64,
-    packets: u64,
+    pub(crate) packets: u64,
     overlimits: u64,
     requeues: u64,
-    backlog: u64,
+    pub(crate) backlog: u64,
-    qlen: u64,
+    pub(crate) qlen: u64,
     memory_used: u64,
     memory_limit: u64,
     capacity_estimate: u64,
@@ -151,8 +151,8 @@ pub(crate) struct TcCake {
     min_adj_size: u64,
     max_adj_size: u64,
     avg_hdr_offset: u64,
-    tins: Vec<TcCakeTin>,
+    pub(crate) tins: Vec<TcCakeTin>,
-    drops: u64,
+    pub(crate) drops: u64,
  }
 
  #[derive(Default, Clone, Debug, Serialize)]
@@ -172,10 +172,10 @@ pub(crate) struct TcCake {
  }
 
  #[derive(Default, Clone, Debug, Serialize)]
- struct TcCakeTin {
+ pub(crate) struct TcCakeTin {
     threshold_rate: u64,
-    sent_bytes: u64,
+    pub(crate) sent_bytes: u64,
-    backlog_bytes: u64,
+    pub(crate) backlog_bytes: u64,
     target_us: u64,
     interval_us: u64,
     peak_delay_us: u64,
@@ -185,8 +185,8 @@ pub(crate) struct TcCake {
     way_indirect_hits: u64,
     way_misses: u64,
     way_collisions: u64,
-    drops: u64,
+    pub(crate) drops: u64,
-    ecn_marks: u64,
+    pub(crate) ecn_marks: u64,
     ack_drops: u64,
     sparse_flows: u64,
     bulk_flows: u64,