Add support for processors in hints based k8s autodiscover

changelog/fragments/1691060806-Processor-support-for-hints-autodiscover.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: feature
+
+# Change summary; a 80ish characters long description of the change.
+summary: Add support for processors in hints' based k8s autodiscover
+
+# 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/elastic/elastic-agent/pull/3107
+
+# 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/elastic/elastic-agent/issues/2959

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

@@ -154,15 +154,6 @@ func GenerateHintsMapping(hints mapstr.M, kubeMeta mapstr.M, logger *logp.Logger
 		_, _ = integrationHints.Put("container_logs.enabled", true)
 	}
 
-	// TODO: add support for processors
-	// Processors should be data_stream specific.
-	// Add a basic processor as a base like:
-	//- add_fields:
-	//	  target: kubernetes
-	//	  fields:
-	//	    hints: true
-	// Blocked by https://github.com/elastic/elastic-agent/issues/735
-
 	integrationHost := builder.getFromMeta(builder.getHost(hints), kubeMeta)
 	if integrationHost != "" {
 		_, _ = integrationHints.Put(host, integrationHost)
@@ -258,3 +249,26 @@ func GenerateHintsMapping(hints mapstr.M, kubeMeta mapstr.M, logger *logp.Logger
 
 	return hintsMapping
 }
+
+// Generates the hints and processor mappings from provided pod annotation map
+func GetHintsMapping(k8sMapping map[string]interface{}, logger *logp.Logger, prefix string, cID string) hintsData {
+	hintData := hintsData{
+		composableMapping: mapstr.M{},
+		processors:        []mapstr.M{},
+	}
+
+	if ann, ok := k8sMapping["annotations"]; ok {
+		annotations, _ := ann.(mapstr.M)
+		hints := utils.GenerateHints(annotations, "", prefix)
+		if len(hints) > 0 {
+			logger.Debugf("Extracted hints are :%v", hints)
+			hintData.composableMapping = GenerateHintsMapping(hints, k8sMapping, logger, cID)
+			logger.Debugf("Generated hints mappings are :%v", hintData.composableMapping)
+
+			hintData.processors = utils.GetConfigs(annotations, prefix, processorhints)
+			logger.Debugf("Generated Processors are :%v", hintData.processors)
+		}
+
+	}
+	return hintData
+}

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

@@ -366,3 +366,108 @@ func TestGenerateHintsMappingWithLogStream(t *testing.T) {
 
 	assert.Equal(t, expected, hintsMapping)
 }
+
+func TestGenerateHintsMappingWithProcessors(t *testing.T) {
+	logger := getLogger()
+	pod := &kubernetes.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "testpod",
+			UID:       types.UID(uid),
+			Namespace: "testns",
+			Labels: map[string]string{
+				"foo":        "bar",
+				"with-dash":  "dash-value",
+				"with/slash": "some/path",
+			},
+			Annotations: map[string]string{
+				"app":                      "production",
+				"co.elastic.hints/package": "apache",
+				"co.elastic.hints/processors.1.add_fields.target":      "project",
+				"co.elastic.hints/processors.1.add_fields.fields.name": "myproject",
+				"co.elastic.hints/processors.rename.fields.0.from":     "a.g",
+				"co.elastic.hints/processors.rename.fields.1.to":       "e.d",
+				"co.elastic.hints/processors.rename.fail_on_error":     "false",
+			},
+		},
+		TypeMeta: metav1.TypeMeta{
+			Kind:       "Pod",
+			APIVersion: "v1",
+		},
+		Spec: kubernetes.PodSpec{
+			NodeName: "testnode",
+		},
+		Status: kubernetes.PodStatus{PodIP: "127.0.0.5"},
+	}
+
+	mapping := map[string]interface{}{
+		"namespace": pod.GetNamespace(),
+		"pod": mapstr.M{
+			"uid":  string(pod.GetUID()),
+			"name": pod.GetName(),
+			"ip":   pod.Status.PodIP,
+		},
+		"namespace_annotations": mapstr.M{
+			"nsa": "nsb",
+		},
+		"labels": mapstr.M{
+			"foo":        "bar",
+			"with-dash":  "dash-value",
+			"with/slash": "some/path",
+		},
+		"annotations": mapstr.M{
+			"app": "production",
+			"co": mapstr.M{
+				"elastic": mapstr.M{
+					"hints/package": "apache",
+					"hints/processors": mapstr.M{
+						"add_fields": mapstr.M{
+							"target": "project",
+							"name":   "myproject",
+						},
+						"rename": mapstr.M{
+							"fail_on_error": "false",
+							"fields": mapstr.M{
+								"from": "a.g",
+								"to":   "e.d",
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+
+	expected_hints := mapstr.M{
+		"container_id": "asdfghjkl",
+		"apache": mapstr.M{
+			"container_logs": mapstr.M{
+				"enabled": true,
+			},
+			"enabled": true,
+		},
+	}
+
+	expected_procesors := []mapstr.M{
+		0: {
+			"add_fields": mapstr.M{
+				"target": "project",
+				"name":   "myproject",
+			},
+		},
+		1: {
+			"rename": mapstr.M{
+				"fail_on_error": "false",
+				"fields": mapstr.M{
+					"from": "a.g",
+					"to":   "e.d",
+				},
+			},
+		},
+	}
+
+	hintData := GetHintsMapping(mapping, logger, "co.elastic", "asdfghjkl")
+
+	assert.Equal(t, expected_hints, hintData.composableMapping)
+	assert.Equal(t, expected_procesors, hintData.processors)
+
+}

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

@@ -24,6 +24,10 @@ import (
 	"github.com/elastic/elastic-agent/internal/pkg/composable"
 )
 
+const (
+	processorhints = "hints/processors"
+)
+
 type pod struct {
 	watcher           kubernetes.Watcher
 	nodeWatcher       kubernetes.Watcher
@@ -50,6 +54,12 @@ type providerData struct {
 	processors []map[string]interface{}
 }
 
+// hintsData hold the generated mapping data needed for hints based autodsicovery
+type hintsData struct {
+	composableMapping mapstr.M
+	processors        []mapstr.M
+}
+
 // NewPodEventer creates an eventer that can discover and process pod objects
 func NewPodEventer(
 	comm composable.DynamicProviderComm,
@@ -210,7 +220,7 @@ func (p *pod) emitRunning(pod *kubernetes.Pod) {
 				if len(hints) > 0 {
 					p.logger.Debugf("Extracted hints are :%v", hints)
 					hintsMapping := GenerateHintsMapping(hints, data.mapping, p.logger, "")
-					p.logger.Debugf("Generated hints mappings are :%v", hintsMapping)
+					p.logger.Debugf("Generated Pods' hints mappings are :%v", hintsMapping)
 					_ = p.comm.AddOrUpdate(
 						data.uid,
 						PodPriority,
@@ -423,12 +433,14 @@ func generateContainerData(
 
 		// add container metadata under kubernetes.container.* to
 		// make them available to dynamic var resolution
+
 		containerMeta := mapstr.M{
 			"id":      c.ID,
 			"name":    c.Spec.Name,
 			"image":   c.Spec.Image,
 			"runtime": c.Runtime,
 		}
+
 		if len(c.Spec.Ports) > 0 {
 			for _, port := range c.Spec.Ports {
 				_, _ = containerMeta.Put("port", fmt.Sprintf("%v", port.ContainerPort))
@@ -437,22 +449,28 @@ func generateContainerData(
 
 				if config.Hints.Enabled { // This is "hints based autodiscovery flow"
 					if !managed {
-						hintsMapping := getHintsMapping(k8sMapping, logger, config.Prefix, c.ID)
-						if len(hintsMapping) > 0 {
+						hintData := GetHintsMapping(k8sMapping, logger, config.Prefix, c.ID)
+						if len(hintData.composableMapping) > 0 {
+							if len(hintData.processors) > 0 {
+								processors = updateProcessors(hintData.processors, processors)
+							}
 							_ = comm.AddOrUpdate(
 								eventID,
 								PodPriority,
-								map[string]interface{}{"hints": hintsMapping},
+								map[string]interface{}{"hints": hintData.composableMapping},
 								processors,
 							)
 						} else if config.Hints.DefaultContainerLogs {
 							// in case of no package detected in the hints fallback to the generic log collection
-							_, _ = hintsMapping.Put("container_logs.enabled", true)
-							_, _ = hintsMapping.Put("container_id", c.ID)
+							_, _ = hintData.composableMapping.Put("container_logs.enabled", true)
+							_, _ = hintData.composableMapping.Put("container_id", c.ID)
+							if len(hintData.processors) > 0 {
+								processors = updateProcessors(hintData.processors, processors)
+							}
 							_ = comm.AddOrUpdate(
 								eventID,
 								PodPriority,
-								map[string]interface{}{"hints": hintsMapping},
+								map[string]interface{}{"hints": hintData.composableMapping},
 								processors,
 							)
 						}
@@ -465,22 +483,28 @@ func generateContainerData(
 			k8sMapping["container"] = containerMeta
 			if config.Hints.Enabled { // This is "hints based autodiscovery flow"
 				if !managed {
-					hintsMapping := getHintsMapping(k8sMapping, logger, config.Prefix, c.ID)
-					if len(hintsMapping) > 0 {
+					hintData := GetHintsMapping(k8sMapping, logger, config.Prefix, c.ID)
+					if len(hintData.composableMapping) > 0 {
+						if len(hintData.processors) > 0 {
+							processors = updateProcessors(hintData.processors, processors)
+						}
 						_ = comm.AddOrUpdate(
 							eventID,
 							PodPriority,
-							map[string]interface{}{"hints": hintsMapping},
+							map[string]interface{}{"hints": hintData.composableMapping},
 							processors,
 						)
 					} else if config.Hints.DefaultContainerLogs {
 						// in case of no package detected in the hints fallback to the generic log collection
-						_, _ = hintsMapping.Put("container_logs.enabled", true)
-						_, _ = hintsMapping.Put("container_id", c.ID)
+						_, _ = hintData.composableMapping.Put("container_logs.enabled", true)
+						_, _ = hintData.composableMapping.Put("container_id", c.ID)
+						if len(hintData.processors) > 0 {
+							processors = updateProcessors(hintData.processors, processors)
+						}
 						_ = comm.AddOrUpdate(
 							eventID,
 							PodPriority,
-							map[string]interface{}{"hints": hintsMapping},
+							map[string]interface{}{"hints": hintData.composableMapping},
 							processors,
 						)
 					}
@@ -492,16 +516,11 @@ func generateContainerData(
 	}
 }
 
-func getHintsMapping(k8sMapping map[string]interface{}, logger *logp.Logger, prefix string, cID string) mapstr.M {
-	hintsMapping := mapstr.M{}
-	if ann, ok := k8sMapping["annotations"]; ok {
-		annotations, _ := ann.(mapstr.M)
-		hints := utils.GenerateHints(annotations, "", prefix)
-		if len(hints) > 0 {
-			logger.Debugf("Extracted hints are :%v", hints)
-			hintsMapping = GenerateHintsMapping(hints, k8sMapping, logger, cID)
-			logger.Debugf("Generated hints mappings are :%v", hintsMapping)
-		}
+// Updates processors map with any additional processors identfied from annotations
+func updateProcessors(newprocessors []mapstr.M, processors []map[string]interface{}) []map[string]interface{} {
+	for _, processor := range newprocessors {
+		processors = append(processors, processor)
 	}
-	return hintsMapping
+
+	return processors
 }