veth support

interface-manager/src/interface/mod.rs

@@ -6,6 +6,7 @@
 mod association;
 mod bridge;
 mod properties;
+mod veth;
 mod vrf;
 mod vtep;
 
@@ -16,6 +17,8 @@ pub use bridge::*;
 #[allow(unused_imports)] // re-export
 pub use properties::*;
 #[allow(unused_imports)] // re-export
+pub use veth::*;
+#[allow(unused_imports)] // re-export
 pub use vrf::*;
 #[allow(unused_imports)] // re-export
 pub use vtep::*;
@@ -35,10 +38,10 @@ use net::route::RouteTableId;
 use net::vxlan::InvalidVni;
 use rekon::{AsRequirement, Create, Op, Reconcile, Remove, Update};
 use rtnetlink::packet_route::link::{
-    InfoBridge, InfoData, InfoVrf, InfoVxlan, LinkAttribute, LinkFlags, LinkInfo, LinkMessage,
-    State,
+    InfoBridge, InfoData, InfoVeth, InfoVrf, InfoVxlan, LinkAttribute, LinkFlags, LinkInfo,
+    LinkMessage, State,
 };
-use rtnetlink::{LinkBridge, LinkUnspec, LinkVrf, LinkVxlan};
+use rtnetlink::{LinkBridge, LinkUnspec, LinkVeth, LinkVrf, LinkVxlan};
 use serde::{Deserialize, Serialize};
 use tracing::{error, warn};
 
@@ -147,6 +150,9 @@ impl Create for Manager<Interface> {
             InterfacePropertiesSpec::Vrf(properties) => {
                 LinkVrf::new(requirement.name.as_ref(), properties.route_table_id.into()).build()
             }
+            InterfacePropertiesSpec::Veth(properties) => {
+                LinkVeth::new(requirement.name.as_ref(), properties.peer.as_ref()).build()
+            }
         };
         if let Some(mac) = requirement.mac {
             message
@@ -332,6 +338,21 @@ impl Update for Manager<InterfaceProperties> {
                     .execute()
                     .await
             }
+            (InterfacePropertiesSpec::Veth(req), InterfaceProperties::Veth(_)) => {
+                // FIXME: how to set header in msg and what attributes
+                let mut msg = LinkMessage::default();
+                msg.attributes
+                    .push(LinkAttribute::IfName(req.peer.to_string()));
+                self.handle
+                    .link()
+                    .set_port(
+                        LinkUnspec::new_with_index(observation.index.to_u32())
+                            .set_info_data(InfoData::Veth(InfoVeth::Peer(msg)))
+                            .build(),
+                    )
+                    .execute()
+                    .await
+            }
             (_, _) => {
                 self.handle
                     .link()

interface-manager/src/interface/properties.rs

@@ -2,7 +2,7 @@
 // Copyright Open Network Fabric Authors
 
 use crate::interface::bridge::BridgePropertiesSpec;
-use crate::interface::{VrfPropertiesSpec, VtepPropertiesSpec};
+use crate::interface::{VethPropertiesSpec, VrfPropertiesSpec, VtepPropertiesSpec};
 use net::interface::InterfaceProperties;
 use rekon::AsRequirement;
 use serde::{Deserialize, Serialize};
@@ -17,6 +17,8 @@ pub enum InterfacePropertiesSpec {
     Vtep(VtepPropertiesSpec),
     /// The planned properties of a vrf
     Vrf(VrfPropertiesSpec),
+    /// The planned properties of a veth
+    Veth(VethPropertiesSpec),
 }
 
 impl AsRequirement<InterfacePropertiesSpec> for InterfaceProperties {
@@ -34,6 +36,9 @@ impl AsRequirement<InterfacePropertiesSpec> for InterfaceProperties {
                 InterfacePropertiesSpec::Vtep(props.as_requirement()?)
             }
             InterfaceProperties::Vrf(props) => InterfacePropertiesSpec::Vrf(props.as_requirement()),
+            InterfaceProperties::Veth(props) => {
+                InterfacePropertiesSpec::Veth(props.as_requirement())
+            }
             InterfaceProperties::Other => return None,
         })
     }

interface-manager/src/interface/veth.rs

@@ -0,0 +1,68 @@
+// SPDX-License-Identifier: Apache-2.0
+// Copyright Open Network Fabric Authors
+
+use derive_builder::Builder;
+use multi_index_map::MultiIndexMap;
+use net::interface::InterfaceName;
+use net::interface::VethProperties;
+use rekon::AsRequirement;
+use serde::{Deserialize, Serialize};
+
+/// The planned properties of a veth interface.
+#[derive(
+    Builder,
+    Clone,
+    Debug,
+    Eq,
+    Hash,
+    MultiIndexMap,
+    Ord,
+    PartialEq,
+    PartialOrd,
+    Deserialize,
+    Serialize,
+)]
+#[multi_index_derive(Debug, Clone, Default, Serialize, Deserialize)]
+pub struct VethPropertiesSpec {
+    /// Name of peer interface
+    #[multi_index(ordered_unique)]
+    pub peer: InterfaceName,
+    /// Name of peer net NS
+    pub peer_ns: Option<String>, // not impl
+}
+
+impl AsRequirement<VethPropertiesSpec> for VethProperties {
+    type Requirement<'a>
+        = VethPropertiesSpec
+    where
+        Self: 'a;
+
+    fn as_requirement<'a>(&self) -> Self::Requirement<'a> {
+        VethPropertiesSpec {
+            peer: self.peer.clone(),
+            peer_ns: self.peer_ns.clone(),
+        }
+    }
+}
+
+impl PartialEq<VethProperties> for VethPropertiesSpec {
+    fn eq(&self, other: &VethProperties) -> bool {
+        self == &other.as_requirement()
+    }
+}
+
+#[cfg(any(test, feature = "bolero"))]
+mod contract {
+    use crate::interface::VethPropertiesSpec;
+    use bolero::{Driver, TypeGenerator};
+    use net::interface::InterfaceName;
+
+    impl TypeGenerator for VethPropertiesSpec {
+        fn generate<D: Driver>(driver: &mut D) -> Option<Self> {
+            Some(Self {
+                peer: InterfaceName::generate(driver)?,
+                peer_ns: None, // TBD later
+            })
+        }
+    }
+}

mgmt/src/vpc_manager/mod.rs

@@ -9,14 +9,14 @@ use interface_manager::Manager;
 use interface_manager::interface::{
     BridgePropertiesSpec, InterfaceAssociationSpec, InterfacePropertiesSpec, InterfaceSpecBuilder,
     MultiIndexInterfaceAssociationSpecMap, MultiIndexInterfaceSpecMap,
-    MultiIndexVrfPropertiesSpecMap, MultiIndexVtepPropertiesSpecMap, TryFromLinkMessage,
-    VrfPropertiesSpec, VtepPropertiesSpec,
+    MultiIndexVethPropertiesSpecMap, MultiIndexVrfPropertiesSpecMap,
+    MultiIndexVtepPropertiesSpecMap, TryFromLinkMessage, VrfPropertiesSpec, VtepPropertiesSpec,
 };
 use multi_index_map::MultiIndexMap;
 use net::eth::ethtype::EthType;
 use net::interface::{
-    AdminState, Interface, InterfaceProperties, MultiIndexInterfaceMap, MultiIndexVrfPropertiesMap,
-    MultiIndexVtepPropertiesMap,
+    AdminState, Interface, InterfaceProperties, MultiIndexInterfaceMap,
+    MultiIndexVethPropertiesMap, MultiIndexVrfPropertiesMap, MultiIndexVtepPropertiesMap,
 };
 use net::ip::UnicastIpAddr;
 use net::route::RouteTableId;
@@ -84,6 +84,7 @@ impl From<Vni> for VpcDiscriminant {
 #[derive(Clone, Debug, Deserialize, Serialize, Default, Builder)]
 pub struct RequiredInformationBase {
     pub interfaces: MultiIndexInterfaceSpecMap,
+    pub veths: MultiIndexVethPropertiesSpecMap,
     pub vrfs: MultiIndexVrfPropertiesSpecMap,
     pub vteps: MultiIndexVtepPropertiesSpecMap,
     pub associations: MultiIndexInterfaceAssociationSpecMap,
@@ -94,6 +95,7 @@ pub struct ObservedInformationBase {
     pub interfaces: MultiIndexInterfaceMap,
     pub vrfs: MultiIndexVrfPropertiesMap,
     pub vteps: MultiIndexVtepPropertiesMap,
+    pub veths: MultiIndexVethPropertiesMap,
 }
 
 impl Observe for VpcManager<RequiredInformationBase> {
@@ -124,6 +126,7 @@ impl Observe for VpcManager<RequiredInformationBase> {
         }
         let mut vtep_properties = MultiIndexVtepPropertiesMap::default();
         let mut vrf_properties = MultiIndexVrfPropertiesMap::default();
+        let mut veth_properties = MultiIndexVethPropertiesMap::default();
         let mut indexes_to_remove = vec![];
         for (_, observation) in observations.iter() {
             match &observation.properties {
@@ -148,6 +151,15 @@ impl Observe for VpcManager<RequiredInformationBase> {
                 InterfaceProperties::Other | InterfaceProperties::Bridge(_) => {
                     /* nothing to index */
                 }
+                InterfaceProperties::Veth(properties) => {
+                    match veth_properties.try_insert(properties.clone()) {
+                        Ok(_) => {}
+                        Err(err) => {
+                            error!("{err:?}");
+                            indexes_to_remove.push(observation.index);
+                        }
+                    }
+                }
             }
         }
         for sliced in indexes_to_remove {
@@ -157,6 +169,7 @@ impl Observe for VpcManager<RequiredInformationBase> {
             .interfaces(observations)
             .vteps(vtep_properties)
             .vrfs(vrf_properties)
+            .veths(veth_properties)
             .build()
         {
             Ok(ob) => Ok(ob),
@@ -303,6 +316,7 @@ impl TryFrom<&InternalConfig> for RequiredInformationBase {
     fn try_from(internal: &InternalConfig) -> Result<Self, Self::Error> {
         let mut rib = RequiredInformationBaseBuilder::default();
         let mut interfaces = MultiIndexInterfaceSpecMap::default();
+        let veths = MultiIndexVethPropertiesSpecMap::default();
         let mut vrfs = MultiIndexVrfPropertiesSpecMap::default();
         let mut vteps = MultiIndexVtepPropertiesSpecMap::default();
         let mut associations = MultiIndexInterfaceAssociationSpecMap::default();
@@ -451,10 +465,13 @@ impl TryFrom<&InternalConfig> for RequiredInformationBase {
                 }
             }
         }
+
+        // TODO: setup veths
         rib.interfaces(interfaces);
         rib.vteps(vteps);
         rib.vrfs(vrfs);
         rib.associations(associations);
+        rib.veths(veths);
         rib.build()
     }
 }
@@ -496,6 +513,7 @@ mod contract {
             let mut bridges = vec![];
             let mut vrfs = vec![];
             let mut vteps = vec![];
+            let mut veths = vec![];
 
             for _ in 0..num_vpcs {
                 let mut interface: InterfaceSpec = driver.produce()?;
@@ -539,6 +557,22 @@ mod contract {
                                 .unwrap();
                         }
                     }
+                    InterfacePropertiesSpec::Veth(props) => {
+                        if requirements
+                            .interfaces
+                            .try_insert(interface.clone())
+                            .is_ok()
+                        {
+                            let Err(_) = requirements.veths.try_insert(props.clone()) else {
+                                veths.push(interface.clone());
+                                continue;
+                            };
+                            requirements
+                                .interfaces
+                                .remove_by_name(&interface.name)
+                                .unwrap();
+                        }
+                    }
                 }
             }
             if !bridges.is_empty() {

mgmt/tests/reconcile.rs

@@ -8,8 +8,8 @@ use fixin::wrap;
 use interface_manager::interface::{
     BridgePropertiesSpec, InterfaceAssociationSpec, InterfacePropertiesSpec, InterfaceSpecBuilder,
     MultiIndexBridgePropertiesSpecMap, MultiIndexInterfaceAssociationSpecMap,
-    MultiIndexInterfaceSpecMap, MultiIndexVrfPropertiesSpecMap, MultiIndexVtepPropertiesSpecMap,
-    VrfPropertiesSpec, VtepPropertiesSpec,
+    MultiIndexInterfaceSpecMap, MultiIndexVethPropertiesSpecMap, MultiIndexVrfPropertiesSpecMap,
+    MultiIndexVtepPropertiesSpecMap, VethPropertiesSpec, VrfPropertiesSpec, VtepPropertiesSpec,
 };
 use mgmt::vpc_manager::{RequiredInformationBase, RequiredInformationBaseBuilder, VpcManager};
 use net::eth::ethtype::EthType;
@@ -144,6 +144,15 @@ async fn reconcile_demo() {
             }))
             .build()
             .unwrap(),
+        InterfaceSpecBuilder::default()
+            .name("veth1".try_into().unwrap())
+            .admin_state(AdminState::Up)
+            .properties(InterfacePropertiesSpec::Veth(VethPropertiesSpec {
+                peer: "veth1-peer".try_into().unwrap(),
+                peer_ns: None, // not impl
+            }))
+            .build()
+            .unwrap(),
     ];
 
     for interface in interfaces {
@@ -153,6 +162,7 @@ async fn reconcile_demo() {
     let mut vtep_props = MultiIndexVtepPropertiesSpecMap::default();
     let mut bridge_props = MultiIndexBridgePropertiesSpecMap::default();
     let mut vrf_props = MultiIndexVrfPropertiesSpecMap::default();
+    let mut veth_props = MultiIndexVethPropertiesSpecMap::default();
 
     for (_, interface) in required_interface_map.iter() {
         match &interface.properties {
@@ -165,6 +175,9 @@ async fn reconcile_demo() {
             InterfacePropertiesSpec::Vrf(prop) => {
                 vrf_props.try_insert(prop.clone()).unwrap();
             }
+            InterfacePropertiesSpec::Veth(prop) => {
+                veth_props.try_insert(prop.clone()).unwrap();
+            }
         }
     }
 
@@ -199,6 +212,7 @@ async fn reconcile_demo() {
         .vteps(vtep_props)
         .vrfs(vrf_props)
         .associations(associations)
+        .veths(veth_props)
         .build()
         .unwrap();
 
@@ -249,6 +263,16 @@ async fn reconcile_demo() {
                 }))
                 .build()
                 .unwrap(),
+            InterfaceSpecBuilder::default()
+                .name("veth2".try_into().unwrap())
+                .admin_state(AdminState::Up)
+                .controller(None)
+                .properties(InterfacePropertiesSpec::Veth(VethPropertiesSpec {
+                    peer: "veth2-peer".try_into().unwrap(),
+                    peer_ns: None,
+                }))
+                .build()
+                .unwrap(),
         ];
         for interface in interfaces {
             match &interface.properties {
@@ -259,6 +283,9 @@ async fn reconcile_demo() {
                 InterfacePropertiesSpec::Vrf(props) => {
                     req.vrfs.try_insert(props.clone()).unwrap();
                 }
+                InterfacePropertiesSpec::Veth(props) => {
+                    req.veths.try_insert(props.clone()).unwrap();
+                }
             }
             req.interfaces.try_insert(interface).unwrap();
         }
@@ -292,27 +319,33 @@ async fn reconcile_demo() {
         req.associations
             .remove_by_name(&"vtep1".to_string().try_into().unwrap())
             .unwrap();
+        req.interfaces
+            .remove_by_name(&"veth2".to_string().try_into().unwrap())
+            .unwrap();
     };
 
     let vpcs = VpcManager::<RequiredInformationBase>::new(Arc::new(handle));
 
-    for _ in 0..10 {
+    for k in 0..10 {
         let observed = vpcs.observe().await.unwrap();
         vpcs.reconcile(&mut required, &observed).await;
+        println!("k:{k}\n{}", observed.interfaces);
         tokio::time::sleep(Duration::from_millis(20)).await;
     }
     info!("injecting new requirements");
     inject_new_requirements(&mut required);
-    for _ in 0..20 {
+    for n in 0..20 {
         let observed = vpcs.observe().await.unwrap();
         vpcs.reconcile(&mut required, &observed).await;
+        println!("n:{n}\n{}", observed.interfaces);
         tokio::time::sleep(Duration::from_millis(20)).await;
     }
     info!("removing some requirements");
     remove_some_requirement(&mut required);
-    for _ in 0..20 {
+    for r in 0..20 {
         let observed = vpcs.observe().await.unwrap();
         vpcs.reconcile(&mut required, &observed).await;
+        println!("r:{r}\n{}", observed.interfaces);
         tokio::time::sleep(Duration::from_millis(20)).await;
     }
 }