[Multicast] Use group as flow actions for multicast traffic (#3508)

1. Add local Pod receivers into an OpenFlow type "all" group for each
multicast group, and use such groups in the flow actions. Remove a Pod
from group buckets if the Pod has left the multicast group or is
deleted before leaving the multicast group.
2. Improve multicast e2e tests.

Signed-off-by: wenyingd <wenyingd@vmware.com>
Co-authored-by: Ruochen Shen <src655@gmail.com>

cmd/antrea-agent/agent.go

@@ -579,11 +579,13 @@ func run(o *Options) error {
 		}
 		mcastController := multicast.NewMulticastController(
 			ofClient,
+			v4GroupIDAllocator,
 			nodeConfig,
 			ifaceStore,
 			multicastSocket,
 			sets.NewString(append(o.config.MulticastInterfaces, nodeConfig.NodeTransportInterfaceName)...),
-			ovsBridgeClient)
+			ovsBridgeClient,
+			podUpdateChannel)
 		if err := mcastController.Initialize(); err != nil {
 			return err
 		}

pkg/agent/cniserver/pod_configuration.go

@@ -32,6 +32,7 @@ import (
 	"antrea.io/antrea/pkg/agent/openflow"
 	"antrea.io/antrea/pkg/agent/route"
 	"antrea.io/antrea/pkg/agent/secondarynetwork/cnipodcache"
+	agenttypes "antrea.io/antrea/pkg/agent/types"
 	"antrea.io/antrea/pkg/agent/util"
 	"antrea.io/antrea/pkg/ovs/ovsconfig"
 	"antrea.io/antrea/pkg/util/channel"
@@ -312,6 +313,7 @@ func (pc *podConfigurator) removeInterfaces(containerID string) error {
 	if err := pc.routeClient.DeleteLocalAntreaFlexibleIPAMPodRule(containerConfig.IPs); err != nil {
 		return err
 	}
+
 	return nil
 }
 
@@ -495,7 +497,13 @@ func (pc *podConfigurator) connectInterfaceToOVSCommon(ovsPortName string, conta
 	// Add containerConfig into local cache
 	pc.ifaceStore.AddInterface(containerConfig)
 	// Notify the Pod update event to required components.
-	pc.podUpdateNotifier.Notify(k8s.NamespacedName(containerConfig.PodNamespace, containerConfig.PodName))
+	event := agenttypes.PodUpdate{
+		PodName:      containerConfig.PodName,
+		PodNamespace: containerConfig.PodNamespace,
+		IsAdd:        true,
+		ContainerID:  containerConfig.ContainerID,
+	}
+	pc.podUpdateNotifier.Notify(event)
 	return nil
 }
 
@@ -518,6 +526,13 @@ func (pc *podConfigurator) disconnectInterfaceFromOVS(containerConfig *interface
 	}
 	// Remove container configuration from cache.
 	pc.ifaceStore.DeleteInterface(containerConfig)
+	event := agenttypes.PodUpdate{
+		PodName:      containerConfig.PodName,
+		PodNamespace: containerConfig.PodNamespace,
+		IsAdd:        false,
+		ContainerID:  containerConfig.ContainerID,
+	}
+	pc.podUpdateNotifier.Notify(event)
 	klog.Infof("Removed interfaces for container %s", containerID)
 	return nil
 }

pkg/agent/cniserver/pod_configuration_windows.go

@@ -24,6 +24,7 @@ import (
 	"k8s.io/klog/v2"
 
 	"antrea.io/antrea/pkg/agent/interfacestore"
+	"antrea.io/antrea/pkg/agent/types"
 	"antrea.io/antrea/pkg/agent/util"
 	"antrea.io/antrea/pkg/util/k8s"
 )
@@ -49,7 +50,13 @@ func (pc *podConfigurator) connectInterfaceToOVSAsync(ifConfig *interfacestore.I
 		// Update interface config with the ofPort.
 		ifConfig.OVSPortConfig.OFPort = ofPort
 		// Notify the Pod update event to required components.
-		pc.podUpdateNotifier.Notify(k8s.NamespacedName(ifConfig.PodNamespace, ifConfig.PodName))
+		event := types.PodUpdate{
+			PodName:      ifConfig.PodName,
+			PodNamespace: ifConfig.PodNamespace,
+			IsAdd:        true,
+			ContainerID:  ifConfig.ContainerID,
+		}
+		pc.podUpdateNotifier.Notify(event)
 		return nil
 	})
 }

pkg/agent/controller/egress/egress_controller.go

@@ -41,6 +41,7 @@ import (
 	"antrea.io/antrea/pkg/agent/memberlist"
 	"antrea.io/antrea/pkg/agent/openflow"
 	"antrea.io/antrea/pkg/agent/route"
+	"antrea.io/antrea/pkg/agent/types"
 	cpv1b2 "antrea.io/antrea/pkg/apis/controlplane/v1beta2"
 	crdv1a2 "antrea.io/antrea/pkg/apis/crd/v1alpha2"
 	clientsetversioned "antrea.io/antrea/pkg/client/clientset/versioned"
@@ -217,9 +218,11 @@ func NewEgressController(
 
 // processPodUpdate will be called when CNIServer publishes a Pod update event.
 // It triggers reconciling the effective Egress of the Pod.
-func (c *EgressController) processPodUpdate(pod string) {
+func (c *EgressController) processPodUpdate(e interface{}) {
 	c.egressBindingsMutex.Lock()
 	defer c.egressBindingsMutex.Unlock()
+	podEvent := e.(types.PodUpdate)
+	pod := k8s.NamespacedName(podEvent.PodNamespace, podEvent.PodName)
 	binding, exists := c.egressBindings[pod]
 	if !exists {
 		return

pkg/agent/controller/egress/egress_controller_test.go

@@ -38,6 +38,7 @@ import (
 	ipassignertest "antrea.io/antrea/pkg/agent/ipassigner/testing"
 	openflowtest "antrea.io/antrea/pkg/agent/openflow/testing"
 	routetest "antrea.io/antrea/pkg/agent/route/testing"
+	"antrea.io/antrea/pkg/agent/types"
 	"antrea.io/antrea/pkg/agent/util"
 	cpv1b2 "antrea.io/antrea/pkg/apis/controlplane/v1beta2"
 	crdv1a2 "antrea.io/antrea/pkg/apis/crd/v1alpha2"
@@ -582,7 +583,11 @@ func TestPodUpdateShouldSyncEgress(t *testing.T) {
 	c.mockIPAssigner.EXPECT().UnassignIP(fakeLocalEgressIP1)
 	// Mock CNIServer
 	addPodInterface(c.ifaceStore, "ns1", "pendingPod", 10)
-	c.podUpdateChannel.Notify("ns1/pendingPod")
+	ev := types.PodUpdate{
+		PodName:      "pendingPod",
+		PodNamespace: "ns1",
+	}
+	c.podUpdateChannel.Notify(ev)
 	require.NoError(t, wait.PollImmediate(10*time.Millisecond, time.Second, func() (done bool, err error) {
 		return c.queue.Len() == 1, nil
 	}))

pkg/agent/controller/networkpolicy/cache.go

@@ -28,6 +28,7 @@ import (
 	"k8s.io/klog/v2"
 
 	"antrea.io/antrea/pkg/agent/metrics"
+	agenttypes "antrea.io/antrea/pkg/agent/types"
 	v1beta "antrea.io/antrea/pkg/apis/controlplane/v1beta2"
 	crdv1alpha1 "antrea.io/antrea/pkg/apis/crd/v1alpha1"
 	"antrea.io/antrea/pkg/querier"
@@ -361,12 +362,12 @@ func newRuleCache(dirtyRuleHandler func(string), podUpdateSubscriber channel.Sub
 // done if antrea-controller has computed the Pods' policies and propagated
 // them to this Node by their labels and NodeName, instead of waiting for their
 // IPs are reported to kube-apiserver and processed by antrea-controller.
-func (c *ruleCache) processPodUpdate(pod string) {
-	namespace, name := k8s.SplitNamespacedName(pod)
+func (c *ruleCache) processPodUpdate(e interface{}) {
+	podEvent := e.(agenttypes.PodUpdate)
 	member := &v1beta.GroupMember{
 		Pod: &v1beta.PodReference{
-			Name:      name,
-			Namespace: namespace,
+			Name:      podEvent.PodName,
+			Namespace: podEvent.PodNamespace,
 		},
 	}
 	c.appliedToSetLock.RLock()

pkg/agent/controller/networkpolicy/cache_test.go

@@ -25,8 +25,10 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/sets"
 
+	"antrea.io/antrea/pkg/agent/types"
 	"antrea.io/antrea/pkg/apis/controlplane/v1beta2"
 	"antrea.io/antrea/pkg/util/channel"
+	"antrea.io/antrea/pkg/util/k8s"
 )
 
 var (
@@ -1184,7 +1186,12 @@ func TestRuleCacheProcessPodUpdates(t *testing.T) {
 			stopCh := make(chan struct{})
 			defer close(stopCh)
 			go podUpdateNotifier.Run(stopCh)
-			podUpdateNotifier.Notify(tt.podUpdate)
+			ns, name := k8s.SplitNamespacedName(tt.podUpdate)
+			e := types.PodUpdate{
+				PodNamespace: ns,
+				PodName:      name,
+			}
+			podUpdateNotifier.Notify(e)
 			func() {
 				// Drain the channel with 10 ms timeout so we can know it's done.
 				for {