plugin/kubernetes: handle tombstones in default processor (#3890)

* handle deletion tombstones in default processor

Signed-off-by: Chris O'Haver <cohaver@infoblox.com>

* fix terminating pod exclusion

Signed-off-by: Chris O'Haver <cohaver@infoblox.com>

1 files changed, 19 insertions, 13 deletions

diff --git a/plugin/kubernetes/object/pod.go b/plugin/kubernetes/object/pod.go
index 9fc9b5726..04cbe1ad2 100644
--- a/plugin/kubernetes/object/pod.go
+++ b/plugin/kubernetes/object/pod.go
@@ -1,6 +1,9 @@
 package object
 
 import (
+	"errors"
+	"fmt"
+
 	api "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 )
@@ -16,30 +19,33 @@ type Pod struct {
 	*Empty
 }
 
+var errPodTerminating = errors.New("pod terminating")
+
 // ToPod returns a function that converts an api.Pod to a *Pod.
 func ToPod(skipCleanup bool) ToFunc {
-	return func(obj interface{}) interface{} {
-		return toPod(skipCleanup, obj)
+	return func(obj interface{}) (interface{}, error) {
+		apiPod, ok := obj.(*api.Pod)
+		if !ok {
+			return nil, fmt.Errorf("unexpected object %v", obj)
+		}
+		pod := toPod(skipCleanup, apiPod)
+		t := apiPod.ObjectMeta.DeletionTimestamp
+		if t != nil && !(*t).Time.IsZero() {
+			// if the pod is in the process of termination, return an error so it can be ignored
+			// during add/update event processing
+			return pod, errPodTerminating
+		}
+		return pod, nil
 	}
 }
 
-func toPod(skipCleanup bool, obj interface{}) interface{} {
-	pod, ok := obj.(*api.Pod)
-	if !ok {
-		return nil
-	}
-
+func toPod(skipCleanup bool, pod *api.Pod) *Pod {
 	p := &Pod{
 		Version:   pod.GetResourceVersion(),
 		PodIP:     pod.Status.PodIP,
 		Namespace: pod.GetNamespace(),
 		Name:      pod.GetName(),
 	}
-	// don't add pods that are being deleted.
-	t := pod.ObjectMeta.DeletionTimestamp
-	if t != nil && !(*t).Time.IsZero() {
-		return nil
-	}
 
 	if !skipCleanup {
 		*pod = api.Pod{}