Pr/daniel noland/vxlan decap

dataplane/src/main.rs

@@ -5,25 +5,35 @@
 
 mod args;
 mod nat;
-mod packet;
 mod pipeline;
+mod vxlan;
 
+use crate::args::{CmdArgs, Parser};
+use crate::pipeline::NetworkFunction;
+use crate::pipeline::StaticChain;
 use dpdk::dev::{Dev, TxOffloadConfig};
 use dpdk::eal::Eal;
 use dpdk::lcore::{LCoreId, WorkerThread};
-use dpdk::mem::{Mbuf, Pool, PoolConfig, PoolParams, RteAllocator};
+use dpdk::mem::{Mbuf, Pool, PoolConfig, PoolParams};
 use dpdk::queue::rx::{RxQueueConfig, RxQueueIndex};
 use dpdk::queue::tx::{TxQueueConfig, TxQueueIndex};
 use dpdk::{dev, eal, socket};
-use tracing::{info, trace, warn};
-
-use crate::args::{CmdArgs, Parser};
-use crate::packet::Packet;
-use crate::pipeline::sample_nfs::Passthrough;
-use crate::pipeline::{DynPipeline, NetworkFunction};
-
-#[global_allocator]
-static GLOBAL_ALLOCATOR: RteAllocator = RteAllocator::new_uninitialized();
+use net::eth::Eth;
+use net::eth::ethtype::EthType;
+use net::eth::mac::{DestinationMac, Mac, SourceMac};
+use net::headers::{Headers, Net, Transport};
+use net::ip::NextHeader;
+use net::ipv4::Ipv4;
+use net::ipv4::addr::UnicastIpv4Addr;
+use net::packet::Packet;
+use net::udp::port::UdpPort;
+use net::udp::{Udp, UdpEncap};
+use net::vxlan::{Vni, Vxlan};
+use net::vxlan::{VxlanDecap, VxlanEncap};
+use std::net::Ipv4Addr;
+use tracing::{debug, error, info, trace, warn};
+// #[global_allocator]
+// static GLOBAL_ALLOCATOR: RteAllocator = RteAllocator::new_uninitialized();
 
 fn init_eal(args: impl IntoIterator<Item = impl AsRef<str>>) -> Eal {
     let rte = eal::init(args);
@@ -38,10 +48,43 @@ fn init_eal(args: impl IntoIterator<Item = impl AsRef<str>>) -> Eal {
 }
 
 // FIXME(mvachhar) construct pipline elsewhere, ideally from config file
-fn setup_pipeline() -> DynPipeline<Mbuf> {
-    let pipeline = DynPipeline::new();
-
-    pipeline.add_stage(Passthrough)
+fn setup_pipeline() -> impl NetworkFunction<Mbuf> {
+    let vxlan_decap = VxlanDecap;
+    let mut encap = Headers::new(Eth::new(
+        SourceMac::new(Mac([0b10, 0, 0, 0, 0, 1])).unwrap(),
+        DestinationMac::new(Mac([0xa0, 0x88, 0xc2, 0x46, 0xa8, 0xdd])).unwrap(),
+        EthType::IPV4,
+    ));
+    let mut ipv4 = Ipv4::default();
+    ipv4.set_source(UnicastIpv4Addr::new(Ipv4Addr::new(192, 168, 32, 53)).unwrap());
+    ipv4.set_destination(Ipv4Addr::new(192, 168, 32, 53));
+    ipv4.set_ttl(64);
+    unsafe {
+        ipv4.set_next_header(NextHeader::UDP);
+    }
+    encap.net = Some(Net::Ipv4(ipv4));
+    let udp = Udp::new(UdpPort::new_checked(10000).unwrap(), Vxlan::PORT);
+    encap.transport = Some(Transport::Udp(udp));
+    encap.udp_encap = Some(UdpEncap::Vxlan(Vxlan::new(Vni::new_checked(1234).unwrap())));
+    let vxlan_encap = VxlanEncap::new(encap).unwrap();
+    let mut encap2 = Headers::new(Eth::new(
+        SourceMac::new(Mac([0b10, 0, 0, 0, 0, 1])).unwrap(),
+        DestinationMac::new(Mac([0xa0, 0x88, 0xc2, 0x46, 0xa8, 0xdd])).unwrap(),
+        EthType::IPV4,
+    ));
+    let mut ipv4 = Ipv4::default();
+    ipv4.set_source(UnicastIpv4Addr::new(Ipv4Addr::new(192, 168, 55, 55)).unwrap());
+    ipv4.set_destination(Ipv4Addr::new(192, 168, 32, 111));
+    ipv4.set_ttl(18);
+    unsafe {
+        ipv4.set_next_header(NextHeader::UDP);
+    }
+    encap2.net = Some(Net::Ipv4(ipv4));
+    let udp = Udp::new(UdpPort::new_checked(12222).unwrap(), Vxlan::PORT);
+    encap2.transport = Some(Transport::Udp(udp));
+    encap2.udp_encap = Some(UdpEncap::Vxlan(Vxlan::new(Vni::new_checked(5432).unwrap())));
+    let vxlan_encap2 = VxlanEncap::new(encap2).unwrap();
+    vxlan_decap.chain(vxlan_encap).chain(vxlan_encap2)
 }
 
 fn init_devices(eal: &Eal) -> Vec<Dev> {
@@ -112,15 +155,26 @@ fn start_rte_workers(devices: &[Dev]) {
             loop {
                 let mbufs = rx_queue.receive();
                 let pkts = mbufs.filter_map(|mbuf| match Packet::new(mbuf) {
-                    Ok(pkt) => Some(pkt),
+                    Ok(pkt) => {
+                        debug!("packet: {pkt:?}");
+                        Some(pkt)
+                    }
                     Err(e) => {
                         trace!("Failed to parse packet: {e:?}");
                         None
                     }
                 });
 
                 let pkts_out = pipeline.process(pkts);
-                tx_queue.transmit(pkts_out.map(Packet::reserialize));
+
+                let buffers = pkts_out.filter_map(|pkt| match pkt.serialize() {
+                    Ok(buf) => Some(buf),
+                    Err(e) => {
+                        error!("{e:?}");
+                        None
+                    }
+                });
+                tx_queue.transmit(buffers);
             }
         });
     });

dataplane/src/pipeline/dyn_nf.rs

@@ -6,8 +6,8 @@ use net::buffer::PacketBufferMut;
 use std::any::Any;
 use std::marker::PhantomData;
 
-use crate::packet::Packet;
 use crate::pipeline::NetworkFunction;
+use net::packet::Packet;
 
 /// Trait for an object that processes a stream of packets.
 ///

dataplane/src/pipeline/pipeline.rs

@@ -1,11 +1,10 @@
 // SPDX-License-Identifier: Apache-2.0
 // Copyright Open Network Fabric Authors
 
+use crate::pipeline::{DynNetworkFunction, NetworkFunction, nf_dyn};
 use dyn_iter::{DynIter, IntoDynIterator};
 use net::buffer::PacketBufferMut;
-
-use crate::packet::Packet;
-use crate::pipeline::{DynNetworkFunction, NetworkFunction, nf_dyn};
+use net::packet::Packet;
 
 /// A dynamic pipeline that can be updated at runtime.
 ///

dataplane/src/pipeline/sample_nfs.rs

@@ -1,14 +1,13 @@
 // SPDX-License-Identifier: Apache-2.0
 // Copyright Open Network Fabric Authors
 
+use crate::pipeline::NetworkFunction;
 use net::buffer::PacketBufferMut;
 use net::eth::mac::{DestinationMac, Mac};
 use net::headers::{TryEthMut, TryHeaders, TryIpv4Mut, TryIpv6Mut};
+use net::packet::Packet;
 use tracing::{debug, trace};
 
-use crate::packet::Packet;
-use crate::pipeline::NetworkFunction;
-
 /// Network function that uses [`debug!`] to print the parsed packet headers.
 pub struct InspectHeaders;
 

dataplane/src/pipeline/static_nf.rs

@@ -2,10 +2,9 @@
 // Copyright Open Network Fabric Authors
 
 use net::buffer::PacketBufferMut;
+use net::packet::Packet;
 use std::marker::PhantomData;
 
-use crate::packet::Packet;
-
 /// Trait for an object that processes a stream of packets.
 pub trait NetworkFunction<Buf: PacketBufferMut> {
     /// The `process` method takes an iterator of [`crate::packet::Packet`] objects,

dataplane/src/pipeline/test_utils.rs

@@ -3,21 +3,20 @@
 
 #![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
 
+use crate::pipeline::sample_nfs::{BroadcastMacs, DecrementTtl, InspectHeaders, Passthrough};
+use crate::pipeline::{DynNetworkFunction, nf_dyn};
 use net::buffer::TestBuffer;
 use net::eth::Eth;
 use net::eth::ethtype::EthType;
 use net::eth::mac::{DestinationMac, Mac, SourceMac};
 use net::headers::{Headers, Net};
 use net::ipv4::Ipv4;
 use net::ipv4::addr::UnicastIpv4Addr;
+use net::packet::{InvalidPacket, Packet};
 use net::parse::DeParse;
 use std::default::Default;
 use std::net::Ipv4Addr;
 
-use crate::packet::{InvalidPacket, Packet};
-use crate::pipeline::sample_nfs::{BroadcastMacs, DecrementTtl, InspectHeaders, Passthrough};
-use crate::pipeline::{DynNetworkFunction, nf_dyn};
-
 /// Generates an infinite sequence of network functions.
 ///
 /// The sequence is a repeating pattern of:

dataplane/src/vxlan/mod.rs

@@ -0,0 +1,41 @@
+// SPDX-License-Identifier: Apache-2.0
+// Copyright Open Network Fabric Authors
+
+use crate::pipeline::NetworkFunction;
+use net::buffer::PacketBufferMut;
+use net::packet::Packet;
+use net::vxlan::{VxlanDecap, VxlanEncap};
+use tracing::{debug, trace};
+
+impl<Buf: PacketBufferMut> NetworkFunction<Buf> for VxlanEncap {
+    fn process<'a, Input: Iterator<Item = Packet<Buf>> + 'a>(
+        &'a mut self,
+        input: Input,
+    ) -> impl Iterator<Item = Packet<Buf>> + 'a {
+        input.filter_map(|packet| packet.vxlan_encap(self).ok())
+    }
+}
+
+// TODO: this will need to actually mark the packet with metadata derived from the vxlan header we
+// removed.
+impl<Buf: PacketBufferMut> NetworkFunction<Buf> for VxlanDecap {
+    fn process<'a, Input: Iterator<Item = Packet<Buf>> + 'a>(
+        &'a mut self,
+        input: Input,
+    ) -> impl Iterator<Item = Packet<Buf>> + 'a {
+        input.map(|mut packet| {
+            match packet.vxlan_decap() {
+                None => {
+                    trace!("skipping packet with no vxlan header: {packet:?}");
+                }
+                Some(Err(invalid)) => {
+                    debug!("invalid packet in vxlan: {invalid:?}");
+                }
+                Some(Ok(vxlan)) => {
+                    trace!("decapsulated vxlan packet: {vxlan:?} inner packet: {packet:?}");
+                }
+            }
+            packet
+        })
+    }
+}

net/src/buffer/mod.rs

@@ -7,13 +7,14 @@
 mod test_buffer;
 
 use core::fmt::Debug;
+use std::error::Error;
 #[allow(unused_imports)] // re-export
 #[cfg(any(test, feature = "test_buffer"))]
 pub use test_buffer::*;
 
 /// Super trait representing the abstract operations which may be performed on a packet buffer.
-pub trait PacketBuffer: AsRef<[u8]> + Headroom + 'static {}
-impl<T> PacketBuffer for T where T: AsRef<[u8]> + Headroom + 'static {}
+pub trait PacketBuffer: AsRef<[u8]> + Headroom + Debug + 'static {}
+impl<T> PacketBuffer for T where T: AsRef<[u8]> + Headroom + Debug + 'static {}
 
 /// Super trait representing the abstract operations which may be performed on mutable a packet buffer.
 pub trait PacketBufferMut: PacketBuffer + AsMut<[u8]> + Prepend + TrimFromStart {}
@@ -34,9 +35,12 @@ pub trait Tailroom {
 /// Trait representing the ability to prepend data to a packet buffer.
 pub trait Prepend {
     /// Error which may occur when attempting to prepend data to the buffer.
-    type Error: Debug;
+    type Error: Debug + Error;
     /// Prepend data to the buffer if possible.
     ///
+    /// If successful, this method returns a slice to the net start of the buffer.
+    /// The contents of the buffer will not be otherwise altered.
+    ///
     /// # Errors
     ///
     /// Returns [`Self::Error`] if an error occurs while performing this operation.