dataplane, stats: pipeline, worker and flow table telemetry

dataplane/src/drivers/dpdk/control.rs

@@ -230,7 +230,7 @@ fn receive<'p>(
         // Copied, as every punt is: the far end is a task on the management runtime, which an
         // mbuf cannot reach. The mbuf goes back to the control pool when this iteration drops it.
         if punt.try_send(frame.to_vec()).is_err() {
-            counters.tx_drops += 1;
+            counters.punt_drops += 1;
             trace!(
                 worker = id,
                 "punt queue for {name} is full; dropping a control frame for the kernel"

dataplane/src/drivers/dpdk/inject.rs

@@ -72,7 +72,7 @@ impl<'p> Injector<'p> {
         let mbufs = match self.pool.alloc_bulk(frames.len()) {
             Ok(mbufs) => mbufs,
             Err(e) => {
-                counters.tx_drops += frames.len() as u64;
+                counters.inject_drops += frames.len() as u64;
                 warn!(
                     worker = id,
                     "could not allocate {} mbuf(s) to inject on {}: {e}. The control plane's \
@@ -90,7 +90,7 @@ impl<'p> Injector<'p> {
         let mut batch = MbufArray::new_empty();
         for (mut mbuf, frame) in mbufs.into_iter().zip(frames) {
             let Ok(len) = u16::try_from(frame.len()) else {
-                counters.tx_drops += 1;
+                counters.inject_drops += 1;
                 warn!(
                     worker = id,
                     "the kernel offered a {} byte frame on {}, which is not a frame; dropping it",
@@ -102,7 +102,7 @@ impl<'p> Injector<'p> {
             match mbuf.append(len) {
                 Ok(room) => room.copy_from_slice(&frame),
                 Err(e) => {
-                    counters.tx_drops += 1;
+                    counters.inject_drops += 1;
                     warn!(
                         worker = id,
                         "the kernel offered a {len} byte frame on {}, which does not fit an mbuf \
@@ -125,9 +125,9 @@ impl<'p> Injector<'p> {
         }
         let unsent = tx.transmit(batch);
         let refused = unsent.len() as u64;
-        counters.tx += attempted - refused;
+        counters.injected += attempted - refused;
         if refused > 0 {
-            counters.tx_drops += refused;
+            counters.inject_drops += refused;
             trace!(
                 worker = id,
                 "tx queue on {} refused {refused} of {attempted} injected frame(s)", self.name

dataplane/src/drivers/dpdk/mod.rs

@@ -41,7 +41,7 @@ use tracing::{info, warn};
 use traffic_control::ControlPlane;
 
 use super::DriverError;
-use super::status::DriverStatusWriter;
+use super::status::{DriverStatusWriter, RxQueue};
 use super::supervisor::{RxTaskMonitor, WorkerMonitor, spawn_supervisor};
 
 pub(crate) use crate::drivers::cpbridge::{CpBridge, DatapathEnds, PortIdentity};
@@ -115,8 +115,10 @@ impl DriverDpdk {
         let placement = Placement::of_this_process();
         let mut monitors = Vec::with_capacity(dealt.workers.len() + 1);
         for (id, queues) in dealt.workers.into_iter().enumerate() {
-            let rx_tasks: Vec<RxTaskMonitor> =
-                queues.iter().map(|q| RxTaskMonitor::new(&q.name)).collect();
+            let rx_tasks: Vec<RxTaskMonitor> = queues
+                .iter()
+                .map(|q| RxTaskMonitor::new(&q.name, RxQueue::Data))
+                .collect();
             let worker_ports = queues
                 .into_iter()
                 .zip(&rx_tasks)
@@ -199,7 +201,7 @@ where
 {
     let rx_tasks: Vec<RxTaskMonitor> = control
         .iter()
-        .map(|q| RxTaskMonitor::new(&format!("{} (control)", q.name)))
+        .map(|q| RxTaskMonitor::new(&q.name, RxQueue::Control))
         .collect();
     let ports = control
         .into_iter()

dataplane/src/drivers/kernel/worker.rs

@@ -38,7 +38,7 @@ use crate::drivers::kernel::fanout::{PacketFanoutType, set_packet_fanout};
 use crate::drivers::kernel::kif::Kif;
 use crate::drivers::kernel::multicast::receive_all_multicast;
 use crate::drivers::kernel::sockstats;
-use crate::drivers::status::WorkerId;
+use crate::drivers::status::{RxQueue, WorkerId};
 use crate::drivers::supervisor::{RxTaskMonitor, TASK_PAT_PERIOD, WorkerMonitor};
 use crate::drivers::watchdog::{RxCounters, Watchdog};
 use crate::packet_processor::{EXPIRY_BUDGET, PipelineFactory, WorkerPipeline};
@@ -374,7 +374,7 @@ impl Worker {
         // collect them. However build_interface_table() needs to be called from the worker's tokio runtime
         let ifmonitors: Vec<_> = interfaces
             .iter()
-            .map(|kif| RxTaskMonitor::new(&kif.name))
+            .map(|kif| RxTaskMonitor::new(&kif.name, RxQueue::Data))
             .collect();
 
         let worker_ifmonitors = ifmonitors.clone();
@@ -489,7 +489,7 @@ fn build_interface_table(
         // find the watchdog for this interface
         let watchdog = ifmonitors
             .iter()
-            .find(|ifm| ifm.ifname.as_ref() == kif.name.as_str())
+            .find(|ifm| ifm.port.as_ref() == kif.name.as_str())
             .map(|ifm| ifm.watchdog.clone())
             .ok_or(io::Error::other("Failed to find interface watchdog"))?;
 

dataplane/src/drivers/status.rs

@@ -14,64 +14,63 @@ use concurrency::sync::Arc;
 use std::fmt::Display;
 
 use crate::drivers::supervisor;
-use crate::drivers::watchdog::{Activity, RxCounters};
+use crate::drivers::watchdog::{Activity, Totals};
 
 // The unique Id of a worker
 pub(crate) type WorkerId = usize;
 
-/// Liveness/activity of one rx task (a worker's reader for one interface),
-/// as of the supervisor's last check.
+/// Which of a port's receive queues an rx task serves.
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
+pub enum RxQueue {
+    /// A queue the data path's traffic arrives on.
+    Data,
+    /// The queue the control plane's traffic is steered to (DPDK driver).
+    Control,
+}
+
+impl RxQueue {
+    /// Its name, as telemetry labels it.
+    #[must_use]
+    pub fn name(self) -> &'static str {
+        match self {
+            RxQueue::Data => "data",
+            RxQueue::Control => "control",
+        }
+    }
+}
+
+/// Liveness/activity of one rx task (a worker's reader for one of a port's queues),
+/// as of the supervisor's last check, and its totals as they stand.
 #[derive(Clone)]
 pub struct RxTaskStatus {
-    pub ifname: Arc<str>,
+    /// The port, by its configured interface name.
+    pub port: Arc<str>,
+    pub queue: RxQueue,
     pub activity: Activity,
     pub misses: u64,
-    pub total_rx: u64,
-    pub total_tx: u64,
-    pub total_ppline_drops: u64,
-    pub total_tx_drops: u64,
-    pub total_punt_drops: u64,
-    pub total_parse_errors: u64,
-    pub total_truncated: u64,
-    pub total_csum_unknown: u64,
-    pub total_zero_len: u64,
-    pub total_kernel_drops: u64,
     pub pps: f64,
+    /// What the task counted since it started, read when wanted rather than when polled.
+    pub totals: Totals,
 }
 impl RxTaskStatus {
     #[must_use]
-    pub fn new(ifname: Arc<str>) -> Self {
+    pub fn new(port: Arc<str>, queue: RxQueue, totals: Totals) -> Self {
         Self {
-            ifname,
+            port,
+            queue,
             activity: Activity::Idle,
             misses: 0,
-            total_rx: 0,
-            total_tx: 0,
-            total_ppline_drops: 0,
-            total_tx_drops: 0,
-            total_punt_drops: 0,
-            total_parse_errors: 0,
-            total_truncated: 0,
-            total_csum_unknown: 0,
-            total_zero_len: 0,
-            total_kernel_drops: 0,
             pps: 0.,
+            totals,
         }
     }
 
-    /// Add the counters read from a rx task watchdog to the totals. Reading a watchdog
-    /// clears it, so this must be called for every read, whatever the activity reported.
-    pub fn accumulate(&mut self, counters: &RxCounters) {
-        self.total_rx += counters.rx;
-        self.total_tx += counters.tx;
-        self.total_ppline_drops += counters.ppline_drops;
-        self.total_tx_drops += counters.tx_drops;
-        self.total_punt_drops += counters.punt_drops;
-        self.total_parse_errors += counters.parse_errors;
-        self.total_truncated += counters.truncated;
-        self.total_csum_unknown += counters.csum_unknown;
-        self.total_zero_len += counters.zero_len;
-        self.total_kernel_drops += counters.kernel_drops;
+    /// The task as the CLI names it: its port, and its queue unless that is a data queue.
+    fn name(&self) -> String {
+        match self.queue {
+            RxQueue::Data => self.port.to_string(),
+            RxQueue::Control => format!("{} (control)", self.port),
+        }
     }
 }
 
@@ -169,13 +168,13 @@ impl Display for WorkerState {
 
 macro_rules! RX_TASK_TBL_FMT {
     () => {
-        "   {:<16}  {:<6}  {:>14}  {:>20}  {:>20}  {:>9}"
+        "   {:<16}  {:<6}  {:>14}  {:>20}  {:>20}  {:>12}  {:>9}"
     };
 }
 
 macro_rules! RX_DROP_TBL_FMT {
     () => {
-        "   {:<16}  {:>12}  {:>10}  {:>10}  {:>10}  {:>10}  {:>12}  {:>10}  {:>12}"
+        "   {:<16}  {:>12}  {:>10}  {:>10}  {:>10}  {:>10}  {:>12}  {:>10}  {:>12}  {:>12}"
     };
 }
 
@@ -185,19 +184,26 @@ fn fmt_rx_task_heading(f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         "{}",
         format_args!(
             RX_TASK_TBL_FMT!(),
-            "iface", "status", "pps", "pkt-rx", "pkt-tx", "wd-misses"
+            "iface", "status", "pps", "pkt-rx", "pkt-tx", "injected", "wd-misses"
         )
     )
 }
 
 fn fmt_rx_task(f: &mut std::fmt::Formatter<'_>, rx: &RxTaskStatus) -> std::fmt::Result {
     let pps = format!("{:.1}", rx.pps);
+    let totals = rx.totals.load();
     writeln!(
         f,
         "{}",
         format_args!(
             RX_TASK_TBL_FMT!(),
-            rx.ifname, rx.activity, pps, rx.total_rx, rx.total_tx, rx.misses
+            rx.name(),
+            rx.activity,
+            pps,
+            totals.rx,
+            totals.tx,
+            totals.injected,
+            rx.misses
         )
     )
 }
@@ -216,26 +222,29 @@ fn fmt_rx_drop_heading(f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
             "truncated",
             "csum-unknown",
             "zero-len",
-            "kernel-drops"
+            "kernel-drops",
+            "inject-drops"
         )
     )
 }
 
 fn fmt_rx_drop(f: &mut std::fmt::Formatter<'_>, rx: &RxTaskStatus) -> std::fmt::Result {
+    let totals = rx.totals.load();
     writeln!(
         f,
         "{}",
         format_args!(
             RX_DROP_TBL_FMT!(),
-            rx.ifname,
-            rx.total_ppline_drops,
-            rx.total_tx_drops,
-            rx.total_punt_drops,
-            rx.total_parse_errors,
-            rx.total_truncated,
-            rx.total_csum_unknown,
-            rx.total_zero_len,
-            rx.total_kernel_drops
+            rx.name(),
+            totals.ppline_drops,
+            totals.tx_drops,
+            totals.punt_drops,
+            totals.parse_errors,
+            totals.truncated,
+            totals.csum_unknown,
+            totals.zero_len,
+            totals.kernel_drops,
+            totals.inject_drops
         )
     )
 }
@@ -274,40 +283,3 @@ impl Display for DriverStatus {
         Ok(())
     }
 }
-
-#[cfg(test)]
-mod test {
-    use super::{Arc, RxCounters, RxTaskStatus};
-
-    /// Every counter read from a watchdog must make it to the totals.
-    #[test]
-    fn accumulate_keeps_every_counter() {
-        let mut status = RxTaskStatus::new(Arc::from("eth0"));
-        let counters = RxCounters {
-            rx: 1,
-            tx: 2,
-            ppline_drops: 3,
-            tx_drops: 4,
-            punt_drops: 9,
-            parse_errors: 5,
-            truncated: 6,
-            csum_unknown: 10,
-            zero_len: 7,
-            kernel_drops: 8,
-        };
-
-        status.accumulate(&counters);
-        status.accumulate(&counters);
-
-        assert_eq!(status.total_rx, 2);
-        assert_eq!(status.total_tx, 4);
-        assert_eq!(status.total_ppline_drops, 6);
-        assert_eq!(status.total_tx_drops, 8);
-        assert_eq!(status.total_punt_drops, 18);
-        assert_eq!(status.total_parse_errors, 10);
-        assert_eq!(status.total_truncated, 12);
-        assert_eq!(status.total_csum_unknown, 20);
-        assert_eq!(status.total_zero_len, 14);
-        assert_eq!(status.total_kernel_drops, 16);
-    }
-}