Add support for port forwarding

config/src/errors.rs

@@ -108,6 +108,9 @@ pub enum ConfigError {
 
     #[error("Community {0} is mapped from distinct priorities")]
     DuplicateCommunity(String),
+
+    #[error("Failed to apply port-forwarding configuration: {0}")]
+    PortForwarding(String),
 }
 
 /// Result-like type for configurations

dataplane/src/main.rs

@@ -204,6 +204,7 @@ fn main() {
             nattablesw: setup.nattablesw,
             natallocatorw: setup.natallocatorw,
             flowfilterw: setup.flowfiltertablesw,
+            portfw_w: setup.portfw_w,
             vpc_stats_store: setup.vpc_stats_store,
             dp_status_r: dp_status.clone(),
             bmp_options: bmp_client_opts,

dataplane/src/packet_processor/ipforward.rs

@@ -28,9 +28,14 @@ use net::packet::VpcDiscriminant;
 use net::udp::UdpEncap;
 use net::vxlan::{Vxlan, VxlanEncap};
 
-use tracectl::trace_target;
+use tracectl::{custom_target, tdebug, trace_target};
 trace_target!("ip-forward", LevelFilter::WARN, &["pipeline"]);
 
+const VXLAN_D: &str = "vxlan-decap";
+const VXLAN_E: &str = "vxlan-encap";
+custom_target!(VXLAN_D, LevelFilter::OFF, &["vxlan"]);
+custom_target!(VXLAN_E, LevelFilter::OFF, &["vxlan"]);
+
 pub struct IpForwarder {
     name: String,
     fibtr: FibTableReader,
@@ -131,8 +136,8 @@ impl IpForwarder {
         match packet.vxlan_decap() {
             Some(Ok(vxlan)) => {
                 let vni = vxlan.vni();
-                debug!("{nfi}: DECAPSULATED vxlan packet:\n {packet}");
-                debug!("{nfi}: Packet comes with vni {vni}");
+                tdebug!(VXLAN_D, "DECAPSULATED vxlan packet (vni={vni}):\n{packet}");
+                debug!("{nfi}: DECAPSULATED vxlan packet, vni = {vni}");
 
                 // access fib for Vni vni
                 let fibkey = FibKey::from_vni(vni);
@@ -274,8 +279,6 @@ impl IpForwarder {
             }
             Ok(vxlan_headers) => match packet.vxlan_encap(&vxlan_headers) {
                 Ok(()) => {
-                    debug!("{nfi}: ENCAPSULATED packet with VxLAN:\n {packet}");
-
                     let vni = vxlan_headers
                         .headers()
                         .udp_encap
@@ -284,6 +287,7 @@ impl IpForwarder {
                         .vxlan_vni();
 
                     packet.meta_mut().dst_vpcd = vni.map(VpcDiscriminant::VNI);
+                    tdebug!(VXLAN_E, "ENCAPSULATED packet with VxLAN:\n{packet}");
                 }
                 Err(e) => {
                     error!("{nfi}: Failed to ENCAPSULATE packet with VxLAN: {e}");

dataplane/src/packet_processor/mod.rs

@@ -15,6 +15,7 @@ use concurrency::sync::Arc;
 use flow_entry::flow_table::{ExpirationsNF, FlowLookup, FlowTable};
 use flow_filter::{FlowFilter, FlowFilterTableWriter};
 
+use nat::portfw::{PortForwarder, PortFwTableWriter};
 use nat::stateful::NatAllocatorWriter;
 use nat::stateless::NatTablesWriter;
 use nat::{StatefulNat, StatelessNat};
@@ -41,6 +42,7 @@ where
     pub flowfiltertablesw: FlowFilterTableWriter,
     pub stats: StatsCollector,
     pub vpc_stats_store: Arc<VpcStatsStore>,
+    pub portfw_w: PortFwTableWriter,
 }
 
 /// Start a router and provide the associated pipeline
@@ -52,6 +54,7 @@ pub(crate) fn start_router<Buf: PacketBufferMut>(
     let flowfiltertablesw = FlowFilterTableWriter::new();
     let router = Router::new(params)?;
     let vpcmapw = VpcMapWriter::<VpcMapName>::new();
+    let portfw_w = PortFwTableWriter::new();
 
     // Allocate the shared VPC stats store (returns Arc<VpcStatsStore>)
     let vpc_stats_store: Arc<VpcStatsStore> = VpcStatsStore::new();
@@ -69,6 +72,7 @@ pub(crate) fn start_router<Buf: PacketBufferMut>(
     let atabler_factory = router.get_atabler_factory();
     let nattabler_factory = nattablesw.get_reader_factory();
     let natallocator_factory = natallocatorw.get_reader_factory();
+    let portfw_factory = portfw_w.reader().factory();
 
     let pipeline_builder = move || {
         // Build network functions
@@ -87,6 +91,11 @@ pub(crate) fn start_router<Buf: PacketBufferMut>(
         let flow_filter = FlowFilter::new("flow-filter", flowfiltertablesr_factory.handle());
         let flow_lookup = FlowLookup::new("flow-lookup", flow_table.clone());
         let flow_expirations_nf = ExpirationsNF::new(flow_table.clone());
+        let portfw = PortForwarder::new(
+            "port-forwarder",
+            portfw_factory.handle(),
+            flow_table.clone(),
+        );
 
         // Build the pipeline for a router. The composition of the pipeline (in stages) is currently
         // hard-coded. In any pipeline, the Stats and ExpirationsNF stages should go last
@@ -95,6 +104,7 @@ pub(crate) fn start_router<Buf: PacketBufferMut>(
             .add_stage(iprouter1)
             .add_stage(flow_lookup)
             .add_stage(flow_filter)
+            .add_stage(portfw)
             .add_stage(stateless_nat)
             .add_stage(stateful_nat)
             .add_stage(iprouter2)
@@ -113,5 +123,6 @@ pub(crate) fn start_router<Buf: PacketBufferMut>(
         flowfiltertablesw,
         stats,
         vpc_stats_store,
+        portfw_w,
     })
 }

flow-entry/src/flow_table/display.rs

@@ -49,44 +49,34 @@ impl Display for FlowKeyData {
         };
 
         match self.src_vpcd() {
-            Some(vpcd) => write!(f, "{{ VPCs({vpcd}->"),
+            Some(vpcd) => write!(f, "{{ VPCs({vpcd}"),
             None => write!(f, "{{ VPCs(None->"),
         }?;
-        match self.dst_vpcd() {
-            Some(vpcd) => write!(f, "{vpcd})"),
-            None => write!(f, "None)"),
-        }?;
         write!(f, " {protocol} ({source}, {destination}){icmp_data} }}")
     }
 }
 
 impl Display for FlowKey {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         match self {
-            FlowKey::Unidirectional(data) | FlowKey::Bidirectional(data) => write!(f, "{data}"),
+            FlowKey::Unidirectional(data) => write!(f, "{data}"),
         }
     }
 }
 
 impl Display for FlowTable {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        let table = self.table.read().unwrap();
-        Heading(format!("Flow Table ({})", table.len())).fmt(f)?;
-        for entry in table.iter() {
-            if let Some(value) = entry.value().upgrade() {
-                let value = value.locked.read().unwrap();
-                let nat_state = value.nat_state.as_ref();
-                let dst_vpcd = value.dst_vpcd.as_ref();
-                write!(f, "{} -> ", entry.key())?;
-                match nat_state {
-                    Some(state) => write!(f, "{{ {state}, "),
-                    None => write!(f, "{{ None, "),
-                }?;
-                match dst_vpcd {
-                    Some(vpcd) => writeln!(f, "dst_vpcd: {vpcd} }}"),
-                    None => writeln!(f, "dst_vpcd: None }}"),
-                }?;
+        if let Ok(table) = self.table.try_read() {
+            Heading(format!("Flow Table ({} entries)", table.len())).fmt(f)?;
+            for entry in table.iter() {
+                let key = entry.key();
+                match entry.value().upgrade() {
+                    Some(value) => writeln!(f, "key = {key}\ndata = {value}")?,
+                    None => writeln!(f, "key = {key} NONE")?,
+                }
             }
+        } else {
+            write!(f, "Failed to lock flow table")?;
         }
         Ok(())
     }