otel(processors): implement PdataProcessor interface for add_kubernetes_metadata

libbeat/processors/add_kubernetes_metadata/kubernetes.go

@@ -32,13 +32,16 @@ import (
 
 	k8sclient "k8s.io/client-go/kubernetes"
 
+	"go.opentelemetry.io/collector/pdata/pcommon"
+
 	"github.com/elastic/elastic-agent-autodiscover/kubernetes"
 	"github.com/elastic/elastic-agent-autodiscover/kubernetes/metadata"
 	"github.com/elastic/elastic-agent-libs/config"
 	"github.com/elastic/elastic-agent-libs/logp"
 	"github.com/elastic/elastic-agent-libs/mapstr"
 
 	"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/processors/shared"
 )
@@ -79,6 +82,8 @@ func init() {
 	Indexing.AddMatcher(FieldFormatMatcherName, NewFieldFormatMatcher)
 }
 
+var _ processors.PdataProcessor = (*kubernetesAnnotator)(nil)
+
 func isKubernetesAvailable(client k8sclient.Interface, logger *logp.Logger) (bool, error) {
 	server, err := client.Discovery().ServerVersion()
 	if err != nil {
@@ -398,34 +403,69 @@ func (k *kubernetesAnnotator) Run(event *beat.Event) (*beat.Event, error) {
 		return event, nil
 	}
 
-	// One full clone for the kubernetes field; one cheap sub-map clone for the OCI
-	// container field. This replaces the original three full clones.
-	kubeMeta := metadata.Clone()
+	kubeMeta, ociContainer := prepareKubeMetadata(metadata)
+	if ociContainer != nil {
+		event.Fields.DeepUpdate(mapstr.M{"container": ociContainer})
+	}
+	event.Fields.DeepUpdate(kubeMeta)
+
+	return event, nil
+}
+
+// RunPdata enriches the given pcommon.Map directly with Kubernetes metadata
+func (k *kubernetesAnnotator) RunPdata(body pcommon.Map) (bool, error) {
+	if _, ok := body.Get("kubernetes"); ok {
+		return false, nil
+	}
+
+	// A nil state means init has not published yet (still running or kubernetes unavailable); the
+	// load pairs with the Store in init for a race-free read.
+	state := k.state.Load()
+	if state == nil {
+		return false, nil
+	}
 
-	// Build the OCI container field by cloning only the container sub-map —
-	// much cheaper than cloning the full metadata. Transform it in place:
-	// drop container.name and rewrite container.image -> container.image.name.
+	index := state.matchers.MetadataIndexPdata(body)
+	if index == "" {
+		k.log.Debug("No container match string, not adding kubernetes data")
+		return false, nil
+	}
+
+	metadata := k.cache.get(index)
+	if metadata == nil {
+		return false, nil
+	}
+
+	kubeMeta, ociContainer := prepareKubeMetadata(metadata)
+	if ociContainer != nil {
+		if err := otelmap.MergeMapstrIntoPdata(mapstr.M{"container": ociContainer}, body, true); err != nil {
+			return false, err
+		}
+	}
+	return false, otelmap.MergeMapstrIntoPdata(kubeMeta, body, true)
+}
+
+// prepareKubeMetadata clones the cached metadata, builds the OCI container
+// sub-map from kubernetes.container (dropping name, rewriting image), and
+// strips the kubernetes-only container fields. container.name is kept in
+// kubeMeta to match original behaviour.
+// ociContainer is nil when the kubernetes.container sub-map is absent.
+func prepareKubeMetadata(metadata mapstr.M) (kubeMeta mapstr.M, ociContainer mapstr.M) {
+	kubeMeta = metadata.Clone()
 	if containerVal, err := kubeMeta.GetValue("kubernetes.container"); err == nil {
 		if cm, ok := containerVal.(mapstr.M); ok {
-			ociContainer := cm.Clone()
+			ociContainer = cm.Clone()
 			_ = ociContainer.Delete("name")
 			if img, imgErr := ociContainer.GetValue("image"); imgErr == nil {
 				_ = ociContainer.Delete("image")
 				ociContainer["image"] = mapstr.M{"name": img}
 			}
-			event.Fields.DeepUpdate(mapstr.M{"container": ociContainer})
 		}
 	}
-
-	// Remove container fields that belong only in the OCI section before writing
-	// kubernetes metadata to the event. container.name is intentionally kept here
-	// to match original behaviour.
 	_ = kubeMeta.Delete("kubernetes.container.id")
 	_ = kubeMeta.Delete("kubernetes.container.runtime")
 	_ = kubeMeta.Delete("kubernetes.container.image")
-	event.Fields.DeepUpdate(kubeMeta)
-
-	return event, nil
+	return
 }
 
 func (k *kubernetesAnnotator) Close() error {

libbeat/processors/add_kubernetes_metadata/kubernetes_test.go

@@ -27,18 +27,41 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.uber.org/goleak"
 	"k8s.io/apimachinery/pkg/runtime"
 	k8sclient "k8s.io/client-go/kubernetes"
 	k8sfake "k8s.io/client-go/kubernetes/fake"
 	k8stesting "k8s.io/client-go/testing"
 
 	"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/elastic-agent-libs/config"
 	"github.com/elastic/elastic-agent-libs/logp/logptest"
 	"github.com/elastic/elastic-agent-libs/mapstr"
 )
 
+// 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")
+}
+
 // Test Annotator is skipped if kubernetes metadata already exist
 func TestAnnotatorSkipped(t *testing.T) {
 	cfg := config.MustNewConfigFrom(map[string]interface{}{
@@ -67,23 +90,7 @@ func TestAnnotatorSkipped(t *testing.T) {
 			},
 		})
 
-	event, err := processor.Run(&beat.Event{
-		Fields: mapstr.M{
-			"kubernetes": mapstr.M{
-				"pod": mapstr.M{
-					"name": "foo",
-					"id":   "pod_id",
-					"metrics": mapstr.M{
-						"a": 1,
-						"b": 2,
-					},
-				},
-			},
-		},
-	})
-	assert.NoError(t, err)
-
-	assert.Equal(t, mapstr.M{
+	input := mapstr.M{
 		"kubernetes": mapstr.M{
 			"pod": mapstr.M{
 				"name": "foo",
@@ -94,7 +101,14 @@ func TestAnnotatorSkipped(t *testing.T) {
 				},
 			},
 		},
-	}, event.Fields)
+	}
+	event, err := processor.Run(&beat.Event{Fields: input.Clone()})
+	assert.NoError(t, err)
+
+	assert.Equal(t, input, event.Fields)
+
+	// RunPdata path: assert Run == RunPdata.
+	assertRunPdataEquivalent(t, &processor, input, event.Fields)
 }
 
 // Test metadata are not included in the event
@@ -123,6 +137,9 @@ func TestAnnotatorWithNoKubernetesAvailable(t *testing.T) {
 	assert.NoError(t, err)
 
 	assert.Equal(t, intialEventMap, event.Fields)
+
+	// RunPdata path: assert Run == RunPdata.
+	assertRunPdataEquivalent(t, &processor, intialEventMap, event.Fields)
 }
 
 // TestNewProcessorConfigDefaultIndexers validates the behaviour of default indexers and
@@ -317,6 +334,9 @@ func TestAnnotatorRunFullContainerMetadata(t *testing.T) {
 	assert.NotContains(t, k8sContainer, "id", "kubernetes.container must NOT have id")
 	assert.NotContains(t, k8sContainer, "runtime", "kubernetes.container must NOT have runtime")
 	assert.NotContains(t, k8sContainer, "image", "kubernetes.container must NOT have image")
+
+	// RunPdata path: assert Run == RunPdata.
+	assertRunPdataEquivalent(t, processor, mapstr.M{"container": mapstr.M{"id": "abc123"}}, event.Fields)
 }
 
 // TestAnnotatorRunContainerWithoutImage verifies that when there is no image in
@@ -345,6 +365,9 @@ func TestAnnotatorRunContainerWithoutImage(t *testing.T) {
 	assert.Equal(t, "abc456", container["id"])
 	assert.Equal(t, "docker", container["runtime"])
 	assert.NotContains(t, container, "image", "container must NOT have image key when no image in metadata")
+
+	// RunPdata path: assert Run == RunPdata.
+	assertRunPdataEquivalent(t, processor, baseEvent("abc456").Fields, event.Fields)
 }
 
 // TestAnnotatorRunContainerWithoutName verifies that missing container.name
@@ -375,6 +398,9 @@ func TestAnnotatorRunContainerWithoutName(t *testing.T) {
 	require.IsType(t, mapstr.M{}, imageRaw)
 	imageMap, _ := imageRaw.(mapstr.M)
 	assert.Equal(t, "busybox:latest", imageMap["name"])
+
+	// RunPdata path: assert Run == RunPdata.
+	assertRunPdataEquivalent(t, processor, baseEvent("abc789").Fields, event.Fields)
 }
 
 // TestAnnotatorRunNoContainerSubMap verifies that when the metadata has no
@@ -426,6 +452,9 @@ func TestAnnotatorRunNoContainerSubMap(t *testing.T) {
 	require.IsType(t, mapstr.M{}, podRaw)
 	pod, _ := podRaw.(mapstr.M)
 	assert.Equal(t, "mypod", pod["name"])
+
+	// RunPdata path: assert Run == RunPdata.
+	assertRunPdataEquivalent(t, processor, mapstr.M{"pod": mapstr.M{"name": "mypod"}}, event.Fields)
 }
 
 // TestAnnotatorRunExtraContainerFieldsPreserved verifies that unknown extra
@@ -454,6 +483,9 @@ func TestAnnotatorRunExtraContainerFieldsPreserved(t *testing.T) {
 	container, _ := containerRaw.(mapstr.M)
 
 	assert.Equal(t, "extra", container["custom_field"], "extra container fields must be preserved in OCI container")
+
+	// RunPdata path: assert Run == RunPdata.
+	assertRunPdataEquivalent(t, processor, baseEvent("xtra001").Fields, event.Fields)
 }
 
 // TestAnnotatorRunCacheNotMutated verifies that running the processor multiple
@@ -536,6 +568,24 @@ func TestAnnotatorRunEventIndependence(t *testing.T) {
 	assert.NotContains(t, container2, "injected", "mutating first result must not affect second result")
 }
 
+// TestAnnotatorRunCacheMiss verifies that when the matcher returns an index key
+// but the cache has no entry for it, both Run and RunPdata leave the event
+// unchanged and return no error.
+func TestAnnotatorRunCacheMiss(t *testing.T) {
+	// Seed the cache with a different key so the matcher can fire but miss.
+	processor := newAnnotatorForTest(t, "known-key", mapstr.M{
+		"kubernetes": mapstr.M{"pod": mapstr.M{"name": "mypod"}},
+	})
+
+	input := mapstr.M{"container": mapstr.M{"id": "unknown-key"}}
+	event, err := processor.Run(&beat.Event{Fields: input.Clone()})
+	require.NoError(t, err)
+	assert.Equal(t, input, event.Fields, "Run must leave the event unchanged on a cache miss")
+
+	// RunPdata path: assert Run == RunPdata.
+	assertRunPdataEquivalent(t, processor, input, event.Fields)
+}
+
 func BenchmarkKubernetesAnnotatorRun(b *testing.B) {
 	cfg := config.MustNewConfigFrom(map[string]interface{}{
 		"lookup_fields": []string{"container.id"},

libbeat/processors/add_kubernetes_metadata/matchers.go

@@ -22,8 +22,11 @@ import (
 	"regexp"
 	"slices"
 
+	"go.opentelemetry.io/collector/pdata/pcommon"
+
 	"github.com/elastic/beats/v7/libbeat/beat"
 	"github.com/elastic/beats/v7/libbeat/common/fmtstr"
+	"github.com/elastic/beats/v7/libbeat/otel/otelmap"
 	"github.com/elastic/beats/v7/libbeat/outputs/codec"
 	"github.com/elastic/beats/v7/libbeat/outputs/codec/format"
 	"github.com/elastic/elastic-agent-libs/config"
@@ -89,6 +92,35 @@ func (m *Matchers) MetadataIndex(event mapstr.M) string {
 	return ""
 }
 
+// pdataMatcher is an optional interface a Matcher can implement to avoid a
+// full pcommon.Map→mapstr.M conversion when looking up the metadata index.
+type pdataMatcher interface {
+	MetadataIndexPdata(body pcommon.Map) string
+}
+
+// MetadataIndexPdata is the pdata-native counterpart of MetadataIndex. For
+// each matcher that implements pdataMatcher the lookup is done directly on the
+// pcommon.Map; the mapstr.M conversion is performed lazily and only once for
+// matchers that do not implement the interface.
+func (m *Matchers) MetadataIndexPdata(body pcommon.Map) string {
+	var fallback mapstr.M
+	for _, matcher := range m.matchers {
+		if pm, ok := matcher.(pdataMatcher); ok {
+			if index := pm.MetadataIndexPdata(body); index != "" {
+				return index
+			}
+		} else {
+			if fallback == nil {
+				fallback = otelmap.ToMapstr(body)
+			}
+			if index := matcher.MetadataIndex(fallback); index != "" {
+				return index
+			}
+		}
+	}
+	return ""
+}
+
 func (m *Matchers) Empty() bool {
 	return len(m.matchers) == 0
 }
@@ -157,6 +189,28 @@ func (f *FieldMatcher) MetadataIndex(event mapstr.M) string {
 	return ""
 }
 
+func (f *FieldMatcher) MetadataIndexPdata(body pcommon.Map) string {
+	for _, field := range f.MatchFields {
+		v, ok := otelmap.GetAtPath(field, body)
+		if !ok || v.Type() != pcommon.ValueTypeStr {
+			continue
+		}
+		fieldValue := v.Str()
+		if f.Regexp == nil {
+			return fieldValue
+		}
+		matches := f.Regexp.FindStringSubmatch(fieldValue)
+		if matches == nil {
+			continue
+		}
+		key := matches[f.Regexp.SubexpIndex(regexKeyGroupName)]
+		if key != "" {
+			return key
+		}
+	}
+	return ""
+}
+
 type FieldFormatMatcher struct {
 	Codec codec.Codec
 }