admin_network_policy.go

package controller

import (
	"errors"
	"fmt"
	"net"
	"reflect"
	"strings"
	"unicode"

	"github.com/scylladb/go-set/strset"
	k8serrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
	"k8s.io/client-go/tools/cache"
	"k8s.io/klog/v2"
	v1alpha1 "sigs.k8s.io/network-policy-api/apis/v1alpha1"

	kubeovnv1 "github.com/kubeovn/kube-ovn/pkg/apis/kubeovn/v1"
	"github.com/kubeovn/kube-ovn/pkg/ovs"
	"github.com/kubeovn/kube-ovn/pkg/ovsdb/ovnnb"
	"github.com/kubeovn/kube-ovn/pkg/util"
)

type ChangedField string

const (
	ChangedSubject     ChangedField = "Subject"
	ChangedIngressRule ChangedField = "IngressRule"
	ChangedEgressRule  ChangedField = "EgressRule"
)

type ChangedName struct {
	// the rule name can be omitted default, add isMatch to append check for rule update
	isMatch     bool
	oldRuleName string
	curRuleName string
}

type AdminNetworkPolicyChangedDelta struct {
	key       string
	ruleNames [util.AnpMaxRules]ChangedName
	field     ChangedField
}

func (c *Controller) enqueueAddAnp(obj interface{}) {
	var key string
	var err error
	if key, err = cache.MetaNamespaceKeyFunc(obj); err != nil {
		utilruntime.HandleError(err)
		return
	}
	klog.V(3).Infof("enqueue add anp %s", key)
	c.addAnpQueue.Add(key)
}

func (c *Controller) enqueueDeleteAnp(obj interface{}) {
	anp := obj.(*v1alpha1.AdminNetworkPolicy)
	klog.V(3).Infof("enqueue delete anp %s", anp.Name)
	c.deleteAnpQueue.Add(anp)
}

func (c *Controller) enqueueUpdateAnp(oldObj, newObj interface{}) {
	oldAnpObj := oldObj.(*v1alpha1.AdminNetworkPolicy)
	newAnpObj := newObj.(*v1alpha1.AdminNetworkPolicy)

	// All the acls should be recreated with the following situations
	if oldAnpObj.Spec.Priority != newAnpObj.Spec.Priority || len(oldAnpObj.Spec.Ingress) != len(newAnpObj.Spec.Ingress) || len(oldAnpObj.Spec.Egress) != len(newAnpObj.Spec.Egress) {
		c.addAnpQueue.Add(newAnpObj.Name)
		return
	}

	// Acls should be updated when action or ports of ingress/egress rule has been changed
	for index, rule := range newAnpObj.Spec.Ingress {
		oldRule := oldAnpObj.Spec.Ingress[index]
		if oldRule.Action != rule.Action || !reflect.DeepEqual(oldRule.Ports, rule.Ports) {
			// It's difficult to distinguish which rule has changed and update acls for that rule, so go through the anp add process to recreate acls.
			// If we want to get fine-grained changes over rule, maybe it's a better way to add a new queue to process the change
			c.addAnpQueue.Add(newAnpObj.Name)
			return
		}
	}

	for index, rule := range newAnpObj.Spec.Egress {
		oldRule := oldAnpObj.Spec.Egress[index]
		if oldRule.Action != rule.Action || !reflect.DeepEqual(oldRule.Ports, rule.Ports) {
			c.addAnpQueue.Add(newAnpObj.Name)
			return
		}
	}

	if oldAnpObj.Annotations[util.ACLActionsLogAnnotation] != newAnpObj.Annotations[util.ACLActionsLogAnnotation] {
		c.addAnpQueue.Add(newAnpObj.Name)
		return
	}
	klog.V(3).Infof("enqueue update anp %s", newAnpObj.Name)

	// The remaining changes do not affect the acls. The port-group or address-set should be updated.
	// The port-group for anp should be updated
	if !reflect.DeepEqual(oldAnpObj.Spec.Subject, newAnpObj.Spec.Subject) {
		c.updateAnpQueue.Add(&AdminNetworkPolicyChangedDelta{key: newAnpObj.Name, field: ChangedSubject})
	}

	// Rule name or peer selector in ingress/egress rule has changed, the corresponding address-set need be updated
	ruleChanged := false
	var changedIngressRuleNames, changedEgressRuleNames [util.AnpMaxRules]ChangedName
	for index, rule := range newAnpObj.Spec.Ingress {
		oldRule := oldAnpObj.Spec.Ingress[index]
		if oldRule.Name != rule.Name {
			changedIngressRuleNames[index] = ChangedName{oldRuleName: oldRule.Name, curRuleName: rule.Name}
			ruleChanged = true
		}
		if !reflect.DeepEqual(oldRule.From, rule.From) {
			changedIngressRuleNames[index] = ChangedName{curRuleName: rule.Name}
			ruleChanged = true
		}
	}
	if ruleChanged {
		c.updateAnpQueue.Add(&AdminNetworkPolicyChangedDelta{key: newAnpObj.Name, ruleNames: changedIngressRuleNames, field: ChangedIngressRule})
	}

	ruleChanged = false
	for index, rule := range newAnpObj.Spec.Egress {
		oldRule := oldAnpObj.Spec.Egress[index]
		if oldRule.Name != rule.Name {
			changedEgressRuleNames[index] = ChangedName{oldRuleName: oldRule.Name, curRuleName: rule.Name}
			ruleChanged = true
		}
		if !reflect.DeepEqual(oldRule.To, rule.To) {
			changedEgressRuleNames[index] = ChangedName{curRuleName: rule.Name}
			ruleChanged = true
		}
	}
	if ruleChanged {
		c.updateAnpQueue.Add(&AdminNetworkPolicyChangedDelta{key: newAnpObj.Name, ruleNames: changedEgressRuleNames, field: ChangedEgressRule})
	}
}

func (c *Controller) handleAddAnp(key string) (err error) {
	c.anpKeyMutex.LockKey(key)
	defer func() { _ = c.anpKeyMutex.UnlockKey(key) }()

	cachedAnp, err := c.anpsLister.Get(key)
	if err != nil {
		if k8serrors.IsNotFound(err) {
			return nil
		}
		klog.Error(err)
		return err
	}
	klog.Infof("handle add anp %s", cachedAnp.Name)
	anp := cachedAnp.DeepCopy()

	if err := c.validateAnpConfig(anp); err != nil {
		klog.Errorf("failed to validate anp %s: %v", anp.Name, err)
		return err
	}
	if priority, exist := c.anpNamePrioMap[anp.Name]; exist && priority != anp.Spec.Priority {
		// anp spec's priority has been changed
		delete(c.anpPrioNameMap, priority)
	}
	// record new created anp after validation
	c.anpPrioNameMap[anp.Spec.Priority] = anp.Name
	c.anpNamePrioMap[anp.Name] = anp.Spec.Priority

	anpName := getAnpName(anp.Name)
	var logActions []string
	if anp.Annotations[util.ACLActionsLogAnnotation] != "" {
		logActions = strings.Split(anp.Annotations[util.ACLActionsLogAnnotation], ",")
	}

	// ovn portGroup/addressSet doesn't support name with '-', so we replace '-' by '.'.
	// This may cause conflict if two anp with name test-anp and test.anp, maybe hash is a better solution, but we do not want to lost the readability now.
	// Make sure all create operations are reentrant.
	pgName := strings.ReplaceAll(anpName, "-", ".")
	if err = c.OVNNbClient.CreatePortGroup(pgName, map[string]string{adminNetworkPolicyKey: anpName}); err != nil {
		klog.Errorf("failed to create port group for anp %s: %v", key, err)
		return err
	}

	ports, err := c.fetchSelectedPods(&anp.Spec.Subject)
	if err != nil {
		klog.Errorf("failed to fetch ports belongs to anp %s: %v", key, err)
		return err
	}

	if err = c.OVNNbClient.PortGroupSetPorts(pgName, ports); err != nil {
		klog.Errorf("failed to set ports %v to port group %s: %v", ports, pgName, err)
		return err
	}

	ingressACLOps, err := c.OVNNbClient.DeleteAclsOps(pgName, portGroupKey, "to-lport", nil, util.NilACLTier)
	if err != nil {
		klog.Errorf("failed to generate clear operations for anp %s ingress acls: %v", key, err)
		return err
	}

	curIngressAddrSet, curEgressAddrSet, err := c.getCurrentAddrSetByName(anpName, false)
	if err != nil {
		klog.Errorf("failed to list address sets for anp %s: %v", key, err)
		return err
	}
	desiredIngressAddrSet := strset.NewWithSize(len(anp.Spec.Ingress) * 2)
	desiredEgressAddrSet := strset.NewWithSize(len(anp.Spec.Egress) * 2)

	// create ingress acl
	for index, anpr := range anp.Spec.Ingress {
		// A single address set must contain addresses of the same type and the name must be unique within table, so IPv4 and IPv6 address set should be different
		ingressAsV4Name, ingressAsV6Name := getAnpAddressSetName(pgName, anpr.Name, index, true)
		desiredIngressAddrSet.Add(ingressAsV4Name, ingressAsV6Name)

		var v4Addrs, v4Addr, v6Addrs, v6Addr []string
		// This field must be defined and contain at least one item.
		for _, anprpeer := range anpr.From {
			if v4Addr, v6Addr, err = c.fetchIngressSelectedAddresses(&anprpeer); err != nil {
				klog.Errorf("failed to fetch admin network policy selected addresses, %v", err)
				return err
			}
			v4Addrs = append(v4Addrs, v4Addr...)
			v6Addrs = append(v6Addrs, v6Addr...)
		}
		klog.Infof("anp %s, ingress rule %s, selected v4 address %v, v6 address %v", anpName, anpr.Name, v4Addrs, v6Addrs)

		if err = c.createAsForAnpRule(anpName, anpr.Name, "ingress", ingressAsV4Name, v4Addrs, false); err != nil {
			klog.Error(err)
			return err
		}
		if err = c.createAsForAnpRule(anpName, anpr.Name, "ingress", ingressAsV6Name, v6Addrs, false); err != nil {
			klog.Error(err)
			return err
		}

		aclPriority := util.AnpACLMaxPriority - int(anp.Spec.Priority*100) - index
		aclAction := convertAction(anpr.Action, "")
		rulePorts := []v1alpha1.AdminNetworkPolicyPort{}
		if anpr.Ports != nil {
			rulePorts = *anpr.Ports
		}

		if len(v4Addrs) != 0 {
			aclName := fmt.Sprintf("anp/%s/ingress/%s/%d", anpName, kubeovnv1.ProtocolIPv4, index)
			ops, err := c.OVNNbClient.UpdateAnpRuleACLOps(pgName, ingressAsV4Name, kubeovnv1.ProtocolIPv4, aclName, aclPriority, aclAction, logActions, rulePorts, true, false)
			if err != nil {
				klog.Errorf("failed to add v4 ingress acls for anp %s: %v", key, err)
				return err
			}
			ingressACLOps = append(ingressACLOps, ops...)
		}

		if len(v6Addrs) != 0 {
			aclName := fmt.Sprintf("anp/%s/ingress/%s/%d", anpName, kubeovnv1.ProtocolIPv6, index)
			ops, err := c.OVNNbClient.UpdateAnpRuleACLOps(pgName, ingressAsV6Name, kubeovnv1.ProtocolIPv6, aclName, aclPriority, aclAction, logActions, rulePorts, true, false)
			if err != nil {
				klog.Errorf("failed to add v6 ingress acls for anp %s: %v", key, err)
				return err
			}
			ingressACLOps = append(ingressACLOps, ops...)
		}
	}

	if err := c.OVNNbClient.Transact("add-ingress-acls", ingressACLOps); err != nil {
		return fmt.Errorf("failed to add ingress acls for anp %s: %w", key, err)
	}
	if err := c.deleteUnusedAddrSetForAnp(curIngressAddrSet, desiredIngressAddrSet); err != nil {
		return fmt.Errorf("failed to delete unused ingress address set for anp %s: %w", key, err)
	}

	egressACLOps, err := c.OVNNbClient.DeleteAclsOps(pgName, portGroupKey, "from-lport", nil, util.NilACLTier)
	if err != nil {
		klog.Errorf("failed to generate clear operations for anp %s egress acls: %v", key, err)
		return err
	}
	// create egress acl
	for index, anpr := range anp.Spec.Egress {
		// A single address set must contain addresses of the same type and the name must be unique within table, so IPv4 and IPv6 address set should be different
		egressAsV4Name, egressAsV6Name := getAnpAddressSetName(pgName, anpr.Name, index, false)
		desiredEgressAddrSet.Add(egressAsV4Name, egressAsV6Name)

		var v4Addrs, v4Addr, v6Addrs, v6Addr []string
		// This field must be defined and contain at least one item.
		for _, anprpeer := range anpr.To {
			if v4Addr, v6Addr, err = c.fetchEgressSelectedAddresses(&anprpeer); err != nil {
				klog.Errorf("failed to fetch admin network policy selected addresses, %v", err)
				return err
			}
			v4Addrs = append(v4Addrs, v4Addr...)
			v6Addrs = append(v6Addrs, v6Addr...)
		}
		klog.Infof("anp %s, egress rule %s, selected v4 address %v, v6 address %v", anpName, anpr.Name, v4Addrs, v6Addrs)

		if err = c.createAsForAnpRule(anpName, anpr.Name, "egress", egressAsV4Name, v4Addrs, false); err != nil {
			klog.Error(err)
			return err
		}
		if err = c.createAsForAnpRule(anpName, anpr.Name, "egress", egressAsV6Name, v6Addrs, false); err != nil {
			klog.Error(err)
			return err
		}

		aclPriority := util.AnpACLMaxPriority - int(anp.Spec.Priority*100) - index
		aclAction := convertAction(anpr.Action, "")
		rulePorts := []v1alpha1.AdminNetworkPolicyPort{}
		if anpr.Ports != nil {
			rulePorts = *anpr.Ports
		}

		if len(v4Addrs) != 0 {
			aclName := fmt.Sprintf("anp/%s/egress/%s/%d", anpName, kubeovnv1.ProtocolIPv4, index)
			ops, err := c.OVNNbClient.UpdateAnpRuleACLOps(pgName, egressAsV4Name, kubeovnv1.ProtocolIPv4, aclName, aclPriority, aclAction, logActions, rulePorts, false, false)
			if err != nil {
				klog.Errorf("failed to add v4 egress acls for anp %s: %v", key, err)
				return err
			}
			egressACLOps = append(egressACLOps, ops...)
		}

		if len(v6Addrs) != 0 {
			aclName := fmt.Sprintf("anp/%s/egress/%s/%d", anpName, kubeovnv1.ProtocolIPv6, index)
			ops, err := c.OVNNbClient.UpdateAnpRuleACLOps(pgName, egressAsV6Name, kubeovnv1.ProtocolIPv6, aclName, aclPriority, aclAction, logActions, rulePorts, false, false)
			if err != nil {
				klog.Errorf("failed to add v6 egress acls for anp %s: %v", key, err)
				return err
			}
			egressACLOps = append(egressACLOps, ops...)
		}
	}

	if err := c.OVNNbClient.Transact("add-egress-acls", egressACLOps); err != nil {
		return fmt.Errorf("failed to add egress acls for anp %s: %w", key, err)
	}
	if err := c.deleteUnusedAddrSetForAnp(curEgressAddrSet, desiredEgressAddrSet); err != nil {
		return fmt.Errorf("failed to delete unused egress address set for anp %s: %w", key, err)
	}

	return nil
}

func (c *Controller) handleDeleteAnp(anp *v1alpha1.AdminNetworkPolicy) error {
	c.anpKeyMutex.LockKey(anp.Name)
	defer func() { _ = c.anpKeyMutex.UnlockKey(anp.Name) }()

	klog.Infof("handle delete admin network policy %s", anp.Name)
	delete(c.anpPrioNameMap, anp.Spec.Priority)
	delete(c.anpNamePrioMap, anp.Name)

	anpName := getAnpName(anp.Name)

	// ACLs releated to port_group will be deleted automatically when port_group is deleted
	pgName := strings.ReplaceAll(anpName, "-", ".")
	if err := c.OVNNbClient.DeletePortGroup(pgName); err != nil {
		klog.Errorf("failed to delete port group for anp %s: %v", anpName, err)
	}

	if err := c.OVNNbClient.DeleteAddressSets(map[string]string{
		adminNetworkPolicyKey: fmt.Sprintf("%s/%s", anpName, "ingress"),
	}); err != nil {
		klog.Errorf("failed to delete ingress address set for anp %s: %v", anpName, err)
		return err
	}

	if err := c.OVNNbClient.DeleteAddressSets(map[string]string{
		adminNetworkPolicyKey: fmt.Sprintf("%s/%s", anpName, "egress"),
	}); err != nil {
		klog.Errorf("failed to delete egress address set for anp %s: %v", anpName, err)
		return err
	}

	return nil
}

func (c *Controller) handleUpdateAnp(changed *AdminNetworkPolicyChangedDelta) error {
	// Only handle updates that do not affect acls.
	c.anpKeyMutex.LockKey(changed.key)
	defer func() { _ = c.anpKeyMutex.UnlockKey(changed.key) }()

	cachedAnp, err := c.anpsLister.Get(changed.key)
	if err != nil {
		if k8serrors.IsNotFound(err) {
			return nil
		}
		klog.Error(err)
		return err
	}
	desiredAnp := cachedAnp.DeepCopy()
	klog.Infof("handle update admin network policy %s", desiredAnp.Name)

	anpName := getAnpName(desiredAnp.Name)
	pgName := strings.ReplaceAll(anpName, "-", ".")

	// The port-group for anp should be updated
	if changed.field == ChangedSubject {
		// The port-group must exist when update anp, this check should never be matched.
		if ok, err := c.OVNNbClient.PortGroupExists(pgName); !ok || err != nil {
			klog.Errorf("port-group for anp %s does not exist when update anp", desiredAnp.Name)
			return err
		}

		ports, err := c.fetchSelectedPods(&desiredAnp.Spec.Subject)
		if err != nil {
			klog.Errorf("failed to fetch ports belongs to anp %s: %v", desiredAnp.Name, err)
			return err
		}

		if err = c.OVNNbClient.PortGroupSetPorts(pgName, ports); err != nil {
			klog.Errorf("failed to set ports %v to port group %s: %v", ports, pgName, err)
			return err
		}
	}

	// Peer selector in ingress/egress rule has changed, so the corresponding address-set need be updated
	if changed.field == ChangedIngressRule {
		for index, rule := range desiredAnp.Spec.Ingress {
			// Make sure the rule is changed and go on update
			if rule.Name == changed.ruleNames[index].curRuleName || changed.ruleNames[index].isMatch {
				if err := c.setAddrSetForAnpRule(anpName, pgName, rule.Name, index, rule.From, []v1alpha1.AdminNetworkPolicyEgressPeer{}, true, false); err != nil {
					klog.Errorf("failed to set ingress address-set for anp rule %s/%s, %v", anpName, rule.Name, err)
					return err
				}

				if changed.ruleNames[index].oldRuleName != "" {
					oldRuleName := changed.ruleNames[index].oldRuleName
					// Normally the name can not be changed, but when the name really changes, the old address set should be deleted
					// There is no description in the Name comments that it cannot be changed
					oldAsV4Name, oldAsV6Name := getAnpAddressSetName(pgName, oldRuleName, index, true)

					if err := c.OVNNbClient.DeleteAddressSet(oldAsV4Name); err != nil {
						klog.Errorf("failed to delete address set %s, %v", oldAsV4Name, err)
						// just record error log
					}
					if err := c.OVNNbClient.DeleteAddressSet(oldAsV6Name); err != nil {
						klog.Errorf("failed to delete address set %s, %v", oldAsV6Name, err)
					}
				}
			}
		}
	}

	if changed.field == ChangedEgressRule {
		for index, rule := range desiredAnp.Spec.Egress {
			// Make sure the rule is changed and go on update
			if rule.Name == changed.ruleNames[index].curRuleName || changed.ruleNames[index].isMatch {
				if err := c.setAddrSetForAnpRule(anpName, pgName, rule.Name, index, []v1alpha1.AdminNetworkPolicyIngressPeer{}, rule.To, false, false); err != nil {
					klog.Errorf("failed to set egress address-set for anp rule %s/%s, %v", anpName, rule.Name, err)
					return err
				}

				if changed.ruleNames[index].oldRuleName != "" {
					oldRuleName := changed.ruleNames[index].oldRuleName
					// Normally the name can not be changed, but when the name really changes, the old address set should be deleted
					// There is no description in the Name comments that it cannot be changed
					oldAsV4Name, oldAsV6Name := getAnpAddressSetName(pgName, oldRuleName, index, false)

					if err := c.OVNNbClient.DeleteAddressSet(oldAsV4Name); err != nil {
						klog.Errorf("failed to delete address set %s, %v", oldAsV4Name, err)
						// just record error log
					}
					if err := c.OVNNbClient.DeleteAddressSet(oldAsV6Name); err != nil {
						klog.Errorf("failed to delete address set %s, %v", oldAsV6Name, err)
					}
				}
			}
		}
	}

	return nil
}

func (c *Controller) validateAnpConfig(anp *v1alpha1.AdminNetworkPolicy) error {
	// The behavior is undefined if two ANP objects have same priority.
	if anpName, exist := c.anpPrioNameMap[anp.Spec.Priority]; exist && anpName != anp.Name {
		err := fmt.Errorf("can not create anp with same priority %d, exist one is %s, new created is %s", anp.Spec.Priority, anpName, anp.Name)
		klog.Error(err)
		return err
	}

	// We have noticed redhat's discussion about ACL priority in https://bugzilla.redhat.com/show_bug.cgi?id=2175752
	// After discussion, we decided to use the same range of priorities(20000-30000). Pay tribute to the developers of redhat.
	if anp.Spec.Priority > util.AnpMaxPriority {
		err := fmt.Errorf("the priority of anp %s is greater than max value %d", anp.Name, util.AnpMaxPriority)
		klog.Error(err)
		return err
	}

	if len(anp.Spec.Ingress) > util.AnpMaxRules || len(anp.Spec.Egress) > util.AnpMaxRules {
		err := fmt.Errorf("at most %d rules can be create in anp ingress/egress, ingress rules num %d and egress rules num %d in anp %s", util.AnpMaxRules, len(anp.Spec.Ingress), len(anp.Spec.Egress), anp.Name)
		klog.Error(err)
		return err
	}

	return nil
}

func (c *Controller) fetchSelectedPods(anpSubject *v1alpha1.AdminNetworkPolicySubject) ([]string, error) {
	var ports []string

	// Exactly one field must be set.
	if anpSubject.Namespaces != nil {
		nsSelector, err := metav1.LabelSelectorAsSelector(anpSubject.Namespaces)
		if err != nil {
			return nil, fmt.Errorf("error creating ns label selector, %w", err)
		}

		ports, _, _, err = c.fetchPods(nsSelector, labels.Everything())
		if err != nil {
			return nil, fmt.Errorf("failed to fetch pods, %w", err)
		}
	} else if anpSubject.Pods != nil {
		nsSelector, err := metav1.LabelSelectorAsSelector(&anpSubject.Pods.NamespaceSelector)
		if err != nil {
			return nil, fmt.Errorf("error creating ns label selector, %w", err)
		}
		podSelector, err := metav1.LabelSelectorAsSelector(&anpSubject.Pods.PodSelector)
		if err != nil {
			return nil, fmt.Errorf("error creating pod label selector, %w", err)
		}

		ports, _, _, err = c.fetchPods(nsSelector, podSelector)
		if err != nil {
			return nil, fmt.Errorf("failed to fetch pods, %w", err)
		}
	}
	klog.Infof("get selected ports for subject, %v", ports)
	return ports, nil
}

func (c *Controller) fetchPods(nsSelector, podSelector labels.Selector) ([]string, []string, []string, error) {
	ports := make([]string, 0, util.AnpMaxRules)
	v4Addresses := make([]string, 0, util.AnpMaxRules)
	v6Addresses := make([]string, 0, util.AnpMaxRules)

	namespaces, err := c.namespacesLister.List(nsSelector)
	if err != nil {
		klog.Errorf("failed to list namespaces: %v", err)
		return nil, nil, nil, err
	}

	klog.V(3).Infof("fetch pod ports/addresses, namespace selector is %s, pod selector is %s", nsSelector.String(), podSelector.String())
	for _, namespace := range namespaces {
		pods, err := c.podsLister.Pods(namespace.Name).List(podSelector)
		if err != nil {
			return nil, nil, nil, fmt.Errorf("failed to list pods, %w", err)
		}

		for _, pod := range pods {
			if pod.Spec.HostNetwork {
				continue
			}
			podName := c.getNameByPod(pod)

			podNets, err := c.getPodKubeovnNets(pod)
			if err != nil {
				return nil, nil, nil, fmt.Errorf("failed to get pod networks, %w", err)
			}

			for _, podNet := range podNets {
				if !isOvnSubnet(podNet.Subnet) {
					continue
				}

				if pod.Annotations[fmt.Sprintf(util.AllocatedAnnotationTemplate, podNet.ProviderName)] == "true" {
					ports = append(ports, ovs.PodNameToPortName(podName, pod.Namespace, podNet.ProviderName))

					podIPAnnotation := pod.Annotations[fmt.Sprintf(util.IPAddressAnnotationTemplate, podNet.ProviderName)]
					podIPs := strings.Split(podIPAnnotation, ",")
					for _, podIP := range podIPs {
						switch util.CheckProtocol(podIP) {
						case kubeovnv1.ProtocolIPv4:
							v4Addresses = append(v4Addresses, podIP)
						case kubeovnv1.ProtocolIPv6:
							v6Addresses = append(v6Addresses, podIP)
						}
					}
				}
			}
		}
	}

	return ports, v4Addresses, v6Addresses, nil
}

func (c *Controller) fetchIngressSelectedAddresses(ingressPeer *v1alpha1.AdminNetworkPolicyIngressPeer) ([]string, []string, error) {
	var v4Addresses, v6Addresses []string

	// Exactly one of the selector pointers must be set for a given peer.
	if ingressPeer.Namespaces != nil {
		nsSelector, err := metav1.LabelSelectorAsSelector(ingressPeer.Namespaces)
		if err != nil {
			return nil, nil, fmt.Errorf("error creating ns label selector, %w", err)
		}

		_, v4Addresses, v6Addresses, err = c.fetchPods(nsSelector, labels.Everything())
		if err != nil {
			return nil, nil, fmt.Errorf("failed to fetch ingress peer addresses, %w", err)
		}
	} else if ingressPeer.Pods != nil {
		nsSelector, err := metav1.LabelSelectorAsSelector(&ingressPeer.Pods.NamespaceSelector)
		if err != nil {
			return nil, nil, fmt.Errorf("error creating ns label selector, %w", err)
		}
		podSelector, err := metav1.LabelSelectorAsSelector(&ingressPeer.Pods.PodSelector)
		if err != nil {
			return nil, nil, fmt.Errorf("error creating pod label selector, %w", err)
		}

		_, v4Addresses, v6Addresses, err = c.fetchPods(nsSelector, podSelector)
		if err != nil {
			return nil, nil, fmt.Errorf("failed to fetch ingress peer addresses, %w", err)
		}
	}

	return v4Addresses, v6Addresses, nil
}

func (c *Controller) fetchEgressSelectedAddresses(egressPeer *v1alpha1.AdminNetworkPolicyEgressPeer) ([]string, []string, error) {
	var v4Addresses, v6Addresses []string

	// Exactly one of the selector pointers must be set for a given peer.
	switch {
	case egressPeer.Namespaces != nil:
		nsSelector, err := metav1.LabelSelectorAsSelector(egressPeer.Namespaces)
		if err != nil {
			return nil, nil, fmt.Errorf("error creating ns label selector, %w", err)
		}

		_, v4Addresses, v6Addresses, err = c.fetchPods(nsSelector, labels.Everything())
		if err != nil {
			return nil, nil, fmt.Errorf("failed to fetch egress peer addresses, %w", err)
		}
	case egressPeer.Pods != nil:
		nsSelector, err := metav1.LabelSelectorAsSelector(&egressPeer.Pods.NamespaceSelector)
		if err != nil {
			return nil, nil, fmt.Errorf("error creating ns label selector, %w", err)
		}
		podSelector, err := metav1.LabelSelectorAsSelector(&egressPeer.Pods.PodSelector)
		if err != nil {
			return nil, nil, fmt.Errorf("error creating pod label selector, %w", err)
		}

		_, v4Addresses, v6Addresses, err = c.fetchPods(nsSelector, podSelector)
		if err != nil {
			return nil, nil, fmt.Errorf("failed to fetch egress peer addresses, %w", err)
		}
	case egressPeer.Nodes != nil:
		nodesSelector, err := metav1.LabelSelectorAsSelector(egressPeer.Nodes)
		if err != nil {
			return nil, nil, fmt.Errorf("error creating nodes label selector, %w", err)
		}
		v4Addresses, v6Addresses, err = c.fetchNodesAddrs(nodesSelector)
		if err != nil {
			return nil, nil, fmt.Errorf("failed to fetch egress peer addresses, %w", err)
		}
	case len(egressPeer.Networks) != 0:
		v4Addresses, v6Addresses = fetchCIDRAddrs(egressPeer.Networks)
	default:
		return nil, nil, errors.New("at least one egressPeer must be specified")
	}

	return v4Addresses, v6Addresses, nil
}

func (c *Controller) createAsForAnpRule(anpName, ruleName, direction, asName string, addresses []string, isBanp bool) error {
	var err error
	if isBanp {
		err = c.OVNNbClient.CreateAddressSet(asName, map[string]string{
			baselineAdminNetworkPolicyKey: fmt.Sprintf("%s/%s", anpName, direction),
		})
	} else {
		err = c.OVNNbClient.CreateAddressSet(asName, map[string]string{
			adminNetworkPolicyKey: fmt.Sprintf("%s/%s", anpName, direction),
		})
	}
	if err != nil {
		klog.Errorf("failed to create ovn address set %s for anp/banp rule %s/%s: %v", asName, anpName, ruleName, err)
		return err
	}

	if err := c.OVNNbClient.AddressSetUpdateAddress(asName, addresses...); err != nil {
		klog.Errorf("failed to set addresses %q to address set %s: %v", strings.Join(addresses, ","), asName, err)
		return err
	}

	return nil
}

func (c *Controller) getCurrentAddrSetByName(anpName string, isBanp bool) (*strset.Set, *strset.Set, error) {
	curIngressAddrSet := strset.New()
	curEgressAddrSet := strset.New()
	var ass []ovnnb.AddressSet
	var err error

	// anp and banp can use same name, so depends on the external_ids key field to distinguish
	if isBanp {
		ass, err = c.OVNNbClient.ListAddressSets(map[string]string{
			baselineAdminNetworkPolicyKey: fmt.Sprintf("%s/%s", anpName, "ingress"),
		})
	} else {
		ass, err = c.OVNNbClient.ListAddressSets(map[string]string{
			adminNetworkPolicyKey: fmt.Sprintf("%s/%s", anpName, "ingress"),
		})
	}
	if err != nil {
		klog.Errorf("failed to list ingress address sets for anp/banp %s: %v", anpName, err)
		return curIngressAddrSet, curEgressAddrSet, err
	}
	for _, as := range ass {
		curIngressAddrSet.Add(as.Name)
	}

	if isBanp {
		ass, err = c.OVNNbClient.ListAddressSets(map[string]string{
			baselineAdminNetworkPolicyKey: fmt.Sprintf("%s/%s", anpName, "egress"),
		})
	} else {
		ass, err = c.OVNNbClient.ListAddressSets(map[string]string{
			adminNetworkPolicyKey: fmt.Sprintf("%s/%s", anpName, "egress"),
		})
	}
	if err != nil {
		klog.Errorf("failed to list egress address sets for anp/banp %s: %v", anpName, err)
		return curIngressAddrSet, curEgressAddrSet, err
	}
	for _, as := range ass {
		curEgressAddrSet.Add(as.Name)
	}

	return curIngressAddrSet, curEgressAddrSet, nil
}

func (c *Controller) deleteUnusedAddrSetForAnp(curAddrSet, desiredAddrSet *strset.Set) error {
	toDel := strset.Difference(curAddrSet, desiredAddrSet).List()

	for _, asName := range toDel {
		if err := c.OVNNbClient.DeleteAddressSet(asName); err != nil {
			klog.Errorf("failed to delete address set %s, %v", asName, err)
			return err
		}
	}

	return nil
}

func (c *Controller) setAddrSetForAnpRule(anpName, pgName, ruleName string, index int, from []v1alpha1.AdminNetworkPolicyIngressPeer, to []v1alpha1.AdminNetworkPolicyEgressPeer, isIngress, isBanp bool) error {
	// A single address set must contain addresses of the same type and the name must be unique within table, so IPv4 and IPv6 address set should be different

	var v4Addrs, v4Addr, v6Addrs, v6Addr []string
	var err error
	if isIngress {
		for _, anprpeer := range from {
			if v4Addr, v6Addr, err = c.fetchIngressSelectedAddresses(&anprpeer); err != nil {
				klog.Errorf("failed to fetch anp/banp ingress selected addresses, %v", err)
				return err
			}
			v4Addrs = append(v4Addrs, v4Addr...)
			v6Addrs = append(v6Addrs, v6Addr...)
		}
		klog.Infof("update anp/banp ingress rule %s, selected v4 address %v, v6 address %v", ruleName, v4Addrs, v6Addrs)

		gressAsV4Name, gressAsV6Name := getAnpAddressSetName(pgName, ruleName, index, true)
		if err = c.createAsForAnpRule(anpName, ruleName, "ingress", gressAsV4Name, v4Addrs, isBanp); err != nil {
			klog.Error(err)
			return err
		}
		if err = c.createAsForAnpRule(anpName, ruleName, "ingress", gressAsV6Name, v6Addrs, isBanp); err != nil {
			klog.Error(err)
			return err
		}
	} else {
		for _, anprpeer := range to {
			if v4Addr, v6Addr, err = c.fetchEgressSelectedAddresses(&anprpeer); err != nil {
				klog.Errorf("failed to fetch anp/banp egress selected addresses, %v", err)
				return err
			}
			v4Addrs = append(v4Addrs, v4Addr...)
			v6Addrs = append(v6Addrs, v6Addr...)
		}
		klog.Infof("update anp/banp egress rule %s, selected v4 address %v, v6 address %v", ruleName, v4Addrs, v6Addrs)

		gressAsV4Name, gressAsV6Name := getAnpAddressSetName(pgName, ruleName, index, false)
		if err = c.createAsForAnpRule(anpName, ruleName, "egress", gressAsV4Name, v4Addrs, isBanp); err != nil {
			klog.Error(err)
			return err
		}
		if err = c.createAsForAnpRule(anpName, ruleName, "egress", gressAsV6Name, v6Addrs, isBanp); err != nil {
			klog.Error(err)
			return err
		}
	}

	return nil
}

func (c *Controller) updateAnpsByLabelsMatch(nsLabels, podLabels map[string]string) {
	anps, _ := c.anpsLister.List(labels.Everything())
	for _, anp := range anps {
		changed := &AdminNetworkPolicyChangedDelta{
			key: anp.Name,
		}

		if isLabelsMatch(anp.Spec.Subject.Namespaces, anp.Spec.Subject.Pods, nsLabels, podLabels) {
			klog.Infof("anp %s, labels matched for anp's subject, nsLabels %s, podLabels %s", anp.Name, labels.Set(nsLabels).String(), labels.Set(podLabels).String())
			changed.field = ChangedSubject
			c.updateAnpQueue.Add(changed)
		}

		ingressRuleNames, egressRuleNames := isLabelsMatchAnpRulePeers(anp.Spec.Ingress, anp.Spec.Egress, nsLabels, podLabels)
		if !isRulesArrayEmpty(ingressRuleNames) {
			klog.Infof("anp %s, labels matched for anp's ingress peer, nsLabels %s, podLabels %s", anp.Name, labels.Set(nsLabels).String(), labels.Set(podLabels).String())
			changed.ruleNames = ingressRuleNames
			changed.field = ChangedIngressRule
			c.updateAnpQueue.Add(changed)
		}

		if !isRulesArrayEmpty(egressRuleNames) {
			klog.Infof("anp %s, labels matched for anp's egress peer, nsLabels %s, podLabels %s", anp.Name, labels.Set(nsLabels).String(), labels.Set(podLabels).String())
			changed.ruleNames = egressRuleNames
			changed.field = ChangedEgressRule
			c.updateAnpQueue.Add(changed)
		}
	}

	banps, _ := c.banpsLister.List(labels.Everything())
	for _, banp := range banps {
		changed := &AdminNetworkPolicyChangedDelta{
			key: banp.Name,
		}

		if isLabelsMatch(banp.Spec.Subject.Namespaces, banp.Spec.Subject.Pods, nsLabels, podLabels) {
			klog.Infof("banp %s, labels matched for banp's subject, nsLabels %s, podLabels %s", banp.Name, labels.Set(nsLabels).String(), labels.Set(podLabels).String())
			changed.field = ChangedSubject
			c.updateBanpQueue.Add(changed)
		}

		ingressRuleNames, egressRuleNames := isLabelsMatchBanpRulePeers(banp.Spec.Ingress, banp.Spec.Egress, nsLabels, podLabels)
		if !isRulesArrayEmpty(ingressRuleNames) {
			klog.Infof("banp %s, labels matched for banp's ingress peer, nsLabels %s, podLabels %s", banp.Name, labels.Set(nsLabels).String(), labels.Set(podLabels).String())
			changed.ruleNames = ingressRuleNames
			changed.field = ChangedIngressRule
			c.updateBanpQueue.Add(changed)
		}

		if !isRulesArrayEmpty(egressRuleNames) {
			klog.Infof("banp %s, labels matched for banp's egress peer, nsLabels %s, podLabels %s", banp.Name, labels.Set(nsLabels).String(), labels.Set(podLabels).String())
			changed.ruleNames = egressRuleNames
			changed.field = ChangedEgressRule
			c.updateBanpQueue.Add(changed)
		}
	}
}

func isLabelsMatch(namespaces *metav1.LabelSelector, pods *v1alpha1.NamespacedPod, nsLabels, podLabels map[string]string) bool {
	// Exactly one field of namespaces/pods must be set.
	if namespaces != nil {
		nsSelector, _ := metav1.LabelSelectorAsSelector(namespaces)
		klog.V(3).Infof("namespaces is not nil, nsSelector %s", nsSelector.String())
		if nsSelector.Matches(labels.Set(nsLabels)) {
			return true
		}
	} else if pods != nil {
		nsSelector, _ := metav1.LabelSelectorAsSelector(&pods.NamespaceSelector)
		podSelector, _ := metav1.LabelSelectorAsSelector(&pods.PodSelector)
		klog.V(3).Infof("pods is not nil, nsSelector %s, podSelector %s", nsSelector.String(), podSelector.String())
		if nsSelector.Matches(labels.Set(nsLabels)) && podSelector.Matches(labels.Set(podLabels)) {
			return true
		}
	}

	return false
}

func isLabelsMatchRulePeers(from []v1alpha1.AdminNetworkPolicyIngressPeer, to []v1alpha1.AdminNetworkPolicyEgressPeer, nsLabels, podLabels map[string]string) bool {
	for _, ingressPeer := range from {
		if isLabelsMatch(ingressPeer.Namespaces, ingressPeer.Pods, nsLabels, podLabels) {
			return true
		}
	}

	for _, egressPeer := range to {
		if isLabelsMatch(egressPeer.Namespaces, egressPeer.Pods, nsLabels, podLabels) {
			return true
		}
	}

	return false
}

func isLabelsMatchAnpRulePeers(ingress []v1alpha1.AdminNetworkPolicyIngressRule, egress []v1alpha1.AdminNetworkPolicyEgressRule, nsLabels, podLabels map[string]string) ([util.AnpMaxRules]ChangedName, [util.AnpMaxRules]ChangedName) {
	var changedIngressRuleNames, changedEgressRuleNames [util.AnpMaxRules]ChangedName

	for index, anpr := range ingress {
		if isLabelsMatchRulePeers(anpr.From, []v1alpha1.AdminNetworkPolicyEgressPeer{}, nsLabels, podLabels) {
			changedIngressRuleNames[index].isMatch = true
			changedIngressRuleNames[index].curRuleName = anpr.Name
		}
	}

	for index, anpr := range egress {
		if isLabelsMatchRulePeers([]v1alpha1.AdminNetworkPolicyIngressPeer{}, anpr.To, nsLabels, podLabels) {
			changedEgressRuleNames[index].isMatch = true
			changedEgressRuleNames[index].curRuleName = anpr.Name
		}
	}

	return changedIngressRuleNames, changedEgressRuleNames
}

func isLabelsMatchBanpRulePeers(ingress []v1alpha1.BaselineAdminNetworkPolicyIngressRule, egress []v1alpha1.BaselineAdminNetworkPolicyEgressRule, nsLabels, podLabels map[string]string) ([util.AnpMaxRules]ChangedName, [util.AnpMaxRules]ChangedName) {
	var changedIngressRuleNames, changedEgressRuleNames [util.AnpMaxRules]ChangedName

	for index, banpr := range ingress {
		if isLabelsMatchRulePeers(banpr.From, []v1alpha1.AdminNetworkPolicyEgressPeer{}, nsLabels, podLabels) {
			changedIngressRuleNames[index].isMatch = true
			changedIngressRuleNames[index].curRuleName = banpr.Name
		}
	}

	for index, banpr := range egress {
		if isLabelsMatchRulePeers([]v1alpha1.AdminNetworkPolicyIngressPeer{}, banpr.To, nsLabels, podLabels) {
			changedEgressRuleNames[index].isMatch = true
			changedEgressRuleNames[index].curRuleName = banpr.Name
		}
	}

	return changedIngressRuleNames, changedEgressRuleNames
}

func getAnpName(name string) string {
	anpName := name
	nameArray := []rune(name)
	if !unicode.IsLetter(nameArray[0]) {
		anpName = "anp" + name
	}
	return anpName
}

func getAnpAddressSetName(pgName, ruleName string, index int, isIngress bool) (string, string) {
	var asV4Name, asV6Name string
	if isIngress {
		// In case ruleName is omitted, add direction and index to distinguish address-set
		asV4Name = strings.ReplaceAll(fmt.Sprintf("%s.ingress.%d.%s.%s", pgName, index, ruleName, kubeovnv1.ProtocolIPv4), "-", ".")
		asV6Name = strings.ReplaceAll(fmt.Sprintf("%s.ingress.%d.%s.%s", pgName, index, ruleName, kubeovnv1.ProtocolIPv6), "-", ".")
	} else {
		asV4Name = strings.ReplaceAll(fmt.Sprintf("%s.egress.%d.%s.%s", pgName, index, ruleName, kubeovnv1.ProtocolIPv4), "-", ".")
		asV6Name = strings.ReplaceAll(fmt.Sprintf("%s.egress.%d.%s.%s", pgName, index, ruleName, kubeovnv1.ProtocolIPv6), "-", ".")
	}

	return asV4Name, asV6Name
}

func convertAction(anpRuleAction v1alpha1.AdminNetworkPolicyRuleAction, banpRuleAction v1alpha1.BaselineAdminNetworkPolicyRuleAction) (aclAction ovnnb.ACLAction) {
	switch anpRuleAction {
	case v1alpha1.AdminNetworkPolicyRuleActionAllow:
		aclAction = ovnnb.ACLActionAllowRelated
	case v1alpha1.AdminNetworkPolicyRuleActionDeny:
		aclAction = ovnnb.ACLActionDrop
	case v1alpha1.AdminNetworkPolicyRuleActionPass:
		aclAction = ovnnb.ACLActionPass
	}

	switch banpRuleAction {
	case v1alpha1.BaselineAdminNetworkPolicyRuleActionAllow:
		aclAction = ovnnb.ACLActionAllowRelated
	case v1alpha1.BaselineAdminNetworkPolicyRuleActionDeny:
		aclAction = ovnnb.ACLActionDrop
	}
	return
}

func isRulesArrayEmpty(ruleNames [util.AnpMaxRules]ChangedName) bool {
	isEmpty := true
	for _, ruleName := range ruleNames {
		// The ruleName can be omitted default
		if ruleName.curRuleName != "" || ruleName.isMatch {
			isEmpty = false
			break
		}
	}
	return isEmpty
}

func (c *Controller) fetchNodesAddrs(nodeSelector labels.Selector) ([]string, []string, error) {
	nodes, err := c.nodesLister.List(nodeSelector)
	if err != nil {
		klog.Errorf("failed to list nodes: %v", err)
		return nil, nil, err
	}
	v4Addresses := make([]string, 0, len(nodes))
	v6Addresses := make([]string, 0, len(nodes))

	klog.V(3).Infof("fetch nodes addresses, selector is %s", nodeSelector.String())
	for _, node := range nodes {
		nodeIPv4, nodeIPv6 := util.GetNodeInternalIP(*node)
		if nodeIPv4 != "" {
			v4Addresses = append(v4Addresses, nodeIPv4)
		}
		if nodeIPv6 != "" {
			v6Addresses = append(v6Addresses, nodeIPv6)
		}
	}

	return v4Addresses, v6Addresses, nil
}

func fetchCIDRAddrs(networks []v1alpha1.CIDR) ([]string, []string) {
	v4Addresses := make([]string, 0, len(networks))
	v6Addresses := make([]string, 0, len(networks))

	for _, network := range networks {
		if _, _, err := net.ParseCIDR(string(network)); err != nil {
			klog.Errorf("invalid cidr %s", string(network))
			continue
		}
		switch util.CheckProtocol(string(network)) {
		case kubeovnv1.ProtocolIPv4:
			v4Addresses = append(v4Addresses, string(network))
		case kubeovnv1.ProtocolIPv6:
			v6Addresses = append(v6Addresses, string(network))
		}
	}

	return v4Addresses, v6Addresses
}