From 70df3c0aeb0e049736c7168cd533ce22237e6c7c Mon Sep 17 00:00:00 2001 From: Mauri de Souza Meneguzzo Date: Tue, 14 Jul 2026 08:14:05 -0300 Subject: [PATCH] otel(processors): implement PdataProcessor for add_docker_metadata (#51679) Adds RunPdata, which enriches a pcommon.Map directly with Docker container metadata. Introduces resolveCIDFromPdata, mirroring Run's three-path CID resolution strategy (log.file.path via sourceProcessor, PID-based cgroup lookup, user-defined match fields) without converting to/from mapstr.M. Parity coverage in TestMatchContainer verifies Run and RunPdata produce identical output. * metricbeat: fix module.NewWrapper call in runner_test.go Upstream commit c2c559cdf0 passed a bare *logp.Logger as the third argument to module.NewWrapper, whose signature expects *beat.Info, breaking build/vet. Wrap the logger in &beat.Info{...} to match the API and the other calls in this file. * reuse code between Run/RunPdata * more coverage for Run == RunPdata for negative cases * linter fix (cherry picked from commit 53993f36bf5a676da55945530dd496453fcea415) --- .../add_docker_metadata.go | 170 ++++++++++++++---- .../add_docker_metadata_test.go | 65 ++++++- 2 files changed, 197 insertions(+), 38 deletions(-) diff --git a/libbeat/processors/add_docker_metadata/add_docker_metadata.go b/libbeat/processors/add_docker_metadata/add_docker_metadata.go index a47f760ceee..adbf92ca87a 100644 --- a/libbeat/processors/add_docker_metadata/add_docker_metadata.go +++ b/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,17 +301,14 @@ 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 == "" { @@ -314,31 +316,137 @@ func (d *addDockerMetadata) Run(event *beat.Event) (*beat.Event, error) { } 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 { diff --git a/libbeat/processors/add_docker_metadata/add_docker_metadata_test.go b/libbeat/processors/add_docker_metadata/add_docker_metadata_test.go index 6ae8be9e05d..cfa6694a090 100644 --- a/libbeat/processors/add_docker_metadata/add_docker_metadata_test.go +++ b/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) {