Add L7 Antrea native NetworkPolicy datapath support

Signed-off-by: Hongliang Liu <[email protected]>

cmd/antrea-agent/agent.go

@@ -137,6 +137,7 @@ func run(o *Options) error {
 
 	ovsDatapathType := ovsconfig.OVSDatapathType(o.config.OVSDatapathType)
 	ovsBridgeClient := ovsconfig.NewOVSBridge(o.config.OVSBridge, ovsDatapathType, ovsdbConnection)
+	ovsCtlClient := ovsctl.NewClient(o.config.OVSBridge)
 	ovsBridgeMgmtAddr := ofconfig.GetMgmtAddress(o.config.OVSRunDir, o.config.OVSBridge)
 	multicastEnabled := features.DefaultFeatureGate.Enabled(features.Multicast)
 	ofClient := openflow.NewClient(o.config.OVSBridge, ovsBridgeMgmtAddr,
@@ -232,6 +233,7 @@ func run(o *Options) error {
 		k8sClient,
 		crdClient,
 		ovsBridgeClient,
+		ovsCtlClient,
 		ofClient,
 		routeClient,
 		ifaceStore,
@@ -645,7 +647,7 @@ func run(o *Options) error {
 		tcController := trafficcontrol.NewTrafficControlController(ofClient,
 			ifaceStore,
 			ovsBridgeClient,
-			ovsctl.NewClient(o.config.OVSBridge),
+			ovsCtlClient,
 			trafficControlInformer,
 			localPodInformer,
 			namespaceInformer,

hack/update-codegen-dockerized.sh

@@ -39,6 +39,7 @@ ANTREA_PROTO_PKG="antrea_io.antrea"
 function generate_mocks {
   # Generate mocks for testing with mockgen.
   MOCKGEN_TARGETS=(
+    "pkg/agent/controller/l7networkpolicy/suricata ReconcilerOperations testing"
     "pkg/agent/cniserver/ipam IPAMDriver testing"
     "pkg/agent/flowexporter/connections ConnTrackDumper,NetFilterConnTrack testing"
     "pkg/agent/interfacestore InterfaceStore testing"

pkg/agent/agent.go

@@ -88,6 +88,7 @@ type Initializer struct {
 	client                clientset.Interface
 	crdClient             versioned.Interface
 	ovsBridgeClient       ovsconfig.OVSBridgeClient
+	ovsCtlClient          ovsctl.OVSCtlClient
 	ofClient              openflow.Client
 	routeClient           route.Interface
 	wireGuardClient       wireguard.Interface
@@ -100,6 +101,7 @@ type Initializer struct {
 	wireGuardConfig       *config.WireGuardConfig
 	egressConfig          *config.EgressConfig
 	serviceConfig         *config.ServiceConfig
+	l7NetworkPolicyConfig *config.L7NetworkPolicyConfig
 	enableProxy           bool
 	connectUplinkToBridge bool
 	// networkReadyCh should be closed once the Node's network is ready.
@@ -115,6 +117,7 @@ func NewInitializer(
 	k8sClient clientset.Interface,
 	crdClient versioned.Interface,
 	ovsBridgeClient ovsconfig.OVSBridgeClient,
+	ovsCtlClient ovsctl.OVSCtlClient,
 	ofClient openflow.Client,
 	routeClient route.Interface,
 	ifaceStore interfacestore.InterfaceStore,
@@ -135,6 +138,7 @@ func NewInitializer(
 ) *Initializer {
 	return &Initializer{
 		ovsBridgeClient:       ovsBridgeClient,
+		ovsCtlClient:          ovsCtlClient,
 		client:                k8sClient,
 		crdClient:             crdClient,
 		ifaceStore:            ifaceStore,
@@ -147,6 +151,7 @@ func NewInitializer(
 		wireGuardConfig:       wireGuardConfig,
 		egressConfig:          egressConfig,
 		serviceConfig:         serviceConfig,
+		l7NetworkPolicyConfig: &config.L7NetworkPolicyConfig{},
 		networkReadyCh:        networkReadyCh,
 		stopCh:                stopCh,
 		nodeType:              nodeType,
@@ -378,6 +383,10 @@ func (i *Initializer) Initialize() error {
 		return err
 	}
 
+	if err := i.prepareL7NetworkPolicyInterfaces(); err != nil {
+		return err
+	}
+
 	// initializeWireGuard must be executed after setupOVSBridge as it requires gateway addresses on the OVS bridge.
 	if i.networkConfig.TrafficEncryptionMode == config.TrafficEncryptionModeWireGuard {
 		if err := i.initializeWireGuard(); err != nil {
@@ -503,7 +512,7 @@ func (i *Initializer) initOpenFlowPipeline() error {
 	roundInfo := getRoundInfo(i.ovsBridgeClient)
 
 	// Set up all basic flows.
-	ofConnCh, err := i.ofClient.Initialize(roundInfo, i.nodeConfig, i.networkConfig, i.egressConfig, i.serviceConfig)
+	ofConnCh, err := i.ofClient.Initialize(roundInfo, i.nodeConfig, i.networkConfig, i.egressConfig, i.serviceConfig, i.l7NetworkPolicyConfig)
 	if err != nil {
 		klog.Errorf("Failed to initialize openflow client: %v", err)
 		return err

pkg/agent/agent_linux.go

@@ -312,3 +312,59 @@ func (i *Initializer) prepareOVSBridgeForVM() error {
 func (i *Initializer) installVMInitialFlows() error {
 	return nil
 }
+
+// prepareL7NetworkPolicyInterfaces creates two internal OVS ports on OVS. L7 NetworkPolicy processing engine Suricata
+// uses two ofPorts to connect to OVS.
+func (i *Initializer) prepareL7NetworkPolicyInterfaces() error {
+	trafficControlPortExternalIDs := map[string]interface{}{
+		interfacestore.AntreaInterfaceTypeKey: interfacestore.AntreaTrafficControl,
+	}
+
+	for _, portName := range []string{config.L7NetworkPolicyTargetOFPort, config.L7NetworkPolicyReturnOFPort} {
+		_, exists := i.ifaceStore.GetInterface(portName)
+		if exists {
+			continue
+		}
+
+		// Create the ofPort if it doesn't exist.
+		portUUID, err := i.ovsBridgeClient.CreateInternalPort(portName, 0, "", trafficControlPortExternalIDs)
+		if err != nil {
+			return err
+		}
+		// Set the ofPort up.
+		if pollErr := wait.PollImmediate(time.Second, 5*time.Second, func() (bool, error) {
+			_, _, err := util.SetLinkUp(portName)
+			if err == nil {
+				return true, nil
+			}
+			if _, ok := err.(util.LinkNotFound); ok {
+				return false, nil
+			}
+			return false, err
+		}); pollErr != nil {
+			return pollErr
+		}
+
+		// Get the port numbers of the ofPorts.
+		ofPort, err := i.ovsBridgeClient.GetOFPort(portName, false)
+		if err != nil {
+			return err
+		}
+
+		// Set the ofPort with no-flood to reject ARP flood packets.
+		if err := i.ovsCtlClient.SetPortNoFlood(int(ofPort)); err != nil {
+			return fmt.Errorf("failed to set port %s with no-flood config: %w", portName, err)
+		}
+
+		itf := interfacestore.NewTrafficControlInterface(portName)
+		itf.OVSPortConfig = &interfacestore.OVSPortConfig{PortUUID: portUUID, OFPort: ofPort}
+		i.ifaceStore.AddInterface(itf)
+	}
+
+	ofPort, _ := i.ifaceStore.GetInterfaceByName(config.L7NetworkPolicyTargetOFPort)
+	i.l7NetworkPolicyConfig.TargetOFPort = uint32(ofPort.OFPort)
+	ofPort, _ = i.ifaceStore.GetInterfaceByName(config.L7NetworkPolicyReturnOFPort)
+	i.l7NetworkPolicyConfig.ReturnOFPort = uint32(ofPort.OFPort)
+
+	return nil
+}

pkg/agent/agent_windows.go

@@ -474,3 +474,7 @@ func (i *Initializer) installVMInitialFlows() error {
 	}
 	return nil
 }
+
+func (i *Initializer) prepareL7NetworkPolicyInterfaces() error {
+	return nil
+}

pkg/agent/config/node_config.go

@@ -44,6 +44,11 @@ const (
 	IPv6ExtraOverhead = 20
 )
 
+const (
+	L7NetworkPolicyTargetOFPort = "antrea-tap0"
+	L7NetworkPolicyReturnOFPort = "antrea-tap1"
+)
+
 var (
 	// VirtualServiceIPv4 or VirtualServiceIPv6 is used in the following scenarios:
 	// - The IP is used to perform SNAT for packets of Service sourced from Antrea gateway and destined for external
@@ -233,3 +238,9 @@ type ServiceConfig struct {
 	NodePortAddressesIPv4 []net.IP
 	NodePortAddressesIPv6 []net.IP
 }
+
+// L7NetworkPolicyConfig includes target and return ofPorts for L7 NetworkPolicy.
+type L7NetworkPolicyConfig struct {
+	TargetOFPort uint32 // Matched L7 traffic is forwarded to Suricata via this ofPort.
+	ReturnOFPort uint32 // Traffic is returned from Suricata via this ofPort.
+}