17228354799

Committed 26 Aug 2025 04:55AM UTC coverage: 21.341% (-0.2%) from 21.508%

Build # 17228354799

Build Type

push

github

Committed by

oilbeater

Commit Message

handle delete final state unknown object in enqueue handler (#5649)

Signed-off-by: Mengxin Liu <liumengxinfly@gmail.com>

Run Details

0 of 443 new or added lines in 27 files covered. (0.0%)

1 existing line in 1 file now uncovered.

10514 of 49267 relevant lines covered (21.34%)

0.25 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

2.03

/pkg/controller/baseline_admin_network_policy.go

package controller

import (
        "fmt"
        "reflect"
        "strings"

        "github.com/scylladb/go-set/strset"
        k8serrors "k8s.io/apimachinery/pkg/api/errors"
        "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/ovsdb/ovnnb"
        "github.com/kubeovn/kube-ovn/pkg/util"
)

func (c *Controller) enqueueAddBanp(obj any) {
        key := cache.MetaObjectToName(obj.(*v1alpha1.BaselineAdminNetworkPolicy)).String()
        klog.V(3).Infof("enqueue add banp %s", key)
        c.addBanpQueue.Add(key)
}

func (c *Controller) enqueueDeleteBanp(obj any) {
        var bnp *v1alpha1.BaselineAdminNetworkPolicy
        switch t := obj.(type) {
        case *v1alpha1.BaselineAdminNetworkPolicy:
                bnp = t
        case cache.DeletedFinalStateUnknown:
                b, ok := t.Obj.(*v1alpha1.BaselineAdminNetworkPolicy)
                if !ok {
                        klog.Warningf("unexpected object type: %T", t.Obj)
                        return
                }
                bnp = b
        default:
                klog.Warningf("unexpected type: %T", obj)
                return
        }

        klog.V(3).Infof("enqueue delete bnp %s", cache.MetaObjectToName(bnp).String())
        c.deleteBanpQueue.Add(bnp)
}

func (c *Controller) enqueueUpdateBanp(oldObj, newObj any) {
        oldBanp := oldObj.(*v1alpha1.BaselineAdminNetworkPolicy)
        newBanp := newObj.(*v1alpha1.BaselineAdminNetworkPolicy)

        // All the acls should be recreated with the following situations
        if len(oldBanp.Spec.Ingress) != len(newBanp.Spec.Ingress) || len(oldBanp.Spec.Egress) != len(newBanp.Spec.Egress) {
                c.addBanpQueue.Add(newBanp.Name)
                return
        }

        // Acls should be updated when action or ports of ingress/egress rule has been changed
        for index, rule := range newBanp.Spec.Ingress {
                oldRule := oldBanp.Spec.Ingress[index]
                if oldRule.Action != rule.Action || !reflect.DeepEqual(oldRule.Ports, rule.Ports) {
                        c.addBanpQueue.Add(newBanp.Name)
                        return
                }
        }

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

        if oldBanp.Annotations[util.ACLActionsLogAnnotation] != newBanp.Annotations[util.ACLActionsLogAnnotation] {
                c.addBanpQueue.Add(newBanp.Name)
                return
        }
        klog.V(3).Infof("enqueue update banp %s", newBanp.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(oldBanp.Spec.Subject, newBanp.Spec.Subject) {
                c.updateBanpQueue.Add(&AdminNetworkPolicyChangedDelta{key: newBanp.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 newBanp.Spec.Ingress {
                oldRule := oldBanp.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.updateBanpQueue.Add(&AdminNetworkPolicyChangedDelta{key: newBanp.Name, ruleNames: changedIngressRuleNames, field: ChangedIngressRule})
        }

        ruleChanged = false
        for index, rule := range newBanp.Spec.Egress {
                oldRule := oldBanp.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.updateBanpQueue.Add(&AdminNetworkPolicyChangedDelta{key: newBanp.Name, ruleNames: changedEgressRuleNames, field: ChangedEgressRule})
        }
}

func banpACLAction(action v1alpha1.BaselineAdminNetworkPolicyRuleAction) ovnnb.ACLAction {
        switch action {
        case v1alpha1.BaselineAdminNetworkPolicyRuleActionAllow:
                return ovnnb.ACLActionAllowRelated
        case v1alpha1.BaselineAdminNetworkPolicyRuleActionDeny:
                return ovnnb.ACLActionDrop
        }
        return ovnnb.ACLActionDrop
}

func (c *Controller) handleAddBanp(key string) (err error) {
        // Only one banp with default name can be created in cluster, no need to check
        c.banpKeyMutex.LockKey(key)
        defer func() { _ = c.banpKeyMutex.UnlockKey(key) }()

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

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

        // ovn portGroup/addressSet doesn't support name with '-', so we replace '-' by '.'.
        pgName := strings.ReplaceAll(banpName, "-", ".")
        if err = c.OVNNbClient.CreatePortGroup(pgName, map[string]string{baselineAdminNetworkPolicyKey: banpName}); err != nil {
                klog.Errorf("failed to create port group for banp %s: %v", key, err)
                return err
        }

        ports, err := c.fetchSelectedPods(&banp.Spec.Subject)
        if err != nil {
                klog.Errorf("failed to fetch ports belongs to banp %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)
        if err != nil {
                klog.Errorf("failed to generate clear operations for banp %s ingress acls: %v", key, err)
                return err
        }

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

        // create ingress acl
        for index, banpr := range banp.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, banpr.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 banpr.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("Banp Ingress Rule %s, selected v4 address %v, v6 address %v", banpr.Name, v4Addrs, v6Addrs)

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

                // use 1700-1800 for banp acl priority
                aclPriority := util.BanpACLMaxPriority - index
                aclAction := banpACLAction(banpr.Action)
                rulePorts := []v1alpha1.AdminNetworkPolicyPort{}
                if banpr.Ports != nil {
                        rulePorts = *banpr.Ports
                }

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

                if len(v6Addrs) != 0 {
                        aclName := fmt.Sprintf("banp/%s/ingress/%s/%d", banpName, kubeovnv1.ProtocolIPv6, index)
                        ops, err := c.OVNNbClient.UpdateAnpRuleACLOps(pgName, ingressAsV6Name, kubeovnv1.ProtocolIPv6, aclName, aclPriority, aclAction, logActions, rulePorts, true, true)
                        if err != nil {
                                klog.Errorf("failed to add v6 ingress acls for banp %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 banp %s: %w", key, err)
        }
        if err := c.deleteUnusedAddrSetForAnp(curIngressAddrSet, desiredIngressAddrSet); err != nil {
                return fmt.Errorf("failed to delete unused ingress address set for banp %s: %w", key, err)
        }

        egressACLOps, err := c.OVNNbClient.DeleteAclsOps(pgName, portGroupKey, "from-lport", nil)
        if err != nil {
                klog.Errorf("failed to generate clear operations for banp %s egress acls: %v", key, err)
                return err
        }
        // create egress acl
        for index, banpr := range banp.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, banpr.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 banpr.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("Banp Egress Rule %s, selected v4 address %v, v6 address %v", banpr.Name, v4Addrs, v6Addrs)

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

                aclPriority := util.BanpACLMaxPriority - index
                aclAction := banpACLAction(banpr.Action)
                rulePorts := []v1alpha1.AdminNetworkPolicyPort{}
                if banpr.Ports != nil {
                        rulePorts = *banpr.Ports
                }

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

                if len(v6Addrs) != 0 {
                        aclName := fmt.Sprintf("banp/%s/egress/%s/%d", banpName, kubeovnv1.ProtocolIPv6, index)
                        ops, err := c.OVNNbClient.UpdateAnpRuleACLOps(pgName, egressAsV6Name, kubeovnv1.ProtocolIPv6, aclName, aclPriority, aclAction, logActions, rulePorts, false, true)
                        if err != nil {
                                klog.Errorf("failed to add v6 egress acls for banp %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 banp %s: %w", key, err)
        }
        if err := c.deleteUnusedAddrSetForAnp(curEgressAddrSet, desiredEgressAddrSet); err != nil {
                return fmt.Errorf("failed to delete unused egress address set for banp %s: %w", key, err)
        }

        return nil
}

func (c *Controller) handleDeleteBanp(banp *v1alpha1.BaselineAdminNetworkPolicy) error {
        c.banpKeyMutex.LockKey(banp.Name)
        defer func() { _ = c.banpKeyMutex.UnlockKey(banp.Name) }()

        klog.Infof("handle delete banp %s", banp.Name)
        banpName := getAnpName(banp.Name)

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

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

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

        return nil
}

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

        cachedBanp, err := c.banpsLister.Get(changed.key)
        if err != nil {
                if k8serrors.IsNotFound(err) {
                        return nil
                }
                klog.Error(err)
                return err
        }
        desiredBanp := cachedBanp.DeepCopy()
        klog.Infof("handle update banp %s", desiredBanp.Name)

        banpName := getAnpName(desiredBanp.Name)
        pgName := strings.ReplaceAll(banpName, "-", ".")

        // 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 banp %s does not exist when update banp", desiredBanp.Name)
                        return err
                }

                ports, err := c.fetchSelectedPods(&desiredBanp.Spec.Subject)
                if err != nil {
                        klog.Errorf("failed to fetch ports belongs to banp %s: %v", desiredBanp.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 desiredBanp.Spec.Ingress {
                        // Make sure the rule is changed and go on update
                        if rule.Name == changed.ruleNames[index].curRuleName {
                                if err := c.setAddrSetForAnpRule(banpName, pgName, rule.Name, index, rule.From, []v1alpha1.AdminNetworkPolicyEgressPeer{}, true, true); err != nil {
                                        klog.Errorf("failed to set ingress address-set for anp rule %s/%s, %v", banpName, rule.Name, err)
                                        return err
                                }

                                if changed.ruleNames[index].oldRuleName != "" {
                                        oldRuleName := changed.ruleNames[index].oldRuleName
                                        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 desiredBanp.Spec.Egress {
                        // Make sure the rule is changed and go on update
                        if rule.Name == changed.ruleNames[index].curRuleName {
                                if err := c.setAddrSetForAnpRule(banpName, pgName, rule.Name, index, []v1alpha1.AdminNetworkPolicyIngressPeer{}, rule.To, false, true); err != nil {
                                        klog.Errorf("failed to set egress address-set for banp rule %s/%s, %v", banpName, rule.Name, err)
                                        return err
                                }

                                if changed.ruleNames[index].oldRuleName != "" {
                                        oldRuleName := changed.ruleNames[index].oldRuleName
                                        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
}

1	package controller
2
3	import (
4	"fmt"
5	"reflect"
6	"strings"
7
8	"github.com/scylladb/go-set/strset"
9	k8serrors "k8s.io/apimachinery/pkg/api/errors"
10	"k8s.io/client-go/tools/cache"
11	"k8s.io/klog/v2"
12	v1alpha1 "sigs.k8s.io/network-policy-api/apis/v1alpha1"
13
14	kubeovnv1 "github.com/kubeovn/kube-ovn/pkg/apis/kubeovn/v1"
15	"github.com/kubeovn/kube-ovn/pkg/ovsdb/ovnnb"
16	"github.com/kubeovn/kube-ovn/pkg/util"
17	)
18
19	func (c *Controller) enqueueAddBanp(obj any) {	×
20	key := cache.MetaObjectToName(obj.(*v1alpha1.BaselineAdminNetworkPolicy)).String()	×
21	klog.V(3).Infof("enqueue add banp %s", key)	×
22	c.addBanpQueue.Add(key)	×
23	}	×
24
25	func (c *Controller) enqueueDeleteBanp(obj any) {	×
NEW 26	var bnp *v1alpha1.BaselineAdminNetworkPolicy	×
NEW 27	switch t := obj.(type) {	×
NEW 28	case *v1alpha1.BaselineAdminNetworkPolicy:	×
NEW 29	bnp = t	×
NEW 30	case cache.DeletedFinalStateUnknown:	×
NEW 31	b, ok := t.Obj.(*v1alpha1.BaselineAdminNetworkPolicy)	×
NEW 32	if !ok {	×
NEW 33	klog.Warningf("unexpected object type: %T", t.Obj)	×
NEW 34	return	×
NEW 35	}	×
NEW 36	bnp = b	×
NEW 37	default:	×
NEW 38	klog.Warningf("unexpected type: %T", obj)	×
NEW 39	return	×
40	}
41
NEW 42	klog.V(3).Infof("enqueue delete bnp %s", cache.MetaObjectToName(bnp).String())	×
NEW 43	c.deleteBanpQueue.Add(bnp)	×
44	}
45
46	func (c *Controller) enqueueUpdateBanp(oldObj, newObj any) {	×
47	oldBanp := oldObj.(*v1alpha1.BaselineAdminNetworkPolicy)	×
48	newBanp := newObj.(*v1alpha1.BaselineAdminNetworkPolicy)	×
49		×
50	// All the acls should be recreated with the following situations	×
51	if len(oldBanp.Spec.Ingress) != len(newBanp.Spec.Ingress) \|\| len(oldBanp.Spec.Egress) != len(newBanp.Spec.Egress) {	×
52	c.addBanpQueue.Add(newBanp.Name)	×
53	return	×
54	}	×
55
56	// Acls should be updated when action or ports of ingress/egress rule has been changed
57	for index, rule := range newBanp.Spec.Ingress {	×
58	oldRule := oldBanp.Spec.Ingress[index]	×
59	if oldRule.Action != rule.Action \|\| !reflect.DeepEqual(oldRule.Ports, rule.Ports) {	×
60	c.addBanpQueue.Add(newBanp.Name)	×
61	return	×
62	}	×
63	}
64
65	for index, rule := range newBanp.Spec.Egress {	×
66	oldRule := oldBanp.Spec.Egress[index]	×
67	if oldRule.Action != rule.Action \|\| !reflect.DeepEqual(oldRule.Ports, rule.Ports) {	×
68	c.addBanpQueue.Add(newBanp.Name)	×
69	return	×
70	}	×
71	}
72
73	if oldBanp.Annotations[util.ACLActionsLogAnnotation] != newBanp.Annotations[util.ACLActionsLogAnnotation] {	×
74	c.addBanpQueue.Add(newBanp.Name)	×
75	return	×
76	}	×
77	klog.V(3).Infof("enqueue update banp %s", newBanp.Name)	×
78		×
79	// The remaining changes do not affect the acls. The port-group or address-set should be updated.	×
80	// The port-group for anp should be updated	×
81	if !reflect.DeepEqual(oldBanp.Spec.Subject, newBanp.Spec.Subject) {	×
82	c.updateBanpQueue.Add(&AdminNetworkPolicyChangedDelta{key: newBanp.Name, field: ChangedSubject})	×
83	}	×
84
85	// Rule name or peer selector in ingress/egress rule has changed, the corresponding address-set need be updated
86	ruleChanged := false	×
87	var changedIngressRuleNames, changedEgressRuleNames [util.AnpMaxRules]ChangedName	×
88		×
89	for index, rule := range newBanp.Spec.Ingress {	×
90	oldRule := oldBanp.Spec.Ingress[index]	×
91	if oldRule.Name != rule.Name {	×
92	changedIngressRuleNames[index] = ChangedName{oldRuleName: oldRule.Name, curRuleName: rule.Name}	×
93	ruleChanged = true	×
94	}	×
95	if !reflect.DeepEqual(oldRule.From, rule.From) {	×
96	changedIngressRuleNames[index] = ChangedName{curRuleName: rule.Name}	×
97	ruleChanged = true	×
98	}	×
99	}
100	if ruleChanged {	×
101	c.updateBanpQueue.Add(&AdminNetworkPolicyChangedDelta{key: newBanp.Name, ruleNames: changedIngressRuleNames, field: ChangedIngressRule})	×
102	}	×
103
104	ruleChanged = false	×
105	for index, rule := range newBanp.Spec.Egress {	×
106	oldRule := oldBanp.Spec.Egress[index]	×
107	if oldRule.Name != rule.Name {	×
108	changedEgressRuleNames[index] = ChangedName{oldRuleName: oldRule.Name, curRuleName: rule.Name}	×
109	ruleChanged = true	×
110	}	×
111	if !reflect.DeepEqual(oldRule.To, rule.To) {	×
112	changedEgressRuleNames[index] = ChangedName{curRuleName: rule.Name}	×
113	ruleChanged = true	×
114	}	×
115	}
116	if ruleChanged {	×
117	c.updateBanpQueue.Add(&AdminNetworkPolicyChangedDelta{key: newBanp.Name, ruleNames: changedEgressRuleNames, field: ChangedEgressRule})	×
118	}	×
119	}
120
121	func banpACLAction(action v1alpha1.BaselineAdminNetworkPolicyRuleAction) ovnnb.ACLAction {	1✔
122	switch action {	1✔
123	case v1alpha1.BaselineAdminNetworkPolicyRuleActionAllow:	1✔
124	return ovnnb.ACLActionAllowRelated	1✔
125	case v1alpha1.BaselineAdminNetworkPolicyRuleActionDeny:	1✔
126	return ovnnb.ACLActionDrop	1✔
127	}
128	return ovnnb.ACLActionDrop	1✔
129	}
130
131	func (c *Controller) handleAddBanp(key string) (err error) {	×
132	// Only one banp with default name can be created in cluster, no need to check	×
133	c.banpKeyMutex.LockKey(key)	×
134	defer func() { _ = c.banpKeyMutex.UnlockKey(key) }()	×
135
136	cachedBanp, err := c.banpsLister.Get(key)	×
137	if err != nil {	×
138	if k8serrors.IsNotFound(err) {	×
139	return nil	×
140	}	×
141	klog.Error(err)	×
142	return err	×
143	}
144	klog.Infof("handle add banp %s", cachedBanp.Name)	×
145	banp := cachedBanp.DeepCopy()	×
146		×
147	banpName := getAnpName(banp.Name)	×
148	var logActions []string	×
149	if banp.Annotations[util.ACLActionsLogAnnotation] != "" {	×
150	logActions = strings.Split(banp.Annotations[util.ACLActionsLogAnnotation], ",")	×
151	}	×
152
153	// ovn portGroup/addressSet doesn't support name with '-', so we replace '-' by '.'.
154	pgName := strings.ReplaceAll(banpName, "-", ".")	×
155	if err = c.OVNNbClient.CreatePortGroup(pgName, map[string]string{baselineAdminNetworkPolicyKey: banpName}); err != nil {	×
156	klog.Errorf("failed to create port group for banp %s: %v", key, err)	×
157	return err	×
158	}	×
159
160	ports, err := c.fetchSelectedPods(&banp.Spec.Subject)	×
161	if err != nil {	×
162	klog.Errorf("failed to fetch ports belongs to banp %s: %v", key, err)	×
163	return err	×
164	}	×
165
166	if err = c.OVNNbClient.PortGroupSetPorts(pgName, ports); err != nil {	×
167	klog.Errorf("failed to set ports %v to port group %s: %v", ports, pgName, err)	×
168	return err	×
169	}	×
170
171	ingressACLOps, err := c.OVNNbClient.DeleteAclsOps(pgName, portGroupKey, "to-lport", nil)	×
172	if err != nil {	×
173	klog.Errorf("failed to generate clear operations for banp %s ingress acls: %v", key, err)	×
174	return err	×
175	}	×
176
177	curIngressAddrSet, curEgressAddrSet, err := c.getCurrentAddrSetByName(banpName, true)	×
178	if err != nil {	×
179	klog.Errorf("failed to list address sets for banp %s: %v", key, err)	×
180	return err	×
181	}	×
182	desiredIngressAddrSet := strset.NewWithSize(len(banp.Spec.Ingress) * 2)	×
183	desiredEgressAddrSet := strset.NewWithSize(len(banp.Spec.Egress) * 2)	×
184		×
185	// create ingress acl	×
186	for index, banpr := range banp.Spec.Ingress {	×
187	// 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	×
188	ingressAsV4Name, ingressAsV6Name := getAnpAddressSetName(pgName, banpr.Name, index, true)	×
189	desiredIngressAddrSet.Add(ingressAsV4Name, ingressAsV6Name)	×
190		×
191	var v4Addrs, v4Addr, v6Addrs, v6Addr []string	×
192	// This field must be defined and contain at least one item.	×
193	for _, anprpeer := range banpr.From {	×
194	if v4Addr, v6Addr, err = c.fetchIngressSelectedAddresses(&anprpeer); err != nil {	×
195	klog.Errorf("failed to fetch admin network policy selected addresses, %v", err)	×
196	return err	×
197	}	×
198	v4Addrs = append(v4Addrs, v4Addr...)	×
199	v6Addrs = append(v6Addrs, v6Addr...)	×
200	}
201	klog.Infof("Banp Ingress Rule %s, selected v4 address %v, v6 address %v", banpr.Name, v4Addrs, v6Addrs)	×
202		×
203	if err = c.createAsForAnpRule(banpName, banpr.Name, "ingress", ingressAsV4Name, v4Addrs, true); err != nil {	×
204	klog.Error(err)	×
205	return err	×
206	}	×
207	if err = c.createAsForAnpRule(banpName, banpr.Name, "ingress", ingressAsV6Name, v6Addrs, true); err != nil {	×
208	klog.Error(err)	×
209	return err	×
210	}	×
211
212	// use 1700-1800 for banp acl priority
213	aclPriority := util.BanpACLMaxPriority - index	×
214	aclAction := banpACLAction(banpr.Action)	×
215	rulePorts := []v1alpha1.AdminNetworkPolicyPort{}	×
216	if banpr.Ports != nil {	×
217	rulePorts = *banpr.Ports	×
218	}	×
219
220	if len(v4Addrs) != 0 {	×
221	aclName := fmt.Sprintf("banp/%s/ingress/%s/%d", banpName, kubeovnv1.ProtocolIPv4, index)	×
222	ops, err := c.OVNNbClient.UpdateAnpRuleACLOps(pgName, ingressAsV4Name, kubeovnv1.ProtocolIPv4, aclName, aclPriority, aclAction, logActions, rulePorts, true, true)	×
223	if err != nil {	×
224	klog.Errorf("failed to add v4 ingress acls for banp %s: %v", key, err)	×
225	return err	×
226	}	×
227	ingressACLOps = append(ingressACLOps, ops...)	×
228	}
229
230	if len(v6Addrs) != 0 {	×
231	aclName := fmt.Sprintf("banp/%s/ingress/%s/%d", banpName, kubeovnv1.ProtocolIPv6, index)	×
232	ops, err := c.OVNNbClient.UpdateAnpRuleACLOps(pgName, ingressAsV6Name, kubeovnv1.ProtocolIPv6, aclName, aclPriority, aclAction, logActions, rulePorts, true, true)	×
233	if err != nil {	×
234	klog.Errorf("failed to add v6 ingress acls for banp %s: %v", key, err)	×
235	return err	×
236	}	×
237	ingressACLOps = append(ingressACLOps, ops...)	×
238	}
239	}
240
241	if err := c.OVNNbClient.Transact("add-ingress-acls", ingressACLOps); err != nil {	×
242	return fmt.Errorf("failed to add ingress acls for banp %s: %w", key, err)	×
243	}	×
244	if err := c.deleteUnusedAddrSetForAnp(curIngressAddrSet, desiredIngressAddrSet); err != nil {	×
245	return fmt.Errorf("failed to delete unused ingress address set for banp %s: %w", key, err)	×
246	}	×
247
248	egressACLOps, err := c.OVNNbClient.DeleteAclsOps(pgName, portGroupKey, "from-lport", nil)	×
249	if err != nil {	×
250	klog.Errorf("failed to generate clear operations for banp %s egress acls: %v", key, err)	×
251	return err	×
252	}	×
253	// create egress acl
254	for index, banpr := range banp.Spec.Egress {	×
255	// 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	×
256	egressAsV4Name, egressAsV6Name := getAnpAddressSetName(pgName, banpr.Name, index, false)	×
257	desiredEgressAddrSet.Add(egressAsV4Name, egressAsV6Name)	×
258		×
259	var v4Addrs, v4Addr, v6Addrs, v6Addr []string	×
260	// This field must be defined and contain at least one item.	×
261	for _, anprpeer := range banpr.To {	×
262	if v4Addr, v6Addr, err = c.fetchEgressSelectedAddresses(&anprpeer); err != nil {	×
263	klog.Errorf("failed to fetch admin network policy selected addresses, %v", err)	×
264	return err	×
265	}	×
266	v4Addrs = append(v4Addrs, v4Addr...)	×
267	v6Addrs = append(v6Addrs, v6Addr...)	×
268	}
269	klog.Infof("Banp Egress Rule %s, selected v4 address %v, v6 address %v", banpr.Name, v4Addrs, v6Addrs)	×
270		×
271	if err = c.createAsForAnpRule(banpName, banpr.Name, "egress", egressAsV4Name, v4Addrs, true); err != nil {	×
272	klog.Error(err)	×
273	return err	×
274	}	×
275	if err = c.createAsForAnpRule(banpName, banpr.Name, "egress", egressAsV6Name, v6Addrs, true); err != nil {	×
276	klog.Error(err)	×
277	return err	×
278	}	×
279
280	aclPriority := util.BanpACLMaxPriority - index	×
281	aclAction := banpACLAction(banpr.Action)	×
282	rulePorts := []v1alpha1.AdminNetworkPolicyPort{}	×
283	if banpr.Ports != nil {	×
284	rulePorts = *banpr.Ports	×
285	}	×
286
287	if len(v4Addrs) != 0 {	×
288	aclName := fmt.Sprintf("banp/%s/egress/%s/%d", banpName, kubeovnv1.ProtocolIPv4, index)	×
289	ops, err := c.OVNNbClient.UpdateAnpRuleACLOps(pgName, egressAsV4Name, kubeovnv1.ProtocolIPv4, aclName, aclPriority, aclAction, logActions, rulePorts, false, true)	×
290	if err != nil {	×
291	klog.Errorf("failed to add v4 egress acls for banp %s: %v", key, err)	×
292	return err	×
293	}	×
294	egressACLOps = append(egressACLOps, ops...)	×
295	}
296
297	if len(v6Addrs) != 0 {	×
298	aclName := fmt.Sprintf("banp/%s/egress/%s/%d", banpName, kubeovnv1.ProtocolIPv6, index)	×
299	ops, err := c.OVNNbClient.UpdateAnpRuleACLOps(pgName, egressAsV6Name, kubeovnv1.ProtocolIPv6, aclName, aclPriority, aclAction, logActions, rulePorts, false, true)	×
300	if err != nil {	×
301	klog.Errorf("failed to add v6 egress acls for banp %s: %v", key, err)	×
302	return err	×
303	}	×
304	egressACLOps = append(egressACLOps, ops...)	×
305	}
306	}
307
308	if err := c.OVNNbClient.Transact("add-egress-acls", egressACLOps); err != nil {	×
309	return fmt.Errorf("failed to add egress acls for banp %s: %w", key, err)	×
310	}	×
311	if err := c.deleteUnusedAddrSetForAnp(curEgressAddrSet, desiredEgressAddrSet); err != nil {	×
312	return fmt.Errorf("failed to delete unused egress address set for banp %s: %w", key, err)	×
313	}	×
314
315	return nil	×
316	}
317
318	func (c Controller) handleDeleteBanp(banp v1alpha1.BaselineAdminNetworkPolicy) error {	×
319	c.banpKeyMutex.LockKey(banp.Name)	×
320	defer func() { _ = c.banpKeyMutex.UnlockKey(banp.Name) }()	×
321
322	klog.Infof("handle delete banp %s", banp.Name)	×
323	banpName := getAnpName(banp.Name)	×
324		×
325	// ACLs releated to port_group will be deleted automatically when port_group is deleted	×
326	pgName := strings.ReplaceAll(banpName, "-", ".")	×
327	if err := c.OVNNbClient.DeletePortGroup(pgName); err != nil {	×
328	klog.Errorf("failed to delete port group for banp %s: %v", banpName, err)	×
329	}	×
330
331	if err := c.OVNNbClient.DeleteAddressSets(map[string]string{	×
332	baselineAdminNetworkPolicyKey: fmt.Sprintf("%s/%s", banpName, "ingress"),	×
333	}); err != nil {	×
334	klog.Errorf("failed to delete ingress address set for banp %s: %v", banpName, err)	×
335	return err	×
336	}	×
337
338	if err := c.OVNNbClient.DeleteAddressSets(map[string]string{	×
339	baselineAdminNetworkPolicyKey: fmt.Sprintf("%s/%s", banpName, "egress"),	×
340	}); err != nil {	×
341	klog.Errorf("failed to delete egress address set for banp %s: %v", banpName, err)	×
342	return err	×
343	}	×
344
345	return nil	×
346	}
347
348	func (c Controller) handleUpdateBanp(changed AdminNetworkPolicyChangedDelta) error {	×
349	// Only handle updates that do not affect acls.	×
350	c.banpKeyMutex.LockKey(changed.key)	×
351	defer func() { _ = c.banpKeyMutex.UnlockKey(changed.key) }()	×
352
353	cachedBanp, err := c.banpsLister.Get(changed.key)	×
354	if err != nil {	×
355	if k8serrors.IsNotFound(err) {	×
356	return nil	×
357	}	×
358	klog.Error(err)	×
359	return err	×
360	}
361	desiredBanp := cachedBanp.DeepCopy()	×
362	klog.Infof("handle update banp %s", desiredBanp.Name)	×
363		×
364	banpName := getAnpName(desiredBanp.Name)	×
365	pgName := strings.ReplaceAll(banpName, "-", ".")	×
366		×
367	// The port-group for anp should be updated	×
368	if changed.field == ChangedSubject {	×
369	// The port-group must exist when update anp, this check should never be matched.	×
370	if ok, err := c.OVNNbClient.PortGroupExists(pgName); !ok \|\| err != nil {	×
371	klog.Errorf("port-group for banp %s does not exist when update banp", desiredBanp.Name)	×
372	return err	×
373	}	×
374
375	ports, err := c.fetchSelectedPods(&desiredBanp.Spec.Subject)	×
376	if err != nil {	×
377	klog.Errorf("failed to fetch ports belongs to banp %s: %v", desiredBanp.Name, err)	×
378	return err	×
379	}	×
380
381	if err = c.OVNNbClient.PortGroupSetPorts(pgName, ports); err != nil {	×
382	klog.Errorf("failed to set ports %v to port group %s: %v", ports, pgName, err)	×
383	return err	×
384	}	×
385	}
386
387	// Peer selector in ingress/egress rule has changed, so the corresponding address-set need be updated
388	if changed.field == ChangedIngressRule {	×
389	for index, rule := range desiredBanp.Spec.Ingress {	×
390	// Make sure the rule is changed and go on update	×
391	if rule.Name == changed.ruleNames[index].curRuleName {	×
392	if err := c.setAddrSetForAnpRule(banpName, pgName, rule.Name, index, rule.From, []v1alpha1.AdminNetworkPolicyEgressPeer{}, true, true); err != nil {	×
393	klog.Errorf("failed to set ingress address-set for anp rule %s/%s, %v", banpName, rule.Name, err)	×
394	return err	×
395	}	×
396
397	if changed.ruleNames[index].oldRuleName != "" {	×
398	oldRuleName := changed.ruleNames[index].oldRuleName	×
399	oldAsV4Name, oldAsV6Name := getAnpAddressSetName(pgName, oldRuleName, index, true)	×
400		×
401	if err := c.OVNNbClient.DeleteAddressSet(oldAsV4Name); err != nil {	×
402	klog.Errorf("failed to delete address set %s, %v", oldAsV4Name, err)	×
403	// just record error log	×
404	}	×
405	if err := c.OVNNbClient.DeleteAddressSet(oldAsV6Name); err != nil {	×
406	klog.Errorf("failed to delete address set %s, %v", oldAsV6Name, err)	×
407	}	×
408	}
409	}
410	}
411	}
412
413	if changed.field == ChangedEgressRule {	×
414	for index, rule := range desiredBanp.Spec.Egress {	×
415	// Make sure the rule is changed and go on update	×
416	if rule.Name == changed.ruleNames[index].curRuleName {	×
417	if err := c.setAddrSetForAnpRule(banpName, pgName, rule.Name, index, []v1alpha1.AdminNetworkPolicyIngressPeer{}, rule.To, false, true); err != nil {	×
418	klog.Errorf("failed to set egress address-set for banp rule %s/%s, %v", banpName, rule.Name, err)	×
419	return err	×
420	}	×
421
422	if changed.ruleNames[index].oldRuleName != "" {	×
423	oldRuleName := changed.ruleNames[index].oldRuleName	×
424	oldAsV4Name, oldAsV6Name := getAnpAddressSetName(pgName, oldRuleName, index, false)	×
425		×
426	if err := c.OVNNbClient.DeleteAddressSet(oldAsV4Name); err != nil {	×
427	klog.Errorf("failed to delete address set %s, %v", oldAsV4Name, err)	×
428	// just record error log	×
429	}	×
430	if err := c.OVNNbClient.DeleteAddressSet(oldAsV6Name); err != nil {	×
431	klog.Errorf("failed to delete address set %s, %v", oldAsV6Name, err)	×
432	}	×
433	}
434	}
435	}
436	}
437	return nil	×
438	}

kubeovn / kube-ovn / 17228354799

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous