Consolidate several L4 OVN ACLs into smaller number of OVN ACLs

If an ingress rule has 5 TCP port specified along with a namespace
selector, then we end up creating 5 OVN ACLs. Instead, we could create
just one ACL by consoldiating the individual ports and/or ranges. For
example:

With this Network Policy,

spec:
  ingress:
    - from:
      - nameSelector:
          matchLabels:
            name: web
      ports:
        - protocol: TCP
          port: 3306
        - protocol: TCP
          port: 80
        - protocol: TCP
          port: 9000
          endPort: 9100

we end up creating following 3 matches
(ip4.src == {$as1} && tcp && tcp.dst==3306 && outport == @pg1)
(ip4.src == {$as1} && tcp && tcp.dst==80 && outport == @pg1)
(ip4.src == {$as1} && tcp && 9000<=tcp.dst<=9100 && outport == @pg1)

With the fix in this commit, we will have only one match
(ip4.src == {$as1} && tcp && (tcp.dst=={3306,80} || \
  9000<=tcp.dst<=9100 && outport == @pg1)

With reduced number of OVN ACLs will result in fewer logical flows
and few OpenFlow flows

Signed-off-by: Xiaobin Qu <[email protected]>
Co-authored-by: Girish Moodalbail <[email protected]>

go-controller/pkg/libovsdbops/db_object_types.go

@@ -34,6 +34,7 @@ const (
 	PortPolicyIndexKey    ExternalIDKey = "port-policy-index"
 	IpBlockIndexKey       ExternalIDKey = "ip-block-index"
 	RuleIndex             ExternalIDKey = "rule-index"
+	MatchHashKey          ExternalIDKey = "match-hash"
 )
 
 // ObjectIDsTypes should only be created here

go-controller/pkg/ovn/base_network_controller_policy.go

@@ -1030,6 +1030,11 @@ func (bnc *BaseNetworkController) createNetworkPolicy(policy *knet.NetworkPolicy
 			return fmt.Errorf("failed to start local pod handler: %v", err)
 		}
 
+		// 8. Clean up non-hashed ACLs for this policy
+		err = bnc.cleanupNonHashedACLs(np)
+		if err != nil {
+			return fmt.Errorf("failed to clean up non-hashed ACLs for %s/%s: %v", np.namespace, np.name, err)
+		}
 		return nil
 	})
 	return np, err
@@ -1456,6 +1461,27 @@ func (bnc *BaseNetworkController) shutdownHandlers(np *networkPolicy) {
 	np.nsHandlerList = make([]*factory.Handler, 0)
 }
 
+func (bnc *BaseNetworkController) cleanupNonHashedACLs(np *networkPolicy) error {
+	predicateForNonHashedACL := func(item *nbdb.ACL) bool {
+		if item.GetExternalIDs()[libovsdbops.OwnerControllerKey.String()] != bnc.controllerName ||
+			item.GetExternalIDs()[libovsdbops.ObjectNameKey.String()] != getACLPolicyKey(np.namespace, np.name) {
+			return false
+		}
+		// non-hashed ACLs don't have hash in external ids
+		return item.GetExternalIDs()[libovsdbops.MatchHashKey.String()] == ""
+	}
+	nonHashedACLs, err := libovsdbops.FindACLsWithPredicate(bnc.nbClient, predicateForNonHashedACL)
+	if err != nil {
+		return err
+	}
+	ops, err := libovsdbops.DeleteACLsFromPortGroupOps(bnc.nbClient, nil, np.portGroupName, nonHashedACLs...)
+	if err != nil {
+		return err
+	}
+	_, err = libovsdbops.TransactAndCheck(bnc.nbClient, ops)
+	return err
+}
+
 // The following 2 functions should return the same key for network policy based on k8s on internal networkPolicy object
 func getPolicyKey(policy *knet.NetworkPolicy) string {
 	return fmt.Sprintf("%v/%v", policy.Namespace, policy.Name)

go-controller/pkg/ovn/default_network_controller.go

@@ -439,6 +439,9 @@ func (oc *DefaultNetworkController) Run(ctx context.Context) error {
 	if err := WithSyncDurationMetric("network policy", oc.WatchNetworkPolicy); err != nil {
 		return err
 	}
+	// now that we have added all the OVN ACLs with optimization, it is time to remove the stale OVN
+	// ACL entries from the database
+	aclsyncer.NewACLSyncer(oc.nbClient, oc.controllerName).CleanStaleNetworkPolicy()
 
 	if config.OVNKubernetesFeature.EnableEgressIP {
 		// This is probably the best starting order for all egress IP handlers.

go-controller/pkg/ovn/external_ids_syncer/acl/acl_sync.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"strconv"
 	"strings"
+	"time"
 
 	libovsdbclient "github.com/ovn-org/libovsdb/client"
 	libovsdb "github.com/ovn-org/libovsdb/ovsdb"
@@ -438,3 +439,49 @@ func (syncer *aclSyncer) updateStaleEgressFirewallACLs(legacyACLs []*nbdb.ACL) [
 	}
 	return updatedACLs
 }
+
+// CleanStaleNetworkPolicy deletes ACLs having l4Match in external-ids (not None)
+func (syncer *aclSyncer) CleanStaleNetworkPolicy() {
+	start := time.Now()
+	defer func() {
+		klog.V(5).Infof("Completed cleaning up stale OVN ACLs in %v", time.Since(start))
+	}()
+
+	klog.V(5).Infof("Cleaning up stale OVN ACLs that are left behind after L4 Port consolidation")
+	// want ACLs that don't have l4fused key and have l4Match set (but not to None)
+	pACL := func(item *nbdb.ACL) bool {
+		aclType := item.ExternalIDs[libovsdbops.OwnerTypeKey.String()]
+		ownerName := item.ExternalIDs[libovsdbops.ObjectNameKey.String()]
+		_, hashExists := item.ExternalIDs[libovsdbops.MatchHashKey.String()]
+		return aclType == string(libovsdbops.NetworkPolicyOwnerType) && ownerName != "" && !hashExists
+	}
+
+	staleACLs, err := libovsdbops.FindACLsWithPredicate(syncer.nbClient, pACL)
+	if err != nil {
+		klog.Warningf("Failed to retrieve stale OVN ACL entries that were not optimized " +
+			"for L4 Ports consolidation: %v, err")
+		return
+	}
+	// it could be that delete all the acls in one go might fail for various reasons,
+	// so lets try to delete one at a time so that we can remove as many stale acls
+	// as possible.
+	klog.V(5).Infof("Number of stale ACLS to be cleaned is %d", len(staleACLs))
+	for _, staleACL := range staleACLs {
+		staleACL := staleACL
+		nsName := staleACL.ExternalIDs[namespaceACLExtIdKey]
+		policyName := staleACL.ExternalIDs[policyACLExtIdKey]
+		pgName := fmt.Sprintf("%s_%s", nsName, policyName)
+		pgName = util.HashForOVN(pgName)
+		aclDesc := fmt.Sprintf("stale ACL %s/%s/%s in port group %s", staleACL.UUID, nsName, policyName, pgName)
+		klog.V(5).Infof("About to delete %s", aclDesc)
+		ops, err := libovsdbops.DeleteACLsFromPortGroupOps(syncer.nbClient, nil, pgName, staleACL)
+		if err != nil {
+			klog.Warningf("Failed to get ops to delete %s: %v", aclDesc, err)
+			continue
+		}
+		_, err = libovsdbops.TransactAndCheck(syncer.nbClient, ops)
+		if err != nil {
+			klog.Warningf("Failed to delete %s: %v", aclDesc, err)
+		}
+	}
+}

go-controller/pkg/ovn/gress_policy.go

@@ -14,6 +14,7 @@ import (
 	"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/util"
 
 	knet "k8s.io/api/networking/v1"
+	"k8s.io/klog/v2"
 	utilnet "k8s.io/utils/net"
 )
 
@@ -62,25 +63,65 @@ type portPolicy struct {
 	endPort  int32
 }
 
-func (pp *portPolicy) getL4Match() (string, error) {
-	var supportedProtocols = []string{TCP, UDP, SCTP}
-	var foundProtocol string
-	for _, protocol := range supportedProtocols {
-		if protocol == pp.protocol {
-			foundProtocol = strings.ToLower(pp.protocol)
-			break
+// for a given ingress/egress rule, captures all the provided port ranges and
+// individual ports
+type gressPolicyPorts struct {
+	portList  []string // list of provided ports as string
+	portRange []string // list of provided port ranges in OVN ACL format
+}
+
+var supportedProtocols = []string{TCP, UDP, SCTP}
+
+func (gp *gressPolicy) getProtocolPortsMap() map[string]*gressPolicyPorts {
+	gressProtoPortsMap := make(map[string]*gressPolicyPorts)
+	for _, pp := range gp.portPolicies {
+		var found bool
+		for _, protocol := range supportedProtocols {
+			if protocol == pp.protocol {
+				found = true
+				break
+			}
+		}
+		if !found {
+			klog.Warningf("Unknown protocol %v, while processing network policy %s/%s",
+				pp.protocol, gp.policyNamespace, gp.policyName)
+			continue
+		}
+		protocol := strings.ToLower(pp.protocol)
+		gpp, ok := gressProtoPortsMap[protocol]
+		if !ok {
+			gpp = &gressPolicyPorts{portList: []string{}, portRange: []string{}}
+			gressProtoPortsMap[protocol] = gpp
+		}
+		if pp.endPort != 0 && pp.endPort != pp.port {
+			gpp.portRange = append(gpp.portRange, fmt.Sprintf("%d<=%s.dst<=%d", pp.port, protocol, pp.endPort))
+		} else if pp.port != 0 {
+			gpp.portList = append(gpp.portList, fmt.Sprintf("%d", pp.port))
 		}
 	}
-	if len(foundProtocol) == 0 {
-		return "", fmt.Errorf("unknown port protocol %v", pp.protocol)
-	}
-	if pp.endPort != 0 && pp.endPort != pp.port {
-		return fmt.Sprintf("%s && %d<=%s.dst<=%d", foundProtocol, pp.port, foundProtocol, pp.endPort), nil
+	return gressProtoPortsMap
+}
 
-	} else if pp.port != 0 {
-		return fmt.Sprintf("%s && %s.dst==%d", foundProtocol, foundProtocol, pp.port), nil
+func getL4Match(protocol string, ports *gressPolicyPorts) string {
+	allL4Matches := []string{}
+	if len(ports.portList) > 0 {
+		// if there is just one port, then don't use `{}`
+		template := "%s.dst==%s"
+		if len(ports.portList) > 1 {
+			template = "%s.dst=={%s}"
+		}
+		allL4Matches = append(allL4Matches, fmt.Sprintf(template, protocol, strings.Join(ports.portList, ",")))
 	}
-	return foundProtocol, nil
+	allL4Matches = append(allL4Matches, ports.portRange...)
+	l4Match := protocol
+	if len(allL4Matches) > 0 {
+		template := "%s && %s"
+		if len(allL4Matches) > 1 {
+			template = "%s && (%s)"
+		}
+		l4Match = fmt.Sprintf(template, protocol, strings.Join(allL4Matches, " || "))
+	}
+	return l4Match
 }
 
 func newGressPolicy(policyType knet.PolicyType, idx int, namespace, name, controllerName string, isNetPolStateless bool, netConfInfo util.NetConfInfo) *gressPolicy {
@@ -295,30 +336,21 @@ func (gp *gressPolicy) buildLocalPodACLs(portGroupName string, aclLogging *ACLLo
 	} else {
 		lportMatch = fmt.Sprintf("inport == @%s", portGroupName)
 	}
-	var portPolicyIdxs []int
-	if len(gp.portPolicies) == 0 {
-		portPolicyIdxs = []int{emptyIdx}
-	} else {
-		portPolicyIdxs = make([]int, 0, len(gp.portPolicies))
-		for i := 0; i < len(gp.portPolicies); i++ {
-			portPolicyIdxs = append(portPolicyIdxs, i)
-		}
-	}
 	var l4Match string
-	var err error
 	action := nbdb.ACLActionAllowRelated
 	if gp.isNetPolStateless {
 		action = nbdb.ACLActionAllowStateless
 	}
-	for _, portPolIdx := range portPolicyIdxs {
-		if portPolIdx != emptyIdx {
-			portPol := gp.portPolicies[portPolIdx]
-			l4Match, err = portPol.getL4Match()
-			if err != nil {
-				continue
-			}
-		} else {
-			l4Match = noneMatch
+	protocolPortsMap := gp.getProtocolPortsMap()
+	portPolIdx := 0
+	if len(protocolPortsMap) == 0 {
+		protocolPortsMap[noneMatch] = nil
+		portPolIdx = -1
+	}
+	for protocol, ports := range protocolPortsMap {
+		l4Match = noneMatch
+		if ports != nil {
+			l4Match = getL4Match(protocol, ports)
 		}
 
 		if len(gp.ipBlocks) > 0 {
@@ -328,6 +360,8 @@ func (gp *gressPolicy) buildLocalPodACLs(portGroupName string, aclLogging *ACLLo
 				aclIDs := gp.getNetpolACLDbIDs(portPolIdx, ipBlockIdx)
 				acl := BuildACL(aclIDs, types.DefaultAllowPriority, ipBlockMatch, action,
 					aclLogging, gp.aclPipeline)
+				// add hash info to indicate that this ACL has been possibly consolidated
+				addMatchHash(acl)
 				createdACLs = append(createdACLs, acl)
 			}
 		}
@@ -349,6 +383,8 @@ func (gp *gressPolicy) buildLocalPodACLs(portGroupName string, aclLogging *ACLLo
 			aclIDs := gp.getNetpolACLDbIDs(portPolIdx, emptyIdx)
 			acl := BuildACL(aclIDs, types.DefaultAllowPriority, addrSetMatch, action,
 				aclLogging, gp.aclPipeline)
+			// add hash info to indicate that this ACL has been possibly consolidated
+			addMatchHash(acl)
 			if l3Match == "" {
 				// if l3Match is empty, then no address sets are selected for a given gressPolicy.
 				// fortunately l3 match is not a part of externalIDs, that means that we can find
@@ -358,6 +394,7 @@ func (gp *gressPolicy) buildLocalPodACLs(portGroupName string, aclLogging *ACLLo
 				createdACLs = append(createdACLs, acl)
 			}
 		}
+		portPolIdx++
 	}
 	return
 }
@@ -395,3 +432,10 @@ func (gp *gressPolicy) getNetpolACLDbIDs(portPolicyIdx, ipBlockIdx int) *libovsd
 			libovsdbops.IpBlockIndexKey:    strconv.Itoa(ipBlockIdx),
 		})
 }
+
+func addMatchHash(acl *nbdb.ACL) {
+	if acl == nil {
+		return
+	}
+	acl.ExternalIDs[libovsdbops.MatchHashKey.String()] = util.HashForOVN(acl.Match)
+}