controller: handle tombstone objects in delete handlers (antrea-io#7964)

Signed-off-by: OmAmbole009 <omambole09@gmail.com>

Author: OmAmbole009

pkg/controller/egress/controller.go

@@ -459,7 +459,19 @@ func (c *EgressController) updateEgress(old, cur interface{}) {
 
 // deleteEgress processes Egress DELETE events and deletes corresponding EgressGroup.
 func (c *EgressController) deleteEgress(obj interface{}) {
-	egress := obj.(*egressv1beta1.Egress)
+	egress, ok := obj.(*egressv1beta1.Egress)
+	if !ok {
+		tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
+		if !ok {
+			klog.V(2).InfoS("Error decoding object when deleting Egress, invalid type", "object", obj)
+			return
+		}
+		egress, ok = tombstone.Obj.(*egressv1beta1.Egress)
+		if !ok {
+			klog.V(2).InfoS("Error decoding object tombstone when deleting Egress, invalid type", "object", tombstone.Obj)
+			return
+		}
+	}
 	klog.InfoS("Processing Egress DELETE event", "egress", egress.Name)
 	c.egressGroupStore.Delete(egress.Name)
 	// Unregister the group from the grouping interface.

pkg/controller/egress/controller_test.go

@@ -934,3 +934,22 @@ func TestUpdateEgressAllocatedCondition(t *testing.T) {
 		})
 	}
 }
+
+func TestDeleteEgress_Tombstone(t *testing.T) {
+	egress := newEgress("egressA", "", "", &metav1.LabelSelector{}, nil, nil)
+	egress.UID = "uidA"
+	controller := newController(nil, []runtime.Object{egress})
+
+	// Simulate the controller having processed the Egress ADD event.
+	controller.addEgress(egress)
+	_, found, _ := controller.egressGroupStore.Get(egress.Name)
+	require.True(t, found)
+
+	// Deliver the delete as a cache.DeletedFinalStateUnknown tombstone, as the
+	// informer does after a watch reconnect. This exercises the tombstone path
+	// in deleteEgress rather than the direct *Egress type assertion.
+	controller.deleteEgress(cache.DeletedFinalStateUnknown{
+		Key: egress.Name,
+		Obj: egress,
+	})
+}

pkg/controller/externalippool/controller.go

@@ -416,7 +416,19 @@ func (c *ExternalIPPoolController) updateExternalIPPool(_, cur interface{}) {
 // deleteExternalIPPool processes ExternalIPPool DELETE events. It deletes the IPAllocator for the pool and triggers
 // reconciliation of all consumers that refer to the pool.
 func (c *ExternalIPPoolController) deleteExternalIPPool(obj interface{}) {
-	pool := obj.(*antreacrds.ExternalIPPool)
+	pool, ok := obj.(*antreacrds.ExternalIPPool)
+	if !ok {
+		tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
+		if !ok {
+			klog.V(2).InfoS("Error decoding object when deleting ExternalIPPool, invalid type", "object", obj)
+			return
+		}
+		pool, ok = tombstone.Obj.(*antreacrds.ExternalIPPool)
+		if !ok {
+			klog.V(2).InfoS("Error decoding object tombstone when deleting ExternalIPPool, invalid type", "object", tombstone.Obj)
+			return
+		}
+	}
 	klog.InfoS("Processing ExternalIPPool DELETE event", "pool", pool.Name, "ipRanges", pool.Spec.IPRanges)
 	c.deleteIPAllocator(pool.Name)
 	// Call consumers to reclaim the IPs allocated from the pool.

pkg/controller/externalippool/controller_test.go

@@ -632,3 +632,19 @@ func TestExternalIPPoolController_RestoreIPAllocations(t *testing.T) {
 		})
 	}
 }
+
+func TestDeleteExternalIPPool_Tombstone(t *testing.T) {
+	pool := newExternalIPPool("pool1", "10.10.10.0/24", "", "")
+	c := newController([]runtime.Object{pool})
+
+	// Simulate the controller having created an allocator for the pool.
+	c.createOrUpdateIPAllocator(pool)
+	require.True(t, c.IPPoolExists(pool.Name))
+
+	// Deliver the delete as a tombstone, as the informer does after a watch reconnect.
+	c.deleteExternalIPPool(cache.DeletedFinalStateUnknown{
+		Key: pool.Name,
+		Obj: pool,
+	})
+	assert.False(t, c.IPPoolExists(pool.Name))
+}

pkg/controller/externalnode/controller.go

@@ -119,7 +119,19 @@ func (c *ExternalNodeController) enqueueExternalNodeUpdate(oldObj interface{}, n
 }
 
 func (c *ExternalNodeController) enqueueExternalNodeDelete(obj interface{}) {
-	en := obj.(*v1alpha1.ExternalNode)
+	en, ok := obj.(*v1alpha1.ExternalNode)
+	if !ok {
+		tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
+		if !ok {
+			klog.V(2).InfoS("Error decoding object when deleting ExternalNode, invalid type", "object", obj)
+			return
+		}
+		en, ok = tombstone.Obj.(*v1alpha1.ExternalNode)
+		if !ok {
+			klog.V(2).InfoS("Error decoding object tombstone when deleting ExternalNode, invalid type", "object", tombstone.Obj)
+			return
+		}
+	}
 	key, _ := keyFunc(en)
 	c.queue.Add(key)
 	klog.InfoS("Enqueued ExternalNode DELETE event", "ExternalNode", klog.KObj(en))

pkg/controller/externalnode/controller_test.go

@@ -28,6 +28,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/util/wait"
 	k8stesting "k8s.io/client-go/testing"
+	"k8s.io/client-go/tools/cache"
 
 	"antrea.io/antrea/v2/pkg/apis/crd/v1alpha1"
 	"antrea.io/antrea/v2/pkg/apis/crd/v1alpha2"
@@ -663,3 +664,24 @@ func newExternalNodeController(objects []runtime.Object) *ExternalNodeController
 	externalEntityInformer := informerFactory.Crd().V1alpha2().ExternalEntities()
 	return NewExternalNodeController(crdClient, externalNodeInformer, externalEntityInformer)
 }
+
+func TestEnqueueExternalNodeDelete_Tombstone(t *testing.T) {
+	externalNode := &v1alpha1.ExternalNode{
+		ObjectMeta: metav1.ObjectMeta{Name: "vm1", Namespace: "ns1"},
+		Spec: v1alpha1.ExternalNodeSpec{
+			Interfaces: []v1alpha1.NetworkInterface{{IPs: []string{"1.1.1.2"}}},
+		},
+	}
+	controller := newExternalNodeController(nil)
+	expectedKey, _ := keyFunc(externalNode)
+
+	// Deliver the delete as a tombstone, as the informer does after a watch reconnect.
+	controller.enqueueExternalNodeDelete(cache.DeletedFinalStateUnknown{
+		Key: expectedKey,
+		Obj: externalNode,
+	})
+
+	require.Equal(t, 1, controller.queue.Len())
+	key, _ := controller.queue.Get()
+	assert.Equal(t, expectedKey, key)
+}

pkg/controller/labelidentity/controller.go

@@ -116,7 +116,19 @@ func (c *Controller) addLabelIdentity(obj interface{}) {
 }
 
 func (c *Controller) deleteLabelIdentity(obj interface{}) {
-	labelIdentity := obj.(*mcv1alpha1.LabelIdentity)
+	labelIdentity, ok := obj.(*mcv1alpha1.LabelIdentity)
+	if !ok {
+		tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
+		if !ok {
+			klog.V(2).InfoS("Error decoding object when deleting LabelIdentity, invalid type", "object", obj)
+			return
+		}
+		labelIdentity, ok = tombstone.Obj.(*mcv1alpha1.LabelIdentity)
+		if !ok {
+			klog.V(2).InfoS("Error decoding object tombstone when deleting LabelIdentity, invalid type", "object", tombstone.Obj)
+			return
+		}
+	}
 	klog.InfoS("Processing LabelIdentity DELETE event", "label", labelIdentity.Spec.Label)
 	c.labelIdentityIndex.DeleteLabelIdentity(labelIdentity.Spec.Label)
 }

pkg/controller/labelidentity/controller_test.go

@@ -21,6 +21,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/tools/cache"
 
 	mcv1alpha1 "antrea.io/antrea/v2/multicluster/apis/multicluster/v1alpha1"
 	fakeversioned "antrea.io/antrea/v2/multicluster/pkg/client/clientset/versioned/fake"
@@ -81,3 +82,21 @@ func TestGroupEntityControllerRun(t *testing.T) {
 		return index.HasSynced()
 	}, 1*time.Second, 10*time.Millisecond, "LabelIdentityIndex hasn't been synced in 1 second after starting LabelIdentityController")
 }
+
+func TestDeleteLabelIdentity_Tombstone(t *testing.T) {
+	index := NewLabelIdentityIndex()
+	mcClient := fakeversioned.NewSimpleClientset()
+	mcInformerFactory := crdinformers.NewSharedInformerFactory(mcClient, informerDefaultResync)
+	c := NewLabelIdentityController(index, mcInformerFactory.Multicluster().V1alpha1().LabelIdentities())
+
+	// Add the label identity to the index first.
+	c.addLabelIdentity(labelIdentityA)
+	assert.Contains(t, index.labelIdentities, labelIdentityA.Spec.Label)
+
+	// Deliver the delete as a tombstone, as the informer does after a watch reconnect.
+	c.deleteLabelIdentity(cache.DeletedFinalStateUnknown{
+		Key: labelIdentityA.Name,
+		Obj: labelIdentityA,
+	})
+	assert.NotContains(t, index.labelIdentities, labelIdentityA.Spec.Label)
+}

pkg/controller/traceflow/controller.go

@@ -162,7 +162,19 @@ func (c *Controller) updateTraceflow(_, curObj interface{}) {
 }
 
 func (c *Controller) deleteTraceflow(old interface{}) {
-	tf := old.(*crdv1beta1.Traceflow)
+	tf, ok := old.(*crdv1beta1.Traceflow)
+	if !ok {
+		tombstone, ok := old.(cache.DeletedFinalStateUnknown)
+		if !ok {
+			klog.V(2).InfoS("Unexpected object type when deleting Traceflow", "object", old)
+			return
+		}
+		tf, ok = tombstone.Obj.(*crdv1beta1.Traceflow)
+		if !ok {
+			klog.V(2).InfoS("Unexpected object type in tombstone when deleting Traceflow", "object", tombstone.Obj)
+			return
+		}
+	}
 	klog.Infof("Processing Traceflow %s DELETE event", tf.Name)
 	c.deallocateTagForTF(tf)
 }

pkg/controller/traceflow/controller_test.go

@@ -29,6 +29,7 @@ import (
 	clientset "k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/kubernetes/fake"
 	k8stesting "k8s.io/client-go/testing"
+	"k8s.io/client-go/tools/cache"
 
 	crdv1beta1 "antrea.io/antrea/v2/pkg/apis/crd/v1beta1"
 	"antrea.io/antrea/v2/pkg/client/clientset/versioned"
@@ -197,3 +198,34 @@ func newCRDClientset() *fakeversioned.Clientset {
 
 	return client
 }
+
+func TestDeleteTraceflow_Tombstone(t *testing.T) {
+	tfc := newController()
+
+	tf := &crdv1beta1.Traceflow{
+		ObjectMeta: metav1.ObjectMeta{Name: "tf1", UID: "uid1"},
+		Spec: crdv1beta1.TraceflowSpec{
+			Source:      crdv1beta1.Source{Namespace: "ns1", Pod: "pod1"},
+			Destination: crdv1beta1.Destination{Namespace: "ns2", Pod: "pod2"},
+		},
+		Status: crdv1beta1.TraceflowStatus{
+			DataplaneTag: 5,
+		},
+	}
+
+	// Simulate the tag being tracked as running.
+	tfc.runningTraceflowsMutex.Lock()
+	tfc.runningTraceflows[uint8(tf.Status.DataplaneTag)] = tf.Name
+	tfc.runningTraceflowsMutex.Unlock()
+
+	// Deliver the delete as a tombstone, as the informer does after a watch reconnect.
+	tfc.deleteTraceflow(cache.DeletedFinalStateUnknown{
+		Key: tf.Name,
+		Obj: tf,
+	})
+
+	tfc.runningTraceflowsMutex.Lock()
+	_, exists := tfc.runningTraceflows[uint8(tf.Status.DataplaneTag)]
+	tfc.runningTraceflowsMutex.Unlock()
+	assert.False(t, exists, "expected tag to be deallocated after tombstone delete")
+}