feat(mgmt,dataplane): define MgmtParams

Define struct MgmtParams to pass all of the parameters that the
mgmt thread requires. This makes it much easier to pass new
stuff, without the need to change several function signatures
every time.

Signed-off-by: Fredi Raspall <[email protected]>

Author: Fredi-raspall

dataplane/src/main.rs

@@ -14,8 +14,7 @@ use crate::statistics::MetricsServer;
 use args::{CmdArgs, Parser};
 
 use drivers::kernel::DriverKernel;
-
-use mgmt::processor::launch::start_mgmt;
+use mgmt::{ConfigProcessorParams, MgmtParams, start_mgmt};
 
 use pyroscope::PyroscopeAgent;
 use pyroscope_pprofrs::{PprofConfig, pprof_backend};
@@ -127,19 +126,21 @@ fn main() {
 
     MetricsServer::new(args.metrics_address(), setup.stats);
 
-    /* pipeline builder */
+    // pipeline builder
     let pipeline_factory = setup.pipeline;
 
     /* start management */
-    start_mgmt(
+    start_mgmt(MgmtParams {
         grpc_addr,
-        setup.router.get_ctl_tx(),
-        setup.nattablew,
-        setup.natallocatorw,
-        setup.vpcdtablesw,
-        setup.vpcmapw,
-        setup.vpc_stats_store,
-    )
+        processor_params: ConfigProcessorParams {
+            router_ctl: setup.router.get_ctl_tx(),
+            vpcmapw: setup.vpcmapw,
+            nattablesw: setup.nattablesw,
+            natallocatorw: setup.natallocatorw,
+            vpcdtablesw: setup.vpcdtablesw,
+            vpc_stats_store: setup.vpc_stats_store,
+        },
+    })
     .expect("Failed to start gRPC server");
 
     /* start driver with the provided pipeline builder */

dataplane/src/packet_processor/mod.rs

@@ -36,7 +36,7 @@ where
     pub router: Router,
     pub pipeline: Arc<dyn Send + Sync + Fn() -> DynPipeline<Buf>>,
     pub vpcmapw: VpcMapWriter<VpcMapName>,
-    pub nattablew: NatTablesWriter,
+    pub nattablesw: NatTablesWriter,
     pub natallocatorw: NatAllocatorWriter,
     pub vpcdtablesw: VpcDiscTablesWriter,
     pub stats: StatsCollector,
@@ -47,7 +47,7 @@ where
 pub(crate) fn start_router<Buf: PacketBufferMut>(
     params: RouterParams,
 ) -> Result<InternalSetup<Buf>, RouterError> {
-    let nattablew = NatTablesWriter::new();
+    let nattablesw = NatTablesWriter::new();
     let natallocatorw = NatAllocatorWriter::new();
     let vpcdtablesw = VpcDiscTablesWriter::new();
     let router = Router::new(params)?;
@@ -67,7 +67,7 @@ pub(crate) fn start_router<Buf: PacketBufferMut>(
     let fibtr_factory = router.get_fibtr_factory();
     let vpcdtablesr_factory = vpcdtablesw.get_reader_factory();
     let atabler_factory = router.get_atabler_factory();
-    let nattabler_factory = nattablew.get_reader_factory();
+    let nattabler_factory = nattablesw.get_reader_factory();
     let natallocator_factory = natallocatorw.get_reader_factory();
 
     let pipeline_builder = move || {
@@ -108,7 +108,7 @@ pub(crate) fn start_router<Buf: PacketBufferMut>(
         router,
         pipeline: Arc::new(pipeline_builder),
         vpcmapw,
-        nattablew,
+        nattablesw,
         natallocatorw,
         vpcdtablesw,
         stats,

mgmt/src/lib.rs

@@ -3,17 +3,13 @@
 
 //! Dataplane management module
 
-/* gRPC entry point */
-pub mod grpc;
-
-/* Configuration processor */
-pub mod processor;
-
-/* VPC manager */
+mod grpc;
+mod processor;
+mod tests;
 pub mod vpc_manager;
 
-#[cfg(test)]
-mod tests;
+pub use processor::launch::{MgmtParams, start_mgmt};
+pub use processor::proc::ConfigProcessorParams;
 
 use tracectl::trace_target;
 trace_target!("mgmt", LevelFilter::DEBUG, &["management"]);

mgmt/src/processor/gwconfigdb.rs

@@ -9,7 +9,7 @@ use tracing::{debug, error, info};
 
 /// Configuration database, keeps a set of [`GwConfig`]s keyed by generation id [`GenId`]
 #[derive(Default)]
-pub struct GwConfigDatabase {
+pub(crate) struct GwConfigDatabase {
     configs: BTreeMap<GenId, GwConfig>, /* collection of configs */
     current: Option<GenId>,             /* [`GenId`] of currently applied config */
 }
@@ -52,7 +52,7 @@ impl GwConfigDatabase {
     pub fn get_mut(&mut self, generation: GenId) -> Option<&mut GwConfig> {
         self.configs.get_mut(&generation)
     }
-
+    #[allow(unused)]
     pub fn remove(&mut self, genid: GenId) -> ConfigResult {
         if genid == ExternalConfig::BLANK_GENID {
             debug!("Will not remove config {genid} as it is protected");

mgmt/src/processor/launch.rs

@@ -3,7 +3,7 @@
 
 use crate::processor::proc::ConfigChannelRequest;
 use crate::processor::proc::ConfigProcessor;
-
+use args::GrpcAddress;
 use std::fmt::Display;
 use std::io::Error;
 use std::net::SocketAddr;
@@ -16,17 +16,10 @@ use tokio::net::UnixListener;
 use tokio::sync::mpsc::Sender;
 use tokio_stream::Stream;
 
-use nat::stateful::NatAllocatorWriter;
-use nat::stateless::NatTablesWriter;
-use pkt_meta::dst_vpcd_lookup::VpcDiscTablesWriter;
-use routing::RouterCtlSender;
-
 use crate::grpc::server::create_config_service;
+use crate::processor::proc::ConfigProcessorParams;
 use tonic::transport::Server;
-
-use stats::VpcMapName;
 use tracing::{debug, error, info, warn};
-use vpcmap::map::VpcMapWriter;
 
 /// Start the gRPC server on TCP
 async fn start_grpc_server_tcp(
@@ -161,20 +154,17 @@ impl Display for ServerAddress {
     }
 }
 
+pub struct MgmtParams {
+    pub grpc_addr: GrpcAddress,
+    pub processor_params: ConfigProcessorParams,
+}
+
 /// Start the mgmt service with either type of socket
-pub fn start_mgmt(
-    grpc_addr: args::GrpcAddress,
-    router_ctl: RouterCtlSender,
-    nattablew: NatTablesWriter,
-    natallocatorw: NatAllocatorWriter,
-    vpcdtablesw: VpcDiscTablesWriter,
-    vpcmapw: VpcMapWriter<VpcMapName>,
-    vps_stats_store: std::sync::Arc<stats::VpcStatsStore>,
-) -> Result<std::thread::JoinHandle<()>, Error> {
+pub fn start_mgmt(params: MgmtParams) -> Result<std::thread::JoinHandle<()>, Error> {
     /* build server address from provided grpc address */
-    let server_address = match grpc_addr {
-        args::GrpcAddress::Tcp(addr) => ServerAddress::Tcp(addr),
-        args::GrpcAddress::UnixSocket(path) => ServerAddress::Unix(path.into()),
+    let server_address = match params.grpc_addr {
+        GrpcAddress::Tcp(addr) => ServerAddress::Tcp(addr),
+        GrpcAddress::UnixSocket(path) => ServerAddress::Unix(path.into()),
     };
     debug!("Will start gRPC listening on {server_address}");
 
@@ -192,15 +182,8 @@ pub fn start_mgmt(
 
             /* block thread to run gRPC and configuration processor */
             rt.block_on(async {
-                let (processor, tx) = ConfigProcessor::new(
-                    router_ctl,
-                    vpcmapw,
-                    nattablew,
-                    natallocatorw,
-                    vpcdtablesw,
-                    vps_stats_store,
-                );
-                tokio::task::spawn(async { processor.run().await });
+                let (processor, tx) = ConfigProcessor::new(params.processor_params);
+                tokio::spawn(async { processor.run().await });
 
                 // Start the appropriate server based on address type
                 let result = match server_address {

mgmt/src/processor/mod.rs

@@ -4,8 +4,8 @@
 //! Dataplane configuration processor.
 //! This module implements the core logic to determine and build internal configurations.
 
-pub mod confbuild;
+pub(crate) mod confbuild;
 mod display;
-pub mod gwconfigdb;
-pub mod launch;
-pub mod proc;
+pub(crate) mod gwconfigdb;
+pub(crate) mod launch;
+pub(crate) mod proc;

mgmt/src/processor/proc.rs

@@ -79,20 +79,6 @@ impl ConfigChannelRequest {
     }
 }
 
-/// A configuration processor entity. This is the RPC-independent entity responsible for
-/// accepting/rejecting configurations, storing them in the configuration database and
-/// applying them.
-pub(crate) struct ConfigProcessor {
-    config_db: GwConfigDatabase,
-    rx: mpsc::Receiver<ConfigChannelRequest>,
-    router_ctl: RouterCtlSender,
-    vpc_mgr: VpcManager<RequiredInformationBase>,
-    vpcmapw: VpcMapWriter<VpcMapName>,
-    nattablew: NatTablesWriter,
-    natallocatorw: NatAllocatorWriter,
-    vnitablesw: VpcDiscTablesWriter,
-    vpc_stats_store: Arc<VpcStatsStore>,
-}
 /// Populate FRR status into the dataplane status structure
 pub async fn populate_status_with_frr(
     status: &mut DataplaneStatus,
@@ -103,25 +89,47 @@ pub async fn populate_status_with_frr(
     if let Ok(Some(FrrAppliedConfig { genid, .. })) = router_ctl.get_frr_applied_config().await {
         frr = frr.set_applied_config_gen(genid);
     }
-
     status.set_frr_status(frr);
 }
 
+/// A configuration processor entity. This is the RPC-independent entity responsible for
+/// accepting/rejecting configurations, storing them in the configuration database and
+/// applying them.
+pub(crate) struct ConfigProcessor {
+    config_db: GwConfigDatabase,
+    rx: mpsc::Receiver<ConfigChannelRequest>,
+    vpc_mgr: VpcManager<RequiredInformationBase>,
+    proc_params: ConfigProcessorParams,
+}
+
+pub struct ConfigProcessorParams {
+    // channel to router
+    pub router_ctl: RouterCtlSender,
+
+    // writer for vpc mapping table
+    pub vpcmapw: VpcMapWriter<VpcMapName>,
+
+    // writer for stateless NAT tables
+    pub nattablesw: NatTablesWriter,
+
+    // writer for stateful NAT allocator
+    pub natallocatorw: NatAllocatorWriter,
+
+    // writer for VPC routing table
+    pub vpcdtablesw: VpcDiscTablesWriter,
+
+    // store for vpc stats
+    pub vpc_stats_store: Arc<VpcStatsStore>,
+}
+
 impl ConfigProcessor {
     const CHANNEL_SIZE: usize = 1; // This should not be changed
 
     /////////////////////////////////////////////////////////////////////////////////
     /// Create a [`ConfigProcessor`]
     /////////////////////////////////////////////////////////////////////////////////
     #[must_use]
-    pub(crate) fn new(
-        router_ctl: RouterCtlSender,
-        vpcmapw: VpcMapWriter<VpcMapName>,
-        nattablew: NatTablesWriter,
-        natallocatorw: NatAllocatorWriter,
-        vnitablesw: VpcDiscTablesWriter,
-        vpc_stats_store: Arc<stats::VpcStatsStore>,
-    ) -> (Self, Sender<ConfigChannelRequest>) {
+    pub(crate) fn new(proc_params: ConfigProcessorParams) -> (Self, Sender<ConfigChannelRequest>) {
         debug!("Creating config processor...");
         let (tx, rx) = mpsc::channel(Self::CHANNEL_SIZE);
 
@@ -133,16 +141,11 @@ impl ConfigProcessor {
         let netlink = Arc::new(netlink);
         let vpc_mgr = VpcManager::<RequiredInformationBase>::new(netlink);
 
-        let processor = Self {
+        let processor = ConfigProcessor {
             config_db: GwConfigDatabase::new(),
             rx,
-            router_ctl,
             vpc_mgr,
-            vpcmapw,
-            nattablew,
-            natallocatorw,
-            vnitablesw,
-            vpc_stats_store,
+            proc_params,
         };
         (processor, tx)
     }
@@ -190,15 +193,16 @@ impl ConfigProcessor {
             debug!("The current config is {}", current.genid());
         }
 
+        // FIXME(fredi): pass &mut self.params
         apply_gw_config(
             &self.vpc_mgr,
             &mut config,
             current.as_deref(),
-            &mut self.router_ctl,
-            &mut self.vpcmapw,
-            &mut self.nattablew,
-            &mut self.natallocatorw,
-            &mut self.vnitablesw,
+            &mut self.proc_params.router_ctl,
+            &mut self.proc_params.vpcmapw,
+            &mut self.proc_params.nattablesw,
+            &mut self.proc_params.natallocatorw,
+            &mut self.proc_params.vpcdtablesw,
         )
         .await?;
 
@@ -219,15 +223,16 @@ impl ConfigProcessor {
         let rollback_cfg = current.unwrap_or(ExternalConfig::BLANK_GENID);
         info!("Rolling back to config '{rollback_cfg}'...");
         if let Some(prior) = self.config_db.get_mut(rollback_cfg) {
+            // FIXME(fredi): pass &mut self.params
             let _ = apply_gw_config(
                 &self.vpc_mgr,
                 prior,
                 None,
-                &mut self.router_ctl,
-                &mut self.vpcmapw,
-                &mut self.nattablew,
-                &mut self.natallocatorw,
-                &mut self.vnitablesw,
+                &mut self.proc_params.router_ctl,
+                &mut self.proc_params.vpcmapw,
+                &mut self.proc_params.nattablesw,
+                &mut self.proc_params.natallocatorw,
+                &mut self.proc_params.vpcdtablesw,
             )
             .await;
         }
@@ -262,9 +267,11 @@ impl ConfigProcessor {
     async fn handle_get_dataplane_status(&mut self) -> ConfigResponse {
         let mut status = DataplaneStatus::new();
 
-        let names = self.vpc_stats_store.snapshot_names().await;
-        let pair_snap = self.vpc_stats_store.snapshot_pairs().await;
-        let vpc_snap = self.vpc_stats_store.snapshot_vpcs().await;
+        let stats_store = &self.proc_params.vpc_stats_store;
+
+        let names = stats_store.snapshot_names().await;
+        let pair_snap = stats_store.snapshot_pairs().await;
+        let vpc_snap = stats_store.snapshot_vpcs().await;
 
         // Helper to check if a flow stats has any traffic
         #[inline]
@@ -375,7 +382,7 @@ impl ConfigProcessor {
         }
 
         // FRR minimal info
-        populate_status_with_frr(&mut status, &mut self.router_ctl).await;
+        populate_status_with_frr(&mut status, &mut self.proc_params.router_ctl).await;
 
         ConfigResponse::GetDataplaneStatus(Box::new(status))
     }
@@ -587,6 +594,7 @@ fn apply_device_config(device: &DeviceConfig) -> ConfigResult {
 
 #[allow(clippy::too_many_arguments)]
 /// Main function to apply a config
+// FIXME(fredi): receive &mut self.params
 async fn apply_gw_config(
     vpc_mgr: &VpcManager<RequiredInformationBase>,
     config: &mut GwConfig,