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,25 @@ type podUpdaterStore interface {
 	List() []interface{}
 }
 
+type UpdateWatcher interface {
+	Deltaobjects() []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 UpdateWatcher
+	locker           sync.Locker
 }
 
 // NewNamespacePodUpdater creates a namespacePodUpdater
-func NewNamespacePodUpdater(handler podUpdaterHandlerFunc, store podUpdaterStore, locker sync.Locker) *namespacePodUpdater {
+func NewNamespacePodUpdater(handler podUpdaterHandlerFunc, store podUpdaterStore, namespacewatcher UpdateWatcher, locker sync.Locker) *namespacePodUpdater {
 	return &namespacePodUpdater{
-		handler: handler,
-		store:   store,
-		locker:  locker,
+		handler:          handler,
+		store:            store,
+		namespacewatcher: namespacewatcher,
+		locker:           locker,
 	}
 }
 
@@ -157,21 +167,34 @@ func (n *namespacePodUpdater) OnUpdate(obj interface{}) {
 		return
 	}
 
-	// n.store.List() returns a snapshot at this point. If a delete is received
-	// from the main watcher, this loop may generate an update event after the
-	// delete is processed, leaving configurations that would never be deleted.
-	// Also this loop can miss updates, what could leave outdated configurations.
-	// Avoid these issues by locking the processing of events from the main watcher.
 	if n.locker != nil {
 		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)
+
+	// Slice includes the old and new version of caching object that changes in the current update event. slice[0] is the old version and slice[1] the new updated one
+	slice := n.namespacewatcher.Deltaobjects()
+	cachednamespaceold, ok := slice[0].(*Namespace)
+
+	if ns.Name == cachednamespaceold.Name && ok {
+		labelscheck := checkMetadata(ns.ObjectMeta.Labels, cachednamespaceold.ObjectMeta.Labels)
+		annotationscheck := checkMetadata(ns.ObjectMeta.Annotations, cachednamespaceold.ObjectMeta.Annotations)
+		// Only if there is a diffrence in Metadata labels or annotations proceed to Pod update
+		if !labelscheck || !annotationscheck {
+			// n.store.List() returns a snapshot at this point. If a delete is received
+			// from the main watcher, this loop may generate an update event after the
+			// delete is processed, leaving configurations that would never be deleted.
+			// Also this loop can miss updates, what could leave outdated configurations.
+			// Avoid these issues by locking the processing of events from the main watcher.
+			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 UpdateWatcher
+	locker      sync.Locker
 }
 
 // NewNodePodUpdater creates a nodePodUpdater
-func NewNodePodUpdater(handler podUpdaterHandlerFunc, store podUpdaterStore, locker sync.Locker) *nodePodUpdater {
+func NewNodePodUpdater(handler podUpdaterHandlerFunc, store podUpdaterStore, nodewatcher UpdateWatcher, locker sync.Locker) *nodePodUpdater {
 	return &nodePodUpdater{
-		handler: handler,
-		store:   store,
-		locker:  locker,
+		handler:     handler,
+		store:       store,
+		nodewatcher: nodewatcher,
+		locker:      locker,
 	}
 }
 
@@ -205,19 +230,31 @@ func (n *nodePodUpdater) OnUpdate(obj interface{}) {
 		return
 	}
 
-	// n.store.List() returns a snapshot at this point. If a delete is received
-	// from the main watcher, this loop may generate an update event after the
-	// delete is processed, leaving configurations that would never be deleted.
-	// Also this loop can miss updates, what could leave outdated configurations.
-	// Avoid these issues by locking the processing of events from the main watcher.
 	if n.locker != nil {
 		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)
+
+	// Slice includes the old and new version of caching object that changes in the current update event. slice[0] is the old version and slice[1] the new updated one
+	slice := n.nodewatcher.Deltaobjects()
+	cachednodeold, ok := slice[0].(*Node)
+
+	if node.Name == cachednodeold.Name && ok {
+		labelscheck := checkMetadata(node.ObjectMeta.Labels, cachednodeold.ObjectMeta.Labels)
+		annotationscheck := checkMetadata(node.ObjectMeta.Annotations, cachednodeold.ObjectMeta.Annotations)
+		// Only if there is a diffrence in Metadata labels or annotations proceed to Pod update
+		if !labelscheck || !annotationscheck {
+			// n.store.List() returns a snapshot at this point. If a delete is received
+			// from the main watcher, this loop may generate an update event after the
+			// delete is processed, leaving configurations that would never be deleted.
+			// Also this loop can miss updates, what could leave outdated configurations.
+			// Avoid these issues by locking the processing of events from the main watcher.
+			for _, pod := range n.store.List() {
+				pod, ok := pod.(*Pod)
+				if ok && pod.Spec.NodeName == node.Name {
+					n.handler(pod)
+				}
+			}
 		}
 	}
 }
@@ -229,3 +266,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{}) {}
+
+// checkMetadata receives labels or annotations maps and checks their equality. Returns True if equal, False if there is a diffrence
+func checkMetadata(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
+
+	// Deltaobjects returns the slice of objects that change during the last updated event
+	Deltaobjects() []runtime.Object
 }
 
 // WatchOptions controls watch behaviors
@@ -91,6 +94,7 @@ type watcher struct {
 	stop     context.CancelFunc
 	handler  ResourceEventHandler
 	logger   *logp.Logger
+	delta    []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 deltaslice []runtime.Object
 	informer, _, err := NewInformer(client, resource, opts, indexers)
 	if err != nil {
 		return nil, err
@@ -132,6 +136,7 @@ func NewNamedWatcher(name string, client kubernetes.Interface, resource Resource
 		store:    store,
 		queue:    queue,
 		ctx:      ctx,
+		delta:    deltaslice,
 		stop:     cancel,
 		logger:   logp.NewLogger("kubernetes"),
 		handler:  NoOpEventHandlerFuncs{},
@@ -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.deltaobjects(o, n)
+
 		},
 	})
 
@@ -178,6 +185,11 @@ func (w *watcher) Client() kubernetes.Interface {
 	return w.client
 }
 
+// Deltaobjects returns the slice of objects that change during the last updated event
+func (w *watcher) Deltaobjects() []runtime.Object {
+	return w.delta
+}
+
 // Start watching pods
 func (w *watcher) Start() error {
 	go w.informer.Run(w.ctx.Done())
@@ -217,6 +229,16 @@ func (w *watcher) enqueue(obj interface{}, state string) {
 	w.queue.Add(&item{key, obj, state})
 }
 
+// deltaobjects creates a slice with the old and the new version of cache objects that are ready to chane on update events
+// returns a delta struct to the watcher. w.delta[0] is the old version and w.delta[1] the new updated one
+func (w *watcher) deltaobjects(o interface{}, n interface{}) {
+	//w.delta[:0] initialises always the delta struct before new assignement
+	w.delta = w.delta[:0]
+	w.delta = append(w.delta, o.(runtime.Object))
+	w.delta = append(w.delta, n.(runtime.Object))
+
+}
+
 // 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()