Conditions on podupdater functions of kubernetes autodiscovery

kubernetes/eventhandler.go

@@ -18,23 +18,27 @@
 package kubernetes
 
 import (
+	"reflect"
 	"sync"
+
+	"k8s.io/apimachinery/pkg/runtime"
 )
 
 // ResourceEventHandler can handle notifications for events that happen to a
 // resource. The events are informational only, so you can't return an
 // error.
-//  * OnAdd is called when an object is added.
-//  * OnUpdate is called when an object is modified. Note that oldObj is the
-//      last known state of the object-- it is possible that several changes
-//      were combined together, so you can't use this to see every single
-//      change. OnUpdate is also called when a re-list happens, and it will
-//      get called even if nothing changed. This is useful for periodically
-//      evaluating or syncing something.
-//  * OnDelete will get the final state of the item if it is known, otherwise
-//      it will get an object of type DeletedFinalStateUnknown. This can
-//      happen if the watch is closed and misses the delete event and we don't
-//      notice the deletion until the subsequent re-list.
+//   - OnAdd is called when an object is added.
+//   - OnUpdate is called when an object is modified. Note that oldObj is the
+//     last known state of the object-- it is possible that several changes
+//     were combined together, so you can't use this to see every single
+//     change. OnUpdate is also called when a re-list happens, and it will
+//     get called even if nothing changed. This is useful for periodically
+//     evaluating or syncing something.
+//   - OnDelete will get the final state of the item if it is known, otherwise
+//     it will get an object of type DeletedFinalStateUnknown. This can
+//     happen if the watch is closed and misses the delete event and we don't
+//     notice the deletion until the subsequent re-list.
+//
 // idea: allow the On* methods to return an error so that the RateLimited WorkQueue
 // idea: can requeue the failed event processing.
 type ResourceEventHandler interface {
@@ -134,19 +138,27 @@ type podUpdaterStore interface {
 	List() []interface{}
 }
 
+// OldobjectWatcher is the interface  that an object needs to implement to be
+// able to use Oldobject cache event function from watcher.
+type OldobjectWatcher interface {
+	Oldobject() runtime.Object
+}
+
 // namespacePodUpdater notifies updates on pods when their namespaces are updated.
 type namespacePodUpdater struct {
-	handler podUpdaterHandlerFunc
-	store   podUpdaterStore
-	locker  sync.Locker
+	handler          podUpdaterHandlerFunc
+	store            podUpdaterStore
+	namespacewatcher OldobjectWatcher
+	locker           sync.Locker
 }
 
 // NewNamespacePodUpdater creates a namespacePodUpdater
-func NewNamespacePodUpdater(handler podUpdaterHandlerFunc, store podUpdaterStore, locker sync.Locker) *namespacePodUpdater {
+func NewNamespacePodUpdater(handler podUpdaterHandlerFunc, store podUpdaterStore, namespacewatcher OldobjectWatcher, locker sync.Locker) *namespacePodUpdater {
 	return &namespacePodUpdater{
-		handler: handler,
-		store:   store,
-		locker:  locker,
+		handler:          handler,
+		store:            store,
+		namespacewatcher: namespacewatcher,
+		locker:           locker,
 	}
 }
 
@@ -166,12 +178,23 @@ func (n *namespacePodUpdater) OnUpdate(obj interface{}) {
 		n.locker.Lock()
 		defer n.locker.Unlock()
 	}
-	for _, pod := range n.store.List() {
-		pod, ok := pod.(*Pod)
-		if ok && pod.Namespace == ns.Name {
-			n.handler(pod)
+	oldobject := n.namespacewatcher.Oldobject()
+	cachednamespaceold, ok := oldobject.(*Namespace)
+
+	if ok && ns.Name == cachednamespaceold.Name {
+		labelscheck := isEqualMetadata(ns.ObjectMeta.Labels, cachednamespaceold.ObjectMeta.Labels)
+		annotationscheck := isEqualMetadata(ns.ObjectMeta.Annotations, cachednamespaceold.ObjectMeta.Annotations)
+		// Only if there is a difference in Metadata labels or annotations proceed to Pod update
+		if !labelscheck || !annotationscheck {
+			for _, pod := range n.store.List() {
+				pod, ok := pod.(*Pod)
+				if ok && pod.Namespace == ns.Name {
+					n.handler(pod)
+				}
+			}
 		}
 	}
+
 }
 
 // OnAdd handles add events on namespaces. Nothing to do, if pods are added to this
@@ -184,17 +207,19 @@ func (*namespacePodUpdater) OnDelete(interface{}) {}
 
 // nodePodUpdater notifies updates on pods when their nodes are updated.
 type nodePodUpdater struct {
-	handler podUpdaterHandlerFunc
-	store   podUpdaterStore
-	locker  sync.Locker
+	handler     podUpdaterHandlerFunc
+	store       podUpdaterStore
+	nodewatcher OldobjectWatcher
+	locker      sync.Locker
 }
 
 // NewNodePodUpdater creates a nodePodUpdater
-func NewNodePodUpdater(handler podUpdaterHandlerFunc, store podUpdaterStore, locker sync.Locker) *nodePodUpdater {
+func NewNodePodUpdater(handler podUpdaterHandlerFunc, store podUpdaterStore, nodewatcher OldobjectWatcher, locker sync.Locker) *nodePodUpdater {
 	return &nodePodUpdater{
-		handler: handler,
-		store:   store,
-		locker:  locker,
+		handler:     handler,
+		store:       store,
+		nodewatcher: nodewatcher,
+		locker:      locker,
 	}
 }
 
@@ -214,10 +239,20 @@ func (n *nodePodUpdater) OnUpdate(obj interface{}) {
 		n.locker.Lock()
 		defer n.locker.Unlock()
 	}
-	for _, pod := range n.store.List() {
-		pod, ok := pod.(*Pod)
-		if ok && pod.Spec.NodeName == node.Name {
-			n.handler(pod)
+	oldobject := n.nodewatcher.Oldobject()
+	cachednodeold, ok := oldobject.(*Node)
+
+	if ok && node.Name == cachednodeold.Name {
+		labelscheck := isEqualMetadata(node.ObjectMeta.Labels, cachednodeold.ObjectMeta.Labels)
+		annotationscheck := isEqualMetadata(node.ObjectMeta.Annotations, cachednodeold.ObjectMeta.Annotations)
+		// Only if there is a difference in Metadata labels or annotations proceed to Pod update
+		if !labelscheck || !annotationscheck {
+			for _, pod := range n.store.List() {
+				pod, ok := pod.(*Pod)
+				if ok && pod.Spec.NodeName == node.Name {
+					n.handler(pod)
+				}
+			}
 		}
 	}
 }
@@ -229,3 +264,9 @@ func (*nodePodUpdater) OnAdd(interface{}) {}
 // OnDelete handles delete events on namespaces. Nothing to do, if pods are deleted from this
 // namespace they will generate their own delete events.
 func (*nodePodUpdater) OnDelete(interface{}) {}
+
+// isEqualMetadata receives labels or annotations maps and checks their equality. Returns True if equal, False if there is a diffrence
+func isEqualMetadata(newmetadata, oldmetadata map[string]string) bool {
+	check := reflect.DeepEqual(newmetadata, oldmetadata)
+	return check
+}

kubernetes/metadata/pod.go

@@ -74,6 +74,7 @@ func (p *pod) Generate(obj kubernetes.Resource, opts ...FieldOptions) mapstr.M {
 		"kubernetes": p.GenerateK8s(obj, opts...),
 	}
 	meta.DeepUpdate(ecsFields)
+
 	return meta
 }
 

kubernetes/watcher.go

@@ -59,6 +59,9 @@ type Watcher interface {
 
 	// Client returns the kubernetes client object used by the watcher
 	Client() kubernetes.Interface
+
+	// Oldobject returns the old object before change during the last updated event
+	Oldobject() runtime.Object
 }
 
 // WatchOptions controls watch behaviors
@@ -83,14 +86,15 @@ type item struct {
 }
 
 type watcher struct {
-	client   kubernetes.Interface
-	informer cache.SharedInformer
-	store    cache.Store
-	queue    workqueue.Interface
-	ctx      context.Context
-	stop     context.CancelFunc
-	handler  ResourceEventHandler
-	logger   *logp.Logger
+	client    kubernetes.Interface
+	informer  cache.SharedInformer
+	store     cache.Store
+	queue     workqueue.Interface
+	ctx       context.Context
+	stop      context.CancelFunc
+	handler   ResourceEventHandler
+	logger    *logp.Logger
+	oldobject runtime.Object
 }
 
 // NewWatcher initializes the watcher client to provide a events handler for
@@ -106,7 +110,7 @@ func NewWatcher(client kubernetes.Interface, resource Resource, opts WatchOption
 func NewNamedWatcher(name string, client kubernetes.Interface, resource Resource, opts WatchOptions, indexers cache.Indexers) (Watcher, error) {
 	var store cache.Store
 	var queue workqueue.Interface
-
+	var oldobject runtime.Object
 	informer, _, err := NewInformer(client, resource, opts, indexers)
 	if err != nil {
 		return nil, err
@@ -127,14 +131,15 @@ func NewNamedWatcher(name string, client kubernetes.Interface, resource Resource
 
 	ctx, cancel := context.WithCancel(context.TODO())
 	w := &watcher{
-		client:   client,
-		informer: informer,
-		store:    store,
-		queue:    queue,
-		ctx:      ctx,
-		stop:     cancel,
-		logger:   logp.NewLogger("kubernetes"),
-		handler:  NoOpEventHandlerFuncs{},
+		client:    client,
+		informer:  informer,
+		store:     store,
+		queue:     queue,
+		ctx:       ctx,
+		oldobject: oldobject,
+		stop:      cancel,
+		logger:    logp.NewLogger("kubernetes"),
+		handler:   NoOpEventHandlerFuncs{},
 	}
 
 	w.informer.AddEventHandler(cache.ResourceEventHandlerFuncs{
@@ -157,6 +162,8 @@ func NewNamedWatcher(name string, client kubernetes.Interface, resource Resource
 				// state should just be deduped by autodiscover and not stop/started periodically as would be the case with an update.
 				w.enqueue(n, add)
 			}
+			w.oldobjectreturn(o)
+
 		},
 	})
 
@@ -178,6 +185,11 @@ func (w *watcher) Client() kubernetes.Interface {
 	return w.client
 }
 
+// Oldbject returns the old object in cache during the last updated event
+func (w *watcher) Oldobject() runtime.Object {
+	return w.oldobject
+}
+
 // Start watching pods
 func (w *watcher) Start() error {
 	go w.informer.Run(w.ctx.Done())
@@ -217,6 +229,15 @@ func (w *watcher) enqueue(obj interface{}, state string) {
 	w.queue.Add(&item{key, obj, state})
 }
 
+// oldobjectreturn returns the old version of cache objects before change on update events
+func (w *watcher) oldobjectreturn(o interface{}) {
+	if old, ok := o.(runtime.Object); !ok {
+		utilruntime.HandleError(fmt.Errorf("expected object in cache  got %#v", o))
+	} else {
+		w.oldobject = old
+	}
+}
+
 // process gets the top of the work queue and processes the object that is received.
 func (w *watcher) process(_ context.Context) bool {
 	obj, quit := w.queue.Get()