Emit Pod data only for running Pods in the Kubernetes provider (#6011)

Author: swiatekm

changelog/fragments/1731500837-pod-status-updates.yaml

@@ -0,0 +1,32 @@
+# Kind can be one of:
+# - breaking-change: a change to previously-documented behavior
+# - deprecation: functionality that is being removed in a later release
+# - bug-fix: fixes a problem in a previous version
+# - enhancement: extends functionality but does not break or fix existing behavior
+# - feature: new functionality
+# - known-issue: problems that we are aware of in a given version
+# - security: impacts on the security of a product or a user’s deployment.
+# - upgrade: important information for someone upgrading from a prior version
+# - other: does not fit into any of the other categories
+kind: enhancement
+
+# Change summary; a 80ish characters long description of the change.
+summary: Emit Pod data only for running Pods in the Kubernetes provider
+
+# Long description; in case the summary is not enough to describe the change
+# this field accommodate a description without length limits.
+# NOTE: This field will be rendered only for breaking-change and known-issue kinds at the moment.
+#description:
+
+# Affected component; usually one of "elastic-agent", "fleet-server", "filebeat", "metricbeat", "auditbeat", "all", etc.
+component: elastic-agent
+
+# PR URL; optional; the PR number that added the changeset.
+# If not present is automatically filled by the tooling finding the PR where this changelog fragment has been added.
+# NOTE: the tooling supports backports, so it's able to fill the original PR number instead of the backport PR number.
+# Please provide it if you are adding a fragment for a different PR.
+#pr: https://github.com/owner/repo/1234
+
+# Issue URL; optional; the GitHub issue related to this changeset (either closes or is part of).
+# If not present is automatically filled by the tooling with the issue linked to the PR number.
+#issue: https://github.com/owner/repo/1234

internal/pkg/composable/providers/kubernetes/pod.go

@@ -9,6 +9,8 @@ import (
 	"sync"
 	"time"
 
+	v1 "k8s.io/api/core/v1"
+
 	"k8s.io/apimachinery/pkg/runtime/schema"
 
 	"github.com/elastic/elastic-agent-autodiscover/kubernetes"
@@ -224,7 +226,9 @@ func (p *pod) Stop() {
 }
 
 func (p *pod) emitRunning(pod *kubernetes.Pod) {
-
+	if pod.Status.Phase == v1.PodPending || pod.Status.Phase == v1.PodUnknown {
+		return
+	}
 	namespaceAnnotations := kubernetes.PodNamespaceAnnotations(pod, p.namespaceWatcher)
 
 	data := generatePodData(pod, p.metagen, namespaceAnnotations)

internal/pkg/composable/providers/kubernetes/pod_test.go

@@ -8,6 +8,7 @@ import (
 	"context"
 	"fmt"
 	"testing"
+	"time"
 
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -516,6 +517,63 @@ func TestPodEventer_Namespace_Node_Watcher(t *testing.T) {
 	}
 }
 
+func TestPodEventer_OnlyRunningPods(t *testing.T) {
+	client := k8sfake.NewSimpleClientset()
+
+	log, err := logger.New("service-eventer-test", true)
+	assert.NoError(t, err)
+
+	providerDataChan := make(chan providerData, 1)
+
+	comm := MockDynamicComm{
+		context.TODO(),
+		providerDataChan,
+	}
+
+	var cfg Config
+	cfg.InitDefaults()
+
+	eventer, err := NewPodEventer(&comm, &cfg, log, client, "cluster", false)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	pod := &kubernetes.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "testpod",
+			UID:       types.UID(uid),
+			Namespace: "testns",
+		},
+		TypeMeta: metav1.TypeMeta{
+			Kind:       "Pod",
+			APIVersion: "v1",
+		},
+		Spec: kubernetes.PodSpec{
+			NodeName: "testnode",
+		},
+		Status: kubernetes.PodStatus{
+			PodIP: "127.0.0.5",
+			Phase: v1.PodPending,
+		},
+	}
+
+	eventer.OnUpdate(pod)
+	select {
+	case <-providerDataChan:
+		assert.Fail(t, "should not receive update for Pending Pod")
+	default:
+	}
+
+	// set status to Running, we should get an update now
+	pod.Status.Phase = v1.PodRunning
+	eventer.OnUpdate(pod)
+	select {
+	case <-providerDataChan:
+	case <-time.After(time.Second * 5):
+		assert.Fail(t, "should receive update for Pending Pod")
+	}
+}
+
 // MockDynamicComm is used in tests.
 type MockDynamicComm struct {
 	context.Context