feat(dataplane): enforce and display rate limits

Enforce limits after ACLs and before NAT. Add show rate-limits and drop
rate-limited packets without punting.

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

Author: daniel-noland

Cargo.lock

@@ -348,6 +348,12 @@ fn cmd_show_link() -> Node {
         .action(CliAction::ShowLink)
 }
 
+fn cmd_show_rate_limits() -> Node {
+    Node::new("rate-limits")
+        .desc("Show rate limits and per-color packet and byte counts")
+        .action(CliAction::ShowRateLimits)
+}
+
 fn cmd_show_capture() -> Node {
     Node::new("capture")
         .desc("Show the running or most recent packet capture")
@@ -388,6 +394,7 @@ fn cmd_show() -> Node {
     root += cmd_show_lldp();
     root += cmd_show_transceivers();
     root += cmd_show_link();
+    root += cmd_show_rate_limits();
     root += cmd_show_tech();
     root
 }

cli/bin/cmdtree_dp.rs

@@ -557,6 +557,9 @@ pub enum CliAction {
 
     // Port links (mstlink)
     ShowLink,
+
+    // Rate limits
+    ShowRateLimits,
 }
 impl CliAction {
     #[must_use]

cli/src/cliproto.rs

@@ -15,6 +15,7 @@ shuttle_dfs = ["concurrency/shuttle_dfs", "shuttle"]
 
 [dependencies]
 acl-filter = { workspace = true }
+rate-limit = { workspace = true }
 afpacket = { workspace = true, features = ["async-tokio"] }
 args = { workspace = true }
 arrayvec = { workspace = true }

dataplane/Cargo.toml

@@ -19,6 +19,8 @@ use concurrency::sync::Arc;
 use acl_filter::{AclFilter, AclFilterContextReaderFactory, AclFilterContextWriter};
 use flow_entry::flow_table::{FlowLookup, FlowTable};
 use flow_filter::{FlowFilter, FlowFilterContextReaderFactory, FlowFilterContextWriter};
+use rate_limit::{RateLimitReader, RateLimitWriter, RateLimiter, RateLimits, VpcDestination};
+use std::num::NonZeroUsize;
 
 use nat::masquerade::{NatAllocatorReaderFactory, NatAllocatorWriter};
 use nat::portfw::{PortForwarder, PortFwTableReaderFactory, PortFwTableWriter};
@@ -50,6 +52,7 @@ pub(crate) struct InternalSetup {
     pub natallocatorw: NatAllocatorWriter,
     pub flow_filter_writer: FlowFilterContextWriter,
     pub aclfiltertablesw: AclFilterContextWriter,
+    pub rate_limits: RateLimitWriter<VpcDestination>,
     pub stats: StatsCollector,
     pub vpc_stats_store: Arc<VpcStatsStore>,
     pub portfw_w: PortFwTableWriter,
@@ -71,6 +74,7 @@ pub(crate) struct PipelineIngredients {
     natallocator_factory: NatAllocatorReaderFactory,
     flow_filter_reader_factory: FlowFilterContextReaderFactory,
     aclfiltertablesr_factory: AclFilterContextReaderFactory,
+    rate_limits: RateLimitReader<VpcDestination>,
     portfw_factory: PortFwTableReaderFactory,
     stats_w: PacketStatsWriter,
     pkt_stats: Arc<PacketStats>,
@@ -115,6 +119,7 @@ impl PipelineIngredients {
             let flow_filter =
                 FlowFilter::new("flow-filter", self.flow_filter_reader_factory.handle());
             let acl_filter = AclFilter::new("acl-filter", self.aclfiltertablesr_factory.handle());
+            let rate_limiter = RateLimiter::new(self.rate_limits.clone());
             let icmp_error_handler = IcmpErrorHandler::new(self.flow_table.clone());
             let flow_lookup = FlowLookup::new("flow-lookup", self.flow_table.clone());
             let portfw = PortForwarder::new(
@@ -141,6 +146,9 @@ impl PipelineIngredients {
                 .add_stage(flow_lookup)
                 .add_stage(flow_filter)
                 .add_stage(acl_filter)
+                // After the ACLs, so that denied packets are not charged; before NAT, so that rate
+                // limits see destinations as the source VPC addresses them.
+                .add_stage(rate_limiter)
                 .add_stage(static_nat)
                 .add_stage(portfw)
                 .add_stage(masquerade)
@@ -161,6 +169,7 @@ pub(crate) fn start_router(
     params: RouterParams,
     driver_status: DriverStatusReader,
     capture_dir: Option<String>,
+    workers: NonZeroUsize,
 ) -> Result<InternalSetup, RouterError> {
     let vpcmapw = VpcMapWriter::<VpcMapName>::new();
     let vpc_stats_store: Arc<VpcStatsStore> = VpcStatsStore::new();
@@ -176,6 +185,7 @@ pub(crate) fn start_router(
     let flow_filter_reader_factory = flow_filter_writer.get_reader_factory();
     let aclfiltertablesw = AclFilterContextWriter::new();
     let aclfiltertablesr_factory = aclfiltertablesw.get_reader_factory();
+    let rate_limits = RateLimitWriter::new(RateLimits::empty(), workers);
     let nattablesw = NatTablesWriter::new();
     let natallocatorw = NatAllocatorWriter::new();
     let nattabler_factory = nattablesw.get_reader_factory();
@@ -209,6 +219,7 @@ pub(crate) fn start_router(
             transceivers_read,
         ))),
         link: Some(Box::new(crate::phy_agent::PhyCli(phy_read))),
+        rate_limits: Some(Box::new(rate_limits.reader())),
     };
 
     // create router
@@ -226,6 +237,7 @@ pub(crate) fn start_router(
         natallocator_factory,
         flow_filter_reader_factory,
         aclfiltertablesr_factory,
+        rate_limits: rate_limits.reader(),
         portfw_factory,
         stats_w,
         pkt_stats,
@@ -242,6 +254,7 @@ pub(crate) fn start_router(
         natallocatorw,
         flow_filter_writer,
         aclfiltertablesw,
+        rate_limits,
         stats,
         vpc_stats_store,
         portfw_w,

dataplane/src/packet_processor/mod.rs

@@ -784,11 +784,13 @@ pub fn main() {
     };
 
     // start router
+    let workers = std::num::NonZeroUsize::from(config.driver.num_workers());
     let mut setup = start_router(
         &shutdown.router,
         router_params,
         driver_status_reader,
         config.cli.capture_dir.clone(),
+        workers,
     )
     .expect("failed to start router");
 
@@ -997,6 +999,7 @@ pub fn main() {
                     natallocatorw: setup.natallocatorw,
                     flow_filter_writer: setup.flow_filter_writer,
                     aclfilterw: setup.aclfiltertablesw,
+                    rate_limits: setup.rate_limits,
                     portfw_w: setup.portfw_w,
                     vpc_stats_store: setup.vpc_stats_store,
                     dp_status_r: dp_status.clone(),

dataplane/src/runtime.rs

@@ -18,6 +18,7 @@ bolero = ["dep:bolero", "interface-manager/bolero", "id/bolero", "net/bolero", "
 [dependencies]
 # internal
 acl-filter = { workspace = true }
+rate-limit = { workspace = true }
 args = { workspace = true }
 clock = { workspace = true }
 config = { workspace = true }
@@ -59,6 +60,7 @@ tracing = { workspace = true, features = ["attributes"] }
 unexpected_cfgs = { level = "warn", check-cfg = ['cfg(fuzzing)'] }
 
 [dev-dependencies]
+policer = { workspace = true }
 # internal
 dpdk = { workspace = true, features = ["test"] } # EAL for tests that build the rte_acl-backed ACL filter and flow-filter context
 fixin = { workspace = true }

mgmt/Cargo.toml

@@ -8,6 +8,7 @@ use acl_filter::AclFilterContextWriter;
 use concurrency::sync::Arc;
 use config::external::overlay::ValidatedOverlay;
 use flow_entry::flow_table::FlowTable;
+use rate_limit::{Backend, RateLimitWriter, VpcDestination};
 use std::collections::{HashMap, HashSet};
 use tokio::sync::RwLock;
 use tokio::sync::mpsc;
@@ -100,6 +101,9 @@ pub struct ConfigProcessorParams {
     // writer for ACL filter tables
     pub aclfilterw: AclFilterContextWriter,
 
+    // writer for rate limits
+    pub rate_limits: RateLimitWriter<VpcDestination>,
+
     // writer for port forwarding table
     pub portfw_w: PortFwTableWriter,
 
@@ -574,6 +578,15 @@ fn apply_acl_filter_config(
     Ok(())
 }
 
+fn apply_rate_limit_config(
+    overlay: &ValidatedOverlay,
+    rate_limits: &RateLimitWriter<VpcDestination>,
+) -> ConfigResult {
+    rate_limits.apply(overlay, Backend::Dpdk)?;
+    debug!("Successfully updated the rate limits");
+    Ok(())
+}
+
 /// Update the Nat tables for static NAT
 fn apply_static_nat_config(
     vpc_table: &ValidatedVpcTable,
@@ -679,6 +692,7 @@ impl ConfigProcessor {
         let natallocatorw = &mut self.proc_params.natallocatorw;
         let flow_filter_writer = &mut self.proc_params.flow_filter_writer;
         let aclfilterw = &mut self.proc_params.aclfilterw;
+        let rate_limits = &self.proc_params.rate_limits;
         let portfw_w = &mut self.proc_params.portfw_w;
         let flow_table = &self.proc_params.flow_table;
 
@@ -725,6 +739,9 @@ impl ConfigProcessor {
         /* apply ACL filter config */
         apply_acl_filter_config(overlay, aclfilterw)?;
 
+        /* apply rate limits */
+        apply_rate_limit_config(overlay, rate_limits)?;
+
         /* apply static NAT config */
         apply_static_nat_config(overlay.vpc_table(), nattablesw)?;
 

mgmt/src/processor/proc.rs

@@ -56,11 +56,13 @@ pub mod test {
 
     use config::external::ExternalConfigBuilder;
     use config::external::overlay::Overlay;
+    use config::external::overlay::ratelimit::{ColorActions, RateLimit, RateLimitTables};
     use config::external::overlay::vpc::{Vpc, VpcTable};
     use config::external::overlay::vpcpeering::{
         VpcExpose, VpcManifest, VpcPeering, VpcPeeringTable,
     };
     use config::external::underlay::Underlay;
+    use policer::{Profile, Rate};
 
     use config::ExternalConfig;
     use config::internal::device::DeviceConfig;
@@ -202,7 +204,25 @@ pub mod test {
         let vpc_table = sample_vpc_table();
         let peering_table = sample_vpc_peering_table();
         /* Overlay config */
-        Overlay::new(vpc_table, peering_table)
+        Overlay::new(vpc_table, peering_table).with_rate_limits(sample_rate_limits())
+    }
+
+    /* A byte limit on what VPC-1 sends VPC-2 */
+    fn sample_rate_limits() -> RateLimitTables {
+        let rate = Rate::new(std::num::NonZeroU64::new(125_000_000).expect("not zero"));
+        let profile = Profile::new(rate, 1_250_000, None, 0).expect("a burst");
+        let mut tables = RateLimitTables::default();
+        tables
+            .bytes
+            .add(RateLimit {
+                name: "vpc-1-to-vpc-2".to_string(),
+                vpc: "VPC-1".to_string(),
+                destination: Prefix::expect_from(("192.168.128.0", 24)),
+                profile,
+                actions: ColorActions::default(),
+            })
+            .expect("one rate limit");
+        tables
     }
 
     /* DEVICE configuration */
@@ -492,6 +512,13 @@ pub mod test {
         /* create AclFilterContext for ACL filtering */
         let aclfilterw = AclFilterContextWriter::new();
 
+        /* create the rate limits */
+        let rate_limits = rate_limit::RateLimitWriter::new(
+            rate_limit::RateLimits::empty(),
+            std::num::NonZeroUsize::MIN,
+        );
+        let enforced = rate_limits.clone();
+
         /* create port forwarding table */
         let portfw_w = PortFwTableWriter::new();
 
@@ -516,6 +543,7 @@ pub mod test {
             natallocatorw,
             flow_filter_writer,
             aclfilterw,
+            rate_limits,
             portfw_w,
             vpc_stats_store,
             dp_status_r,
@@ -543,6 +571,17 @@ pub mod test {
             Some("test-gw"),
             "an applied config tells its watchers the gateway's name"
         );
+        let enforced = enforced.load();
+        assert_eq!(
+            enforced
+                .bytes()
+                .rules()
+                .iter()
+                .map(rate_limit::Rule::name)
+                .collect::<Vec<_>>(),
+            ["vpc-1-to-vpc-2"],
+            "an applied config hands its rate limits to the workers"
+        );
 
         /* stop the router */
         debug!("Stopping the router...");

mgmt/src/tests/mgmt.rs

@@ -598,6 +598,7 @@ fn do_handle_cli_request(
         CliAction::ShowLldpNeighbors => show_provider(request, sources.lldp_neighbors.as_deref()),
         CliAction::ShowTransceivers => show_provider(request, sources.transceivers.as_deref()),
         CliAction::ShowLink => show_provider(request, sources.link.as_deref()),
+        CliAction::ShowRateLimits => show_provider(request, sources.rate_limits.as_deref()),
 
         /* prefetching */
         CliAction::Prefetch => prefetch(request, gwconfig, db),