17227113524

Committed 26 Aug 2025 03:33AM UTC coverage: 21.226% (-0.2%) from 21.39%

Build # 17227113524

Build Type

push

github

Committed by

web-flow

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.

10589 of 49886 relevant lines covered (21.23%)

0.25 hits per line

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

0.0

/pkg/controller/ippool.go

package controller

import (
        "context"
        "reflect"
        "slices"

        corev1 "k8s.io/api/core/v1"
        k8serrors "k8s.io/apimachinery/pkg/api/errors"
        metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
        "k8s.io/apimachinery/pkg/labels"
        "k8s.io/apimachinery/pkg/types"
        "k8s.io/client-go/tools/cache"
        "k8s.io/klog/v2"

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

func (c *Controller) enqueueAddIPPool(obj any) {
        key := cache.MetaObjectToName(obj.(*kubeovnv1.IPPool)).String()
        klog.V(3).Infof("enqueue add ippool %s", key)
        c.addOrUpdateIPPoolQueue.Add(key)
}

func (c *Controller) enqueueDeleteIPPool(obj any) {
        var ippool *kubeovnv1.IPPool
        switch t := obj.(type) {
        case *kubeovnv1.IPPool:
                ippool = t
        case cache.DeletedFinalStateUnknown:
                i, ok := t.Obj.(*kubeovnv1.IPPool)
                if !ok {
                        klog.Warningf("unexpected object type: %T", t.Obj)
                        return
                }
                ippool = i
        default:
                klog.Warningf("unexpected type: %T", obj)
                return
        }

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

func (c *Controller) enqueueUpdateIPPool(oldObj, newObj any) {
        oldIPPool := oldObj.(*kubeovnv1.IPPool)
        newIPPool := newObj.(*kubeovnv1.IPPool)
        if !slices.Equal(oldIPPool.Spec.Namespaces, newIPPool.Spec.Namespaces) ||
                !slices.Equal(oldIPPool.Spec.IPs, newIPPool.Spec.IPs) {
                key := cache.MetaObjectToName(newIPPool).String()
                klog.V(3).Infof("enqueue update ippool %s", key)
                c.addOrUpdateIPPoolQueue.Add(key)
        }
}

func (c *Controller) handleAddOrUpdateIPPool(key string) error {
        c.ippoolKeyMutex.LockKey(key)
        defer func() { _ = c.ippoolKeyMutex.UnlockKey(key) }()

        cachedIPPool, err := c.ippoolLister.Get(key)
        if err != nil {
                if k8serrors.IsNotFound(err) {
                        return nil
                }
                klog.Error(err)
                return err
        }
        klog.Infof("handle add/update ippool %s", cachedIPPool.Name)

        ippool := cachedIPPool.DeepCopy()
        ippool.Status.EnsureStandardConditions()
        if err = c.ipam.AddOrUpdateIPPool(ippool.Spec.Subnet, ippool.Name, ippool.Spec.IPs); err != nil {
                klog.Errorf("failed to add/update ippool %s with IPs %v in subnet %s: %v", ippool.Name, ippool.Spec.IPs, ippool.Spec.Subnet, err)
                if patchErr := c.patchIPPoolStatusCondition(ippool, "UpdateIPAMFailed", err.Error()); patchErr != nil {
                        klog.Error(patchErr)
                }
                return err
        }

        v4a, v4u, v6a, v6u, v4as, v4us, v6as, v6us := c.ipam.IPPoolStatistics(ippool.Spec.Subnet, ippool.Name)
        ippool.Status.V4AvailableIPs = v4a
        ippool.Status.V4UsingIPs = v4u
        ippool.Status.V6AvailableIPs = v6a
        ippool.Status.V6UsingIPs = v6u
        ippool.Status.V4AvailableIPRange = v4as
        ippool.Status.V4UsingIPRange = v4us
        ippool.Status.V6AvailableIPRange = v6as
        ippool.Status.V6UsingIPRange = v6us

        if err = c.patchIPPoolStatusCondition(ippool, "UpdateIPAMSucceeded", ""); err != nil {
                klog.Error(err)
                return err
        }

        for _, ns := range ippool.Spec.Namespaces {
                c.addNamespaceQueue.Add(ns)
        }

        return nil
}

func (c *Controller) handleDeleteIPPool(ippool *kubeovnv1.IPPool) error {
        c.ippoolKeyMutex.LockKey(ippool.Name)
        defer func() { _ = c.ippoolKeyMutex.UnlockKey(ippool.Name) }()

        klog.Infof("handle delete ippool %s", ippool.Name)
        c.ipam.RemoveIPPool(ippool.Spec.Subnet, ippool.Name)

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

        for _, ns := range namespaces {
                if len(ns.Annotations) == 0 {
                        continue
                }
                if ns.Annotations[util.IPPoolAnnotation] == ippool.Name {
                        c.enqueueAddNamespace(ns)
                }
        }

        return nil
}

func (c *Controller) handleUpdateIPPoolStatus(key string) error {
        c.ippoolKeyMutex.LockKey(key)
        defer func() { _ = c.ippoolKeyMutex.UnlockKey(key) }()

        cachedIPPool, err := c.ippoolLister.Get(key)
        if err != nil {
                if k8serrors.IsNotFound(err) {
                        return nil
                }
                klog.Error(err)
                return err
        }

        ippool := cachedIPPool.DeepCopy()
        v4a, v4u, v6a, v6u, v4as, v4us, v6as, v6us := c.ipam.IPPoolStatistics(ippool.Spec.Subnet, ippool.Name)
        ippool.Status.V4AvailableIPs = v4a
        ippool.Status.V4UsingIPs = v4u
        ippool.Status.V6AvailableIPs = v6a
        ippool.Status.V6UsingIPs = v6u
        ippool.Status.V4AvailableIPRange = v4as
        ippool.Status.V4UsingIPRange = v4us
        ippool.Status.V6AvailableIPRange = v6as
        ippool.Status.V6UsingIPRange = v6us
        if reflect.DeepEqual(ippool.Status, cachedIPPool.Status) {
                return nil
        }

        return c.patchIPPoolStatus(ippool)
}

func (c Controller) patchIPPoolStatusCondition(ippool *kubeovnv1.IPPool, reason, errMsg string) error {
        if errMsg != "" {
                ippool.Status.SetError(reason, errMsg)
                ippool.Status.NotReady(reason, errMsg)
                c.recorder.Eventf(ippool, corev1.EventTypeWarning, reason, errMsg)
        } else {
                ippool.Status.Ready(reason, "")
                c.recorder.Eventf(ippool, corev1.EventTypeNormal, reason, errMsg)
        }

        return c.patchIPPoolStatus(ippool)
}

func (c Controller) patchIPPoolStatus(ippool *kubeovnv1.IPPool) error {
        bytes, err := ippool.Status.Bytes()
        if err != nil {
                klog.Errorf("failed to generate json representation for status of ippool %s: %v", ippool.Name, err)
                return err
        }
        if _, err = c.config.KubeOvnClient.KubeovnV1().IPPools().Patch(context.Background(), ippool.Name, types.MergePatchType, bytes, metav1.PatchOptions{}, "status"); err != nil {
                klog.Errorf("failed to patch status of ippool %s: %v", ippool.Name, err)
                return err
        }

        return nil
}

1	package controller
2
3	import (
4	"context"
5	"reflect"
6	"slices"
7
8	corev1 "k8s.io/api/core/v1"
9	k8serrors "k8s.io/apimachinery/pkg/api/errors"
10	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
11	"k8s.io/apimachinery/pkg/labels"
12	"k8s.io/apimachinery/pkg/types"
13	"k8s.io/client-go/tools/cache"
14	"k8s.io/klog/v2"
15
16	kubeovnv1 "github.com/kubeovn/kube-ovn/pkg/apis/kubeovn/v1"
17	"github.com/kubeovn/kube-ovn/pkg/util"
18	)
19
20	func (c *Controller) enqueueAddIPPool(obj any) {	×
21	key := cache.MetaObjectToName(obj.(*kubeovnv1.IPPool)).String()	×
22	klog.V(3).Infof("enqueue add ippool %s", key)	×
23	c.addOrUpdateIPPoolQueue.Add(key)	×
24	}	×
25
26	func (c *Controller) enqueueDeleteIPPool(obj any) {	×
NEW 27	var ippool *kubeovnv1.IPPool	×
NEW 28	switch t := obj.(type) {	×
NEW 29	case *kubeovnv1.IPPool:	×
NEW 30	ippool = t	×
NEW 31	case cache.DeletedFinalStateUnknown:	×
NEW 32	i, ok := t.Obj.(*kubeovnv1.IPPool)	×
NEW 33	if !ok {	×
NEW 34	klog.Warningf("unexpected object type: %T", t.Obj)	×
NEW 35	return	×
NEW 36	}	×
NEW 37	ippool = i	×
NEW 38	default:	×
NEW 39	klog.Warningf("unexpected type: %T", obj)	×
NEW 40	return	×
41	}
42
43	klog.V(3).Infof("enqueue delete ippool %s", cache.MetaObjectToName(ippool).String())	×
44	c.deleteIPPoolQueue.Add(ippool)	×
45	}
46
47	func (c *Controller) enqueueUpdateIPPool(oldObj, newObj any) {	×
48	oldIPPool := oldObj.(*kubeovnv1.IPPool)	×
49	newIPPool := newObj.(*kubeovnv1.IPPool)	×
50	if !slices.Equal(oldIPPool.Spec.Namespaces, newIPPool.Spec.Namespaces) \|\|	×
51	!slices.Equal(oldIPPool.Spec.IPs, newIPPool.Spec.IPs) {	×
52	key := cache.MetaObjectToName(newIPPool).String()	×
53	klog.V(3).Infof("enqueue update ippool %s", key)	×
54	c.addOrUpdateIPPoolQueue.Add(key)	×
55	}	×
56	}
57
58	func (c *Controller) handleAddOrUpdateIPPool(key string) error {	×
59	c.ippoolKeyMutex.LockKey(key)	×
60	defer func() { _ = c.ippoolKeyMutex.UnlockKey(key) }()	×
61
62	cachedIPPool, err := c.ippoolLister.Get(key)	×
63	if err != nil {	×
64	if k8serrors.IsNotFound(err) {	×
65	return nil	×
66	}	×
67	klog.Error(err)	×
68	return err	×
69	}
70	klog.Infof("handle add/update ippool %s", cachedIPPool.Name)	×
71		×
72	ippool := cachedIPPool.DeepCopy()	×
73	ippool.Status.EnsureStandardConditions()	×
74	if err = c.ipam.AddOrUpdateIPPool(ippool.Spec.Subnet, ippool.Name, ippool.Spec.IPs); err != nil {	×
75	klog.Errorf("failed to add/update ippool %s with IPs %v in subnet %s: %v", ippool.Name, ippool.Spec.IPs, ippool.Spec.Subnet, err)	×
76	if patchErr := c.patchIPPoolStatusCondition(ippool, "UpdateIPAMFailed", err.Error()); patchErr != nil {	×
77	klog.Error(patchErr)	×
78	}	×
79	return err	×
80	}
81
82	v4a, v4u, v6a, v6u, v4as, v4us, v6as, v6us := c.ipam.IPPoolStatistics(ippool.Spec.Subnet, ippool.Name)	×
83	ippool.Status.V4AvailableIPs = v4a	×
84	ippool.Status.V4UsingIPs = v4u	×
85	ippool.Status.V6AvailableIPs = v6a	×
86	ippool.Status.V6UsingIPs = v6u	×
87	ippool.Status.V4AvailableIPRange = v4as	×
88	ippool.Status.V4UsingIPRange = v4us	×
89	ippool.Status.V6AvailableIPRange = v6as	×
90	ippool.Status.V6UsingIPRange = v6us	×
91		×
92	if err = c.patchIPPoolStatusCondition(ippool, "UpdateIPAMSucceeded", ""); err != nil {	×
93	klog.Error(err)	×
94	return err	×
95	}	×
96
97	for _, ns := range ippool.Spec.Namespaces {	×
98	c.addNamespaceQueue.Add(ns)	×
99	}	×
100
101	return nil	×
102	}
103
104	func (c Controller) handleDeleteIPPool(ippool kubeovnv1.IPPool) error {	×
105	c.ippoolKeyMutex.LockKey(ippool.Name)	×
106	defer func() { _ = c.ippoolKeyMutex.UnlockKey(ippool.Name) }()	×
107
108	klog.Infof("handle delete ippool %s", ippool.Name)	×
109	c.ipam.RemoveIPPool(ippool.Spec.Subnet, ippool.Name)	×
110		×
111	namespaces, err := c.namespacesLister.List(labels.Everything())	×
112	if err != nil {	×
113	klog.Errorf("failed to list namespaces: %v", err)	×
114	return err	×
115	}	×
116
117	for _, ns := range namespaces {	×
118	if len(ns.Annotations) == 0 {	×
119	continue	×
120	}
121	if ns.Annotations[util.IPPoolAnnotation] == ippool.Name {	×
122	c.enqueueAddNamespace(ns)	×
123	}	×
124	}
125
126	return nil	×
127	}
128
129	func (c *Controller) handleUpdateIPPoolStatus(key string) error {	×
130	c.ippoolKeyMutex.LockKey(key)	×
131	defer func() { _ = c.ippoolKeyMutex.UnlockKey(key) }()	×
132
133	cachedIPPool, err := c.ippoolLister.Get(key)	×
134	if err != nil {	×
135	if k8serrors.IsNotFound(err) {	×
136	return nil	×
137	}	×
138	klog.Error(err)	×
139	return err	×
140	}
141
142	ippool := cachedIPPool.DeepCopy()	×
143	v4a, v4u, v6a, v6u, v4as, v4us, v6as, v6us := c.ipam.IPPoolStatistics(ippool.Spec.Subnet, ippool.Name)	×
144	ippool.Status.V4AvailableIPs = v4a	×
145	ippool.Status.V4UsingIPs = v4u	×
146	ippool.Status.V6AvailableIPs = v6a	×
147	ippool.Status.V6UsingIPs = v6u	×
148	ippool.Status.V4AvailableIPRange = v4as	×
149	ippool.Status.V4UsingIPRange = v4us	×
150	ippool.Status.V6AvailableIPRange = v6as	×
151	ippool.Status.V6UsingIPRange = v6us	×
152	if reflect.DeepEqual(ippool.Status, cachedIPPool.Status) {	×
153	return nil	×
154	}	×
155
156	return c.patchIPPoolStatus(ippool)	×
157	}
158
159	func (c Controller) patchIPPoolStatusCondition(ippool *kubeovnv1.IPPool, reason, errMsg string) error {	×
160	if errMsg != "" {	×
161	ippool.Status.SetError(reason, errMsg)	×
162	ippool.Status.NotReady(reason, errMsg)	×
163	c.recorder.Eventf(ippool, corev1.EventTypeWarning, reason, errMsg)	×
164	} else {	×
165	ippool.Status.Ready(reason, "")	×
166	c.recorder.Eventf(ippool, corev1.EventTypeNormal, reason, errMsg)	×
167	}	×
168
169	return c.patchIPPoolStatus(ippool)	×
170	}
171
172	func (c Controller) patchIPPoolStatus(ippool *kubeovnv1.IPPool) error {	×
173	bytes, err := ippool.Status.Bytes()	×
174	if err != nil {	×
175	klog.Errorf("failed to generate json representation for status of ippool %s: %v", ippool.Name, err)	×
176	return err	×
177	}	×
178	if _, err = c.config.KubeOvnClient.KubeovnV1().IPPools().Patch(context.Background(), ippool.Name, types.MergePatchType, bytes, metav1.PatchOptions{}, "status"); err != nil {	×
179	klog.Errorf("failed to patch status of ippool %s: %v", ippool.Name, err)	×
180	return err	×
181	}	×
182
183	return nil	×
184	}

kubeovn / kube-ovn / 17227113524

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