fix(stats): complete rework

This is a complete rework of the entire stats crate.

The prior design was all but fundamentally broken in the thread model used.

Note that while rate calculations are now supported in this crate, that feature is deliberately not yet activated.

Signed-off-by: Daniel Noland <[email protected]>

# Conflicts:
#	Cargo.toml

Author: daniel-noland

Cargo.lock

@@ -18,7 +18,8 @@ members = [
 	"pipeline",
 	"pkt-meta",
 	"rekon",
-	"routing", "stats",
+	"routing",
+	"stats",
 	"test-utils",
 	"vpcmap",
 ]
@@ -35,7 +36,7 @@ config = { path = "./config", package = "dataplane-config" }
 dpdk = { path = "./dpdk", package = "dataplane-dpdk" }
 dpdk-sys = { path = "./dpdk-sys", package = "dataplane-dpdk-sys" }
 dpdk-sysroot-helper = { path = "./dpdk-sysroot-helper", package = "dataplane-dpdk-sysroot-helper" }
-dplane-rpc = { git = "https://github.com/githedgehog/dplane-rpc.git", version = "1.1.2"}
+dplane-rpc = { git = "https://github.com/githedgehog/dplane-rpc.git", version = "1.1.2" }
 errno = { path = "./errno", package = "dataplane-errno" }
 gateway_config = { git = "https://github.com/githedgehog/gateway-proto", tag = "v0.12.0", version = "0.12.0" }
 id = { path = "./id", package = "dataplane-id" }
@@ -58,7 +59,7 @@ ahash = { git = "https://github.com/githedgehog/aHash", branch = "pr/daniel-nola
 arc-swap = { version = "1.7.1", default-features = false, features = [] }
 arrayvec = { version = "0.7.6", default-features = false, features = [] }
 async-trait = { version = "0.1.88", default-features = false, features = [] }
-axum = { version = "0.8.4", default-features = false, features = ["http1", "json"] }
+axum = { version = "0.8.4", default-features = false, features = [] }
 axum-server = { version = "0.7.2", default-features = false, features = [] }
 bindgen = { version = "0.72.0", default-features = false, features = [] }
 bitflags = { version = "2.9.1", default-features = false, features = [] }
@@ -75,9 +76,11 @@ dyn-iter = { version = "1.0.1", default-features = false, features = [] }
 etherparse = { version = "0.19.0", default-features = false, features = [] }
 fixin = { git = "https://github.com/githedgehog/fixin", branch = "main" }
 futures = { version = "0.3.31", default-features = false, features = [] }
+hashbrown = { version = "0.15.5", default-features = false, features = [] }
 hyper = { version = "1.6.0", default-features = false, features = ["http1", "server"] }
-hyper-util = { version = "0.1.16", features = ["tokio"]}
+hyper-util = { version = "0.1.16", features = ["tokio"] }
 ipnet = { version = "2.11.0", default-features = false, features = [] }
+kanal = { version = "0.1.1", default-features = false, features = [] }
 left-right = { version = "0.11.5" }
 libc = { version = "1.0.0-alpha.1", default-features = false, features = [] }
 linux-raw-sys = { version = "0.10.0", default-features = false, features = [] }
@@ -96,13 +99,14 @@ parking_lot = { version = "0.12.4", default-features = false, features = [] }
 prefix-trie = { version = "0.8.1", default-features = false, features = [] }
 pretty_assertions = { version = "1.4.1", default-features = false, features = ["std"] }
 procfs = { version = "0.17.0", default-features = false, features = [] }
-rand = { version = "0.9.1", default-features = false, features = ["thread_rng"] }
+rand = { version = "0.9.2", default-features = false, features = ["thread_rng"] }
 roaring = { version = "0.11.0", default-features = false, features = [] }
 rtnetlink = { git = "https://github.com/githedgehog/rtnetlink.git", branch = "hh/tc-actions", default-features = false, features = [] }
 rustyline = { version = "17.0.0", default-features = false, features = [] }
 serde = { version = "1.0.219", default-features = false, features = [] }
 serde_yml = { version = "0.0.12", default-features = false, features = [] }
 shuttle = { version = "0.8.1", default-features = false, features = [] }
+small-map = { version = "0.1.3", default-features = false, features = [] }
 static_assertions = { version = "1.1.0", default-features = false, features = [] }
 strum = { version = "0.27.2", features = ["derive"] }
 thiserror = { version = "2.0.12", default-features = false, features = [] }

Cargo.toml

@@ -8,7 +8,7 @@ license = "Apache-2.0"
 [dependencies]
 afpacket = { workspace = true }
 arrayvec = { workspace = true }
-axum = { workspace = true, features = ["http1", "json"] }
+axum = { workspace = true, features = ["http1", "tokio"] }
 axum-server = { workspace = true }
 clap = { workspace = true, features = ["std", "derive", "usage"] }
 ctrlc = { workspace = true, features = ["termination"] }
@@ -32,8 +32,8 @@ pkt-meta = { workspace = true }
 routing = { workspace = true }
 serde = { workspace = true, features = ["derive"] }
 serde_yml = { workspace = true }
-tokio = { workspace = true }
 stats = { workspace = true }
+tokio = { workspace = true }
 tracing = { workspace = true }
 tracing-subscriber = { workspace = true, features = ["default"] }
 vpcmap = { workspace = true }

dataplane/Cargo.toml

@@ -78,14 +78,10 @@ pub(crate) struct CmdArgs {
     #[arg(
         long,
         value_name = "Metrics Address and Port",
-        default_value = "127.0.0.1:9090",
-        help = "Bind address and port for Prometheus metrics HTTP endpoint (e.g., 127.0.0.1:9090, 0.0.0.0:8080)"
+        default_value_t = SocketAddr::from(([127, 0, 0, 1], 9090)),
+        help = "Bind address and port for Prometheus metrics HTTP endpoint"
     )]
-    metrics_address: String,
-
-    /// Disable Prometheus metrics server
-    #[arg(long, help = "Disable the Prometheus metrics HTTP endpoint")]
-    disable_metrics: bool,
+    metrics_address: SocketAddr,
 }
 
 impl CmdArgs {
@@ -190,28 +186,7 @@ impl CmdArgs {
     }
 
     /// Get the metrics bind address, returns None if metrics are disabled
-    pub fn metrics_address(&self) -> Option<Result<SocketAddr, String>> {
-        if self.disable_metrics {
-            None
-        } else {
-            Some(self.parse_metrics_address())
-        }
-    }
-
-    /// Parse the metrics address string into a socket address
-    fn parse_metrics_address(&self) -> Result<SocketAddr, String> {
+    pub fn metrics_address(&self) -> SocketAddr {
         self.metrics_address
-            .parse::<SocketAddr>()
-            .map_err(|e| format!("Invalid metrics address '{}': {}", self.metrics_address, e))
-    }
-
-    /// Check if metrics are enabled
-    pub fn metrics_enabled(&self) -> bool {
-        !self.disable_metrics
-    }
-
-    /// Get the raw metrics address string
-    pub fn metrics_address_string(&self) -> &str {
-        &self.metrics_address
     }
 }

dataplane/src/args.rs

@@ -11,22 +11,18 @@ mod packet_processor;
 mod statistics; // Add statistics module
 
 use crate::args::{CmdArgs, Parser};
+use crate::packet_processor::start_router;
+use crate::statistics::MetricsServer;
 use drivers::dpdk::DriverDpdk;
 use drivers::kernel::DriverKernel;
+use mgmt::processor::launch::start_mgmt;
 use net::buffer::PacketBufferMut;
 use net::packet::Packet;
 use pipeline::DynPipeline;
 use pipeline::sample_nfs::PacketDumper;
-#[allow(unused)]
-use tracing::{debug, error, info, warn};
-use tracing_subscriber::EnvFilter;
-
-use crate::packet_processor::start_router;
-use mgmt::processor::launch::start_mgmt;
 use routing::RouterParamsBuilder;
-
-// Import statistics functions
-use crate::statistics::start_metrics_server;
+use tracing::{error, info};
+use tracing_subscriber::EnvFilter;
 
 fn init_logging() {
     tracing_subscriber::fmt()
@@ -78,6 +74,7 @@ fn main() {
 
     /* router parameters */
     let Ok(config) = RouterParamsBuilder::default()
+        .metrics_addr(args.metrics_address())
         .cli_sock_path(args.cli_sock_path())
         .cpi_sock_path(args.cpi_sock_path())
         .frr_agent_path(args.frr_agent_path())
@@ -89,61 +86,20 @@ fn main() {
 
     // start the router and build a pipeline. `start_router` returns `InternalSetup` object
     // that we deconstruct here to feed different components.
-    // TODO(fredi): reduce the number of args needed to start components by letting
-    // `start_router` already provide those grouped in the proper types.
-    let setup = match start_router(config) {
-        Ok(setup) => setup,
-        Err(e) => {
-            error!("Failed to start router: {e}");
-            panic!("Failed to start router: {e}");
-        }
-    };
-
+    let setup = start_router(config).expect("failed to start router");
+    MetricsServer::new(args.metrics_address(), setup.stats);
     /* pipeline builder */
     let builder = move || setup.pipeline;
 
-    /* mgmt: router objects */
-    let router = setup.router;
-    let router_ctl = router.get_ctl_tx();
-
-    /* mgmt: nat table */
-    let nattablew = setup.nattable;
-    let vnitablesw = setup.vnitablesw;
-    let vpcmapw = setup.vpcmapw;
-    let statsr = setup.statsr;
-
     /* start management */
-    if let Err(e) = start_mgmt(grpc_addr, router_ctl, nattablew, vnitablesw, vpcmapw) {
-        error!("Failed to start gRPC server: {e}");
-        panic!("Failed to start gRPC server: {e}");
-    } else {
-        info!("Management gRPC server started successfully");
-    }
-
-    // Start metrics server early in the process
-    let _metrics_handle = if let Some(metrics_addr_result) = args.metrics_address() {
-        match metrics_addr_result {
-            Ok(metrics_addr) => match start_metrics_server(metrics_addr, statsr.clone()) {
-                Ok(handle) => {
-                    info!("Metrics server started on http://{metrics_addr}/metrics");
-                    Some(handle)
-                }
-                Err(e) => {
-                    error!("Failed to start metrics server: {}", e);
-                    warn!("Continuing without metrics...");
-                    None
-                }
-            },
-            Err(e) => {
-                error!("Invalid metrics address configuration: {}", e);
-                warn!("Continuing without metrics...");
-                None
-            }
-        }
-    } else {
-        info!("Metrics server disabled");
-        None
-    };
+    start_mgmt(
+        grpc_addr,
+        setup.router.get_ctl_tx(),
+        setup.nattable,
+        setup.vnitablesw,
+        setup.vpcmapw,
+    )
+    .expect("Failed to start gRPC server");
 
     /* start driver with the provided pipeline builder */
     match args.get_driver_name() {