Move flows installed by InstallMulticastInitialFlows to feature `in…

…itFlows` (#4393)

Unlike `InstallMulticastGroup` and `InstallMulticastRemoteReportFlows`, Flows installed by
`InstallMulticastInitialFlows` don't need any arguments provided by multicast controller,
and they can be installed by method `initFlows` of `featureMulticast` to unify the
installation of feature initial flows as possible.

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

pkg/agent/multicast/mcast_controller.go

@@ -340,16 +340,6 @@ func (c *Controller) Initialize() error {
 	if err != nil {
 		return err
 	}
-	// Install flows on OVS for IGMP packets and multicast traffic forwarding:
-	// 1) send the IGMP report messages to Antrea Agent,
-	// 2) forward the IGMP query messages to all local Pods,
-	// 3) forward the multicast traffic to antrea-gw0 if no local Pods have joined in the group, and this is to ensure
-	//    local Pods can access the external multicast receivers.
-	err = c.ofClient.InstallMulticastInitialFlows(uint8(openflow.PacketInReasonMC))
-	if err != nil {
-		klog.ErrorS(err, "Failed to install multicast initial flows")
-		return err
-	}
 	err = c.initQueryGroup()
 	if err != nil {
 		return err

pkg/agent/multicast/mcast_controller_test.go

@@ -775,7 +775,6 @@ func newMockMulticastController(t *testing.T, isEncap bool) *Controller {
 }
 
 func (c *Controller) initialize(t *testing.T) error {
-	mockOFClient.EXPECT().InstallMulticastInitialFlows(uint8(0)).Times(1)
 	mockOFClient.EXPECT().InstallMulticastGroup(c.queryGroupId, gomock.Any(), gomock.Any()).Times(1)
 	mockOFClient.EXPECT().InstallMulticastFlows(gomock.Any(), gomock.Any())
 	mockIfaceStore.EXPECT().GetInterfacesByType(interfacestore.InterfaceType(0)).Times(1).Return([]*interfacestore.InterfaceConfig{})

pkg/agent/openflow/client.go

@@ -278,9 +278,6 @@ type Client interface {
 	AddAddressToDNSConjunction(id uint32, addrs []types.Address) error
 	// DeleteAddressFromDNSConjunction removes addresses from the toAddresses of the dns packetIn conjunction.
 	DeleteAddressFromDNSConjunction(id uint32, addrs []types.Address) error
-	// InstallMulticastInitialFlows installs OpenFlow to packetIn the IGMP messages and output the Multicast traffic to
-	// antrea-gw0 so that local Pods could access external Multicast servers.
-	InstallMulticastInitialFlows(pktInReason uint8) error
 
 	// InstallMulticastFlows installs the flow to forward Multicast traffic normally, and output it to antrea-gw0
 	// to ensure it can be forwarded to the external addresses.
@@ -1217,19 +1214,6 @@ func (c *client) SendUDPPacketOut(
 	return c.bridge.SendPacketOut(packetOutObj)
 }
 
-func (c *client) InstallMulticastInitialFlows(pktInReason uint8) error {
-	flows := c.featureMulticast.igmpPktInFlows(pktInReason)
-	flows = append(flows, c.featureMulticast.externalMulticastReceiverFlow())
-	flows = append(flows, c.featureMulticast.multicastSkipIGMPMetricFlows()...)
-	if c.enableAntreaPolicy {
-		flows = append(flows, c.featureMulticast.igmpEgressFlow())
-	}
-	cacheKey := "multicast"
-	c.replayMutex.RLock()
-	defer c.replayMutex.RUnlock()
-	return c.addFlows(c.featureMulticast.cachedFlows, cacheKey, flows)
-}
-
 func (c *client) InstallMulticastFlows(multicastIP net.IP, groupID binding.GroupIDType) error {
 	flows := c.featureMulticast.localMulticastForwardFlows(multicastIP, groupID)
 	cacheKey := fmt.Sprintf("multicast_%s", multicastIP.String())

pkg/agent/openflow/multicast.go

@@ -73,8 +73,19 @@ func multicastPipelineClassifyFlow(cookieID uint64, pipeline binding.Pipeline) b
 }
 
 func (f *featureMulticast) initFlows() []binding.Flow {
-	cookieID := f.cookieAllocator.Request(f.category).Raw()
-	return f.multicastOutputFlows(cookieID)
+	// Install flows to send the IGMP report messages to Antrea Agent.
+	flows := f.igmpPktInFlows(uint8(PacketInReasonMC))
+	// Install flow to forward the IGMP query messages to all local Pods.
+	flows = append(flows, f.externalMulticastReceiverFlow())
+	// Install flows to forward the multicast traffic to antrea-gw0 if no local Pods have joined in the group, and this
+	// is to ensure local Pods can access the external multicast receivers.
+	flows = append(flows, f.multicastSkipIGMPMetricFlows()...)
+	if f.enableAntreaPolicy {
+		flows = append(flows, f.igmpEgressFlow())
+	}
+	// Install flows to output multicast packets.
+	flows = append(flows, f.multicastOutputFlows()...)
+	return flows
 }
 
 func (f *featureMulticast) replayFlows() []binding.Flow {
@@ -114,7 +125,8 @@ func (f *featureMulticast) multicastReceiversGroup(groupID binding.GroupIDType,
 	return nil
 }
 
-func (f *featureMulticast) multicastOutputFlows(cookieID uint64) []binding.Flow {
+func (f *featureMulticast) multicastOutputFlows() []binding.Flow {
+	cookieID := f.cookieAllocator.Request(f.category).Raw()
 	flows := []binding.Flow{
 		MulticastOutputTable.ofTable.BuildFlow(priorityNormal).
 			Cookie(cookieID).