Conditions on podupdater function of kubernetes autodiscovery

CHANGELOG.next.asciidoc

@@ -151,6 +151,8 @@ https://github.com/elastic/beats/compare/v8.8.1\...main[Check the HEAD diff]
 - elasticsearch output now supports `idle_connection_timeout`. {issue}35616[35615] {pull}36843[36843]
 - Upgrade golang/x/net to v0.17.0. Updates the publicsuffix table used by the registered_domain processor. {pull}36969[36969]
 Setting environmental variable ELASTIC_NETINFO:false in Elastic Agent pod will disable the netinfo.enabled option of add_host_metadata processor
+- Upgrade to Go 1.20.11. {pull}37123[37123]
+- Make more selective the Pod autodiscovery upon node and namespace update events. {issue}3738[3738] {pull}37431[37431]
 - Upgrade to Go 1.21.5. {pull}37550[37550]
 - The Elasticsearch output can now configure performance presets with the `preset` configuration field. {pull}37259[37259]
 - Upgrade to elastic-agent-libs v0.7.3 and golang.org/x/crypto v0.17.0. {pull}37544[37544]

libbeat/autodiscover/providers/kubernetes/pod.go

@@ -153,12 +153,12 @@ func NewPodEventer(uuid uuid.UUID, cfg *conf.C, client k8s.Interface, publish fu
 	watcher.AddEventHandler(p)
 
 	if nodeWatcher != nil && (config.Hints.Enabled() || metaConf.Node.Enabled()) {
-		updater := kubernetes.NewNodePodUpdater(p.unlockedUpdate, watcher.Store(), &p.crossUpdate)
+		updater := kubernetes.NewNodePodUpdater(p.unlockedUpdate, watcher.Store(), p.nodeWatcher, &p.crossUpdate)
 		nodeWatcher.AddEventHandler(updater)
 	}
 
 	if namespaceWatcher != nil && (config.Hints.Enabled() || metaConf.Namespace.Enabled()) {
-		updater := kubernetes.NewNamespacePodUpdater(p.unlockedUpdate, watcher.Store(), &p.crossUpdate)
+		updater := kubernetes.NewNamespacePodUpdater(p.unlockedUpdate, watcher.Store(), p.namespaceWatcher, &p.crossUpdate)
 		namespaceWatcher.AddEventHandler(updater)
 	}
 

libbeat/autodiscover/providers/kubernetes/pod_test.go

@@ -26,9 +26,13 @@ import (
 	"github.com/stretchr/testify/assert"
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 	k8sfake "k8s.io/client-go/kubernetes/fake"
 
+	interfaces "k8s.io/client-go/kubernetes"
+	caches "k8s.io/client-go/tools/cache"
+
 	"github.com/elastic/beats/v7/libbeat/autodiscover/template"
 	"github.com/elastic/elastic-agent-autodiscover/bus"
 	"github.com/elastic/elastic-agent-autodiscover/kubernetes"
@@ -1988,6 +1992,11 @@ func TestNamespacePodUpdater(t *testing.T) {
 		}
 	}
 
+	namespace := &kubernetes.Namespace{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "foo",
+		}}
+
 	cases := map[string]struct {
 		pods     []interface{}
 		expected []interface{}
@@ -2014,14 +2023,19 @@ func TestNamespacePodUpdater(t *testing.T) {
 		t.Run(title, func(t *testing.T) {
 			handler := &mockUpdaterHandler{}
 			store := &mockUpdaterStore{objects: c.pods}
-			updater := kubernetes.NewNamespacePodUpdater(handler.OnUpdate, store, &sync.Mutex{})
-
-			namespace := &kubernetes.Namespace{
+			//We simulate an update on the namespace with the addition of one label
+			namespace1 := &kubernetes.Namespace{
 				ObjectMeta: metav1.ObjectMeta{
 					Name: "foo",
-				},
-			}
-			updater.OnUpdate(namespace)
+					Labels: map[string]string{
+						"beta.kubernetes.io/arch": "arm64",
+					},
+				}}
+
+			watcher := &mockUpdaterWatcher{cachedObject: namespace}
+			updater := kubernetes.NewNamespacePodUpdater(handler.OnUpdate, store, watcher, &sync.Mutex{})
+
+			updater.OnUpdate(namespace1)
 
 			assert.EqualValues(t, c.expected, handler.objects)
 		})
@@ -2040,8 +2054,15 @@ func TestNodePodUpdater(t *testing.T) {
 		}
 	}
 
+	node := &kubernetes.Node{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "foo",
+		},
+	}
+
 	cases := map[string]struct {
-		pods     []interface{}
+		pods []interface{}
+
 		expected []interface{}
 	}{
 		"no pods": {},
@@ -2066,14 +2087,21 @@ func TestNodePodUpdater(t *testing.T) {
 		t.Run(title, func(t *testing.T) {
 			handler := &mockUpdaterHandler{}
 			store := &mockUpdaterStore{objects: c.pods}
-			updater := kubernetes.NewNodePodUpdater(handler.OnUpdate, store, &sync.Mutex{})
 
-			node := &kubernetes.Node{
+			//We simulate an update on the node with the addition of one label
+			node1 := &kubernetes.Node{
 				ObjectMeta: metav1.ObjectMeta{
 					Name: "foo",
-				},
-			}
-			updater.OnUpdate(node)
+					Annotations: map[string]string{
+						"beta.kubernetes.io/arch": "arm64",
+					},
+				}}
+
+			watcher := &mockUpdaterWatcher{cachedObject: node}
+			updater := kubernetes.NewNodePodUpdater(handler.OnUpdate, store, watcher, &sync.Mutex{})
+
+			//This is when the update happens.
+			updater.OnUpdate(node1)
 
 			assert.EqualValues(t, c.expected, handler.objects)
 		})
@@ -2092,10 +2120,60 @@ type mockUpdaterStore struct {
 	objects []interface{}
 }
 
+var store caches.Store
+var client interfaces.Interface
+var err error
+
+type mockUpdaterWatcher struct {
+	cachedObject runtime.Object
+}
+
+func (s *mockUpdaterWatcher) CachedObject() runtime.Object {
+	return s.cachedObject
+}
+
+func (s *mockUpdaterWatcher) Client() interfaces.Interface {
+	return client
+}
+
+func (s *mockUpdaterWatcher) Start() error {
+	return err
+}
+
+func (s *mockUpdaterWatcher) Stop() {
+}
+
+func (s *mockUpdaterWatcher) Store() caches.Store {
+	return store
+}
+
+func (s *mockUpdaterWatcher) AddEventHandler(kubernetes.ResourceEventHandler) {
+}
+
 func (s *mockUpdaterStore) List() []interface{} {
 	return s.objects
 }
 
+type mockNamespaceStore struct {
+	objects interface{}
+	exist   bool
+	err     error
+}
+
+func (s *mockNamespaceStore) GetByKey(string) (interface{}, bool, error) {
+	return s.objects, s.exist, s.err
+}
+
+type mockNodeStore struct {
+	objects interface{}
+	exist   bool
+	err     error
+}
+
+func (s *mockNodeStore) GetByKey(string) (interface{}, bool, error) {
+	return s.objects, s.exist, s.err
+}
+
 func NewMockPodEventerManager(pod *pod) EventManager {
 	em := &eventerManager{}
 	em.eventer = pod