[8.19](backport #47729) remove otel.component.id and otel.component.kind fields from beat receivers (#47786)

* remove otel.component.id and otel.component.kind fields from beat receivers (#47729)


(cherry picked from commit ca1c17b)

# Conflicts:
#	x-pack/filebeat/tests/integration/otel_lsexporter_test.go
#	x-pack/libbeat/outputs/otelconsumer/otelconsumer.go
#	x-pack/libbeat/outputs/otelconsumer/otelconsumer_test.go

* fix backport

* make linter happy

* fix conflict resolution

* fix integration test

---------

Co-authored-by: Lee E Hinman <57081003+leehinman@users.noreply.github.com>
Co-authored-by: Lee E. Hinman <lee.e.hinman@elastic.co>

Author: mergify[bot]

changelog/fragments/1763578431-remove-otel.component.id-and-otel.component.kind-from-beat-receiver-events.yaml

@@ -0,0 +1,45 @@
+# REQUIRED
+# 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: breaking-change
+
+# REQUIRED for all kinds
+# Change summary; a 80ish characters long description of the change.
+summary: remove otel.component.id and otel.component.kind from beat receiver events
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# Long description; in case the summary is not enough to describe the change
+# this field accommodate a description without length limits.
+# description:
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# impact:
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# action:
+
+# REQUIRED for all kinds
+# Affected component; usually one of "elastic-agent", "fleet-server", "filebeat", "metricbeat", "auditbeat", "all", etc.
+component: all
+
+# AUTOMATED
+# OPTIONAL to manually add other PR URLs
+# PR URL: A link the PR 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/beats/pull/47729
+
+# AUTOMATED
+# OPTIONAL to manually add other issue URLs
+# 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/beats/issues/47600

x-pack/filebeat/fbreceiver/receiver_test.go

@@ -83,8 +83,7 @@ func TestNewReceiver(t *testing.T) {
 		AssertFunc: func(c *assert.CollectT, logs map[string][]mapstr.M, zapLogs *observer.ObservedLogs) {
 			_ = zapLogs
 			require.Lenf(c, logs["r1"], 1, "expected 1 log, got %d", len(logs["r1"]))
-			assert.Equal(c, "filebeatreceiver/r1", logs["r1"][0].Flatten()["agent.otelcol.component.id"], "expected agent.otelcol.component.id field in log record")
-			assert.Equal(c, "receiver", logs["r1"][0].Flatten()["agent.otelcol.component.kind"], "expected agent.otelcol.component.kind field in log record")
+			assert.Equal(c, "test", logs["r1"][0].Flatten()["message"], "expected message field to contain string 'test'")
 			var lastError strings.Builder
 			assert.Conditionf(c, func() bool {
 				return getFromSocket(t, &lastError, monitorSocket, "stats")
@@ -264,9 +263,7 @@ func TestMultipleReceivers(t *testing.T) {
 				writeFile(c, helper.ingest, "A log line")
 
 				require.Greaterf(c, len(logs[helper.name]), 0, "receiver %v does not have any logs", helper)
-
-				assert.Equalf(c, "filebeatreceiver/"+helper.name, logs[helper.name][0].Flatten()["agent.otelcol.component.id"], "expected agent.otelcol.component.id field in %v log record", helper)
-				assert.Equalf(c, "receiver", logs[helper.name][0].Flatten()["agent.otelcol.component.kind"], "expected agent.otelcol.component.kind field in %v log record", helper)
+				assert.Equal(c, "test", logs[helper.name][0].Flatten()["message"], "expected message field to contain string 'test'")
 
 				// Verify that each receiver used its own JavaScript processor script.
 				// This demonstrates path isolation: each receiver loads processor.js from its own path.config.

x-pack/filebeat/input/gcppubsub/otel_test.go

@@ -211,10 +211,10 @@ service:
 			defer findCancel()
 
 			otelDocs, err = estools.PerformQueryForRawQuery(findCtx, rawQuery, ".ds-logs-integration-"+otelNamespace+"*", es)
-			msg.WriteString(fmt.Sprintf("failed to query ES for beat documents: %v", err))
+			fmt.Fprintf(msg, "failed to query ES for beat documents: %v", err)
 
 			filebeatDocs, err = estools.PerformQueryForRawQuery(findCtx, rawQuery, ".ds-logs-integration-"+fbNameSpace+"*", es)
-			msg.WriteString(fmt.Sprintf("failed to query ES for beat documents: %v", err))
+			fmt.Fprintf(msg, "failed to query ES for beat documents: %v", err)
 
 			return otelDocs.Hits.Total.Value >= 1 && filebeatDocs.Hits.Total.Value >= 1
 		},
@@ -228,11 +228,7 @@ service:
 		"agent.ephemeral_id",
 		"agent.id",
 		"event.created",
-		// only present in beats receivers
-		"agent.otelcol.component.id",
-		"agent.otelcol.component.kind",
 	}
 
 	oteltest.AssertMapsEqual(t, filebeatDoc, otelDoc, ignoredFields, "expected documents to be equal")
-
 }

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

@@ -110,7 +110,7 @@ service:
 	writeEventsToLogFile(t, logFilePath, numEvents)
 	oteltestcol.New(t, fmt.Sprintf(otelCfgFile, logFilePath, tmpdir, otelMonitoringPort, fbOtelIndex))
 
-	var beatsCfgFile = `
+	beatsCfgFile := `
 filebeat.inputs:
   - type: filestream
     id: filestream-input-id
@@ -184,18 +184,13 @@ http.port: %d
 		"agent.id",
 		"log.file.inode",
 		"log.file.path",
-		// only present in beats receivers
-		"agent.otelcol.component.id",
-		"agent.otelcol.component.kind",
 		"log.file.device_id", // changes value between filebeat and otel receiver
 	}
 
 	oteltest.AssertMapsEqual(t, filebeatDoc, otelDoc, ignoredFields, "expected documents to be equal")
 
-	assert.Equal(t, "filebeatreceiver", otelDoc.Flatten()["agent.otelcol.component.id"], "expected agent.otelcol.component.id field in log record")
-	assert.Equal(t, "receiver", otelDoc.Flatten()["agent.otelcol.component.kind"], "expected agent.otelcol.component.kind field in log record")
-	assert.NotContains(t, filebeatDoc.Flatten(), "agent.otelcol.component.id", "expected agent.otelcol.component.id field not to be present in filebeat log record")
-	assert.NotContains(t, filebeatDoc.Flatten(), "agent.otelcol.component.kind", "expected agent.otelcol.component.kind field not to be present in filebeat log record")
+	assert.Equal(t, "filebeat", otelDoc.Flatten()["agent.type"], "expected agent.type field to be 'filebeat' in otel docs")
+	assert.Equal(t, "filebeat", filebeatDoc.Flatten()["agent.type"], "expected agent.type field to be 'filebeat' in filebeat docs")
 	assertMonitoring(t, otelMonitoringPort)
 }
 
@@ -367,9 +362,6 @@ service:
 		"agent.ephemeral_id",
 		"agent.id",
 		"event.created",
-		// only present in beats receivers
-		"agent.otelcol.component.id",
-		"agent.otelcol.component.kind",
 	}
 
 	oteltest.AssertMapsEqual(t, filebeatDoc, otelDoc, ignoredFields, "expected documents to be equal")
@@ -572,17 +564,13 @@ http.port: %d
 		"log.file.inode",
 		"log.file.path",
 		// only present in beats receivers
-		"agent.otelcol.component.id",
-		"agent.otelcol.component.kind",
 		"log.file.device_id", // changes value between filebeat and otel receiver
 		"container.id",       // only present in filebeat
 	}
 
 	oteltest.AssertMapsEqual(t, filebeatDoc, otelDoc, ignoredFields, "expected documents to be equal")
-	assert.Equal(t, "filebeatreceiver/filestream", otelDoc.Flatten()["agent.otelcol.component.id"], "expected agent.otelcol.component.id field in log record")
-	assert.Equal(t, "receiver", otelDoc.Flatten()["agent.otelcol.component.kind"], "expected agent.otelcol.component.kind field in log record")
-	assert.NotContains(t, filebeatDoc.Flatten(), "agent.otelcol.component.id", "expected agent.otelcol.component.id field not to be present in filebeat log record")
-	assert.NotContains(t, filebeatDoc.Flatten(), "agent.otelcol.component.kind", "expected agent.otelcol.component.kind field not to be present in filebeat log record")
+	assert.Equal(t, "filebeat", otelDoc.Flatten()["agent.type"], "expected agent.type field to be 'filebeat' in otel docs")
+	assert.Equal(t, "filebeat", filebeatDoc.Flatten()["agent.type"], "expected agent.type field to be 'filebeat' in filebeat docs")
 	assertMonitoring(t, otelConfig.MonitoringPort)
 	assertMonitoring(t, filebeatMonitoringPort) // filebeat
 }
@@ -1006,7 +994,6 @@ service:
 }
 
 func TestFileBeatKerberos(t *testing.T) {
-
 	wantEvents := 1
 	krbURL := "http://localhost:9203" // this is kerberos client - we've hardcoded the URL here
 	tempFile := t.TempDir()
@@ -1113,7 +1100,6 @@ service:
 			assert.GreaterOrEqual(ct, otelDocs.Hits.Total.Value, wantEvents, "expected at least %d events, got %d", wantEvents, otelDocs.Hits.Total.Value)
 		},
 		2*time.Minute, 1*time.Second)
-
 }
 
 func TestFilebeatOTelBeatProcessorE2E(t *testing.T) {
@@ -1305,7 +1291,6 @@ exporters:
 
 // setupRoleMapping sets up role mapping for the Kerberos user beats@elastic
 func setupRoleMapping(t *testing.T, client *elasticsearch.Client) {
-
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
 
@@ -1339,5 +1324,4 @@ func setupRoleMapping(t *testing.T, client *elasticsearch.Client) {
 	defer resp.Body.Close()
 
 	require.Equal(t, resp.StatusCode, http.StatusOK, "incorrect response code")
-
 }

x-pack/libbeat/outputs/otelconsumer/otelconsumer.go

@@ -32,10 +32,6 @@ const (
 	esDocumentIDAttribute = "elasticsearch.document_id"
 	beatNameCtxKey        = "beat_name"
 	beatVersionCtxtKey    = "beat_version"
-	// otelComponentIDKey is the key used to store the Beat receiver's component id in the beat event.
-	otelComponentIDKey = "otelcol.component.id"
-	// otelComponentKindKey is the key used to store the Beat receiver's component kind in the beat event. This is always "receiver".
-	otelComponentKindKey = "otelcol.component.kind"
 )
 
 func init() {
@@ -143,15 +139,6 @@ func (out *otelConsumer) logsPublish(ctx context.Context, batch publisher.Batch)
 		}
 		logRecord.SetObservedTimestamp(observedTimestamp)
 
-		if agent, _ := beatEvent.GetValue("agent"); agent != nil {
-			switch agent := agent.(type) {
-			case mapstr.M:
-				agent[otelComponentIDKey] = out.beatInfo.ComponentID
-				agent[otelComponentKindKey] = "receiver"
-				beatEvent["agent"] = agent
-			}
-		}
-
 		otelmap.ConvertNonPrimitive(beatEvent)
 
 		// if data_stream field is set on beatEvent. Add it to logrecord.Attributes to support dynamic indexing

x-pack/libbeat/outputs/otelconsumer/otelconsumer_test.go

@@ -281,73 +281,4 @@ func TestPublish(t *testing.T) {
 		assert.Len(t, batch.Signals, 1)
 		assert.Equal(t, outest.BatchACK, batch.Signals[0].Tag)
 	})
-	t.Run("sets otel specific-fields", func(t *testing.T) {
-		testCases := []struct {
-			name                  string
-			componentID           string
-			componentKind         string
-			expectedComponentID   string
-			expectedComponentKind string
-		}{
-			{
-				name:                  "sets beat component ID",
-				componentID:           "filebeatreceiver/1",
-				expectedComponentID:   "filebeatreceiver/1",
-				expectedComponentKind: "receiver",
-			},
-		}
-
-		for _, tc := range testCases {
-			t.Run(tc.name, func(t *testing.T) {
-				event := beat.Event{
-					Fields: mapstr.M{
-						"field": 1,
-						"agent": mapstr.M{},
-					},
-					Meta: mapstr.M{
-						"_id": "abc123",
-					},
-				}
-				ch := make(chan plog.Logs, 1)
-				batch := outest.NewBatch(event)
-				var countLogs int
-				otelConsumer := makeOtelConsumer(t, func(ctx context.Context, ld plog.Logs) error {
-					countLogs = countLogs + ld.LogRecordCount()
-					ch <- ld
-					return nil
-				})
-				otelConsumer.beatInfo.ComponentID = tc.componentID
-				err := otelConsumer.Publish(ctx, batch)
-				assert.NoError(t, err)
-				assert.Len(t, batch.Signals, 1)
-				assert.Equal(t, outest.BatchACK, batch.Signals[0].Tag)
-				assert.Equal(t, len(batch.Events()), countLogs, "all events should be consumed")
-				log := <-ch
-				for i := 0; i < log.ResourceLogs().Len(); i++ {
-					resourceLog := log.ResourceLogs().At(i)
-					for j := 0; j < resourceLog.ScopeLogs().Len(); j++ {
-						scopeLog := resourceLog.ScopeLogs().At(j)
-						for k := 0; k < scopeLog.LogRecords().Len(); k++ {
-							logRecord := scopeLog.LogRecords().At(k)
-							body := logRecord.Body().Map()
-
-							// Traverse nested "agent.otelcol.component" structure
-							agentVal, ok := body.Get("agent")
-							require.True(t, ok, "expected 'agent' in log body")
-
-							agentMap := agentVal.Map()
-							idVal, ok := agentMap.Get("otelcol.component.id")
-							require.True(t, ok, "expected 'agent.otelcol.component.id' in log body")
-							assert.Equal(t, tc.expectedComponentID, idVal.AsString())
-
-							kindVal, ok := agentMap.Get("otelcol.component.kind")
-							require.True(t, ok, "expected 'agent.otelcol.component.kind' in log body")
-							assert.Equal(t, tc.expectedComponentKind, kindVal.AsString())
-						}
-					}
-				}
-
-			})
-		}
-	})
 }

x-pack/metricbeat/mbreceiver/receiver_test.go

@@ -87,8 +87,8 @@ func TestNewReceiver(t *testing.T) {
 			require.Conditionf(c, func() bool {
 				return len(logs["r1"]) > 0
 			}, "expected at least one ingest log, got logs: %v", logs["r1"])
-			assert.Equal(c, "metricbeatreceiver/r1", logs["r1"][0].Flatten()["agent.otelcol.component.id"], "expected agent.otelcol.component.id field in log record")
-			assert.Equal(c, "receiver", logs["r1"][0].Flatten()["agent.otelcol.component.kind"], "expected agent.otelcol.component.kind field in log record")
+			assert.Equal(c, "metricbeat", logs["r1"][0].Flatten()["agent.type"], "expected agent.type field in to be 'metricbeat'")
+
 			var lastError strings.Builder
 			assert.Conditionf(c, func() bool {
 				return getFromSocket(t, &lastError, monitorSocket, "stats")
@@ -198,10 +198,8 @@ func TestMultipleReceivers(t *testing.T) {
 			require.Conditionf(c, func() bool {
 				return len(logs["r1"]) > 0 && len(logs["r2"]) > 0
 			}, "expected at least one ingest log for each receiver, got logs: %v", logs)
-			assert.Equal(c, "metricbeatreceiver/r1", logs["r1"][0].Flatten()["agent.otelcol.component.id"], "expected agent.otelcol.component.id field in r1 log record")
-			assert.Equal(c, "receiver", logs["r1"][0].Flatten()["agent.otelcol.component.kind"], "expected agent.otelcol.component.kind field in r1 log record")
-			assert.Equal(c, "metricbeatreceiver/r2", logs["r2"][0].Flatten()["agent.otelcol.component.id"], "expected agent.otelcol.component.id field in r2 log record")
-			assert.Equal(c, "receiver", logs["r2"][0].Flatten()["agent.otelcol.component.kind"], "expected otelcol.component.kind field in r2 log record")
+			assert.Equal(c, "metricbeat", logs["r1"][0].Flatten()["agent.type"], "expected agent.type field to be 'metricbeat' in r1")
+			assert.Equal(c, "metricbeat", logs["r2"][0].Flatten()["agent.type"], "expected agent.type field to be 'metricbeat' in r2")
 
 			// Make sure that each receiver has a separate logger
 			// instance and does not interfere with others. Previously, the

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

@@ -122,7 +122,7 @@ service:
 
 	oteltestcol.New(t, configBuffer.String())
 
-	var beatsCfgFile = `
+	beatsCfgFile := `
 metricbeat:
    modules:
    - module: system
@@ -205,16 +205,9 @@ http.port: {{.MonitoringPort}}
 	var metricbeatDoc, otelDoc mapstr.M
 	otelDoc = otelDocs.Hits.Hits[0].Source
 	metricbeatDoc = metricbeatDocs.Hits.Hits[0].Source
-	ignoredFields := []string{
-		// only present in beats receivers
-		"agent.otelcol.component.id",
-		"agent.otelcol.component.kind",
-	}
-	assert.Equal(t, "metricbeatreceiver", otelDoc.Flatten()["agent.otelcol.component.id"], "expected agent.otelcol.component.id field in log record")
-	assert.Equal(t, "receiver", otelDoc.Flatten()["agent.otelcol.component.kind"], "expected agent.otelcol.component.kind field in log record")
-	assert.NotContains(t, metricbeatDoc.Flatten(), "agent.otelcol.component.id", "expected agent.otelcol.component.id field not to be present in metricbeat log record")
-	assert.NotContains(t, metricbeatDoc.Flatten(), "agent.otelcol.component.kind", "expected agent.otelcol.component.kind field not to be present in metricbeat log record")
-	assertMapstrKeysEqual(t, otelDoc, metricbeatDoc, ignoredFields, "expected documents keys to be equal")
+	assert.Equal(t, "metricbeat", otelDoc.Flatten()["agent.type"], "expected agent.type field to be 'metricbeat' in otel docs")
+	assert.Equal(t, "metricbeat", metricbeatDoc.Flatten()["agent.type"], "expected agent.type field to be 'metricbeat' in metricbeat docs")
+	assertMapstrKeysEqual(t, otelDoc, metricbeatDoc, nil, "expected documents keys to be equal")
 	assertMonitoring(t, metricbeatMonitoringPort)
 }
 
@@ -543,12 +536,7 @@ service:
 			assert.GreaterOrEqualf(ct, r1Docs.Hits.Total.Value, 1, "expected at least 1 log for receiver 1, got %d", r1Docs.Hits.Total.Value)
 		},
 		1*time.Minute, 100*time.Millisecond, "expected at least 1 log for each receiver")
-	ignoredFields := []string{
-		// only present in beats receivers
-		"agent.otelcol.component.id",
-		"agent.otelcol.component.kind",
-	}
-	assertMapstrKeysEqual(t, r0Docs.Hits.Hits[0].Source, r1Docs.Hits.Hits[0].Source, ignoredFields, "expected documents keys to be equal")
+	assertMapstrKeysEqual(t, r0Docs.Hits.Hits[0].Source, r1Docs.Hits.Hits[0].Source, nil, "expected documents keys to be equal")
 	for _, rec := range otelConfig.Receivers {
 		assertMonitoring(t, rec.MonitoringPort)
 	}