libbeat/processors: track RunPdata coverage for beats processors

## Background

The `beatprocessor` OTel component (`x-pack/otel/processor/beatprocessor`) takes a
zero-copy pdata fast path when every processor in the configured chain implements
`processors.PdataProcessor` (`RunPdata(pcommon.Map) (bool, error)`). If any processor
in the chain is missing it, the whole chain falls back to a per-event mapstr round-trip.

This optimization was implemented for the global processors in https://github.com/elastic/beats/issues/50744.

This issue tracks which processors have coverage so progress is visible over time.

Beatprocessor has an allowlist that should ONLY contain processors that implement PdataProcessor.

Processors that wrap another processor (such as safe, wrap, or shared processors) can also silently disable the optimization if they don't implement RunPdata.

To check if a given beatprocessor processors definition is taking the fast path, simply run a collector with the intended configuration and look for this warning log line, if present, the fast path is disabled and should be fixed by porting the offending processors:

https://github.com/elastic/beats/blob/5a66062f2fa2cd547041397cd251b7d77a025153/x-pack/otel/processor/beatprocessor/processor.go#L65-L66

checkpdatacoverage.go

```go
//go:build ignore

// checkpdatacoverage reports which beat.Processor implementations under
// libbeat/processors and x-pack/libbeat/processors have (or are missing)
// a RunPdata method.
//
// A type is considered covered if it directly implements RunPdata, or if
// another type in the same package embeds it and provides RunPdata (the
// pdata-sibling pattern used by WhenPdataProcessor, safePdataProcessorWithClose, etc.).
//
// Usage (run from the repo root):
//
// go run ./checkpdataprocessor.go
package main

import (
"fmt"
"go/ast"
"go/parser"
"go/token"
"os"
"path/filepath"
"sort"
"strings"
)

type typeKey struct {
dir string
name string
}

type typeInfo struct {
hasRun bool
hasRunPdata bool
// embeds is the set of type names this type embeds (anonymous fields).
embeds []string
}

func main() {
roots := []string{
"libbeat/processors",
"x-pack/libbeat/processors",
}

types := map[typeKey]*typeInfo{}

for _, root := range roots {
if err := filepath.WalkDir(root, func(path string, d os.DirEntry, err error) error {
if err != nil {
return err
}
if d.IsDir() || !strings.HasSuffix(path, ".go") || strings.HasSuffix(path, "_test.go") {
return nil
}
collectMethods(path, types)
return nil
}); err != nil {
fmt.Fprintf(os.Stderr, "walk %s: %v\n", root, err)
os.Exit(1)
}
}

// transitiveEmbeds returns all type names reachable via the embed chain
// from startName within the same dir (breadth-first, cycle-safe).
transitiveEmbeds := func(dir, startName string) map[string]bool {
visited := map[string]bool{startName: true}
queue := []string{startName}
for len(queue) > 0 {
name := queue[0]
queue = queue[1:]
for _, emb := range types[typeKey{dir, name}].embeds {
if !visited[emb] {
visited[emb] = true
queue = append(queue, emb)
}
}
}
return visited
}

// A type is "pdata-covered" if it directly has RunPdata OR if another type
// in the same package has RunPdata and reaches it via the transitive embed chain
// (the pdata-sibling pattern: WhenPdataProcessor, safePdataProcessorWithClose, …).
covered := func(k typeKey) bool {
if types[k].hasRunPdata {
return true
}
for other, oti := range types {
if other.dir != k.dir || !oti.hasRunPdata {
continue
}
if transitiveEmbeds(k.dir, other.name)[k.name] {
return true
}
}
return false
}

var implemented, missing []string
for k, ti := range types {
if !ti.hasRun {
continue
}
label := fmt.Sprintf("%s.%s", k.dir, k.name)
if covered(k) {
implemented = append(implemented, label)
} else {
missing = append(missing, label)
}
}

sort.Strings(implemented)
sort.Strings(missing)

fmt.Println("=== RunPdata implemented ===")
for _, s := range implemented {
fmt.Println(" ✅", s)
}
fmt.Println()
fmt.Println("=== RunPdata missing ===")
for _, s := range missing {
fmt.Println(" ❌", s)
}
}

func collectMethods(path string, types map[typeKey]*typeInfo) {
fset := token.NewFileSet()
f, err := parser.ParseFile(fset, path, nil, 0)
if err != nil {
return
}

dir := filepath.Dir(path)

// Collect embedded fields from struct type declarations.
for _, decl := range f.Decls {
genDecl, ok := decl.(*ast.GenDecl)
if !ok {
continue
}
for _, spec := range genDecl.Specs {
typeSpec, ok := spec.(*ast.TypeSpec)
if !ok {
continue
}
structType, ok := typeSpec.Type.(*ast.StructType)
if !ok {
continue
}
key := typeKey{dir: dir, name: typeSpec.Name.Name}
if types[key] == nil {
types[key] = &typeInfo{}
}
for _, field := range structType.Fields.List {
if len(field.Names) != 0 {
continue // named field, not an embed
}
// Concrete type embed (e.g. sharedProcessorWithClose).
if name := receiverTypeName(field.Type); name != "" {
types[key].embeds = append(types[key].embeds, name)
continue
}
// Interface embed (e.g. beat.Processor): the struct satisfies
// beat.Processor via promotion, so treat it as having Run.
if sel, ok := field.Type.(*ast.SelectorExpr); ok {
if sel.Sel.Name == "Processor" {
types[key].hasRun = true
}
}
}
}
}

// Collect method declarations.
for _, decl := range f.Decls {
fn, ok := decl.(*ast.FuncDecl)
if !ok || fn.Recv == nil || len(fn.Recv.List) == 0 {
continue
}

typeName := receiverTypeName(fn.Recv.List[0].Type)
if typeName == "" {
continue
}

key := typeKey{dir: dir, name: typeName}
if types[key] == nil {
types[key] = &typeInfo{}
}

switch fn.Name.Name {
case "Run":
// beat.Processor.Run: func(event *beat.Event) (*beat.Event, error)
// Require exactly one parameter whose type contains "Event".
if fn.Type.Params != nil && len(fn.Type.Params.List) == 1 &&
nodeContains(fn.Type.Params.List[0].Type, "Event") {
types[key].hasRun = true
}
case "RunPdata":
types[key].hasRunPdata = true
}
}
}

func receiverTypeName(expr ast.Expr) string {
switch t := expr.(type) {
case *ast.StarExpr:
return receiverTypeName(t.X)
case *ast.Ident:
return t.Name
}
return ""
}

// nodeContains reports whether the printed form of expr contains substr.
func nodeContains(expr ast.Expr, substr string) bool {
switch t := expr.(type) {
case *ast.StarExpr:
return nodeContains(t.X, substr)
case *ast.SelectorExpr:
return nodeContains(t.X, substr) || strings.Contains(t.Sel.Name, substr)
case *ast.Ident:
return strings.Contains(t.Name, substr)
}
return false
}

```

## Status

```bash
$ go run checkpdatacoverage.go
=== RunPdata implemented ===
✅ libbeat/processors.SafeProcessor
✅ libbeat/processors.WhenProcessor
✅ libbeat/processors/actions.dropFields
✅ libbeat/processors/actions.mimeTypeProcessor
✅ libbeat/processors/actions/addfields.addFields
✅ libbeat/processors/add_cloud_metadata.addCloudMetadata
✅ libbeat/processors/add_docker_metadata.addDockerMetadata
✅ libbeat/processors/add_host_metadata.addHostMetadata
✅ libbeat/processors/add_kubernetes_metadata.kubernetesAnnotator
✅ libbeat/processors/shared.sharedProcessorWithClose

=== RunPdata missing ===
❌ libbeat/processors.IfThenElseProcessor
❌ libbeat/processors.Processors
❌ libbeat/processors/actions.addTags
❌ libbeat/processors/actions.alterFieldProcessor
❌ libbeat/processors/actions.appendProcessor
❌ libbeat/processors/actions.copyFields
❌ libbeat/processors/actions.decodeBase64Field
❌ libbeat/processors/actions.decodeJSONFields
❌ libbeat/processors/actions.decompressGzipField
❌ libbeat/processors/actions.dropEvent
❌ libbeat/processors/actions.extract_field
❌ libbeat/processors/actions.includeFields
❌ libbeat/processors/actions.networkDirectionProcessor
❌ libbeat/processors/actions.renameFields
❌ libbeat/processors/actions.replaceString
❌ libbeat/processors/actions.truncateFields
❌ libbeat/processors/actions/addagentmetadata.addAgentMetadata
❌ libbeat/processors/add_data_stream.AddDataStream
❌ libbeat/processors/add_formatted_index.AddFormattedIndex
❌ libbeat/processors/add_id.addID
❌ libbeat/processors/add_locale.addLocale
❌ libbeat/processors/add_observer_metadata.observerMetadata
❌ libbeat/processors/add_process_metadata.addProcessMetadata
❌ libbeat/processors/cache.cache
❌ libbeat/processors/communityid.processor
❌ libbeat/processors/convert.processor
❌ libbeat/processors/decode_csv_fields.decodeCSVFields
❌ libbeat/processors/decode_duration.decodeDuration
❌ libbeat/processors/decode_xml.decodeXML
❌ libbeat/processors/decode_xml_wineventlog.processor
❌ libbeat/processors/dissect.processor
❌ libbeat/processors/dns.processor
❌ libbeat/processors/extract_array.extractArrayProcessor
❌ libbeat/processors/fingerprint.fingerprint
❌ libbeat/processors/move_fields.moveFields
❌ libbeat/processors/now.now
❌ libbeat/processors/ratelimit.rateLimit
❌ libbeat/processors/registered_domain.processor
❌ libbeat/processors/script/javascript.jsProcessor
❌ libbeat/processors/script/javascript/module/processor.nativeProcessor
❌ libbeat/processors/syslog.processor
❌ libbeat/processors/timeseries.timeseriesProcessor
❌ libbeat/processors/timestamp.processor
❌ libbeat/processors/translate_ldap_attribute.processor
❌ libbeat/processors/translate_sid.processor
❌ libbeat/processors/urldecode.urlDecode
❌ x-pack/libbeat/processors/add_cloudfoundry_metadata.addCloudFoundryMetadata
❌ x-pack/libbeat/processors/add_nomad_metadata.nomadAnnotator
```

## Notes

- Each implementation should include a compile-time interface guard
(`var _ processors.PdataProcessor = (*T)(nil)`) and parity tests following the
pattern in `libbeat/processors/actions/drop_fields_test.go`.

Start with checkpdatacoverage.go and run it from the repository root to identify an uncovered processor. Read the corresponding processor implementation and libbeat/processors/actions/drop_fields_test.go for the parity-test pattern. Done means the selected processor has RunPdata coverage, a compile-time interface guard, parity tests, and no longer disables the beatprocessor fast path.

Written by the indexing model from the issue text.