[8.17](backport #6730) Add conditions to copy_fields processor

NOTICE.txt

@@ -9697,45 +9697,6 @@ OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 THE SOFTWARE.
 
 
---------------------------------------------------------------------------------
-Dependency : github.com/rcrowley/go-metrics
-Version: v0.0.0-20201227073835-cf1acfcdf475
-Licence type (autodetected): BSD-2-Clause-FreeBSD
---------------------------------------------------------------------------------
-
-Contents of probable licence file $GOMODCACHE/github.com/rcrowley/[email protected]/LICENSE:
-
-Copyright 2012 Richard Crowley. All rights reserved.
-
-Redistribution and use in source and binary forms, with or without
-modification, are permitted provided that the following conditions are
-met:
-
-    1.  Redistributions of source code must retain the above copyright
-        notice, this list of conditions and the following disclaimer.
-
-    2.  Redistributions in binary form must reproduce the above
-        copyright notice, this list of conditions and the following
-        disclaimer in the documentation and/or other materials provided
-        with the distribution.
-
-THIS SOFTWARE IS PROVIDED BY RICHARD CROWLEY ``AS IS'' AND ANY EXPRESS
-OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
-WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
-DISCLAIMED. IN NO EVENT SHALL RICHARD CROWLEY OR CONTRIBUTORS BE LIABLE
-FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
-CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
-SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
-INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
-CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
-ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF
-THE POSSIBILITY OF SUCH DAMAGE.
-
-The views and conclusions contained in the software and documentation
-are those of the authors and should not be interpreted as representing
-official policies, either expressed or implied, of Richard Crowley.
-
-
 --------------------------------------------------------------------------------
 Dependency : github.com/rednafi/link-patrol
 Version: v0.0.0-20240826150821-057643e74d4d
@@ -75045,6 +75006,45 @@ Contents of probable licence file $GOMODCACHE/github.com/prometheus/prometheus@v
    limitations under the License.
 
 
+--------------------------------------------------------------------------------
+Dependency : github.com/rcrowley/go-metrics
+Version: v0.0.0-20201227073835-cf1acfcdf475
+Licence type (autodetected): BSD-2-Clause-FreeBSD
+--------------------------------------------------------------------------------
+
+Contents of probable licence file $GOMODCACHE/github.com/rcrowley/[email protected]/LICENSE:
+
+Copyright 2012 Richard Crowley. All rights reserved.
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are
+met:
+
+    1.  Redistributions of source code must retain the above copyright
+        notice, this list of conditions and the following disclaimer.
+
+    2.  Redistributions in binary form must reproduce the above
+        copyright notice, this list of conditions and the following
+        disclaimer in the documentation and/or other materials provided
+        with the distribution.
+
+THIS SOFTWARE IS PROVIDED BY RICHARD CROWLEY ``AS IS'' AND ANY EXPRESS
+OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+DISCLAIMED. IN NO EVENT SHALL RICHARD CROWLEY OR CONTRIBUTORS BE LIABLE
+FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF
+THE POSSIBILITY OF SUCH DAMAGE.
+
+The views and conclusions contained in the software and documentation
+are those of the authors and should not be interpreted as representing
+official policies, either expressed or implied, of Richard Crowley.
+
+
 --------------------------------------------------------------------------------
 Dependency : github.com/rivo/uniseg
 Version: v0.4.4

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

go.mod

@@ -50,7 +50,6 @@ require (
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/prometheusreceiver v0.119.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/zipkinreceiver v0.119.0
 	github.com/otiai10/copy v1.14.0
-	github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475
 	github.com/rednafi/link-patrol v0.0.0-20240826150821-057643e74d4d
 	github.com/rs/zerolog v1.27.0
 	github.com/sajari/regression v1.0.1
@@ -460,6 +459,7 @@ require (
 	github.com/prometheus/common/sigv4 v0.1.0 // indirect
 	github.com/prometheus/procfs v0.15.1 // indirect
 	github.com/prometheus/prometheus v0.54.1 // indirect
+	github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 // indirect
 	github.com/rivo/uniseg v0.4.4 // indirect
 	github.com/rogpeppe/go-internal v1.13.1 // indirect
 	github.com/rs/cors v1.11.1 // indirect

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

@@ -38,7 +38,6 @@ import (
 	"github.com/elastic/elastic-agent/testing/installtest"
 	"github.com/elastic/elastic-transport-go/v8/elastictransport"
 
-	"github.com/rcrowley/go-metrics"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
@@ -120,17 +119,22 @@ func TestLogIngestionFleetManaged(t *testing.T) {
 	})
 }
 
-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 {
 	uid := uuid.Must(uuid.NewV4())
 	clusterUUID := uuid.Must(uuid.NewV4()).String()
 
 	mux := http.NewServeMux()
 	mux.Handle("/", mockes.NewAPIHandler(
 		uid,
 		clusterUUID,
-		registry,
-		time.Now().Add(time.Hour), 0, 0, 0, 100, 0))
+		nil,
+		time.Now().Add(time.Hour),
+		0,
+		percentDuplicate, percentTooMany, percentNonIndex, percentTooLarge))
 
 	s := httptest.NewServer(mux)
 	t.Cleanup(s.Close)