Add conditions to copy_fields processor (#6730)

This commit adds conditions to the `copy_fields` processor from the
monitoring Filebeat to prevent it from failing and spamming the event
logger at debug level with:
`target field xxx already exists, drop or rename this field first`

---------

Co-authored-by: Pierre HILBERT <[email protected]>
(cherry picked from commit acf1098)

# Conflicts:
#	testing/integration/logs_ingestion_test.go

Author: belimawr

changelog/fragments/1738855375-Add-conditions-to-copy_fields-processors.yaml

@@ -0,0 +1,33 @@
+# 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: >
+  Add conditions to copy_fields processors to prevent spamming the debug logs
+
+# 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/6730
+
+# 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/5299

internal/pkg/agent/application/monitoring/v1_monitor.go

@@ -450,6 +450,13 @@ func (b *BeatsMonitor) injectLogsInput(cfg map[string]interface{}, components []
 				// possible it's a log message from agent itself (doesn't have component.dataset)
 				map[string]interface{}{
 					"copy_fields": map[string]interface{}{
+						"when": map[string]any{
+							"not": map[string]any{
+								"has_fields": []any{
+									"data_stream.dataset",
+								},
+							},
+						},
 						"fields": []interface{}{
 							map[string]interface{}{
 								"from": "data_stream.dataset_original",

testing/integration/container_cmd_test.go

@@ -330,7 +330,7 @@ func TestContainerCMDEventToStderr(t *testing.T) {
 	// We call agentFixture.Prepare to set the workdir
 	require.NoError(t, agentFixture.Prepare(ctx), "failed preparing agent fixture")
 
-	_, outputID := createMockESOutput(t, info)
+	_, outputID := createMockESOutput(t, info, 0, 0, 100, 0)
 	policyID, enrollmentAPIKey := createPolicy(
 		t,
 		ctx,
@@ -390,8 +390,13 @@ func TestContainerCMDEventToStderr(t *testing.T) {
 	}, 3*time.Minute, 10*time.Second, "cannot find events on stderr")
 }
 
-func createMockESOutput(t *testing.T, info *define.Info) (string, string) {
-	mockesURL := startMockES(t)
+// createMockESOutput creates an output configuration pointing to a mockES
+// started in a random port and a cleanup function is registered to close
+// the server at the end of the test.
+// The server will respond with the passed error probabilities. If they add
+// up to zero, all requests are a success.
+func createMockESOutput(t *testing.T, info *define.Info, percentDuplicate, percentTooMany, percentNonIndex, percentTooLarge uint) (string, string) {
+	mockesURL := startMockES(t, percentDuplicate, percentTooMany, percentNonIndex, percentTooLarge)
 	createOutputBody := `
 {
   "id": "mock-es-%[1]s",

testing/integration/event_logging_test.go

@@ -10,6 +10,7 @@ import (
 	"bufio"
 	"bytes"
 	"context"
+	"encoding/json"
 	"fmt"
 	"net/http"
 	"net/http/httputil"
@@ -37,11 +38,11 @@ outputs:
     hosts:
       - %s
     protocol: http
-    preset: balanced
-
+    preset: latency
 inputs:
   - type: filestream
     id: your-input-id
+    log_level: debug
     streams:
       - id: your-filestream-stream-id
         data_stream:
@@ -84,10 +85,10 @@ func TestEventLogFile(t *testing.T) {
 	agentFixture, err := define.NewFixtureFromLocalBuild(t, define.Version())
 	require.NoError(t, err)
 
-	esURL := startMockES(t)
+	esURL := startMockES(t, 0, 0, 0, 0)
 
 	logFilepath := path.Join(t.TempDir(), t.Name())
-	generateLogFile(t, logFilepath, time.Millisecond*100, 1)
+	generateLogFile(t, logFilepath, time.Millisecond*100, 20)
 
 	cfg := fmt.Sprintf(eventLogConfig, esURL, logFilepath)
 
@@ -126,6 +127,7 @@ func TestEventLogFile(t *testing.T) {
 
 	// Now the Elastic-Agent is running, so validate the Event log file.
 	requireEventLogFileExistsWithData(t, agentFixture)
+	requireNoCopyProcessorError(t, agentFixture)
 
 	// The diagnostics command is already tested by another test,
 	// here we just want to validate the events log behaviour
@@ -169,7 +171,7 @@ func TestEventLogOutputConfiguredViaFleet(t *testing.T) {
 	agentFixture, err := define.NewFixtureFromLocalBuild(t, define.Version())
 	require.NoError(t, err)
 
-	_, outputID := createMockESOutput(t, info)
+	_, outputID := createMockESOutput(t, info, 0, 0, 100, 0)
 	policyName := fmt.Sprintf("%s-%s", t.Name(), uuid.Must(uuid.NewV4()).String())
 	policyID, enrollmentAPIKey := createPolicy(
 		t,
@@ -307,7 +309,7 @@ func addOverwriteToPolicy(t *testing.T, info *define.Info, policyName, policyID
 	}
 }
 
-func requireEventLogFileExistsWithData(t *testing.T, agentFixture *atesting.Fixture) {
+func readEventLogFile(t *testing.T, agentFixture *atesting.Fixture) string {
 	// Now the Elastic-Agent is running, so validate the Event log file.
 	// Because the path changes based on the Elastic-Agent version, we
 	// use glob to find the file
@@ -338,8 +340,39 @@ func requireEventLogFileExistsWithData(t *testing.T, agentFixture *atesting.Fixt
 		t.Fatalf("cannot read file '%s': %s", logFileName, err)
 	}
 
-	logEntry := string(logEntryBytes)
-	expectedStr := "Cannot index event"
+	return string(logEntryBytes)
+}
+
+func requireNoCopyProcessorError(t *testing.T, agentFixture *atesting.Fixture) {
+	data := readEventLogFile(t, agentFixture)
+	for _, line := range strings.Split(data, "\n") {
+		logEntry := struct {
+			LogLogger string `json:"log.logger"`
+			Message   string `json:"message"`
+		}{}
+
+		if len(line) == 0 {
+			continue
+		}
+		if err := json.Unmarshal([]byte(line), &logEntry); err != nil {
+			t.Fatalf("could not parse log entry: %q", line)
+		}
+
+		if logEntry.LogLogger == "copy_fields" {
+			if strings.Contains(logEntry.Message, "Failed to copy fields") {
+				if strings.Contains(logEntry.Message, "already exists, drop or rename this field first") {
+					t.Fatal("copy_fields processor must not fail")
+				}
+			}
+		}
+	}
+}
+
+func requireEventLogFileExistsWithData(t *testing.T, agentFixture *atesting.Fixture) {
+	logEntry := readEventLogFile(t, agentFixture)
+	// That's part of the generated event that is logged by the 'processor'
+	// logger at level debug
+	expectedStr := "TestEventLogFile"
 	if !strings.Contains(logEntry, expectedStr) {
 		t.Errorf(
 			"did not find the expected log entry ('%s') in the events log file",

testing/integration/logs_ingestion_test.go

@@ -120,17 +120,32 @@ func TestLogIngestionFleetManaged(t *testing.T) {
 	})
 }
 
+<<<<<<< HEAD
 func startMockES(t *testing.T) string {
 	registry := metrics.NewRegistry()
+=======
+// startMockES starts a MockES on a random port using httptest.NewServer.
+// It registers a cleanup function to close the server when the test finishes.
+// The server will respond with the passed error probabilities. If they add
+// up to zero, all requests are a success.
+func startMockES(t *testing.T, percentDuplicate, percentTooMany, percentNonIndex, percentTooLarge uint) string {
+>>>>>>> acf109882 (Add conditions to copy_fields processor (#6730))
 	uid := uuid.Must(uuid.NewV4())
 	clusterUUID := uuid.Must(uuid.NewV4()).String()
 
 	mux := http.NewServeMux()
 	mux.Handle("/", mockes.NewAPIHandler(
 		uid,
 		clusterUUID,
+<<<<<<< HEAD
 		registry,
 		time.Now().Add(time.Hour), 0, 0, 0, 100, 0))
+=======
+		nil,
+		time.Now().Add(time.Hour),
+		0,
+		percentDuplicate, percentTooMany, percentNonIndex, percentTooLarge))
+>>>>>>> acf109882 (Add conditions to copy_fields processor (#6730))
 
 	s := httptest.NewServer(mux)
 	t.Cleanup(s.Close)