Add support for multicast in user defined networks.

.github/workflows/test.yml

@@ -441,7 +441,7 @@ jobs:
     env:
       JOB_NAME: "${{ matrix.target }}-${{ matrix.ha }}-${{ matrix.gateway-mode }}-${{ matrix.ipfamily }}-${{ matrix.disable-snat-multiple-gws }}-${{ matrix.second-bridge }}-${{ matrix.ic }}"
       OVN_HYBRID_OVERLAY_ENABLE: ${{ (matrix.target == 'control-plane' || matrix.target == 'control-plane-helm') && (matrix.ipfamily == 'ipv4' || matrix.ipfamily == 'dualstack' ) }}
-      OVN_MULTICAST_ENABLE:  "${{ matrix.target == 'control-plane' || matrix.target == 'control-plane-helm' }}"
+      OVN_MULTICAST_ENABLE:  "${{ matrix.target == 'control-plane' || matrix.target == 'control-plane-helm' || matrix.target == 'network-segmentation' }}"
       OVN_EMPTY_LB_EVENTS: "${{ matrix.target == 'control-plane' || matrix.target == 'control-plane-helm' }}"
       OVN_HA: "${{ matrix.ha == 'HA' }}"
       OVN_DISABLE_SNAT_MULTIPLE_GWS: "${{ matrix.disable-snat-multiple-gws == 'noSnatGW' }}"

go-controller/pkg/ovn/base_network_controller.go

@@ -1,6 +1,7 @@
 package ovn
 
 import (
+	"errors"
 	"fmt"
 	"net"
 	"reflect"
@@ -680,16 +681,12 @@ func (bnc *BaseNetworkController) syncNodeManagementPort(node *kapi.Node, switch
 		return nil, err
 	}
 
-	// TODO(dceara): The cluster port group must be per network. So for now skip adding management port to the cluster port
-	// group for secondary network's because the cluster port group is not yet created for secondary networks.
-	if bnc.IsDefault() {
-		clusterPortGroupName := bnc.getClusterPortGroupName(types.ClusterPortGroupNameBase)
-		if err = libovsdbops.AddPortsToPortGroup(bnc.nbClient, clusterPortGroupName, logicalSwitchPort.UUID); err != nil {
-			err1 := fmt.Errorf("failed to add port %s to cluster port group %s (%s): %w",
-				logicalSwitchPort.Name, types.ClusterPortGroupNameBase, clusterPortGroupName, err)
-			klog.Error(err1)
-			return nil, err1
-		}
+	clusterPortGroupName := bnc.getClusterPortGroupName(types.ClusterPortGroupNameBase)
+	if err = libovsdbops.AddPortsToPortGroup(bnc.nbClient, clusterPortGroupName, logicalSwitchPort.UUID); err != nil {
+		err1 := fmt.Errorf("failed to add port %s to cluster port group %s (%s): %w",
+			logicalSwitchPort.Name, types.ClusterPortGroupNameBase, clusterPortGroupName, err)
+		klog.Error(err1)
+		return nil, err1
 	}
 
 	if v4Subnet != nil {
@@ -701,6 +698,26 @@ func (bnc *BaseNetworkController) syncNodeManagementPort(node *kapi.Node, switch
 	return mgmtPortIPs, nil
 }
 
+// addLocalPodToNamespaceLocked returns the ops needed to add the pod's IP to the namespace
+// address set and the port UUID (if applicable) to the namespace port group.
+// This function must be called with the nsInfo lock taken.
+func (bnc *BaseNetworkController) addLocalPodToNamespaceLocked(nsInfo *namespaceInfo, ips []*net.IPNet, portUUID string) ([]ovsdb.Operation, error) {
+	var ops []ovsdb.Operation
+	var err error
+
+	if ops, err = nsInfo.addressSet.AddAddressesReturnOps(util.IPNetsIPToStringSlice(ips)); err != nil {
+		return nil, err
+	}
+
+	if portUUID != "" && nsInfo.portGroupName != "" {
+		if ops, err = libovsdbops.AddPortsToPortGroupOps(bnc.nbClient, ops, nsInfo.portGroupName, portUUID); err != nil {
+			return nil, err
+		}
+	}
+
+	return ops, nil
+}
+
 // WatchNodes starts the watching of the nodes resource and calls back the appropriate handler logic
 func (bnc *BaseNetworkController) WatchNodes() error {
 	if bnc.nodeHandler != nil {
@@ -1025,6 +1042,47 @@ func initLoadBalancerGroups(nbClient libovsdbclient.Client, netInfo util.NetInfo
 	return
 }
 
+func (bnc *BaseNetworkController) setupClusterPortGroups() error {
+	pgIDs := bnc.getClusterPortGroupDbIDs(types.ClusterPortGroupNameBase)
+	pg := &nbdb.PortGroup{
+		Name: libovsdbutil.GetPortGroupName(pgIDs),
+	}
+	pg, err := libovsdbops.GetPortGroup(bnc.nbClient, pg)
+	if err != nil && !errors.Is(err, libovsdbclient.ErrNotFound) {
+		return fmt.Errorf("failed to query cluster port group for network %s: %w", bnc.GetNetworkName(), err)
+	}
+	if pg == nil {
+		// we didn't find an existing clusterPG, let's create a new empty PG (fresh cluster install)
+		// Create a cluster-wide port group that all logical switch ports are part of
+		pg := libovsdbutil.BuildPortGroup(pgIDs, nil, nil)
+		err = libovsdbops.CreateOrUpdatePortGroups(bnc.nbClient, pg)
+		if err != nil {
+			return fmt.Errorf("failed to create cluster port group for network %s: %w", bnc.GetNetworkName(), err)
+		}
+	}
+
+	pgIDs = bnc.getClusterPortGroupDbIDs(types.ClusterRtrPortGroupNameBase)
+	pg = &nbdb.PortGroup{
+		Name: libovsdbutil.GetPortGroupName(pgIDs),
+	}
+	pg, err = libovsdbops.GetPortGroup(bnc.nbClient, pg)
+	if err != nil && !errors.Is(err, libovsdbclient.ErrNotFound) {
+		return fmt.Errorf("failed to query cluster router port group for network %s: %w", bnc.GetNetworkName(), err)
+	}
+	if pg == nil {
+		// we didn't find an existing clusterRtrPG, let's create a new empty PG (fresh cluster install)
+		// Create a cluster-wide port group with all node-to-cluster router
+		// logical switch ports. Currently the only user is multicast but it might
+		// be used for other features in the future.
+		pg = libovsdbutil.BuildPortGroup(pgIDs, nil, nil)
+		err = libovsdbops.CreateOrUpdatePortGroups(bnc.nbClient, pg)
+		if err != nil {
+			return fmt.Errorf("failed to create cluster router port group for network %s: %w", bnc.GetNetworkName(), err)
+		}
+	}
+	return nil
+}
+
 func (bnc *BaseNetworkController) GetSamplingConfig() *libovsdbops.SamplingConfig {
 	if bnc.observManager != nil {
 		return bnc.observManager.SamplingConfig()

go-controller/pkg/ovn/base_network_controller_multicast.go

@@ -322,3 +322,28 @@ func (bnc *BaseNetworkController) syncNsMulticast(k8sNamespaces map[string]bool)
 
 	return nil
 }
+
+func (bnc *BaseNetworkController) syncDefaultMulticastPolicies() error {
+	// If supported, enable IGMP relay on the router to forward multicast
+	// traffic between nodes.
+	if bnc.multicastSupport {
+		// Drop IP multicast globally. Multicast is allowed only if explicitly
+		// enabled in a namespace.
+		if err := bnc.createDefaultDenyMulticastPolicy(); err != nil {
+			klog.Errorf("Failed to create default deny multicast policy, error: %v", err)
+			return err
+		}
+
+		// Allow IP multicast from node switch to cluster router and from
+		// cluster router to node switch.
+		if err := bnc.createDefaultAllowMulticastPolicy(); err != nil {
+			klog.Errorf("Failed to create default deny multicast policy, error: %v", err)
+			return err
+		}
+	} else {
+		if err := bnc.disableMulticast(); err != nil {
+			return fmt.Errorf("failed to delete default multicast policy, error: %v", err)
+		}
+	}
+	return nil
+}

go-controller/pkg/ovn/base_network_controller_policy.go

@@ -220,13 +220,6 @@ func (bnc *BaseNetworkController) syncNetworkPolicies(networkPolicies []interfac
 		return err
 	}
 
-	// FIXME: For primary user defined networks, we need the hairpin allow ACL.
-	// The port group is not found error is thrown at line 269. Is that expected ?
-	// (or) should this be fixed https://github.com/ovn-org/ovn-kubernetes/blob/master/go-controller/pkg/ovn/base_network_controller.go#L632 ?
-	if bnc.NetInfo.IsSecondary() {
-		return nil
-	}
-
 	// add default hairpin allow acl
 	err = bnc.addHairpinAllowACL()
 	if err != nil {

go-controller/pkg/ovn/base_network_controller_secondary.go

@@ -333,10 +333,13 @@ func (bsnc *BaseSecondaryNetworkController) addLogicalPortToNetworkForNAD(pod *k
 			return err
 		}
 	}
-
-	if bsnc.doesNetworkRequireIPAM() && util.IsMultiNetworkPoliciesSupportEnabled() {
+	if bsnc.doesNetworkRequireIPAM() && (util.IsMultiNetworkPoliciesSupportEnabled() || bsnc.multicastSupport) {
 		// Ensure the namespace/nsInfo exists
-		addOps, err := bsnc.addPodToNamespaceForSecondaryNetwork(pod.Namespace, podAnnotation.IPs)
+		portUUID := ""
+		if lsp != nil {
+			portUUID = lsp.UUID
+		}
+		addOps, err := bsnc.addPodToNamespaceForSecondaryNetwork(pod.Namespace, podAnnotation.IPs, portUUID)
 		if err != nil {
 			return err
 		}
@@ -635,8 +638,7 @@ func (bsnc *BaseSecondaryNetworkController) syncPodsForSecondaryNetwork(pods []i
 }
 
 // addPodToNamespaceForSecondaryNetwork returns the ops needed to add pod's IP to the namespace's address set.
-func (bsnc *BaseSecondaryNetworkController) addPodToNamespaceForSecondaryNetwork(ns string, ips []*net.IPNet) ([]ovsdb.Operation, error) {
-	var ops []ovsdb.Operation
+func (bsnc *BaseSecondaryNetworkController) addPodToNamespaceForSecondaryNetwork(ns string, ips []*net.IPNet, portUUID string) ([]ovsdb.Operation, error) {
 	var err error
 	nsInfo, nsUnlock, err := bsnc.ensureNamespaceLockedForSecondaryNetwork(ns, true, nil)
 	if err != nil {
@@ -645,11 +647,7 @@ func (bsnc *BaseSecondaryNetworkController) addPodToNamespaceForSecondaryNetwork
 
 	defer nsUnlock()
 
-	if ops, err = nsInfo.addressSet.AddAddressesReturnOps(util.IPNetsIPToStringSlice(ips)); err != nil {
-		return nil, err
-	}
-
-	return ops, nil
+	return bsnc.addLocalPodToNamespaceLocked(nsInfo, ips, portUUID)
 }
 
 // AddNamespaceForSecondaryNetwork creates corresponding addressset in ovn db for secondary network

go-controller/pkg/ovn/master.go

@@ -18,7 +18,6 @@ import (
 	"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/config"
 	"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/kube"
 	libovsdbops "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/libovsdb/ops"
-	libovsdbutil "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/libovsdb/util"
 	"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/nbdb"
 	"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/sbdb"
 	"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/types"
@@ -48,67 +47,13 @@ func (oc *DefaultNetworkController) SetupMaster(existingNodeNames []string) erro
 		return err
 	}
 
-	pgIDs := oc.getClusterPortGroupDbIDs(types.ClusterPortGroupNameBase)
-	pg := &nbdb.PortGroup{
-		Name: libovsdbutil.GetPortGroupName(pgIDs),
-	}
-	pg, err = libovsdbops.GetPortGroup(oc.nbClient, pg)
-	if err != nil && !errors.Is(err, libovsdbclient.ErrNotFound) {
+	if err := oc.setupClusterPortGroups(); err != nil {
 		return err
 	}
-	if pg == nil {
-		// we didn't find an existing clusterPG, let's create a new empty PG (fresh cluster install)
-		// Create a cluster-wide port group that all logical switch ports are part of
-		pg := libovsdbutil.BuildPortGroup(pgIDs, nil, nil)
-		err = libovsdbops.CreateOrUpdatePortGroups(oc.nbClient, pg)
-		if err != nil {
-			klog.Errorf("Failed to create cluster port group: %v", err)
-			return err
-		}
-	}
 
-	pgIDs = oc.getClusterPortGroupDbIDs(types.ClusterRtrPortGroupNameBase)
-	pg = &nbdb.PortGroup{
-		Name: libovsdbutil.GetPortGroupName(pgIDs),
-	}
-	pg, err = libovsdbops.GetPortGroup(oc.nbClient, pg)
-	if err != nil && !errors.Is(err, libovsdbclient.ErrNotFound) {
+	if err := oc.syncDefaultMulticastPolicies(); err != nil {
 		return err
 	}
-	if pg == nil {
-		// we didn't find an existing clusterRtrPG, let's create a new empty PG (fresh cluster install)
-		// Create a cluster-wide port group with all node-to-cluster router
-		// logical switch ports. Currently the only user is multicast but it might
-		// be used for other features in the future.
-		pg = libovsdbutil.BuildPortGroup(pgIDs, nil, nil)
-		err = libovsdbops.CreateOrUpdatePortGroups(oc.nbClient, pg)
-		if err != nil {
-			klog.Errorf("Failed to create cluster port group: %v", err)
-			return err
-		}
-	}
-
-	// If supported, enable IGMP relay on the router to forward multicast
-	// traffic between nodes.
-	if oc.multicastSupport {
-		// Drop IP multicast globally. Multicast is allowed only if explicitly
-		// enabled in a namespace.
-		if err := oc.createDefaultDenyMulticastPolicy(); err != nil {
-			klog.Errorf("Failed to create default deny multicast policy, error: %v", err)
-			return err
-		}
-
-		// Allow IP multicast from node switch to cluster router and from
-		// cluster router to node switch.
-		if err := oc.createDefaultAllowMulticastPolicy(); err != nil {
-			klog.Errorf("Failed to create default deny multicast policy, error: %v", err)
-			return err
-		}
-	} else {
-		if err = oc.disableMulticast(); err != nil {
-			return fmt.Errorf("failed to delete default multicast policy, error: %v", err)
-		}
-	}
 
 	// Create OVNJoinSwitch that will be used to connect gateway routers to the distributed router.
 	return oc.gatewayTopologyFactory.NewJoinSwitch(logicalRouter, oc.NetInfo, oc.ovnClusterLRPToJoinIfAddrs)

go-controller/pkg/ovn/master_test.go

@@ -1834,22 +1834,32 @@ func newClusterJoinSwitch() *nbdb.LogicalSwitch {
 	}
 }
 
-func newClusterPortGroup() *nbdb.PortGroup {
-	fakeController := getFakeController(DefaultNetworkControllerName)
+func newNetworkClusterPortGroup(netInfo util.NetInfo) *nbdb.PortGroup {
+	netControllerName := getNetworkControllerName(netInfo.GetNetworkName())
+	fakeController := getFakeController(netControllerName)
 	pgIDs := fakeController.getClusterPortGroupDbIDs(types.ClusterPortGroupNameBase)
 	pg := libovsdbutil.BuildPortGroup(pgIDs, nil, nil)
 	pg.UUID = pgIDs.String()
 	return pg
 }
 
-func newRouterPortGroup() *nbdb.PortGroup {
-	fakeController := getFakeController(DefaultNetworkControllerName)
+func newNetworkRouterPortGroup(netInfo util.NetInfo) *nbdb.PortGroup {
+	netControllerName := getNetworkControllerName(netInfo.GetNetworkName())
+	fakeController := getFakeController(netControllerName)
 	pgIDs := fakeController.getClusterPortGroupDbIDs(types.ClusterRtrPortGroupNameBase)
 	pg := libovsdbutil.BuildPortGroup(pgIDs, nil, nil)
 	pg.UUID = pgIDs.String()
 	return pg
 }
 
+func newClusterPortGroup() *nbdb.PortGroup {
+	return newNetworkClusterPortGroup(&util.DefaultNetInfo{})
+}
+
+func newRouterPortGroup() *nbdb.PortGroup {
+	return newNetworkRouterPortGroup(&util.DefaultNetInfo{})
+}
+
 func newOVNClusterRouter() *nbdb.LogicalRouter {
 	return &nbdb.LogicalRouter{
 		UUID: types.OVNClusterRouter + "-UUID",