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

0.0

/pkg/controller/namespace.go

package controller

import (
        "maps"
        "slices"
        "strings"

        "github.com/scylladb/go-set/strset"
        v1 "k8s.io/api/core/v1"
        "k8s.io/apimachinery/pkg/api/errors"
        metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
        "k8s.io/apimachinery/pkg/labels"
        "k8s.io/client-go/tools/cache"
        "k8s.io/klog/v2"

        "github.com/kubeovn/kube-ovn/pkg/util"
)

func (c *Controller) enqueueAddNamespace(obj any) {
        if c.config.EnableNP {
                for _, np := range c.namespaceMatchNetworkPolicies(obj.(*v1.Namespace)) {
                        c.updateNpQueue.Add(np)
                }
        }

        key := cache.MetaObjectToName(obj.(*v1.Namespace)).String()
        c.addNamespaceQueue.Add(key)
}

func (c *Controller) enqueueDeleteNamespace(obj any) {
        var ns *v1.Namespace
        switch t := obj.(type) {
        case *v1.Namespace:
                ns = t
        case cache.DeletedFinalStateUnknown:
                n, ok := t.Obj.(*v1.Namespace)
                if !ok {
                        klog.Warningf("unexpected object type: %T", t.Obj)
                        return
                }
                ns = n
        default:
                klog.Warningf("unexpected type: %T", obj)
                return
        }

        if c.config.EnableNP {
                for _, np := range c.namespaceMatchNetworkPolicies(ns) {
                        c.updateNpQueue.Add(np)
                }
        }
        if c.config.EnableANP {
                c.updateAnpsByLabelsMatch(ns.Labels, nil)
        }
}

func (c *Controller) enqueueUpdateNamespace(oldObj, newObj any) {
        oldNs := oldObj.(*v1.Namespace)
        newNs := newObj.(*v1.Namespace)
        if oldNs.ResourceVersion == newNs.ResourceVersion {
                return
        }

        if !maps.Equal(oldNs.Labels, newNs.Labels) {
                if c.config.EnableNP {
                        oldNp := c.namespaceMatchNetworkPolicies(oldNs)
                        newNp := c.namespaceMatchNetworkPolicies(newNs)
                        for _, np := range util.DiffStringSlice(oldNp, newNp) {
                                c.updateNpQueue.Add(np)
                        }
                }

                if c.config.EnableANP {
                        c.updateAnpsByLabelsMatch(newObj.(*v1.Namespace).Labels, nil)
                }

                expectSubnets, err := c.getNsExpectSubnets(newNs)
                if err != nil {
                        klog.Errorf("failed to list expected subnets for namespace %s, %v", newNs.Name, err)
                        return
                }

                expectSubnetsSet := strset.New(expectSubnets...)
                existSubnetsSet := strset.New(strings.Split(newNs.Annotations[util.LogicalSwitchAnnotation], ",")...)
                if !expectSubnetsSet.IsEqual(existSubnetsSet) {
                        c.addNamespaceQueue.Add(newNs.Name)
                }
        }

        // in case annotations are removed by other controllers
        if newNs.Annotations == nil || newNs.Annotations[util.LogicalSwitchAnnotation] == "" {
                klog.Warningf("no logical switch annotation for ns %s", newNs.Name)
                c.addNamespaceQueue.Add(newNs.Name)
        }
}

func (c *Controller) handleAddNamespace(key string) error {
        c.nsKeyMutex.LockKey(key)
        defer func() { _ = c.nsKeyMutex.UnlockKey(key) }()
        klog.Infof("handle add/update namespace %s", key)

        cachedNs, err := c.namespacesLister.Get(key)
        if err != nil {
                if errors.IsNotFound(err) {
                        return nil
                }
                klog.Error(err)
                return err
        }
        namespace := cachedNs.DeepCopy()

        var ls string
        var lss, cidrs, excludeIps, ipPoolsAnnotation []string
        subnets, err := c.subnetsLister.List(labels.Everything())
        if err != nil {
                klog.Errorf("failed to list subnets %v", err)
                return err
        }
        ipPoolList, err := c.ippoolLister.List(labels.Everything())
        if err != nil {
                klog.Errorf("failed to list ippools: %v", err)
                return err
        }

        // check if subnet bind ns
        for _, s := range subnets {
                if slices.Contains(s.Spec.Namespaces, key) {
                        lss = append(lss, s.Name)
                        cidrs = append(cidrs, s.Spec.CIDRBlock)
                        excludeIps = append(excludeIps, strings.Join(s.Spec.ExcludeIps, ","))
                }

                // bind subnet with namespaceLabelSeletcor which select the namespace
                for _, nsSelector := range s.Spec.NamespaceSelectors {
                        matchSelector, err := metav1.LabelSelectorAsSelector(&nsSelector)
                        if err != nil {
                                klog.Errorf("failed to convert label selector, %v", err)
                                return err
                        }

                        if matchSelector.Matches(labels.Set(namespace.Labels)) {
                                if slices.Contains(lss, s.Name) {
                                        break
                                }
                                lss = append(lss, s.Name)
                                cidrs = append(cidrs, s.Spec.CIDRBlock)
                                excludeIps = append(excludeIps, strings.Join(s.Spec.ExcludeIps, ","))
                                break
                        }
                }

                // check if subnet is in custom vpc with configured defaultSubnet, then annotate the namespace with this subnet
                if s.Spec.Vpc != "" && s.Spec.Vpc != c.config.ClusterRouter {
                        vpc, err := c.vpcsLister.Get(s.Spec.Vpc)
                        if err != nil {
                                if errors.IsNotFound(err) {
                                        // this subnet is broken (it references a non-existent VPC) - we just ignore it.
                                        klog.Errorf("vpc %q is not found. Ignoring subnet %q: %v", s.Spec.Vpc, s.Name, err)
                                        break
                                }
                                klog.Errorf("failed to get vpc %q: %v", s.Spec.Vpc, err)
                                return err
                        }
                        if s.Name == vpc.Spec.DefaultSubnet {
                                if slices.Contains(vpc.Spec.Namespaces, key) && key != metav1.NamespaceSystem {
                                        lss = append([]string{s.Name}, lss...)
                                }
                        }
                }
        }

        for _, ipPool := range ipPoolList {
                if slices.Contains(ipPool.Spec.Namespaces, key) {
                        ipPoolsAnnotation = append(ipPoolsAnnotation, ipPool.Name)
                }
        }

        if lss == nil {
                // If NS does not belong to any custom VPC, then this NS belongs to the default VPC
                vpc, err := c.vpcsLister.Get(c.config.ClusterRouter)
                if err != nil {
                        klog.Errorf("failed to get default vpc %v", err)
                        return err
                }
                vpcs, err := c.vpcsLister.List(labels.Everything())
                if err != nil {
                        klog.Errorf("failed to list vpc %v", err)
                        return err
                }
                for _, v := range vpcs {
                        if slices.Contains(v.Spec.Namespaces, key) {
                                vpc = v
                                break
                        }
                }

                if vpc.Status.DefaultLogicalSwitch != "" {
                        ls = vpc.Status.DefaultLogicalSwitch
                } else {
                        ls = c.config.DefaultLogicalSwitch
                }
                subnet, err := c.subnetsLister.Get(ls)
                if err != nil {
                        klog.Errorf("failed to get default subnet %v", err)
                        return err
                }
                lss = append(lss, subnet.Name)
                cidrs = append(cidrs, subnet.Spec.CIDRBlock)
                excludeIps = append(excludeIps, strings.Join(subnet.Spec.ExcludeIps, ","))
        }

        if namespace.Annotations[util.LogicalSwitchAnnotation] == strings.Join(lss, ",") &&
                namespace.Annotations[util.CidrAnnotation] == strings.Join(cidrs, ";") &&
                namespace.Annotations[util.ExcludeIpsAnnotation] == strings.Join(excludeIps, ";") &&
                namespace.Annotations[util.IPPoolAnnotation] == strings.Join(ipPoolsAnnotation, ",") {
                return nil
        }

        patch := util.KVPatch{
                util.LogicalSwitchAnnotation: strings.Join(lss, ","),
                util.CidrAnnotation:          strings.Join(cidrs, ";"),
                util.ExcludeIpsAnnotation:    strings.Join(excludeIps, ";"),
        }

        if len(ipPoolsAnnotation) == 0 {
                patch[util.IPPoolAnnotation] = nil
        } else {
                patch[util.IPPoolAnnotation] = strings.Join(ipPoolsAnnotation, ",")
        }

        if err = util.PatchAnnotations(c.config.KubeClient.CoreV1().Namespaces(), key, patch); err != nil {
                klog.Errorf("patch namespace %s failed %v", key, err)
        }
        return err
}

func (c *Controller) getNsExpectSubnets(newNs *v1.Namespace) ([]string, error) {
        var expectSubnets []string

        subnets, err := c.subnetsLister.List(labels.Everything())
        if err != nil {
                klog.Errorf("failed to list subnets %v", err)
                return expectSubnets, err
        }
        for _, subnet := range subnets {
                // ns labels match subnet's selector
                for _, nsSelector := range subnet.Spec.NamespaceSelectors {
                        matchSelector, err := metav1.LabelSelectorAsSelector(&nsSelector)
                        if err != nil {
                                klog.Errorf("failed to convert label selector, %v", err)
                                return expectSubnets, err
                        }

                        if matchSelector.Matches(labels.Set(newNs.Labels)) {
                                expectSubnets = append(expectSubnets, subnet.Name)
                                break
                        }
                }

                // ns included in subnet's namespaces
                if slices.Contains(subnet.Spec.Namespaces, newNs.Name) && !slices.Contains(expectSubnets, subnet.Name) {
                        expectSubnets = append(expectSubnets, subnet.Name)
                }
        }

        return expectSubnets, nil
}

1	package controller
2
3	import (
4	"maps"
5	"slices"
6	"strings"
7
8	"github.com/scylladb/go-set/strset"
9	v1 "k8s.io/api/core/v1"
10	"k8s.io/apimachinery/pkg/api/errors"
11	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
12	"k8s.io/apimachinery/pkg/labels"
13	"k8s.io/client-go/tools/cache"
14	"k8s.io/klog/v2"
15
16	"github.com/kubeovn/kube-ovn/pkg/util"
17	)
18
19	func (c *Controller) enqueueAddNamespace(obj any) {	×
20	if c.config.EnableNP {	×
21	for _, np := range c.namespaceMatchNetworkPolicies(obj.(*v1.Namespace)) {	×
22	c.updateNpQueue.Add(np)	×
23	}	×
24	}
25
26	key := cache.MetaObjectToName(obj.(*v1.Namespace)).String()	×
27	c.addNamespaceQueue.Add(key)	×
28	}
29
30	func (c *Controller) enqueueDeleteNamespace(obj any) {	×
NEW 31	var ns *v1.Namespace	×
NEW 32	switch t := obj.(type) {	×
NEW 33	case *v1.Namespace:	×
NEW 34	ns = t	×
NEW 35	case cache.DeletedFinalStateUnknown:	×
NEW 36	n, ok := t.Obj.(*v1.Namespace)	×
NEW 37	if !ok {	×
NEW 38	klog.Warningf("unexpected object type: %T", t.Obj)	×
NEW 39	return	×
NEW 40	}	×
NEW 41	ns = n	×
NEW 42	default:	×
NEW 43	klog.Warningf("unexpected type: %T", obj)	×
NEW 44	return	×
45	}
46
47	if c.config.EnableNP {	×
NEW 48	for _, np := range c.namespaceMatchNetworkPolicies(ns) {	×
49	c.updateNpQueue.Add(np)	×
50	}	×
51	}
52	if c.config.EnableANP {	×
NEW 53	c.updateAnpsByLabelsMatch(ns.Labels, nil)	×
54	}	×
55	}
56
57	func (c *Controller) enqueueUpdateNamespace(oldObj, newObj any) {	×
58	oldNs := oldObj.(*v1.Namespace)	×
59	newNs := newObj.(*v1.Namespace)	×
60	if oldNs.ResourceVersion == newNs.ResourceVersion {	×
61	return	×
62	}	×
63
64	if !maps.Equal(oldNs.Labels, newNs.Labels) {	×
65	if c.config.EnableNP {	×
66	oldNp := c.namespaceMatchNetworkPolicies(oldNs)	×
67	newNp := c.namespaceMatchNetworkPolicies(newNs)	×
68	for _, np := range util.DiffStringSlice(oldNp, newNp) {	×
69	c.updateNpQueue.Add(np)	×
70	}	×
71	}
72
73	if c.config.EnableANP {	×
74	c.updateAnpsByLabelsMatch(newObj.(*v1.Namespace).Labels, nil)	×
75	}	×
76
77	expectSubnets, err := c.getNsExpectSubnets(newNs)	×
78	if err != nil {	×
79	klog.Errorf("failed to list expected subnets for namespace %s, %v", newNs.Name, err)	×
80	return	×
81	}	×
82
83	expectSubnetsSet := strset.New(expectSubnets...)	×
84	existSubnetsSet := strset.New(strings.Split(newNs.Annotations[util.LogicalSwitchAnnotation], ",")...)	×
85	if !expectSubnetsSet.IsEqual(existSubnetsSet) {	×
86	c.addNamespaceQueue.Add(newNs.Name)	×
87	}	×
88	}
89
90	// in case annotations are removed by other controllers
91	if newNs.Annotations == nil \|\| newNs.Annotations[util.LogicalSwitchAnnotation] == "" {	×
92	klog.Warningf("no logical switch annotation for ns %s", newNs.Name)	×
93	c.addNamespaceQueue.Add(newNs.Name)	×
94	}	×
95	}
96
97	func (c *Controller) handleAddNamespace(key string) error {	×
98	c.nsKeyMutex.LockKey(key)	×
99	defer func() { _ = c.nsKeyMutex.UnlockKey(key) }()	×
100	klog.Infof("handle add/update namespace %s", key)	×
101		×
102	cachedNs, err := c.namespacesLister.Get(key)	×
103	if err != nil {	×
104	if errors.IsNotFound(err) {	×
105	return nil	×
106	}	×
107	klog.Error(err)	×
108	return err	×
109	}
110	namespace := cachedNs.DeepCopy()	×
111		×
112	var ls string	×
113	var lss, cidrs, excludeIps, ipPoolsAnnotation []string	×
114	subnets, err := c.subnetsLister.List(labels.Everything())	×
115	if err != nil {	×
116	klog.Errorf("failed to list subnets %v", err)	×
117	return err	×
118	}	×
119	ipPoolList, err := c.ippoolLister.List(labels.Everything())	×
120	if err != nil {	×
121	klog.Errorf("failed to list ippools: %v", err)	×
122	return err	×
123	}	×
124
125	// check if subnet bind ns
126	for _, s := range subnets {	×
127	if slices.Contains(s.Spec.Namespaces, key) {	×
128	lss = append(lss, s.Name)	×
129	cidrs = append(cidrs, s.Spec.CIDRBlock)	×
130	excludeIps = append(excludeIps, strings.Join(s.Spec.ExcludeIps, ","))	×
131	}	×
132
133	// bind subnet with namespaceLabelSeletcor which select the namespace
134	for _, nsSelector := range s.Spec.NamespaceSelectors {	×
135	matchSelector, err := metav1.LabelSelectorAsSelector(&nsSelector)	×
136	if err != nil {	×
137	klog.Errorf("failed to convert label selector, %v", err)	×
138	return err	×
139	}	×
140
141	if matchSelector.Matches(labels.Set(namespace.Labels)) {	×
142	if slices.Contains(lss, s.Name) {	×
143	break	×
144	}
145	lss = append(lss, s.Name)	×
146	cidrs = append(cidrs, s.Spec.CIDRBlock)	×
147	excludeIps = append(excludeIps, strings.Join(s.Spec.ExcludeIps, ","))	×
148	break	×
149	}
150	}
151
152	// check if subnet is in custom vpc with configured defaultSubnet, then annotate the namespace with this subnet
153	if s.Spec.Vpc != "" && s.Spec.Vpc != c.config.ClusterRouter {	×
154	vpc, err := c.vpcsLister.Get(s.Spec.Vpc)	×
155	if err != nil {	×
156	if errors.IsNotFound(err) {	×
157	// this subnet is broken (it references a non-existent VPC) - we just ignore it.	×
158	klog.Errorf("vpc %q is not found. Ignoring subnet %q: %v", s.Spec.Vpc, s.Name, err)	×
159	break	×
160	}
161	klog.Errorf("failed to get vpc %q: %v", s.Spec.Vpc, err)	×
162	return err	×
163	}
164	if s.Name == vpc.Spec.DefaultSubnet {	×
165	if slices.Contains(vpc.Spec.Namespaces, key) && key != metav1.NamespaceSystem {	×
166	lss = append([]string{s.Name}, lss...)	×
167	}	×
168	}
169	}
170	}
171
172	for _, ipPool := range ipPoolList {	×
173	if slices.Contains(ipPool.Spec.Namespaces, key) {	×
174	ipPoolsAnnotation = append(ipPoolsAnnotation, ipPool.Name)	×
175	}	×
176	}
177
178	if lss == nil {	×
179	// If NS does not belong to any custom VPC, then this NS belongs to the default VPC	×
180	vpc, err := c.vpcsLister.Get(c.config.ClusterRouter)	×
181	if err != nil {	×
182	klog.Errorf("failed to get default vpc %v", err)	×
183	return err	×
184	}	×
185	vpcs, err := c.vpcsLister.List(labels.Everything())	×
186	if err != nil {	×
187	klog.Errorf("failed to list vpc %v", err)	×
188	return err	×
189	}	×
190	for _, v := range vpcs {	×
191	if slices.Contains(v.Spec.Namespaces, key) {	×
192	vpc = v	×
193	break	×
194	}
195	}
196
197	if vpc.Status.DefaultLogicalSwitch != "" {	×
198	ls = vpc.Status.DefaultLogicalSwitch	×
199	} else {	×
200	ls = c.config.DefaultLogicalSwitch	×
201	}	×
202	subnet, err := c.subnetsLister.Get(ls)	×
203	if err != nil {	×
204	klog.Errorf("failed to get default subnet %v", err)	×
205	return err	×
206	}	×
207	lss = append(lss, subnet.Name)	×
208	cidrs = append(cidrs, subnet.Spec.CIDRBlock)	×
209	excludeIps = append(excludeIps, strings.Join(subnet.Spec.ExcludeIps, ","))	×
210	}
211
212	if namespace.Annotations[util.LogicalSwitchAnnotation] == strings.Join(lss, ",") &&	×
213	namespace.Annotations[util.CidrAnnotation] == strings.Join(cidrs, ";") &&	×
214	namespace.Annotations[util.ExcludeIpsAnnotation] == strings.Join(excludeIps, ";") &&	×
215	namespace.Annotations[util.IPPoolAnnotation] == strings.Join(ipPoolsAnnotation, ",") {	×
216	return nil	×
217	}	×
218
219	patch := util.KVPatch{	×
220	util.LogicalSwitchAnnotation: strings.Join(lss, ","),	×
221	util.CidrAnnotation: strings.Join(cidrs, ";"),	×
222	util.ExcludeIpsAnnotation: strings.Join(excludeIps, ";"),	×
223	}	×
224		×
225	if len(ipPoolsAnnotation) == 0 {	×
226	patch[util.IPPoolAnnotation] = nil	×
227	} else {	×
228	patch[util.IPPoolAnnotation] = strings.Join(ipPoolsAnnotation, ",")	×
229	}	×
230
231	if err = util.PatchAnnotations(c.config.KubeClient.CoreV1().Namespaces(), key, patch); err != nil {	×
232	klog.Errorf("patch namespace %s failed %v", key, err)	×
233	}	×
234	return err	×
235	}
236
237	func (c Controller) getNsExpectSubnets(newNs v1.Namespace) ([]string, error) {	×
238	var expectSubnets []string	×
239		×
240	subnets, err := c.subnetsLister.List(labels.Everything())	×
241	if err != nil {	×
242	klog.Errorf("failed to list subnets %v", err)	×
243	return expectSubnets, err	×
244	}	×
245	for _, subnet := range subnets {	×
246	// ns labels match subnet's selector	×
247	for _, nsSelector := range subnet.Spec.NamespaceSelectors {	×
248	matchSelector, err := metav1.LabelSelectorAsSelector(&nsSelector)	×
249	if err != nil {	×
250	klog.Errorf("failed to convert label selector, %v", err)	×
251	return expectSubnets, err	×
252	}	×
253
254	if matchSelector.Matches(labels.Set(newNs.Labels)) {	×
255	expectSubnets = append(expectSubnets, subnet.Name)	×
256	break	×
257	}
258	}
259
260	// ns included in subnet's namespaces
261	if slices.Contains(subnet.Spec.Namespaces, newNs.Name) && !slices.Contains(expectSubnets, subnet.Name) {	×
262	expectSubnets = append(expectSubnets, subnet.Name)	×
263	}	×
264	}
265
266	return expectSubnets, nil	×
267	}

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