Fix handling of pods having DeletionTimestamp set (#7119) (#7131)

Signed-off-by: Bartosz Borkowski <bartebor@wp.pl>
Co-authored-by: Bartosz Borkowski <bartosz.borkowski@grupawp.pl>

plugin/kubernetes/informer_test.go

@@ -3,6 +3,7 @@ package kubernetes
 import (
 	"fmt"
 	"testing"
+	"time"
 
 	"github.com/coredns/coredns/plugin/kubernetes/object"
 
@@ -118,3 +119,75 @@ func testProcessor(t *testing.T, processor cache.ProcessFunc, idx cache.Indexer)
 		t.Fatal("tombstone deleted object found in index")
 	}
 }
+
+func TestDefaultProcessorWithPod(t *testing.T) {
+	pbuild := object.DefaultProcessor(object.ToPod, nil)
+	reh := cache.ResourceEventHandlerFuncs{}
+	idx := cache.NewIndexer(cache.DeletionHandlingMetaNamespaceKeyFunc, cache.Indexers{})
+	processor := pbuild(idx, reh)
+	testProcessorWithPod(t, processor, idx)
+}
+
+func testProcessorWithPod(t *testing.T, processor cache.ProcessFunc, idx cache.Indexer) {
+	obj := &api.Pod{
+		ObjectMeta: metav1.ObjectMeta{Name: "pod1", Namespace: "test1"},
+		Spec: api.PodSpec{
+			Containers: []api.Container{
+				api.Container{},
+			},
+		},
+	}
+	obj2 := &api.Pod{
+		ObjectMeta: metav1.ObjectMeta{Name: "pod1", Namespace: "test1", DeletionTimestamp: &metav1.Time{Time: time.Now()}},
+		Spec: api.PodSpec{
+			Containers: []api.Container{
+				api.Container{},
+			},
+		},
+	}
+
+	// Add the pod
+	err := processor(cache.Deltas{
+		{
+			Type:   cache.Added,
+			Object: obj.DeepCopy(),
+		},
+	}, false)
+	if err != nil {
+		t.Fatalf("add pod failed: %v", err)
+	}
+	got, exists, err := idx.Get(obj)
+	if err != nil {
+		t.Fatalf("get added pod failed: %v", err)
+	}
+	if !exists {
+		t.Fatal("added pod not found in index")
+	}
+	_, ok := got.(*object.Pod)
+	if !ok {
+		t.Fatal("pod in index was incorrect type")
+	}
+
+	// Update and delete pod having DeletionTimestamp set in one batch
+	err = processor(cache.Deltas{
+		{
+			Type:   cache.Updated,
+			Object: obj2.DeepCopy(),
+		},
+		{
+			Type:   cache.Deleted,
+			Object: obj2.DeepCopy(),
+		},
+	}, false)
+
+	if err != nil {
+		t.Fatalf("update or delete failed: %v", err)
+	}
+	_, exists, err = idx.Get(obj)
+	if err != nil {
+		t.Fatalf("get pod failed: %v", err)
+	}
+	if exists {
+		t.Fatal("deleted pod still exists in the index")
+	}
+}

plugin/kubernetes/object/informer.go

@@ -40,6 +40,9 @@ func DefaultProcessor(convert ToFunc, recordLatency *EndpointLatencyRecorder) Pr
 				case cache.Sync, cache.Added, cache.Updated:
 					obj, err := convert(d.Object.(meta.Object))
 					if err != nil {
+						if err == errPodTerminating {
+							continue
+						}
 						return err
 					}
 					if old, exists, err := clientState.Get(obj); err == nil && exists {