feat(dataplane): a configurable supervisor poll, and telemetry as of the scrape

Worker counters reached telemetry two hops late: the driver's supervisor
collected them once a second, and the telemetry publisher copied them
into the exporter on a second-long ticker of its own. A scrape could see
counts two seconds old, so anything checking them -- a test after a
burst of traffic, the flowgen runner against a live gateway -- had to
wait that long for them to settle.

- `--supervisor-poll-ms` (10 ms to 10 s, default 1000) sets how often
  the supervisor collects the workers' counters, for either driver. The
  rate it reports is now over the time actually elapsed since the last
  poll rather than the nominal period, and the driver status says what
  the period is.
- The `/metrics` handler publishes the pipeline's, the flow table's and
  the workers' counts before it renders, instead of a ticker doing so
  between scrapes: a scrape sees them as of the scrape, the workers' as
  of the supervisor's last poll.

The watchdog's pat and check periods are unchanged: they are about
liveness, not freshness.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Signed-off-by: Daniel Noland <daniel@githedgehog.com>

Author: daniel-noland

args/src/lib.rs

@@ -241,6 +241,9 @@ pub const DEFAULT_DP_UX_PATH: &str = "/var/run/frr/hh/dataplane.sock";
 /// runtime inspection and control.
 pub const DEFAULT_DP_UX_PATH_CLI: &str = "/var/run/dataplane/cli.sock";
 
+/// How often, in milliseconds, a driver's supervisor collects its workers' counters by default.
+pub const DEFAULT_SUPERVISOR_POLL_MS: u32 = 1000;
+
 /// Default path to the FRR agent socket.
 ///
 /// This socket is used to connect to the FRR agent that controls FRR
@@ -619,6 +622,8 @@ pub struct DpdkDriverConfigSection {
     pub eal_args: Vec<String>,
     /// Packet-processing worker threads to run, each owning one rx/tx queue pair per port
     pub num_workers: NonZero<u16>,
+    /// How often, in milliseconds, the driver's supervisor collects its workers' counters.
+    pub supervisor_poll_ms: u32,
     /// Hugepage placement selected by init, or `None` for EAL defaults.
     pub hugepages: Option<HugepagePlan>,
     /// Hugepage memory to require on each NIC NUMA node, in GiB, overriding what init computes
@@ -647,6 +652,8 @@ pub struct KernelDriverConfigSection {
     pub interfaces: Vec<InterfaceArg>,
     /// Packet-processing worker threads to run
     pub num_workers: NonZero<u16>,
+    /// How often, in milliseconds, the driver's supervisor collects its workers' counters.
+    pub supervisor_poll_ms: u32,
 }
 
 /// Configuration for the dataplane's command-line interface (CLI).
@@ -925,6 +932,16 @@ impl DriverConfigSection {
             DriverConfigSection::Kernel(kernel) => kernel.num_workers,
         }
     }
+
+    /// How often the driver's supervisor collects its workers' counters and checks on them.
+    #[must_use]
+    pub fn supervisor_poll(&self) -> Duration {
+        let ms = match self {
+            DriverConfigSection::Dpdk(dpdk) => dpdk.supervisor_poll_ms,
+            DriverConfigSection::Kernel(kernel) => kernel.supervisor_poll_ms,
+        };
+        Duration::from_millis(u64::from(ms))
+    }
 }
 
 /// Probe a raw descriptor without constructing a borrow of a possibly closed file.
@@ -1515,11 +1532,13 @@ impl TryFrom<CmdArgs> for LaunchConfiguration {
                     hugepages: None,
                     hugepages_gib: value.hugepages_gib,
                     num_workers: value.num_workers,
+                    supervisor_poll_ms: value.supervisor_poll_ms,
                 })
             }
             Some("kernel") => DriverConfigSection::Kernel(KernelDriverConfigSection {
                 interfaces: value.interfaces().collect(),
                 num_workers: value.num_workers,
+                supervisor_poll_ms: value.supervisor_poll_ms,
             }),
             Some(other) => return Err(InvalidCmdArguments::InvalidDriver(other.to_string())),
             None => return Err(InvalidCmdArguments::NoDriverSpecified),
@@ -1629,6 +1648,17 @@ Separate interfaces with commas and options with slashes."
     )]
     hugepages_gib: Option<u32>,
 
+    /// How often the driver's supervisor collects its workers' counters, in milliseconds.
+    #[arg(
+        long,
+        value_name = "MS",
+        default_value_t = DEFAULT_SUPERVISOR_POLL_MS,
+        value_parser = clap::value_parser!(u32).range(10..=10_000),
+        help = "How often, in milliseconds in [10..10000], the driver's supervisor collects its \
+                workers' counters: how stale the worker counters in telemetry and the CLI may be"
+    )]
+    supervisor_poll_ms: u32,
+
     #[arg(
         long,
         value_name = "path to a network namespace",
@@ -1883,6 +1913,12 @@ impl CmdArgs {
         self.num_workers
     }
 
+    /// How often the driver's supervisor collects its workers' counters, in milliseconds.
+    #[must_use]
+    pub fn supervisor_poll_ms(&self) -> u32 {
+        self.supervisor_poll_ms
+    }
+
     /// Get the list of kernel network interfaces to use.
     ///
     /// Returns the interfaces specified via `--interface` arguments.
@@ -2180,6 +2216,22 @@ mod tests {
         assert!(parse("0").is_err());
         assert!(parse("65").is_err());
     }
+
+    #[test]
+    fn a_supervisor_poll_outside_ten_ms_to_ten_s_is_refused_at_parse_time() {
+        use crate::{CmdArgs, DEFAULT_SUPERVISOR_POLL_MS, Parser};
+        let parse = |n: &str| CmdArgs::try_parse_from(["dataplane", "--supervisor-poll-ms", n]);
+        assert_eq!(
+            CmdArgs::try_parse_from(["dataplane"])
+                .unwrap()
+                .supervisor_poll_ms(),
+            DEFAULT_SUPERVISOR_POLL_MS
+        );
+        assert_eq!(parse("10").unwrap().supervisor_poll_ms(), 10);
+        assert_eq!(parse("10000").unwrap().supervisor_poll_ms(), 10_000);
+        assert!(parse("9").is_err());
+        assert!(parse("10001").is_err());
+    }
 }
 
 #[cfg(test)]

dataplane/src/drivers/dpdk/mod.rs

@@ -24,6 +24,7 @@ mod worker;
 use crate::packet_processor::PipelineFactory;
 use std::convert::Infallible;
 use std::num::NonZero;
+use std::time::Duration;
 
 use concurrency::thread;
 #[allow(unused_imports)] // used under loom/shuttle backends
@@ -73,6 +74,7 @@ impl DriverDpdk {
         workers_subsystem: &Subsystem,
         ports: &'p [Port<'_>],
         num_workers: NonZero<u16>,
+        supervisor_poll: Duration,
         setup_pipeline: &PipelineFactory<'p, Mbuf<'p>>,
         status_writer: DriverStatusWriter,
         bridge: Option<&mut DatapathEnds>,
@@ -165,6 +167,7 @@ impl DriverDpdk {
             scope,
             "dpdk",
             MBUF_BURST,
+            supervisor_poll,
             workers_subsystem,
             monitors,
             status_writer,

dataplane/src/drivers/kernel/mod.rs

@@ -20,6 +20,8 @@ mod sockstats;
 mod worker;
 
 use crate::packet_processor::PipelineFactory;
+use std::time::Duration;
+
 use concurrency::thread;
 use lifecycle::Subsystem;
 use lldp::Link;
@@ -123,11 +125,13 @@ impl DriverKernel {
     ///
     /// # Errors
     /// Returns [`DriverError`] on interface setup or thread spawn failure.
+    #[allow(clippy::too_many_arguments)] // each is the driver's to give
     pub fn start<'scope>(
         scope: &'scope thread::Scope<'scope, '_>,
         workers_subsystem: &Subsystem,
         args: impl IntoIterator<Item = args::InterfaceArg>,
         num_workers: usize,
+        supervisor_poll: Duration,
         setup_pipeline: &PipelineFactory<'static, TestBuffer>,
         status_writer: DriverStatusWriter,
         mut cp_ends: Option<DatapathEnds>,
@@ -205,6 +209,7 @@ impl DriverKernel {
             scope,
             "kernel",
             Self::MAX_RX_PKT_BATCH,
+            supervisor_poll,
             workers_subsystem,
             worker_monitors,
             status_writer,

dataplane/src/drivers/status.rs

@@ -109,6 +109,8 @@ impl WorkerStatus {
 pub struct DriverStatus {
     /// The most frames one receive call can return, which differs between drivers.
     pub max_rx_batch: usize,
+    /// How often the supervisor collects the workers' counters into this status.
+    pub poll_period: std::time::Duration,
     pub workers: Vec<WorkerStatus>,
 }
 
@@ -239,7 +241,7 @@ impl Display for DriverStatus {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         Heading("Packet driver status").fmt(f)?;
         writeln!(f, " max rx batch: {} pkts", self.max_rx_batch)?;
-        write!(f, " activity poll: {} s", supervisor::TASK_POLL_PERIOD)?;
+        write!(f, " activity poll: {:?}", self.poll_period)?;
         write!(f, "  watchdog pat: {} s", supervisor::TASK_PAT_PERIOD)?;
         writeln!(f, "  watchdog check: {} s", supervisor::TASK_CHECK_PERIOD)?;
 

dataplane/src/drivers/supervisor.rs

@@ -47,9 +47,6 @@ pub(crate) const TASK_GRACE_PERIOD: u16 = 4;
 /// Interval, in seconds, at which the supervisor will check rx task watchdogs
 pub(crate) const TASK_CHECK_PERIOD: u16 = TASK_PAT_PERIOD + TASK_GRACE_PERIOD;
 
-/// Interval, in seconds, at which the supervisor checks rx task activity, ignoring watchdogs
-pub(crate) const TASK_POLL_PERIOD: u16 = 1;
-
 /// The supervisor's view of one rx task: the interface it serves and the watchdog it reports on.
 #[derive(Clone)]
 pub(crate) struct RxTaskMonitor {
@@ -93,23 +90,25 @@ impl<'scope, E> WorkerMonitor<'scope, E> {
 ///
 /// `driver` names the driver in the supervisor's thread name and logs. The supervisor
 /// just joins-and-logs on worker termination; worker fatal reporting is left to the driver.
-/// `publish_counters` runs on each status poll and once after all workers have stopped.
+/// The supervisor polls every `poll_period`, collecting the workers' counters into the
+/// [`DriverStatus`]; `publish_counters` runs on each poll and once after all workers have stopped.
 ///
 /// # Errors
 ///
 /// Returns an error if the supervisor thread cannot be spawned.
+#[allow(clippy::too_many_arguments)] // each is the driver's to give
 pub(crate) fn spawn_supervisor<'scope, E: Display + Send + Sync + 'scope>(
     scope: &'scope thread::Scope<'scope, '_>,
     driver: &'static str,
     max_rx_batch: usize,
+    poll_period: Duration,
     subsystem: &Subsystem,
     mut monitors: Vec<WorkerMonitor<'scope, E>>,
     status_writer: DriverStatusWriter,
     mut publish_counters: impl FnMut() + Send + 'scope,
 ) -> std::io::Result<()> {
     // the two time intervals that matter for liveness detection
     let check_period = Duration::from_secs(u64::from(TASK_CHECK_PERIOD));
-    let poll_period = Duration::from_secs(u64::from(TASK_POLL_PERIOD));
 
     let subsystem = subsystem.clone();
 
@@ -135,6 +134,8 @@ pub(crate) fn spawn_supervisor<'scope, E: Display + Send + Sync + 'scope>(
 
             // the next instant when the rx tasks watchdogs should be checked.
             let mut next_watchdog_check = clock::now().add(check_period);
+            // when the counters were last collected, for rates over the time actually elapsed
+            let mut last_poll = clock::now();
 
             loop {
                 // check if we must run. Otherwise (got cancelled) join all workers
@@ -145,6 +146,8 @@ pub(crate) fn spawn_supervisor<'scope, E: Display + Send + Sync + 'scope>(
                 // check the current time and decide if we should check whether the rx tasks patted the watchdogs.
                 // If so, compute the next time we should check them again in the future.
                 let now = clock::now();
+                let elapsed = now.saturating_duration_since(last_poll);
+                last_poll = now;
                 let check_watchdog = now >= next_watchdog_check;
                 if check_watchdog {
                     while next_watchdog_check <= now {
@@ -174,7 +177,13 @@ pub(crate) fn spawn_supervisor<'scope, E: Display + Send + Sync + 'scope>(
                             any_running = true;
 
                             // check the worker's rx tasks and update the corresponding status
-                            check_worker_rx_tasks(driver, monitor, wk_status, check_watchdog);
+                            check_worker_rx_tasks(
+                                driver,
+                                monitor,
+                                wk_status,
+                                check_watchdog,
+                                elapsed,
+                            );
                         }
                     } else {
                         // A worker monitor without a handle means that the worker was joined already
@@ -192,6 +201,7 @@ pub(crate) fn spawn_supervisor<'scope, E: Display + Send + Sync + 'scope>(
                 // publish the status of the driver
                 status_writer.publish(DriverStatus {
                     max_rx_batch,
+                    poll_period,
                     workers: workers_status.clone(),
                 });
 
@@ -214,6 +224,7 @@ pub(crate) fn spawn_supervisor<'scope, E: Display + Send + Sync + 'scope>(
             // want to log the last state
             let last = DriverStatus {
                 max_rx_batch,
+                poll_period,
                 workers: workers_status.clone(),
             };
             status_writer.publish(last);
@@ -278,13 +289,15 @@ fn must_run<E: Display>(
 }
 
 /// Check the activity of the rx tasks for a worker (watchdog, if `check_watchdog` is true)
-/// from its `WorkerMonitor` and update the corresponding `WorkerStatus`
+/// from its `WorkerMonitor` and update the corresponding `WorkerStatus`, with rates over `elapsed`,
+/// the time since the last check.
 #[allow(clippy::cast_precision_loss)]
 fn check_worker_rx_tasks<E>(
     driver: &str,
     monitor: &WorkerMonitor<'_, E>,
     wk_status: &mut WorkerStatus,
     check_watchdog: bool,
+    elapsed: Duration,
 ) {
     for (idx, task) in monitor.rx_tasks.iter().enumerate() {
         let rx_task_status = &mut wk_status.rx_tasks[idx];
@@ -309,7 +322,11 @@ fn check_worker_rx_tasks<E>(
                 );
             }
             Activity::Active => {
-                rx_task_status.pps = counters.rx as f64 / f64::from(TASK_POLL_PERIOD);
+                rx_task_status.pps = if elapsed.is_zero() {
+                    0.0
+                } else {
+                    counters.rx as f64 / elapsed.as_secs_f64()
+                };
                 debug!(
                     "{driver} worker {} on {}: rx {} ({:.0} pps), tx {}, pipeline drops {}, \
                      tx drops {}, punt drops {}, parse errors {}",
@@ -366,6 +383,7 @@ mod tests {
                 scope,
                 "test",
                 32,
+                Duration::from_secs(1),
                 &subsystem,
                 vec![WorkerMonitor::new(0, worker, vec![task])],
                 writer,

dataplane/src/runtime.rs

@@ -567,6 +567,7 @@ fn run_kernel_driver(
             workers,
             config.driver.interfaces().cloned().collect::<Vec<_>>(),
             config.driver.num_workers().get().into(),
+            config.driver.supervisor_poll(),
             &ingredients.factory(),
             status_writer,
             bridge,
@@ -624,6 +625,7 @@ fn run_dpdk_driver(
             workers,
             &ports,
             config.driver.num_workers(),
+            config.driver.supervisor_poll(),
             &ingredients.factory(),
             status_writer,
             bridge.as_mut(),