Skip to content

Commit a67a106

Browse files
mergify[bot]mauri870
mergify[bot]
and
mauri870
authored
otel(processors): implement PdataProcessor for add_docker_metadata (#51679) (#51916)
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 c2c559c 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 53993f3) Co-authored-by: Mauri de Souza Meneguzzo <mauri870@gmail.com>
1 parent df2e9bf commit a67a106

2 files changed

Lines changed: 197 additions & 38 deletions

File tree

‎libbeat/processors/add_docker_metadata/add_docker_metadata.go‎

Lines changed: 139 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -30,8 +30,11 @@ import (
3030
"sync/atomic"
3131
"time"
3232

33+
"go.opentelemetry.io/collector/pdata/pcommon"
34+
3335
"github.com/elastic/beats/v7/libbeat/beat"
3436
"github.com/elastic/beats/v7/libbeat/common"
37+
"github.com/elastic/beats/v7/libbeat/otel/otelmap"
3538
"github.com/elastic/beats/v7/libbeat/processors"
3639
"github.com/elastic/beats/v7/libbeat/processors/actions"
3740
"github.com/elastic/elastic-agent-autodiscover/docker"
@@ -62,6 +65,8 @@ func init() {
6265
processors.RegisterPlugin(processorName, New)
6366
}
6467

68+
var _ processors.PdataProcessor = (*addDockerMetadata)(nil)
69+
6570
type addDockerMetadata struct {
6671
log *logp.Logger
6772
watcher docker.Watcher
@@ -264,20 +269,20 @@ func (d *addDockerMetadata) Run(event *beat.Event) (*beat.Event, error) {
264269
return event, nil
265270
}
266271
var cid string
267-
var err error
268272

269273
// Extract CID from the filepath contained in the "log.file.path" field.
270274
if d.sourceProcessor != nil {
271275
lfp, _ := event.Fields.GetValue("log.file.path")
272276
if lfp != nil {
273-
event, err = d.sourceProcessor.Run(event)
277+
id, err := d.resolveCIDFromSourcePath(lfp)
274278
if err != nil {
275279
d.log.Debugf("Error while extracting container ID from source path: %v", err)
276280
return event, nil
277281
}
278282

279-
if v, err := event.GetValue(dockerContainerIDKey); err == nil {
280-
cid, _ = v.(string)
283+
if id != "" {
284+
cid = id
285+
_, _ = event.PutValue(dockerContainerIDKey, cid)
281286
}
282287
}
283288
}
@@ -296,49 +301,152 @@ func (d *addDockerMetadata) Run(event *beat.Event) (*beat.Event, error) {
296301

297302
// Lookup CID from the user defined field names.
298303
if cid == "" && len(d.fields) > 0 {
299-
for _, field := range d.fields {
304+
cid = matchFieldCID(d.fields, func(field string) (string, bool) {
300305
value, err := event.GetValue(field)
301306
if err != nil {
302-
continue
303-
}
304-
305-
if strValue, ok := value.(string); ok {
306-
cid = strValue
307-
break
307+
return "", false
308308
}
309-
}
309+
strValue, ok := value.(string)
310+
return strValue, ok
311+
})
310312
}
311313

312314
if cid == "" {
313315
return event, nil
314316
}
315317

316318
container := d.watcher.Container(cid)
317-
if container != nil {
318-
meta := mapstr.M{}
319-
320-
if len(container.Labels) > 0 {
321-
labels := mapstr.M{}
322-
for k, v := range container.Labels {
323-
if d.dedot {
324-
label := common.DeDot(k)
325-
_, _ = labels.Put(label, v)
326-
} else {
327-
_ = safemapstr.Put(labels, k, v)
319+
if container == nil {
320+
d.log.Debugf("Container not found: cid=%s", cid)
321+
return event, nil
322+
}
323+
324+
event.Fields.DeepUpdate(d.buildContainerMeta(container))
325+
return event, nil
326+
}
327+
328+
// RunPdata enriches the given pcommon.Map directly with Docker container metadata
329+
func (d *addDockerMetadata) RunPdata(body pcommon.Map) (bool, error) {
330+
if !d.dockerAvailable.Load() {
331+
return false, nil
332+
}
333+
334+
cid, err := d.resolveCIDFromPdata(body)
335+
if err != nil {
336+
return false, err
337+
}
338+
if cid == "" {
339+
return false, nil
340+
}
341+
342+
container := d.watcher.Container(cid)
343+
if container == nil {
344+
d.log.Debugf("Container not found: cid=%s", cid)
345+
return false, nil
346+
}
347+
348+
return false, otelmap.MergeMapstrIntoPdata(d.buildContainerMeta(container), body, true)
349+
}
350+
351+
// resolveCIDFromPdata extracts the container ID from a pcommon.Map
352+
func (d *addDockerMetadata) resolveCIDFromPdata(body pcommon.Map) (string, error) {
353+
// Extract CID from log.file.path via sourceProcessor.
354+
if d.sourceProcessor != nil {
355+
if lfpVal, ok := otelmap.GetAtPath("log.file.path", body); ok && lfpVal.Type() == pcommon.ValueTypeStr {
356+
id, err := d.resolveCIDFromSourcePath(lfpVal.Str())
357+
if err != nil {
358+
d.log.Debugf("Error while extracting container ID from source path: %v", err)
359+
return "", nil
360+
}
361+
if id != "" {
362+
if err := otelmap.PutAtPath(dockerContainerIDKey, id, body); err != nil {
363+
return "", err
328364
}
365+
return id, nil
329366
}
330-
_, _ = meta.Put("container.labels", labels)
331367
}
368+
}
332369

333-
_, _ = meta.Put("container.id", container.ID)
334-
_, _ = meta.Put("container.image.name", container.Image)
335-
_, _ = meta.Put("container.name", container.Name)
336-
event.Fields.DeepUpdate(meta)
337-
} else {
338-
d.log.Debugf("Container not found: cid=%s", cid)
370+
// Lookup CID via process cgroup membership.
371+
if len(d.pidFields) > 0 {
372+
miniEvent := &beat.Event{Fields: make(mapstr.M, len(d.pidFields))}
373+
for _, field := range d.pidFields {
374+
if v, ok := otelmap.GetAtPath(field, body); ok {
375+
_, _ = miniEvent.Fields.Put(field, v.AsRaw())
376+
}
377+
}
378+
id, err := d.lookupContainerIDByPID(miniEvent)
379+
if err != nil {
380+
return "", fmt.Errorf("error reading container ID: %w", err)
381+
}
382+
if id != "" {
383+
if err := otelmap.PutAtPath(dockerContainerIDKey, id, body); err != nil {
384+
return "", err
385+
}
386+
return id, nil
387+
}
339388
}
340389

341-
return event, nil
390+
// Lookup CID from user-defined match fields.
391+
cid := matchFieldCID(d.fields, func(field string) (string, bool) {
392+
v, ok := otelmap.GetAtPath(field, body)
393+
if !ok || v.Type() != pcommon.ValueTypeStr {
394+
return "", false
395+
}
396+
return v.Str(), true
397+
})
398+
399+
return cid, nil
400+
}
401+
402+
// resolveCIDFromSourcePath runs the configured source processor against a
403+
// synthetic event carrying only the log.file.path value, returning the
404+
// container ID
405+
func (d *addDockerMetadata) resolveCIDFromSourcePath(logFilePath interface{}) (string, error) {
406+
miniEvent := &beat.Event{Fields: mapstr.M{"log": mapstr.M{"file": mapstr.M{"path": logFilePath}}}}
407+
result, err := d.sourceProcessor.Run(miniEvent)
408+
if err != nil {
409+
return "", err
410+
}
411+
if result == nil {
412+
return "", nil
413+
}
414+
if v, err := result.GetValue(dockerContainerIDKey); err == nil {
415+
cid, _ := v.(string)
416+
return cid, nil
417+
}
418+
return "", nil
419+
}
420+
421+
// matchFieldCID returns the value of the first field in fields for which get
422+
// reports a match.
423+
func matchFieldCID(fields []string, get func(field string) (string, bool)) string {
424+
for _, field := range fields {
425+
if v, ok := get(field); ok {
426+
return v
427+
}
428+
}
429+
return ""
430+
}
431+
432+
func (d *addDockerMetadata) buildContainerMeta(container *docker.Container) mapstr.M {
433+
meta := mapstr.M{}
434+
if len(container.Labels) > 0 {
435+
labels := mapstr.M{}
436+
for k, v := range container.Labels {
437+
if d.dedot {
438+
label := common.DeDot(k)
439+
_, _ = labels.Put(label, v)
440+
} else {
441+
_ = safemapstr.Put(labels, k, v)
442+
}
443+
}
444+
_, _ = meta.Put("container.labels", labels)
445+
}
446+
_, _ = meta.Put("container.id", container.ID)
447+
_, _ = meta.Put("container.image.name", container.Image)
448+
_, _ = meta.Put("container.name", container.Name)
449+
return meta
342450
}
343451

344452
func (d *addDockerMetadata) Close() error {

‎libbeat/processors/add_docker_metadata/add_docker_metadata_test.go‎

Lines changed: 58 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -31,8 +31,10 @@ import (
3131

3232
"github.com/stretchr/testify/assert"
3333
"github.com/stretchr/testify/require"
34+
"go.opentelemetry.io/collector/pdata/pcommon"
3435

3536
"github.com/elastic/beats/v7/libbeat/beat"
37+
"github.com/elastic/beats/v7/libbeat/otel/otelmap"
3638
"github.com/elastic/beats/v7/libbeat/processors"
3739
"github.com/elastic/beats/v7/libbeat/tests/resources"
3840
"github.com/elastic/elastic-agent-autodiscover/bus"
@@ -45,6 +47,26 @@ import (
4547
"github.com/elastic/elastic-agent-system-metrics/metric/system/resolve"
4648
)
4749

50+
// assertRunPdataEquivalent verifies that RunPdata, given the same input fields
51+
// used to produce result via Run, enriches a pcommon.Map with identical output.
52+
func assertRunPdataEquivalent(t *testing.T, p beat.Processor, input, result mapstr.M) {
53+
t.Helper()
54+
55+
pp, ok := p.(processors.PdataProcessor)
56+
require.True(t, ok, "processor must implement PdataProcessor")
57+
58+
body := pcommon.NewMap()
59+
require.NoError(t, otelmap.FromMapstr(body, input))
60+
drop, err := pp.RunPdata(body)
61+
require.NoError(t, err)
62+
require.False(t, drop)
63+
64+
legacyNorm := pcommon.NewMap()
65+
require.NoError(t, otelmap.FromMapstr(legacyNorm, result))
66+
assert.Equal(t, otelmap.ToMapstr(legacyNorm), otelmap.ToMapstr(body),
67+
"Run and RunPdata must produce identical output")
68+
}
69+
4870
type testCGReader struct {
4971
}
5072

@@ -152,6 +174,9 @@ func TestNoMatch(t *testing.T) {
152174
assert.NoError(t, err, "processing an event")
153175

154176
assert.Equal(t, mapstr.M{"field": "value"}, result.Fields)
177+
178+
// RunPdata path (no CID resolves): assert Run == RunPdata.
179+
assertRunPdataEquivalent(t, p, input, result.Fields)
155180
}
156181

157182
func TestMatchNoContainer(t *testing.T) {
@@ -170,6 +195,9 @@ func TestMatchNoContainer(t *testing.T) {
170195
assert.NoError(t, err, "processing an event")
171196

172197
assert.Equal(t, mapstr.M{"foo": "garbage"}, result.Fields)
198+
199+
// RunPdata path (container not found): assert Run == RunPdata.
200+
assertRunPdataEquivalent(t, p, input, result.Fields)
173201
}
174202

175203
func TestMatchContainer(t *testing.T) {
@@ -194,10 +222,9 @@ func TestMatchContainer(t *testing.T) {
194222
}, nil))
195223
assert.NoError(t, err, "initializing add_docker_metadata processor")
196224

197-
input := mapstr.M{
198-
"foo": "container_id",
199-
}
200-
result, err := p.Run(&beat.Event{Fields: input})
225+
input := mapstr.M{"foo": "container_id"}
226+
227+
result, err := p.Run(&beat.Event{Fields: input.Clone()})
201228
assert.NoError(t, err, "processing an event")
202229

203230
assert.EqualValues(t, mapstr.M{
@@ -219,6 +246,9 @@ func TestMatchContainer(t *testing.T) {
219246
},
220247
"foo": "container_id",
221248
}, result.Fields)
249+
250+
// RunPdata path (match_fields case): assert Run == RunPdata.
251+
assertRunPdataEquivalent(t, p, input, result.Fields)
222252
}
223253

224254
func TestMatchContainerWithDedot(t *testing.T) {
@@ -299,7 +329,7 @@ func TestMatchSource(t *testing.T) {
299329
},
300330
}
301331

302-
result, err := p.Run(&beat.Event{Fields: input})
332+
result, err := p.Run(&beat.Event{Fields: input.Clone()})
303333
assert.NoError(t, err, "processing an event")
304334

305335
assert.EqualValues(t, mapstr.M{
@@ -320,6 +350,9 @@ func TestMatchSource(t *testing.T) {
320350
},
321351
},
322352
}, result.Fields)
353+
354+
// RunPdata path (log.file.path/sourceProcessor case): assert Run == RunPdata.
355+
assertRunPdataEquivalent(t, p, input, result.Fields)
323356
}
324357

325358
func TestDisableSource(t *testing.T) {
@@ -395,6 +428,9 @@ func TestMatchPIDs(t *testing.T) {
395428
result, err := p.Run(&beat.Event{Fields: input})
396429
assert.NoError(t, err, "processing an event")
397430
assert.EqualValues(t, expected, result.Fields)
431+
432+
// RunPdata path (pid is not containerized): assert Run == RunPdata.
433+
assertRunPdataEquivalent(t, p, input, result.Fields)
398434
})
399435

400436
t.Run("pid does not exist", func(t *testing.T) {
@@ -407,6 +443,9 @@ func TestMatchPIDs(t *testing.T) {
407443
result, err := p.Run(&beat.Event{Fields: input})
408444
assert.NoError(t, err, "processing an event")
409445
assert.EqualValues(t, expected, result.Fields)
446+
447+
// RunPdata path (pid does not exist): assert Run == RunPdata.
448+
assertRunPdataEquivalent(t, p, input, result.Fields)
410449
})
411450

412451
t.Run("pid is containerized", func(t *testing.T) {
@@ -417,9 +456,12 @@ func TestMatchPIDs(t *testing.T) {
417456
expected.DeepUpdate(dockerMetadata)
418457
expected.DeepUpdate(fields)
419458

420-
result, err := p.Run(&beat.Event{Fields: fields})
459+
result, err := p.Run(&beat.Event{Fields: fields.Clone()})
421460
assert.NoError(t, err, "processing an event")
422461
assert.EqualValues(t, expected, result.Fields)
462+
463+
// RunPdata path (PID/cgroup lookup case): assert Run == RunPdata.
464+
assertRunPdataEquivalent(t, p, fields, result.Fields)
423465
})
424466

425467
t.Run("pid exited and ppid is containerized", func(t *testing.T) {
@@ -431,9 +473,12 @@ func TestMatchPIDs(t *testing.T) {
431473
expected.DeepUpdate(dockerMetadata)
432474
expected.DeepUpdate(fields)
433475

434-
result, err := p.Run(&beat.Event{Fields: fields})
476+
result, err := p.Run(&beat.Event{Fields: fields.Clone()})
435477
assert.NoError(t, err, "processing an event")
436478
assert.EqualValues(t, expected, result.Fields)
479+
480+
// RunPdata path (PID/cgroup lookup case): assert Run == RunPdata.
481+
assertRunPdataEquivalent(t, p, fields, result.Fields)
437482
})
438483

439484
t.Run("cgroup error", func(t *testing.T) {
@@ -446,6 +491,9 @@ func TestMatchPIDs(t *testing.T) {
446491
result, err := p.Run(&beat.Event{Fields: fields})
447492
assert.NoError(t, err, "processing an event")
448493
assert.EqualValues(t, expected, result.Fields)
494+
495+
// RunPdata path (cgroup error): assert Run == RunPdata.
496+
assertRunPdataEquivalent(t, p, fields, result.Fields)
449497
})
450498
}
451499

@@ -547,6 +595,9 @@ func TestWatcherError(t *testing.T) {
547595
result, err := p.Run(&beat.Event{Fields: input})
548596
assert.NoError(t, err, "processing an event")
549597
assert.Equal(t, mapstr.M{"field": "value"}, result.Fields)
598+
599+
// RunPdata path (docker unavailable): assert Run == RunPdata.
600+
assertRunPdataEquivalent(t, p, input, result.Fields)
550601
}
551602

552603
func TestConfigValidate(t *testing.T) {

0 commit comments

Comments
 (0)