fix(otel): fix beatprocessor to honor when conditions (#50555) (#50561)

* otel(beatprocessor): honor  conditions on inner Beat processors

The beat processor instantiated each processor by calling
its New constructor directly, bypassing the conditional wrapping
that processors.Namespace.Register applies on the standard libbeat
path. The `when` field was unpacked into the config but silently
dropped, so processors ran unconditionally.

Wrap each constructor with `processors.NewConditional` before invoking
it so `when` configuration is honored, matching the behavior of
standalone beats.

* add changelog

* better wording

(cherry picked from commit 7bbe8ee)

Co-authored-by: Mauri de Souza Meneguzzo <mauri870@gmail.com>

Author: mergify[bot]

changelog/fragments/1778182489-honor-when-conditions-in-beatprocessor.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: bug-fix
+
+# Change summary; a 80ish characters long description of the change.
+summary: Fix OTel Beat processor to honor `when` conditions
+
+# 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: all
+
+# 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/issues/1234

x-pack/filebeat/tests/integration/otel_test.go

@@ -2174,3 +2174,65 @@ exporters:
 	processorInstanceCount := col.ObservedLogs().FilterMessageSnippet("Configured Beat processor").Len()
 	assert.Equal(t, 1, processorInstanceCount, "expected beat processor to be configured once (shared instance), but got %d", processorInstanceCount)
 }
+
+// TestBeatProcessorWhenCondition verifies that `when` conditions are
+// honored for beat processors.
+func TestBeatProcessorWhenCondition(t *testing.T) {
+	cfg := `service:
+  pipelines:
+    logs:
+      receivers:
+        - filebeatreceiver
+      processors:
+        - beat
+      exporters:
+        - debug
+  telemetry:
+    logs:
+      level: debug
+    metrics:
+      level: none
+receivers:
+  filebeatreceiver:
+    filebeat:
+      inputs:
+        - type: benchmark
+          enabled: true
+          message: "marker test message"
+          count: 1
+    queue.mem.flush.timeout: 0s
+processors:
+  beat:
+    processors:
+      - add_fields:
+          target: ""
+          fields:
+            should_be_added: "yes"
+          when.contains.message: "marker"
+      - add_fields:
+          target: ""
+          fields:
+            should_not_be_added: "yes"
+          when.not.contains.message: "marker"
+exporters:
+  debug:
+    verbosity: detailed
+`
+	col := oteltestcol.New(t, cfg)
+	require.NotNil(t, col)
+
+	require.Eventually(t, func() bool {
+		return col.ObservedLogs().
+			FilterMessageSnippet("Body: Map({").
+			FilterMessageSnippet(`"message":"marker test message"`).
+			FilterMessageSnippet(`"should_be_added":"yes"`).
+			Len() == 1
+	}, 30*time.Second, 100*time.Millisecond, "expected event to be enriched")
+
+	// The processor whose condition does not match must not enrich the event.
+	matchingNotAdded := col.ObservedLogs().
+		FilterMessageSnippet("Body: Map({").
+		FilterMessageSnippet(`"should_not_be_added"`).
+		Len()
+	assert.Equal(t, 0, matchingNotAdded, "expected `should_not_be_added` field to be absent")
+}

x-pack/otel/processor/beatprocessor/processor.go

@@ -13,6 +13,7 @@ import (
 
 	"github.com/elastic/beats/v7/libbeat/beat"
 	"github.com/elastic/beats/v7/libbeat/otel/otelmap"
+	"github.com/elastic/beats/v7/libbeat/processors"
 	"github.com/elastic/beats/v7/libbeat/processors/actions/addfields"
 	"github.com/elastic/beats/v7/libbeat/processors/add_cloud_metadata"
 	"github.com/elastic/beats/v7/libbeat/processors/add_docker_metadata"
@@ -82,24 +83,26 @@ func createProcessor(processorNameAndConfig map[string]any, logpLogger *logp.Log
 			return nil, fmt.Errorf("failed to create config for processor '%s': %w", processorName, configError)
 		}
 
-		var processorInstance beat.Processor
-		var createProcessorError error
+		var constructor processors.Constructor
 
 		switch processorName {
 		case "add_cloud_metadata":
-			processorInstance, createProcessorError = add_cloud_metadata.New(processorConfig, logpLogger)
+			constructor = add_cloud_metadata.New
 		case "add_docker_metadata":
-			processorInstance, createProcessorError = add_docker_metadata.New(processorConfig, logpLogger)
+			constructor = add_docker_metadata.New
 		case "add_fields":
-			processorInstance, createProcessorError = addfields.CreateAddFields(processorConfig, logpLogger)
+			constructor = addfields.CreateAddFields
 		case "add_host_metadata":
-			processorInstance, createProcessorError = add_host_metadata.New(processorConfig, logpLogger)
+			constructor = add_host_metadata.New
 		case "add_kubernetes_metadata":
-			processorInstance, createProcessorError = add_kubernetes_metadata.New(processorConfig, logpLogger)
+			constructor = add_kubernetes_metadata.New
 		default:
 			return nil, fmt.Errorf("invalid processor name '%s'", processorName)
 		}
 
+		// Wrap the constructor with NewConditional so that `when` conditions
+		// configured on the processor are honored.
+		processorInstance, createProcessorError := processors.NewConditional(constructor)(processorConfig, logpLogger)
 		if createProcessorError != nil {
 			return nil, fmt.Errorf("failed to create processor '%s': %w", processorName, createProcessorError)
 		}

x-pack/otel/processor/beatprocessor/processor_test.go

@@ -144,6 +144,87 @@ func TestCreateProcessor(t *testing.T) {
 		require.NotNil(t, processor)
 		assert.Equal(t, "add_kubernetes_metadata", processor.String()[:len("add_kubernetes_metadata")])
 	})
+
+	t.Run("when condition is honored and processor is skipped when condition is false", func(t *testing.T) {
+		processor, err := createProcessor(map[string]any{
+			"add_fields": map[string]any{
+				"target": "",
+				"fields": map[string]any{
+					"enriched": "yes",
+				},
+				"when": map[string]any{
+					"contains": map[string]any{
+						"tags": "forwarded",
+					},
+				},
+			},
+		}, testLogger())
+		require.NoError(t, err)
+		require.NotNil(t, processor)
+		assert.Contains(t, processor.String(), "condition=", "expected processor to be wrapped with a condition")
+
+		event := &beat.Event{Fields: mapstr.M{"message": "hello"}}
+		out, err := processor.Run(event)
+		require.NoError(t, err)
+		_, lookupErr := out.Fields.GetValue("enriched")
+		assert.Error(t, lookupErr, "expected 'enriched' field to be absent when condition is not met")
+	})
+
+	t.Run("when condition is honored and processor runs when condition is true", func(t *testing.T) {
+		processor, err := createProcessor(map[string]any{
+			"add_fields": map[string]any{
+				"target": "",
+				"fields": map[string]any{
+					"enriched": "yes",
+				},
+				"when": map[string]any{
+					"contains": map[string]any{
+						"tags": "forwarded",
+					},
+				},
+			},
+		}, testLogger())
+		require.NoError(t, err)
+		require.NotNil(t, processor)
+
+		event := &beat.Event{Fields: mapstr.M{"message": "hello", "tags": []string{"forwarded"}}}
+		out, err := processor.Run(event)
+		require.NoError(t, err)
+		val, err := out.Fields.GetValue("enriched")
+		require.NoError(t, err, "expected 'enriched' field to be added when condition is met")
+		assert.Equal(t, "yes", val)
+	})
+
+	t.Run("when.not.contains skips processor when matching tag is present", func(t *testing.T) {
+		processor, err := createProcessor(map[string]any{
+			"add_fields": map[string]any{
+				"target": "",
+				"fields": map[string]any{
+					"enriched": "yes",
+				},
+				"when.not.contains.tags": "forwarded",
+			},
+		}, testLogger())
+		require.NoError(t, err)
+		require.NotNil(t, processor)
+
+		event := &beat.Event{Fields: mapstr.M{"message": "hello", "tags": []string{"forwarded"}}}
+		out, err := processor.Run(event)
+		require.NoError(t, err)
+		_, lookupErr := out.Fields.GetValue("enriched")
+		assert.Error(t, lookupErr, "expected 'enriched' field to be absent when 'forwarded' tag is present")
+	})
+
+	t.Run("invalid when condition returns error", func(t *testing.T) {
+		_, err := createProcessor(map[string]any{
+			"add_host_metadata": map[string]any{
+				"when": map[string]any{
+					"not_a_real_condition": map[string]any{},
+				},
+			},
+		}, testLogger())
+		require.Error(t, err)
+	})
 }
 
 func testLogger() *logp.Logger {