Conditions on podupdater function of kubernetes autodiscovery (#37431)

* first update for nodePodUpdater with key function * first update for namespacePodUpdater * updating elastic-agent-autodiscovery library to v0.6.7 * updating NOTICE.txt * updating kubernetes_test.go interface functions * updating pod_test.go by removing unused functions * updating receiver name in tests * fixing lint events --------- Co-authored-by: Michal Pristas <[email protected]>
elastic · Jan 15, 2024 · 10b1f7a · 10b1f7a
1 parent d4f5d9f
commit 10b1f7a
Show file tree

Hide file tree

Showing 7 changed files with 84 additions and 20 deletions.
diff --git a/CHANGELOG.next.asciidoc b/CHANGELOG.next.asciidoc
@@ -140,6 +140,7 @@ Setting environmental variable ELASTIC_NETINFO:false in Elastic Agent pod will d
 - Upgrade to Go 1.21.6. {pull}37615[37615]
 - 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]
+- Make more selective the Pod autodiscovery upon node and namespace update events. {issue}37338[37338] {pull}37431[37431]
 
 *Auditbeat*
 

diff --git a/NOTICE.txt b/NOTICE.txt
@@ -12257,11 +12257,11 @@ SOFTWARE.
 
 --------------------------------------------------------------------------------
 Dependency : github.com/elastic/elastic-agent-autodiscover
-Version: v0.6.6
+Version: v0.6.7
 Licence type (autodetected): Apache-2.0
 --------------------------------------------------------------------------------
 
-Contents of probable licence file $GOMODCACHE/github.com/elastic/[email protected].6/LICENSE:
+Contents of probable licence file $GOMODCACHE/github.com/elastic/[email protected].7/LICENSE:
 
                                  Apache License
                            Version 2.0, January 2004

diff --git a/go.mod b/go.mod
@@ -200,7 +200,7 @@ require (
 	github.com/aws/smithy-go v1.13.5
 	github.com/awslabs/kinesis-aggregation/go/v2 v2.0.0-20220623125934-28468a6701b5
 	github.com/elastic/bayeux v1.0.5
-	github.com/elastic/elastic-agent-autodiscover v0.6.6
+	github.com/elastic/elastic-agent-autodiscover v0.6.7
 	github.com/elastic/elastic-agent-libs v0.7.3
 	github.com/elastic/elastic-agent-shipper-client v0.5.1-0.20230228231646-f04347b666f3
 	github.com/elastic/elastic-agent-system-metrics v0.9.1

diff --git a/go.sum b/go.sum
@@ -658,8 +658,8 @@ github.com/elastic/bayeux v1.0.5 h1:UceFq01ipmT3S8DzFK+uVAkbCdiPR0Bqei8qIGmUeY0=
 github.com/elastic/bayeux v1.0.5/go.mod h1:CSI4iP7qeo5MMlkznGvYKftp8M7qqP/3nzmVZoXHY68=
 github.com/elastic/dhcp v0.0.0-20200227161230-57ec251c7eb3 h1:lnDkqiRFKm0rxdljqrj3lotWinO9+jFmeDXIC4gvIQs=
 github.com/elastic/dhcp v0.0.0-20200227161230-57ec251c7eb3/go.mod h1:aPqzac6AYkipvp4hufTyMj5PDIphF3+At8zr7r51xjY=
-github.com/elastic/elastic-agent-autodiscover v0.6.6 h1:P1y0dDpbhJc7Uw/xe85irPEad4Vljygc+y4iSxtqW7A=
-github.com/elastic/elastic-agent-autodiscover v0.6.6/go.mod h1:chulyCAyZb/njMHgzkhC/yWnt8v/Y6eCRUhmFVnsA5o=
+github.com/elastic/elastic-agent-autodiscover v0.6.7 h1:+KVjltN0rPsBrU8b156gV4lOTBgG/vt0efFCFARrf3g=
+github.com/elastic/elastic-agent-autodiscover v0.6.7/go.mod h1:hFeFqneS2r4jD0/QzGkrNk0YVdN0JGh7lCWdsH7zcI4=
 github.com/elastic/elastic-agent-client/v7 v7.6.0 h1:FEn6FjzynW4TIQo5G096Tr7xYK/P5LY9cSS6wRbXZTc=
 github.com/elastic/elastic-agent-client/v7 v7.6.0/go.mod h1:GlUKrbVd/O1CRAZonpBeN3J0RlVqP6VGcrBjFWca+aM=
 github.com/elastic/elastic-agent-libs v0.7.3 h1:tc6JDXYR+2XFMHJVv+7+M0OwAbZPxm3caLJEd943dlE=

diff --git a/libbeat/autodiscover/providers/kubernetes/pod.go b/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)
 	}
 
@@ -407,7 +407,7 @@ func (p *pod) containerPodEvents(flag string, pod *kubernetes.Pod, c *kubernetes
 		ports = []kubernetes.ContainerPort{{ContainerPort: 0}}
 	}
 
-	var events []bus.Event
+	events := []bus.Event{}
 	portsMap := mapstr.M{}
 
 	ShouldPut(meta, "container", cmeta, p.logger)

diff --git a/libbeat/autodiscover/providers/kubernetes/pod_test.go b/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,6 +2120,36 @@ 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
 }

diff --git a/metricbeat/module/kubernetes/util/kubernetes_test.go b/metricbeat/module/kubernetes/util/kubernetes_test.go
@@ -28,6 +28,7 @@ import (
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 
 	"github.com/elastic/elastic-agent-autodiscover/kubernetes"
@@ -180,3 +181,7 @@ func (m *mockWatcher) Store() cache.Store {
 func (m *mockWatcher) Client() k8s.Interface {
 	return nil
 }
+
+func (m *mockWatcher) CachedObject() runtime.Object {
+	return nil
+}