Keep the cached Endpoints the same as those installed in OVS (#4691)

The main purpose of this PR is to avoid potential inconsistencies between
the cached Endpoints and those installed in OVS, like #4681, #4692.

This PR also updates:

- Method UninstallEndpointFlows of ofClient, support deleting flows of
  multiple Endpoints.
- Remove possible groups when a Service is deleted.
- Log something when a group for a Service is not created.
- Optimize and unify log.

Signed-off-by: Hongliang Liu <lhongliang@vmware.com>

pkg/agent/openflow/client.go

@@ -95,7 +95,7 @@ type Client interface {
 	InstallEndpointFlows(protocol binding.Protocol, endpoints []proxy.Endpoint) error
 	// UninstallEndpointFlows removes flows of the Endpoint installed by
 	// InstallEndpointFlows.
-	UninstallEndpointFlows(protocol binding.Protocol, endpoint proxy.Endpoint) error
+	UninstallEndpointFlows(protocol binding.Protocol, endpoints []proxy.Endpoint) error
 
 	// InstallServiceFlows installs flows for accessing Service NodePort, LoadBalancer and ClusterIP. It installs the
 	// flow that uses the group/bucket to do service LB. If the affinityTimeout is not zero, it also installs the flow
@@ -452,21 +452,36 @@ func (c *client) modifyFlows(cache *flowCategoryCache, flowCacheKey string, flow
 
 // deleteFlows deletes all the flows in the flow cache indexed by the provided flowCacheKey.
 func (c *client) deleteFlows(cache *flowCategoryCache, flowCacheKey string) error {
-	fCacheI, ok := cache.Load(flowCacheKey)
-	if !ok {
-		// no matching flows found in the cache
-		return nil
+	return c.deleteFlowsWithMultipleKeys(cache, []string{flowCacheKey})
+}
+
+// deleteFlowsWithMultipleKeys uninstalls the flows with different flowCache keys and remove them from the cache on success.
+// It will skip the keys which are not in the cache. All flows will be uninstalled via a bundle.
+func (c *client) deleteFlowsWithMultipleKeys(cache *flowCategoryCache, keys []string) error {
+	// allFlows keeps the flows we will delete via a bundle.
+	var allFlows []binding.Flow
+	for _, key := range keys {
+		flows, ok := cache.Load(key)
+		// If a flow cache entry of the key does not exist, skip it.
+		if !ok {
+			klog.V(2).InfoS("Cached flow with provided key was not found", "key", key)
+			continue
+		}
+		for _, flow := range flows.(flowCache) {
+			allFlows = append(allFlows, flow)
+		}
 	}
-	fCache := fCacheI.(flowCache)
-	// Delete flows from OVS.
-	delFlows := make([]binding.Flow, 0, len(fCache))
-	for _, flow := range fCache {
-		delFlows = append(delFlows, flow)
+	if len(allFlows) == 0 {
+		return nil
 	}
-	if err := c.ofEntryOperations.DeleteAll(delFlows); err != nil {
+	err := c.ofEntryOperations.DeleteAll(allFlows)
+	if err != nil {
 		return err
 	}
-	cache.Delete(flowCacheKey)
+	// Delete the keys and corresponding flows from the flow cache.
+	for _, key := range keys {
+		cache.Delete(key)
+	}
 	return nil
 }
 
@@ -684,16 +699,22 @@ func (c *client) InstallEndpointFlows(protocol binding.Protocol, endpoints []pro
 	return c.addFlowsWithMultipleKeys(c.featureService.cachedFlows, keyToFlows)
 }
 
-func (c *client) UninstallEndpointFlows(protocol binding.Protocol, endpoint proxy.Endpoint) error {
+func (c *client) UninstallEndpointFlows(protocol binding.Protocol, endpoints []proxy.Endpoint) error {
 	c.replayMutex.RLock()
 	defer c.replayMutex.RUnlock()
 
-	port, err := endpoint.Port()
-	if err != nil {
-		return fmt.Errorf("error when getting port: %w", err)
+	// keyToFlows is a map from the flows' cache key to the flows.
+	flowCacheKeys := make([]string, 0, len(endpoints))
+
+	for _, endpoint := range endpoints {
+		port, err := endpoint.Port()
+		if err != nil {
+			return fmt.Errorf("error when getting port: %w", err)
+		}
+		flowCacheKeys = append(flowCacheKeys, generateEndpointFlowCacheKey(endpoint.IP(), port, protocol))
 	}
-	cacheKey := generateEndpointFlowCacheKey(endpoint.IP(), port, protocol)
-	return c.deleteFlows(c.featureService.cachedFlows, cacheKey)
+
+	return c.deleteFlowsWithMultipleKeys(c.featureService.cachedFlows, flowCacheKeys)
 }
 
 func (c *client) InstallServiceFlows(groupID binding.GroupIDType, svcIP net.IP, svcPort uint16, protocol binding.Protocol, affinityTimeout uint16, nodeLocalExternal bool, svcType v1.ServiceType, nested bool) error {

pkg/agent/openflow/client_test.go

@@ -1037,7 +1037,7 @@ func Test_client_InstallEndpointFlows(t *testing.T) {
 			defer resetPipelines()
 
 			m.EXPECT().AddAll(gomock.Any()).Return(nil).Times(1)
-			m.EXPECT().DeleteAll(gomock.Any()).Return(nil).Times(len(tc.endpoints))
+			m.EXPECT().DeleteAll(gomock.Any()).Return(nil).Times(1)
 
 			assert.NoError(t, fc.InstallEndpointFlows(tc.protocol, tc.endpoints))
 			var flows []string
@@ -1050,8 +1050,8 @@ func Test_client_InstallEndpointFlows(t *testing.T) {
 			}
 			assert.ElementsMatch(t, tc.expectedFlows, flows)
 
+			assert.NoError(t, fc.UninstallEndpointFlows(tc.protocol, tc.endpoints))
 			for _, ep := range tc.endpoints {
-				assert.NoError(t, fc.UninstallEndpointFlows(tc.protocol, ep))
 				endpointPort, _ := ep.Port()
 				cacheKey := generateEndpointFlowCacheKey(ep.IP(), endpointPort, tc.protocol)
 				_, ok := fc.featureService.cachedFlows.Load(cacheKey)