Enhance bundle to support both group and flow modifications (antrea-i…

…o#559)

hack/update-codegen-dockerized.sh

@@ -55,7 +55,7 @@ $GOPATH/bin/openapi-gen  \
 MOCKGEN_TARGETS=(
   "pkg/agent/cniserver/ipam IPAMDriver"
   "pkg/agent/interfacestore InterfaceStore"
-  "pkg/agent/openflow Client,FlowOperations"
+  "pkg/agent/openflow Client,OFEntryOperations"
   "pkg/agent/route Interface"
   "pkg/ovs/openflow Bridge,Table,Flow,Action,FlowBuilder"
   "pkg/ovs/ovsconfig OVSBridgeClient"

pkg/agent/openflow/client.go

@@ -139,7 +139,7 @@ func (c *client) addFlows(cache *flowCategoryCache, flowCacheKey string, flows [
 		klog.V(2).Infof("Flows with cache key %s are already installed", flowCacheKey)
 		return nil
 	}
-	err := c.flowOperations.AddAll(flows)
+	err := c.ofEntryOperations.AddAll(flows)
 	if err != nil {
 		return err
 	}
@@ -165,7 +165,7 @@ func (c *client) deleteFlows(cache *flowCategoryCache, flowCacheKey string) erro
 	for _, flow := range fCache {
 		delFlows = append(delFlows, flow)
 	}
-	if err := c.flowOperations.DeleteAll(delFlows); err != nil {
+	if err := c.ofEntryOperations.DeleteAll(delFlows); err != nil {
 		return err
 	}
 	cache.Delete(flowCacheKey)
@@ -235,7 +235,7 @@ func (c *client) UninstallPodFlows(containerID string) error {
 
 func (c *client) InstallClusterServiceCIDRFlows(serviceNet *net.IPNet, gatewayMAC net.HardwareAddr, gatewayOFPort uint32) error {
 	flow := c.serviceCIDRDNATFlow(serviceNet, gatewayMAC, gatewayOFPort, cookie.Service)
-	if err := c.flowOperations.Add(flow); err != nil {
+	if err := c.ofEntryOperations.Add(flow); err != nil {
 		return err
 	}
 	c.clusterServiceCIDRFlows = []binding.Flow{flow}
@@ -261,7 +261,7 @@ func (c *client) InstallGatewayFlows(gatewayAddr net.IP, gatewayMAC net.Hardware
 		flows = append(flows, c.reEntranceBypassCTFlow(gatewayOFPort, gatewayOFPort, cookie.Default))
 	}
 
-	if err := c.flowOperations.AddAll(flows); err != nil {
+	if err := c.ofEntryOperations.AddAll(flows); err != nil {
 		return err
 	}
 	c.gatewayFlows = flows
@@ -270,32 +270,32 @@ func (c *client) InstallGatewayFlows(gatewayAddr net.IP, gatewayMAC net.Hardware
 
 func (c *client) InstallDefaultTunnelFlows(tunnelOFPort uint32) error {
 	flow := c.tunnelClassifierFlow(tunnelOFPort, cookie.Default)
-	if err := c.flowOperations.Add(flow); err != nil {
+	if err := c.ofEntryOperations.Add(flow); err != nil {
 		return err
 	}
 	c.defaultTunnelFlows = []binding.Flow{flow}
 	return nil
 }
 
 func (c *client) initialize() error {
-	if err := c.flowOperations.AddAll(c.defaultFlows()); err != nil {
+	if err := c.ofEntryOperations.AddAll(c.defaultFlows()); err != nil {
 		return fmt.Errorf("failed to install default flows: %v", err)
 	}
-	if err := c.flowOperations.Add(c.arpNormalFlow(cookie.Default)); err != nil {
+	if err := c.ofEntryOperations.Add(c.arpNormalFlow(cookie.Default)); err != nil {
 		return fmt.Errorf("failed to install arp normal flow: %v", err)
 	}
-	if err := c.flowOperations.Add(c.l2ForwardOutputFlow(cookie.Default)); err != nil {
+	if err := c.ofEntryOperations.Add(c.l2ForwardOutputFlow(cookie.Default)); err != nil {
 		return fmt.Errorf("failed to install L2 forward output flows: %v", err)
 	}
-	if err := c.flowOperations.AddAll(c.connectionTrackFlows(cookie.Default)); err != nil {
+	if err := c.ofEntryOperations.AddAll(c.connectionTrackFlows(cookie.Default)); err != nil {
 		return fmt.Errorf("failed to install connection track flows: %v", err)
 	}
-	if err := c.flowOperations.AddAll(c.establishedConnectionFlows(cookie.Default)); err != nil {
+	if err := c.ofEntryOperations.AddAll(c.establishedConnectionFlows(cookie.Default)); err != nil {
 		return fmt.Errorf("failed to install flows to skip established connections: %v", err)
 	}
 
 	if c.encapMode.SupportsNoEncap() {
-		if err := c.flowOperations.Add(c.l2ForwardOutputReentInPortFlow(c.gatewayPort, cookie.Default)); err != nil {
+		if err := c.ofEntryOperations.Add(c.l2ForwardOutputReentInPortFlow(c.gatewayPort, cookie.Default)); err != nil {
 			return fmt.Errorf("failed to install L2 forward same in-port and out-port flow: %v", err)
 		}
 	}
@@ -347,7 +347,7 @@ func (c *client) ReplayFlows() {
 		for _, flow := range flows {
 			flow.Reset()
 		}
-		if err := c.flowOperations.AddAll(flows); err != nil {
+		if err := c.ofEntryOperations.AddAll(flows); err != nil {
 			klog.Errorf("Error when replaying fixed flows: %v", err)
 		}
 
@@ -366,7 +366,7 @@ func (c *client) ReplayFlows() {
 			cachedFlows = append(cachedFlows, flow)
 		}
 
-		if err := c.flowOperations.AddAll(cachedFlows); err != nil {
+		if err := c.ofEntryOperations.AddAll(cachedFlows); err != nil {
 			klog.Errorf("Error when replaying cached flows: %v", err)
 		}
 		return true
@@ -400,7 +400,7 @@ func (c *client) setupPolicyOnlyFlows() error {
 		// Replies any ARP request with the same global virtual MAC.
 		c.arpResponderStaticFlow(cookie.Default),
 	}
-	if err := c.flowOperations.AddAll(flows); err != nil {
+	if err := c.ofEntryOperations.AddAll(flows); err != nil {
 		return fmt.Errorf("failed to setup policy-only flows: %w", err)
 	}
 	return nil

pkg/agent/openflow/client_test.go

@@ -83,12 +83,12 @@ func TestIdempotentFlowInstallation(t *testing.T) {
 		t.Run(tc.name, func(t *testing.T) {
 			ctrl := gomock.NewController(t)
 			defer ctrl.Finish()
-			m := oftest.NewMockFlowOperations(ctrl)
+			m := oftest.NewMockOFEntryOperations(ctrl)
 			ofClient := NewClient(bridgeName)
 			client := ofClient.(*client)
 			client.cookieAllocator = cookie.NewAllocator(0)
 			client.nodeConfig = &config.NodeConfig{}
-			client.flowOperations = m
+			client.ofEntryOperations = m
 
 			m.EXPECT().AddAll(gomock.Any()).Return(nil).Times(1)
 			// Installing the flows should succeed, and all the flows should be added into the cache.
@@ -111,12 +111,12 @@ func TestIdempotentFlowInstallation(t *testing.T) {
 		t.Run(tc.name, func(t *testing.T) {
 			ctrl := gomock.NewController(t)
 			defer ctrl.Finish()
-			m := oftest.NewMockFlowOperations(ctrl)
+			m := oftest.NewMockOFEntryOperations(ctrl)
 			ofClient := NewClient(bridgeName)
 			client := ofClient.(*client)
 			client.cookieAllocator = cookie.NewAllocator(0)
 			client.nodeConfig = &config.NodeConfig{}
-			client.flowOperations = m
+			client.ofEntryOperations = m
 
 			errorCall := m.EXPECT().AddAll(gomock.Any()).Return(errors.New("Bundle error")).Times(1)
 			m.EXPECT().AddAll(gomock.Any()).Return(nil).After(errorCall)
@@ -152,12 +152,12 @@ func TestFlowInstallationFailed(t *testing.T) {
 		t.Run(tc.name, func(t *testing.T) {
 			ctrl := gomock.NewController(t)
 			defer ctrl.Finish()
-			m := oftest.NewMockFlowOperations(ctrl)
+			m := oftest.NewMockOFEntryOperations(ctrl)
 			ofClient := NewClient(bridgeName)
 			client := ofClient.(*client)
 			client.cookieAllocator = cookie.NewAllocator(0)
 			client.nodeConfig = &config.NodeConfig{}
-			client.flowOperations = m
+			client.ofEntryOperations = m
 
 			// We generate an error for AddAll call.
 			m.EXPECT().AddAll(gomock.Any()).Return(errors.New("Bundle error"))
@@ -186,12 +186,12 @@ func TestConcurrentFlowInstallation(t *testing.T) {
 		t.Run(tc.name, func(t *testing.T) {
 			ctrl := gomock.NewController(t)
 			defer ctrl.Finish()
-			m := oftest.NewMockFlowOperations(ctrl)
+			m := oftest.NewMockOFEntryOperations(ctrl)
 			ofClient := NewClient(bridgeName)
 			client := ofClient.(*client)
 			client.cookieAllocator = cookie.NewAllocator(0)
 			client.nodeConfig = &config.NodeConfig{}
-			client.flowOperations = m
+			client.ofEntryOperations = m
 
 			var concurrentCalls atomic.Value // set to true if we observe concurrent calls
 			timeoutCh := make(chan struct{})

pkg/agent/openflow/network_policy.go

@@ -730,7 +730,7 @@ func (c *client) InstallPolicyRuleFlows(ruleID uint32, rule *types.PolicyRule) e
 				actionFlows = append(actionFlows, flow)
 			}
 		}
-		if err := c.flowOperations.AddAll(actionFlows); err != nil {
+		if err := c.ofEntryOperations.AddAll(actionFlows); err != nil {
 			return nil
 		}
 		// Add the action flows after the Openflow entries are installed on the OVS bridge successfully.
@@ -909,7 +909,7 @@ func (c *client) UninstallPolicyRuleFlows(ruleID uint32) error {
 	}
 
 	// Delete action flows from the OVS bridge.
-	if err := c.flowOperations.DeleteAll(conj.actionFlows); err != nil {
+	if err := c.ofEntryOperations.DeleteAll(conj.actionFlows); err != nil {
 		return err
 	}
 
@@ -955,7 +955,7 @@ func (c *client) replayPolicyFlows() {
 	for _, ctx := range c.globalConjMatchFlowCache {
 		addMatchFlows(ctx)
 	}
-	if err := c.flowOperations.AddAll(flows); err != nil {
+	if err := c.ofEntryOperations.AddAll(flows); err != nil {
 		klog.Errorf("Error when replaying flows: %v", err)
 	}
 }

pkg/agent/openflow/network_policy_test.go

@@ -415,9 +415,9 @@ func prepareClient(ctrl *gomock.Controller) *client {
 		bridge:                   bridge,
 	}
 	c.cookieAllocator = cookie.NewAllocator(0)
-	m := oftest.NewMockFlowOperations(ctrl)
+	m := oftest.NewMockOFEntryOperations(ctrl)
 	m.EXPECT().AddAll(gomock.Any()).Return(nil).AnyTimes()
 	m.EXPECT().DeleteAll(gomock.Any()).Return(nil).AnyTimes()
-	c.flowOperations = m
+	c.ofEntryOperations = m
 	return c
 }

pkg/agent/openflow/pipeline.go

@@ -96,12 +96,14 @@ var (
 	ReentranceMAC, _    = net.ParseMAC("de:ad:be:ef:de:ad")
 )
 
-type FlowOperations interface {
+type OFEntryOperations interface {
 	Add(flow binding.Flow) error
 	Modify(flow binding.Flow) error
 	Delete(flow binding.Flow) error
 	AddAll(flows []binding.Flow) error
 	DeleteAll(flows []binding.Flow) error
+	AddOFEntries(ofEntries []binding.OFEntry) error
+	DeleteOFEntries(ofEntries []binding.OFEntry) error
 }
 
 type flowCache map[string]binding.Flow
@@ -119,9 +121,9 @@ type client struct {
 	// "fixed" flows installed by the agent after initialization and which do not change during
 	// the lifetime of the client.
 	gatewayFlows, clusterServiceCIDRFlows, defaultTunnelFlows []binding.Flow
-	// flowOperations is a wrapper interface for flow Add / Modify / Delete operations. It
+	// ofEntryOperations is a wrapper interface for OpenFlow entry Add / Modify / Delete operations. It
 	// enables convenient mocking in unit tests.
-	flowOperations FlowOperations
+	ofEntryOperations OFEntryOperations
 	// policyCache is a map from PolicyRule ID to policyRuleConjunction. It's guaranteed that one policyRuleConjunction
 	// is processed by at most one goroutine at any given time.
 	policyCache       sync.Map
@@ -156,6 +158,14 @@ func (c *client) DeleteAll(flows []binding.Flow) error {
 	return c.bridge.AddFlowsInBundle(nil, nil, flows)
 }
 
+func (c *client) AddOFEntries(ofEntries []binding.OFEntry) error {
+	return c.bridge.AddOFEntriesInBundle(ofEntries, nil, nil)
+}
+
+func (c *client) DeleteOFEntries(ofEntries []binding.OFEntry) error {
+	return c.bridge.AddOFEntriesInBundle(nil, nil, ofEntries)
+}
+
 // defaultFlows generates the default flows of all tables.
 func (c *client) defaultFlows() (flows []binding.Flow) {
 	for _, table := range c.pipeline {
@@ -650,6 +660,6 @@ func NewClient(bridgeName string) Client {
 		policyCache:              sync.Map{},
 		globalConjMatchFlowCache: map[string]*conjMatchFlowContext{},
 	}
-	c.flowOperations = c
+	c.ofEntryOperations = c
 	return c
 }