[9.5](backport #51679) otel(processors): implement PdataProcessor for add_docker_metadata

libbeat/processors/add_docker_metadata/add_docker_metadata.go

@@ -30,8 +30,11 @@ import (
 	"sync/atomic"
 	"time"
 
+	"go.opentelemetry.io/collector/pdata/pcommon"
+
 	"github.com/elastic/beats/v7/libbeat/beat"
 	"github.com/elastic/beats/v7/libbeat/common"
+	"github.com/elastic/beats/v7/libbeat/otel/otelmap"
 	"github.com/elastic/beats/v7/libbeat/processors"
 	"github.com/elastic/beats/v7/libbeat/processors/actions"
 	"github.com/elastic/elastic-agent-autodiscover/docker"
@@ -62,6 +65,8 @@ func init() {
 	processors.RegisterPlugin(processorName, New)
 }
 
+var _ processors.PdataProcessor = (*addDockerMetadata)(nil)
+
 type addDockerMetadata struct {
 	log             *logp.Logger
 	watcher         docker.Watcher
@@ -264,20 +269,20 @@ func (d *addDockerMetadata) Run(event *beat.Event) (*beat.Event, error) {
 		return event, nil
 	}
 	var cid string
-	var err error
 
 	// Extract CID from the filepath contained in the "log.file.path" field.
 	if d.sourceProcessor != nil {
 		lfp, _ := event.Fields.GetValue("log.file.path")
 		if lfp != nil {
-			event, err = d.sourceProcessor.Run(event)
+			id, err := d.resolveCIDFromSourcePath(lfp)
 			if err != nil {
 				d.log.Debugf("Error while extracting container ID from source path: %v", err)
 				return event, nil
 			}
 
-			if v, err := event.GetValue(dockerContainerIDKey); err == nil {
-				cid, _ = v.(string)
+			if id != "" {
+				cid = id
+				_, _ = event.PutValue(dockerContainerIDKey, cid)
 			}
 		}
 	}
@@ -296,49 +301,152 @@ func (d *addDockerMetadata) Run(event *beat.Event) (*beat.Event, error) {
 
 	// Lookup CID from the user defined field names.
 	if cid == "" && len(d.fields) > 0 {
-		for _, field := range d.fields {
+		cid = matchFieldCID(d.fields, func(field string) (string, bool) {
 			value, err := event.GetValue(field)
 			if err != nil {
-				continue
-			}
-
-			if strValue, ok := value.(string); ok {
-				cid = strValue
-				break
+				return "", false
 			}
-		}
+			strValue, ok := value.(string)
+			return strValue, ok
+		})
 	}
 
 	if cid == "" {
 		return event, nil
 	}
 
 	container := d.watcher.Container(cid)
-	if container != nil {
-		meta := mapstr.M{}
-
-		if len(container.Labels) > 0 {
-			labels := mapstr.M{}
-			for k, v := range container.Labels {
-				if d.dedot {
-					label := common.DeDot(k)
-					_, _ = labels.Put(label, v)
-				} else {
-					_ = safemapstr.Put(labels, k, v)
+	if container == nil {
+		d.log.Debugf("Container not found: cid=%s", cid)
+		return event, nil
+	}
+
+	event.Fields.DeepUpdate(d.buildContainerMeta(container))
+	return event, nil
+}
+
+// RunPdata enriches the given pcommon.Map directly with Docker container metadata
+func (d *addDockerMetadata) RunPdata(body pcommon.Map) (bool, error) {
+	if !d.dockerAvailable.Load() {
+		return false, nil
+	}
+
+	cid, err := d.resolveCIDFromPdata(body)
+	if err != nil {
+		return false, err
+	}
+	if cid == "" {
+		return false, nil
+	}
+
+	container := d.watcher.Container(cid)
+	if container == nil {
+		d.log.Debugf("Container not found: cid=%s", cid)
+		return false, nil
+	}
+
+	return false, otelmap.MergeMapstrIntoPdata(d.buildContainerMeta(container), body, true)
+}
+
+// resolveCIDFromPdata extracts the container ID from a pcommon.Map
+func (d *addDockerMetadata) resolveCIDFromPdata(body pcommon.Map) (string, error) {
+	// Extract CID from log.file.path via sourceProcessor.
+	if d.sourceProcessor != nil {
+		if lfpVal, ok := otelmap.GetAtPath("log.file.path", body); ok && lfpVal.Type() == pcommon.ValueTypeStr {
+			id, err := d.resolveCIDFromSourcePath(lfpVal.Str())
+			if err != nil {
+				d.log.Debugf("Error while extracting container ID from source path: %v", err)
+				return "", nil
+			}
+			if id != "" {
+				if err := otelmap.PutAtPath(dockerContainerIDKey, id, body); err != nil {
+					return "", err
 				}
+				return id, nil
 			}
-			_, _ = meta.Put("container.labels", labels)
 		}
+	}
 
-		_, _ = meta.Put("container.id", container.ID)
-		_, _ = meta.Put("container.image.name", container.Image)
-		_, _ = meta.Put("container.name", container.Name)
-		event.Fields.DeepUpdate(meta)
-	} else {
-		d.log.Debugf("Container not found: cid=%s", cid)
+	// Lookup CID via process cgroup membership.
+	if len(d.pidFields) > 0 {
+		miniEvent := &beat.Event{Fields: make(mapstr.M, len(d.pidFields))}
+		for _, field := range d.pidFields {
+			if v, ok := otelmap.GetAtPath(field, body); ok {
+				_, _ = miniEvent.Fields.Put(field, v.AsRaw())
+			}
+		}
+		id, err := d.lookupContainerIDByPID(miniEvent)
+		if err != nil {
+			return "", fmt.Errorf("error reading container ID: %w", err)
+		}
+		if id != "" {
+			if err := otelmap.PutAtPath(dockerContainerIDKey, id, body); err != nil {
+				return "", err
+			}
+			return id, nil
+		}
 	}
 
-	return event, nil
+	// Lookup CID from user-defined match fields.
+	cid := matchFieldCID(d.fields, func(field string) (string, bool) {
+		v, ok := otelmap.GetAtPath(field, body)
+		if !ok || v.Type() != pcommon.ValueTypeStr {
+			return "", false
+		}
+		return v.Str(), true
+	})
+
+	return cid, nil
+}
+
+// resolveCIDFromSourcePath runs the configured source processor against a
+// synthetic event carrying only the log.file.path value, returning the
+// container ID
+func (d *addDockerMetadata) resolveCIDFromSourcePath(logFilePath interface{}) (string, error) {
+	miniEvent := &beat.Event{Fields: mapstr.M{"log": mapstr.M{"file": mapstr.M{"path": logFilePath}}}}
+	result, err := d.sourceProcessor.Run(miniEvent)
+	if err != nil {
+		return "", err
+	}
+	if result == nil {
+		return "", nil
+	}
+	if v, err := result.GetValue(dockerContainerIDKey); err == nil {
+		cid, _ := v.(string)
+		return cid, nil
+	}
+	return "", nil
+}
+
+// matchFieldCID returns the value of the first field in fields for which get
+// reports a match.
+func matchFieldCID(fields []string, get func(field string) (string, bool)) string {
+	for _, field := range fields {
+		if v, ok := get(field); ok {
+			return v
+		}
+	}
+	return ""
+}
+
+func (d *addDockerMetadata) buildContainerMeta(container *docker.Container) mapstr.M {
+	meta := mapstr.M{}
+	if len(container.Labels) > 0 {
+		labels := mapstr.M{}
+		for k, v := range container.Labels {
+			if d.dedot {
+				label := common.DeDot(k)
+				_, _ = labels.Put(label, v)
+			} else {
+				_ = safemapstr.Put(labels, k, v)
+			}
+		}
+		_, _ = meta.Put("container.labels", labels)
+	}
+	_, _ = meta.Put("container.id", container.ID)
+	_, _ = meta.Put("container.image.name", container.Image)
+	_, _ = meta.Put("container.name", container.Name)
+	return meta
 }
 
 func (d *addDockerMetadata) Close() error {

libbeat/processors/add_docker_metadata/add_docker_metadata_test.go

@@ -31,8 +31,10 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/pdata/pcommon"
 
 	"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/tests/resources"
 	"github.com/elastic/elastic-agent-autodiscover/bus"
@@ -45,6 +47,26 @@ import (
 	"github.com/elastic/elastic-agent-system-metrics/metric/system/resolve"
 )
 
+// assertRunPdataEquivalent verifies that RunPdata, given the same input fields
+// used to produce result via Run, enriches a pcommon.Map with identical output.
+func assertRunPdataEquivalent(t *testing.T, p beat.Processor, input, result mapstr.M) {
+	t.Helper()
+
+	pp, ok := p.(processors.PdataProcessor)
+	require.True(t, ok, "processor must implement PdataProcessor")
+
+	body := pcommon.NewMap()
+	require.NoError(t, otelmap.FromMapstr(body, input))
+	drop, err := pp.RunPdata(body)
+	require.NoError(t, err)
+	require.False(t, drop)
+
+	legacyNorm := pcommon.NewMap()
+	require.NoError(t, otelmap.FromMapstr(legacyNorm, result))
+	assert.Equal(t, otelmap.ToMapstr(legacyNorm), otelmap.ToMapstr(body),
+		"Run and RunPdata must produce identical output")
+}
+
 type testCGReader struct {
 }
 
@@ -152,6 +174,9 @@ func TestNoMatch(t *testing.T) {
 	assert.NoError(t, err, "processing an event")
 
 	assert.Equal(t, mapstr.M{"field": "value"}, result.Fields)
+
+	// RunPdata path (no CID resolves): assert Run == RunPdata.
+	assertRunPdataEquivalent(t, p, input, result.Fields)
 }
 
 func TestMatchNoContainer(t *testing.T) {
@@ -170,6 +195,9 @@ func TestMatchNoContainer(t *testing.T) {
 	assert.NoError(t, err, "processing an event")
 
 	assert.Equal(t, mapstr.M{"foo": "garbage"}, result.Fields)
+
+	// RunPdata path (container not found): assert Run == RunPdata.
+	assertRunPdataEquivalent(t, p, input, result.Fields)
 }
 
 func TestMatchContainer(t *testing.T) {
@@ -194,10 +222,9 @@ func TestMatchContainer(t *testing.T) {
 		}, nil))
 	assert.NoError(t, err, "initializing add_docker_metadata processor")
 
-	input := mapstr.M{
-		"foo": "container_id",
-	}
-	result, err := p.Run(&beat.Event{Fields: input})
+	input := mapstr.M{"foo": "container_id"}
+
+	result, err := p.Run(&beat.Event{Fields: input.Clone()})
 	assert.NoError(t, err, "processing an event")
 
 	assert.EqualValues(t, mapstr.M{
@@ -219,6 +246,9 @@ func TestMatchContainer(t *testing.T) {
 		},
 		"foo": "container_id",
 	}, result.Fields)
+
+	// RunPdata path (match_fields case): assert Run == RunPdata.
+	assertRunPdataEquivalent(t, p, input, result.Fields)
 }
 
 func TestMatchContainerWithDedot(t *testing.T) {
@@ -299,7 +329,7 @@ func TestMatchSource(t *testing.T) {
 		},
 	}
 
-	result, err := p.Run(&beat.Event{Fields: input})
+	result, err := p.Run(&beat.Event{Fields: input.Clone()})
 	assert.NoError(t, err, "processing an event")
 
 	assert.EqualValues(t, mapstr.M{
@@ -320,6 +350,9 @@ func TestMatchSource(t *testing.T) {
 			},
 		},
 	}, result.Fields)
+
+	// RunPdata path (log.file.path/sourceProcessor case): assert Run == RunPdata.
+	assertRunPdataEquivalent(t, p, input, result.Fields)
 }
 
 func TestDisableSource(t *testing.T) {
@@ -395,6 +428,9 @@ func TestMatchPIDs(t *testing.T) {
 		result, err := p.Run(&beat.Event{Fields: input})
 		assert.NoError(t, err, "processing an event")
 		assert.EqualValues(t, expected, result.Fields)
+
+		// RunPdata path (pid is not containerized): assert Run == RunPdata.
+		assertRunPdataEquivalent(t, p, input, result.Fields)
 	})
 
 	t.Run("pid does not exist", func(t *testing.T) {
@@ -407,6 +443,9 @@ func TestMatchPIDs(t *testing.T) {
 		result, err := p.Run(&beat.Event{Fields: input})
 		assert.NoError(t, err, "processing an event")
 		assert.EqualValues(t, expected, result.Fields)
+
+		// RunPdata path (pid does not exist): assert Run == RunPdata.
+		assertRunPdataEquivalent(t, p, input, result.Fields)
 	})
 
 	t.Run("pid is containerized", func(t *testing.T) {
@@ -417,9 +456,12 @@ func TestMatchPIDs(t *testing.T) {
 		expected.DeepUpdate(dockerMetadata)
 		expected.DeepUpdate(fields)
 
-		result, err := p.Run(&beat.Event{Fields: fields})
+		result, err := p.Run(&beat.Event{Fields: fields.Clone()})
 		assert.NoError(t, err, "processing an event")
 		assert.EqualValues(t, expected, result.Fields)
+
+		// RunPdata path (PID/cgroup lookup case): assert Run == RunPdata.
+		assertRunPdataEquivalent(t, p, fields, result.Fields)
 	})
 
 	t.Run("pid exited and ppid is containerized", func(t *testing.T) {
@@ -431,9 +473,12 @@ func TestMatchPIDs(t *testing.T) {
 		expected.DeepUpdate(dockerMetadata)
 		expected.DeepUpdate(fields)
 
-		result, err := p.Run(&beat.Event{Fields: fields})
+		result, err := p.Run(&beat.Event{Fields: fields.Clone()})
 		assert.NoError(t, err, "processing an event")
 		assert.EqualValues(t, expected, result.Fields)
+
+		// RunPdata path (PID/cgroup lookup case): assert Run == RunPdata.
+		assertRunPdataEquivalent(t, p, fields, result.Fields)
 	})
 
 	t.Run("cgroup error", func(t *testing.T) {
@@ -446,6 +491,9 @@ func TestMatchPIDs(t *testing.T) {
 		result, err := p.Run(&beat.Event{Fields: fields})
 		assert.NoError(t, err, "processing an event")
 		assert.EqualValues(t, expected, result.Fields)
+
+		// RunPdata path (cgroup error): assert Run == RunPdata.
+		assertRunPdataEquivalent(t, p, fields, result.Fields)
 	})
 }
 
@@ -547,6 +595,9 @@ func TestWatcherError(t *testing.T) {
 	result, err := p.Run(&beat.Event{Fields: input})
 	assert.NoError(t, err, "processing an event")
 	assert.Equal(t, mapstr.M{"field": "value"}, result.Fields)
+
+	// RunPdata path (docker unavailable): assert Run == RunPdata.
+	assertRunPdataEquivalent(t, p, input, result.Fields)
 }
 
 func TestConfigValidate(t *testing.T) {