Skip add_kubernetes_metadata processor if kubernetes metadata are ale…

…ady there (#27689) (#27738)

* Skip add_kubernetes_metadata processor if kubernetes metadata are already there

(cherry picked from commit 4d068b4)

Co-authored-by: Michael Katsoulis <michaelkatsoulis88@gmail.com>

CHANGELOG.next.asciidoc

@@ -21,6 +21,8 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Add daemonset.name in pods controlled by DaemonSets {pull}26808[26808], {issue}25816[25816]
 - Kubernetes autodiscover fails in node scope if node name cannot be discovered {pull}26947[26947]
 - Loading Kibana assets (dashboards, index templates) rely on Saved Object API. So to provide a reliable service, Beats can only import and export dashboards using at least Kibana 7.15. {issue}20672[20672] {pull}27220[27220]
+- Loading Kibana assets (dashboards, index templates) rely on Saved Object API. So to provide a reliable service, Beats can only import and export dasbhboards using at least Kibana 7.15. {issue}20672[20672] {pull}27220[27220]
+- Skip add_kubernetes_metadata processor when kubernetes metadata are already present {pull}27689[27689]
 
 *Auditbeat*
 

libbeat/processors/add_kubernetes_metadata/kubernetes.go

@@ -88,6 +88,14 @@ func isKubernetesAvailableWithRetry(client k8sclient.Interface) bool {
 	}
 }
 
+// kubernetesMetadataExist checks whether an event is already enriched with kubernetes metadata
+func kubernetesMetadataExist(event *beat.Event) bool {
+	if _, err := event.GetValue("kubernetes"); err != nil {
+		return false
+	}
+	return true
+}
+
 // New constructs a new add_kubernetes_metadata processor.
 func New(cfg *common.Config) (processors.Processor, error) {
 	config, err := newProcessorConfig(cfg, Indexing)
@@ -251,6 +259,10 @@ func (k *kubernetesAnnotator) Run(event *beat.Event) (*beat.Event, error) {
 	if !k.kubernetesAvailable {
 		return event, nil
 	}
+	if kubernetesMetadataExist(event) {
+		k.log.Debug("Skipping add_kubernetes_metadata processor as kubernetes metadata already exist")
+		return event, nil
+	}
 	index := k.matchers.MetadataIndex(event.Fields)
 	if index == "" {
 		k.log.Debug("No container match string, not adding kubernetes data")

libbeat/processors/add_kubernetes_metadata/kubernetes_test.go

@@ -29,8 +29,8 @@ import (
 	"github.com/elastic/beats/v7/libbeat/logp"
 )
 
-// Test metadata updates don't replace existing pod metrics
-func TestAnnotatorDeepUpdate(t *testing.T) {
+// Test Annotator is skipped if kubernetes metadata already exist
+func TestAnnotatorSkipped(t *testing.T) {
 	cfg := common.MustNewConfigFrom(map[string]interface{}{
 		"lookup_fields": []string{"kubernetes.pod.name"},
 	})
@@ -53,8 +53,7 @@ func TestAnnotatorDeepUpdate(t *testing.T) {
 			"kubernetes": common.MapStr{
 				"pod": common.MapStr{
 					"labels": common.MapStr{
-						"dont":     "replace",
-						"original": "fields",
+						"added": "should not",
 					},
 				},
 			},
@@ -85,10 +84,6 @@ func TestAnnotatorDeepUpdate(t *testing.T) {
 					"a": 1,
 					"b": 2,
 				},
-				"labels": common.MapStr{
-					"dont":     "replace",
-					"original": "fields",
-				},
 			},
 		},
 	}, event.Fields)