[8.19](backport #45328) feat(otel): add skeleton of Beat processor (#46760)

* feat(otel): add skeleton of Beat processor (#45328)

* feat(otel): add skeleton of Beat processor

* refactor: separate the OTel Beat processor from Beat processors in code

* docs: add note on compatibility with receivers

* docs: make the processor description clearer

* docs: add README in `x-pack/otel` directory

(cherry picked from commit 65c0dc2)

# Conflicts:
#	NOTICE.txt
#	go.mod
#	go.sum

* fix conflicts

---------

Co-authored-by: Andrzej Stencel <[email protected]>

Author: mergify[bot]

NOTICE.txt

@@ -136,7 +136,7 @@ func (e *inputTestingEnvironment) waitUntilEventCount(count int) {
 		fmt.Fprintf(&msg, "too few events; expected: %d, actual: %d", count, sum)
 
 		return false
-	}, 5*time.Second, 10*time.Millisecond, &msg)
+	}, 10*time.Second, 10*time.Millisecond, &msg)
 }
 
 func (e *inputTestingEnvironment) RequireStatuses(expected []statusUpdate) {

filebeat/input/journald/environment_test.go

@@ -116,7 +116,7 @@ require (
 	github.com/shopspring/decimal v1.3.1 // indirect
 	github.com/spf13/cobra v1.9.1
 	github.com/spf13/pflag v1.0.6
-	github.com/stretchr/testify v1.10.0
+	github.com/stretchr/testify v1.11.1
 	github.com/ugorji/go/codec v1.1.8
 	github.com/vmware/govmomi v0.51.0
 	go.elastic.co/ecszap v1.0.2
@@ -135,8 +135,8 @@ require (
 	golang.org/x/tools v0.35.0
 	google.golang.org/api v0.238.0
 	google.golang.org/genproto v0.0.0-20250505200425-f936aa4a68b2 // indirect
-	google.golang.org/grpc v1.75.0
-	google.golang.org/protobuf v1.36.7
+	google.golang.org/grpc v1.75.1
+	google.golang.org/protobuf v1.36.9
 	gopkg.in/inf.v0 v0.9.1
 	gopkg.in/yaml.v2 v2.4.0
 	gotest.tools/gotestsum v1.7.0
@@ -241,6 +241,8 @@ require (
 	go.opentelemetry.io/collector/client v1.38.0
 	go.opentelemetry.io/collector/component/componenttest v0.132.0
 	go.opentelemetry.io/collector/consumer/consumertest v0.132.0
+	go.opentelemetry.io/collector/processor v1.38.0
+	go.opentelemetry.io/collector/processor/processorhelper v0.132.0
 	go.uber.org/goleak v1.3.0
 )
 
@@ -451,7 +453,6 @@ require (
 	go.opentelemetry.io/collector/pdata/xpdata v0.132.0 // indirect
 	go.opentelemetry.io/collector/pipeline v1.38.0 // indirect
 	go.opentelemetry.io/collector/pipeline/xpipeline v0.132.0 // indirect
-	go.opentelemetry.io/collector/processor v1.38.0 // indirect
 	go.opentelemetry.io/collector/processor/processortest v0.132.0 // indirect
 	go.opentelemetry.io/collector/processor/xprocessor v0.132.0 // indirect
 	go.opentelemetry.io/collector/receiver/receivertest v0.132.0 // indirect
@@ -465,7 +466,7 @@ require (
 	go.opentelemetry.io/contrib/otelconf v0.16.0 // indirect
 	go.opentelemetry.io/contrib/propagators/b3 v1.36.0 // indirect
 	go.opentelemetry.io/ebpf-profiler v0.0.0-20250212075250-7bf12d3f962f // indirect
-	go.opentelemetry.io/otel v1.37.0 // indirect
+	go.opentelemetry.io/otel v1.38.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploggrpc v0.13.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploghttp v0.13.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc v1.37.0 // indirect
@@ -477,12 +478,12 @@ require (
 	go.opentelemetry.io/otel/exporters/stdout/stdoutlog v0.13.0 // indirect
 	go.opentelemetry.io/otel/exporters/stdout/stdoutmetric v1.37.0 // indirect
 	go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.37.0 // indirect
-	go.opentelemetry.io/otel/log v0.13.0 // indirect
-	go.opentelemetry.io/otel/metric v1.37.0 // indirect
-	go.opentelemetry.io/otel/sdk v1.37.0 // indirect
+	go.opentelemetry.io/otel/log v0.14.0 // indirect
+	go.opentelemetry.io/otel/metric v1.38.0 // indirect
+	go.opentelemetry.io/otel/sdk v1.38.0 // indirect
 	go.opentelemetry.io/otel/sdk/log v0.13.0 // indirect
-	go.opentelemetry.io/otel/sdk/metric v1.37.0 // indirect
-	go.opentelemetry.io/otel/trace v1.37.0 // indirect
+	go.opentelemetry.io/otel/sdk/metric v1.38.0 // indirect
+	go.opentelemetry.io/otel/trace v1.38.0 // indirect
 	go.opentelemetry.io/proto/otlp v1.7.0 // indirect
 	go.uber.org/ratelimit v0.3.1 // indirect
 	go.yaml.in/yaml/v2 v2.4.2 // indirect

go.mod

@@ -956,8 +956,8 @@ github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/
 github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
 github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
 github.com/stretchr/testify v1.8.2/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
-github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA=
-github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
+github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
+github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
 github.com/teambition/rrule-go v1.8.2 h1:lIjpjvWTj9fFUZCmuoVDrKVOtdiyzbzc93qTmRVe/J8=
 github.com/teambition/rrule-go v1.8.2/go.mod h1:Ieq5AbrKGciP1V//Wq8ktsTXwSwJHDD5mD/wLBGl3p4=
 github.com/tidwall/gjson v1.18.0 h1:FIDeeyB800efLX89e5a8Y0BNH+LOngJyGrIWxG2FKQY=
@@ -1137,6 +1137,8 @@ go.opentelemetry.io/collector/pipeline/xpipeline v0.132.0 h1:ISE9c9TvywcnIGIPfLO
 go.opentelemetry.io/collector/pipeline/xpipeline v0.132.0/go.mod h1:aneg0Kepxwa2RoTSGJx1bg6JKl6dlKTijmqloR0hbC8=
 go.opentelemetry.io/collector/processor v1.38.0 h1:OGZ+2ku4cyzlSehCJb4QdSrBOYeWgM0zPHHlq7qBZqM=
 go.opentelemetry.io/collector/processor v1.38.0/go.mod h1:wFky0NRSLlwvuHQOzP/DUIKUL1A/YKj5rezF9lzTAGM=
+go.opentelemetry.io/collector/processor/processorhelper v0.132.0 h1:PsKrdBj6E0qxEDMUvaWlHEeIhsL+f7IhWuYtGe8eQuQ=
+go.opentelemetry.io/collector/processor/processorhelper v0.132.0/go.mod h1:InJZfNrIuu5d/rEvvDJTcrcFejGiQ+PCubDgar+RjhI=
 go.opentelemetry.io/collector/processor/processortest v0.132.0 h1:p8vk2ICOB2LlpVd7Y8JF0uvtNxJA57XOG4/EDi3zlgA=
 go.opentelemetry.io/collector/processor/processortest v0.132.0/go.mod h1:hYYON5yz+EDdvM0RRCXKCAaoJn149hrUHZCd/zMngMo=
 go.opentelemetry.io/collector/processor/xprocessor v0.132.0 h1:cuEJqX5hZf/N27nPgnl0tm0ECOMHQqhmsoVDmAVfeYg=
@@ -1167,8 +1169,8 @@ go.opentelemetry.io/contrib/zpages v0.62.0 h1:9fUYTLmrK0x/lweM2uM+BOx069jLx8PxVq
 go.opentelemetry.io/contrib/zpages v0.62.0/go.mod h1:C8kXoiC1Ytvereztus2R+kqdSa6W/MZ8FfS8Zwj+LiM=
 go.opentelemetry.io/ebpf-profiler v0.0.0-20250212075250-7bf12d3f962f h1:DqRQ7JaRjf3TwWwfwHIvsBB/aLUs+kgrX+MrAIllALI=
 go.opentelemetry.io/ebpf-profiler v0.0.0-20250212075250-7bf12d3f962f/go.mod h1:hfAVBjRN6FZjSgZUBsNzvRDJWlS46R5Y0SGVr4Jl86s=
-go.opentelemetry.io/otel v1.37.0 h1:9zhNfelUvx0KBfu/gb+ZgeAfAgtWrfHJZcAqFC228wQ=
-go.opentelemetry.io/otel v1.37.0/go.mod h1:ehE/umFRLnuLa/vSccNq9oS1ErUlkkK71gMcN34UG8I=
+go.opentelemetry.io/otel v1.38.0 h1:RkfdswUDRimDg0m2Az18RKOsnI8UDzppJAtj01/Ymk8=
+go.opentelemetry.io/otel v1.38.0/go.mod h1:zcmtmQ1+YmQM9wrNsTGV/q/uyusom3P8RxwExxkZhjM=
 go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploggrpc v0.13.0 h1:z6lNIajgEBVtQZHjfw2hAccPEBDs+nx58VemmXWa2ec=
 go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploggrpc v0.13.0/go.mod h1:+kyc3bRx/Qkq05P6OCu3mTEIOxYRYzoIg+JsUp5X+PM=
 go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploghttp v0.13.0 h1:zUfYw8cscHHLwaY8Xz3fiJu+R59xBnkgq2Zr1lwmK/0=
@@ -1191,22 +1193,22 @@ go.opentelemetry.io/otel/exporters/stdout/stdoutmetric v1.37.0 h1:6VjV6Et+1Hd2iL
 go.opentelemetry.io/otel/exporters/stdout/stdoutmetric v1.37.0/go.mod h1:u8hcp8ji5gaM/RfcOo8z9NMnf1pVLfVY7lBY2VOGuUU=
 go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.37.0 h1:SNhVp/9q4Go/XHBkQ1/d5u9P/U+L1yaGPoi0x+mStaI=
 go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.37.0/go.mod h1:tx8OOlGH6R4kLV67YaYO44GFXloEjGPZuMjEkaaqIp4=
-go.opentelemetry.io/otel/log v0.13.0 h1:yoxRoIZcohB6Xf0lNv9QIyCzQvrtGZklVbdCoyb7dls=
-go.opentelemetry.io/otel/log v0.13.0/go.mod h1:INKfG4k1O9CL25BaM1qLe0zIedOpvlS5Z7XgSbmN83E=
+go.opentelemetry.io/otel/log v0.14.0 h1:2rzJ+pOAZ8qmZ3DDHg73NEKzSZkhkGIua9gXtxNGgrM=
+go.opentelemetry.io/otel/log v0.14.0/go.mod h1:5jRG92fEAgx0SU/vFPxmJvhIuDU9E1SUnEQrMlJpOno=
 go.opentelemetry.io/otel/log/logtest v0.13.0 h1:xxaIcgoEEtnwdgj6D6Uo9K/Dynz9jqIxSDu2YObJ69Q=
 go.opentelemetry.io/otel/log/logtest v0.13.0/go.mod h1:+OrkmsAH38b+ygyag1tLjSFMYiES5UHggzrtY1IIEA8=
-go.opentelemetry.io/otel/metric v1.37.0 h1:mvwbQS5m0tbmqML4NqK+e3aDiO02vsf/WgbsdpcPoZE=
-go.opentelemetry.io/otel/metric v1.37.0/go.mod h1:04wGrZurHYKOc+RKeye86GwKiTb9FKm1WHtO+4EVr2E=
-go.opentelemetry.io/otel/sdk v1.37.0 h1:ItB0QUqnjesGRvNcmAcU0LyvkVyGJ2xftD29bWdDvKI=
-go.opentelemetry.io/otel/sdk v1.37.0/go.mod h1:VredYzxUvuo2q3WRcDnKDjbdvmO0sCzOvVAiY+yUkAg=
+go.opentelemetry.io/otel/metric v1.38.0 h1:Kl6lzIYGAh5M159u9NgiRkmoMKjvbsKtYRwgfrA6WpA=
+go.opentelemetry.io/otel/metric v1.38.0/go.mod h1:kB5n/QoRM8YwmUahxvI3bO34eVtQf2i4utNVLr9gEmI=
+go.opentelemetry.io/otel/sdk v1.38.0 h1:l48sr5YbNf2hpCUj/FoGhW9yDkl+Ma+LrVl8qaM5b+E=
+go.opentelemetry.io/otel/sdk v1.38.0/go.mod h1:ghmNdGlVemJI3+ZB5iDEuk4bWA3GkTpW+DOoZMYBVVg=
 go.opentelemetry.io/otel/sdk/log v0.13.0 h1:I3CGUszjM926OphK8ZdzF+kLqFvfRY/IIoFq/TjwfaQ=
 go.opentelemetry.io/otel/sdk/log v0.13.0/go.mod h1:lOrQyCCXmpZdN7NchXb6DOZZa1N5G1R2tm5GMMTpDBw=
 go.opentelemetry.io/otel/sdk/log/logtest v0.13.0 h1:9yio6AFZ3QD9j9oqshV1Ibm9gPLlHNxurno5BreMtIA=
 go.opentelemetry.io/otel/sdk/log/logtest v0.13.0/go.mod h1:QOGiAJHl+fob8Nu85ifXfuQYmJTFAvcrxL6w5/tu168=
-go.opentelemetry.io/otel/sdk/metric v1.37.0 h1:90lI228XrB9jCMuSdA0673aubgRobVZFhbjxHHspCPc=
-go.opentelemetry.io/otel/sdk/metric v1.37.0/go.mod h1:cNen4ZWfiD37l5NhS+Keb5RXVWZWpRE+9WyVCpbo5ps=
-go.opentelemetry.io/otel/trace v1.37.0 h1:HLdcFNbRQBE2imdSEgm/kwqmQj1Or1l/7bW6mxVK7z4=
-go.opentelemetry.io/otel/trace v1.37.0/go.mod h1:TlgrlQ+PtQO5XFerSPUYG0JSgGyryXewPGyayAWSBS0=
+go.opentelemetry.io/otel/sdk/metric v1.38.0 h1:aSH66iL0aZqo//xXzQLYozmWrXxyFkBJ6qT5wthqPoM=
+go.opentelemetry.io/otel/sdk/metric v1.38.0/go.mod h1:dg9PBnW9XdQ1Hd6ZnRz689CbtrUp0wMMs9iPcgT9EZA=
+go.opentelemetry.io/otel/trace v1.38.0 h1:Fxk5bKrDZJUH+AMyyIXGcFAPah0oRcT+LuNtJrmcNLE=
+go.opentelemetry.io/otel/trace v1.38.0/go.mod h1:j1P9ivuFsTceSWe1oY+EeW3sc+Pp42sO++GHkg4wwhs=
 go.opentelemetry.io/proto/otlp v1.7.0 h1:jX1VolD6nHuFzOYso2E73H85i92Mv8JQYk0K9vz09os=
 go.opentelemetry.io/proto/otlp v1.7.0/go.mod h1:fSKjH6YJ7HDlwzltzyMj036AJ3ejJLCgCSHGj4efDDo=
 go.uber.org/atomic v1.7.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc=
@@ -1420,8 +1422,8 @@ google.golang.org/grpc v1.23.0/go.mod h1:Y5yQAOtifL1yxbo5wqy6BxZv8vAUGQwXBOALyac
 google.golang.org/grpc v1.25.1/go.mod h1:c3i+UQWmh7LiEpx4sFZnkU36qjEYZ0imhYfXVyQciAY=
 google.golang.org/grpc v1.27.0/go.mod h1:qbnxyOmOxrQa7FizSgH+ReBfzJrCY1pSN7KXBS8abTk=
 google.golang.org/grpc v1.33.2/go.mod h1:JMHMWHQWaTccqQQlmk3MJZS+GWXOdAesneDmEnv2fbc=
-google.golang.org/grpc v1.75.0 h1:+TW+dqTd2Biwe6KKfhE5JpiYIBWq865PhKGSXiivqt4=
-google.golang.org/grpc v1.75.0/go.mod h1:JtPAzKiq4v1xcAB2hydNlWI2RnF85XXcV0mhKXr2ecQ=
+google.golang.org/grpc v1.75.1 h1:/ODCNEuf9VghjgO3rqLcfg8fiOP0nSluljWFlDxELLI=
+google.golang.org/grpc v1.75.1/go.mod h1:JtPAzKiq4v1xcAB2hydNlWI2RnF85XXcV0mhKXr2ecQ=
 google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8=
 google.golang.org/protobuf v0.0.0-20200221191635-4d8936d0db64/go.mod h1:kwYJMbMJ01Woi6D6+Kah6886xMZcty6N08ah7+eCXa0=
 google.golang.org/protobuf v0.0.0-20200228230310-ab0ca4ff8a60/go.mod h1:cfTl7dwQJ+fmap5saPgwCLgHXTUD7jkjRqWcaiX5VyM=
@@ -1433,8 +1435,8 @@ google.golang.org/protobuf v1.23.1-0.20200526195155-81db48ad09cc/go.mod h1:EGpAD
 google.golang.org/protobuf v1.25.0/go.mod h1:9JNX74DMeImyA3h4bdi1ymwjUzf21/xIlbajtzgsN7c=
 google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
 google.golang.org/protobuf v1.27.1/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
-google.golang.org/protobuf v1.36.7 h1:IgrO7UwFQGJdRNXH/sQux4R1Dj1WAKcLElzeeRaXV2A=
-google.golang.org/protobuf v1.36.7/go.mod h1:jduwjTPXsFjZGTmRluh+L6NjiWu7pchiJ2/5YcXBHnY=
+google.golang.org/protobuf v1.36.9 h1:w2gp2mA27hUeUzj9Ex9FBjsBm40zfaDtEWow293U7Iw=
+google.golang.org/protobuf v1.36.9/go.mod h1:fuxRtAxBytpl4zzqUh6/eyUujkJdNiuEkXntxiD/uRU=
 gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=

go.sum

@@ -126,7 +126,7 @@ logging.level: debug
 
 	// 1. wait mockbeat to start
 	mockbeat.WaitForLogs(
-		fmt.Sprint("mockbeat start running"),
+		"mockbeat start running",
 		10*time.Second,
 		"did not find 'mockbeat start running' log")
 
@@ -148,7 +148,7 @@ logging.level: debug
 
 	// 5. wait mockbeat to start again
 	mockbeat.WaitForLogs(
-		fmt.Sprint("mockbeat start running"),
+		"mockbeat start running",
 		10*time.Second,
 		"did not find 'mockbeat start running' log again")
 }
@@ -160,19 +160,19 @@ func startMockES(t *testing.T, addr string) (*http.Server, metrics.Registry) {
 
 	s := http.Server{Addr: addr, Handler: es, ReadHeaderTimeout: time.Second}
 	go func() {
-		if err := s.ListenAndServe(); !errors.Is(http.ErrServerClosed, err) {
+		if err := s.ListenAndServe(); !errors.Is(err, http.ErrServerClosed) {
 			t.Errorf("could not start mock-es server: %s", err)
 		}
 	}()
 
 	require.Eventually(t, func() bool {
-		resp, err := http.Get("http://" + addr) // nolint: noctx // It's just a test
+		resp, err := http.Get("http://" + addr) //nolint: noctx // It's just a test
 		if err != nil {
-			// nolint: errcheck // We're just draining the body, we can ignore the error
-			io.Copy(io.Discard, resp.Body)
-			resp.Body.Close()
 			return false
 		}
+		//nolint: errcheck // We're just draining the body, we can ignore the error
+		io.Copy(io.Discard, resp.Body)
+		resp.Body.Close()
 		return true
 	},
 		time.Second, time.Millisecond, "mock-es server did not start on '%s'", addr)

libbeat/tests/integration/elasticsearch_test.go

@@ -0,0 +1,12 @@
+# OpenTelemetry Collector Components in Beats
+
+This is the home of OpenTelemetry Collector components like receivers, processors, exporters, extensions, connectors, etc. that are related to Beats.
+
+The intended structure of this directory is to mimic the structure in the [OpenTelemetry Collector Contrib] repository, specifically:
+
+- Put X receiver in `receiver/xreceiver` subdirectory,
+- Put Y processor in `processor/yprocessor` subdirectory,
+- Put Z exporter in `exporter/zexporter` subdirectory,
+- and so on.
+
+There should be no need to put any Go files directly in this directory.

x-pack/otel/README.md

@@ -0,0 +1,64 @@
+# Beat Processor
+
+| Status    |                     |
+| --------- | ------------------- |
+| Stability | [development]: logs |
+
+[development]: https://github.com/open-telemetry/opentelemetry-collector/blob/main/docs/component-stability.md#development
+
+> [!NOTE]
+> This component is currently in development and no functionality is implemented.
+> Including it in a pipeline is a no-op.
+> The documentation describes the intended state after the functionality is implemented.
+
+The Beat processor (`beat`) is an OpenTelemetry Collector processor that wraps the [Beat processors].
+This allows you to use Beat processorss like e.g. [add_host_metadata] anywhere in the OpenTelemetry Collector's pipeline, independently of Beat receivers.
+
+> [!NOTE]
+> This component is only expected to work correctly with data from the Beat receivers: [Filebeat receiver], [Metricbeat receiver].
+> This is because it relies on the specific structure of telemetry emitted by those components.
+> Using it with data coming from other components is not recommended and may result in unexpected behavior.
+
+## Example
+
+The following [Filebeat receiver] configuration
+
+```yaml
+receivers:
+  filebeatreceiver:
+    filebeat:
+      inputs:
+        - type: filestream
+          id: host-logs
+          paths:
+            - /var/log/*.log
+    processors:
+      - add_host_metadata: ~
+    output:
+      otelconsumer:
+```
+
+is functionally equivalent to this one, using the Beat processor:
+
+```yaml
+receivers:
+  filebeatreceiver:
+    filebeat:
+      inputs:
+        - type: filestream
+          id: host-logs
+          paths:
+            - /var/log/*.log
+    output:
+      otelconsumer:
+
+processors:
+  beat:
+    processors:
+      - add_host_metadata: ~
+```
+
+[Beat processors]: https://www.elastic.co/docs/reference/beats/filebeat/filtering-enhancing-data#using-processors
+[Filebeat receiver]: https://github.com/elastic/beats/tree/main/x-pack/filebeat/fbreceiver
+[Metricbeat receiver]: https://github.com/elastic/beats/tree/main/x-pack/metricbeat/mbreceiver
+[add_host_metadata]: https://www.elastic.co/docs/reference/beats/filebeat/add-host-metadata

x-pack/otel/processor/beatprocessor/README.md

@@ -0,0 +1,7 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package beatprocessor
+
+type Config struct{}

x-pack/otel/processor/beatprocessor/config.go

@@ -0,0 +1,49 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package beatprocessor
+
+import (
+	"context"
+
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/pdata/plog"
+	"go.opentelemetry.io/collector/processor"
+	"go.opentelemetry.io/collector/processor/processorhelper"
+)
+
+const (
+	Name = "beat"
+)
+
+func NewFactory() processor.Factory {
+	return processor.NewFactory(
+		component.MustNewType(Name),
+		createDefaultConfig,
+		processor.WithLogs(createLogsProcessor, component.StabilityLevelDevelopment),
+	)
+}
+
+func createDefaultConfig() component.Config {
+	return &Config{}
+}
+
+func createLogsProcessor(
+	ctx context.Context,
+	set processor.Settings,
+	cfg component.Config,
+	nextConsumer consumer.Logs,
+) (processor.Logs, error) {
+	return processorhelper.NewLogs(
+		ctx,
+		set,
+		cfg,
+		nextConsumer,
+		func(_ context.Context, logs plog.Logs) (plog.Logs, error) {
+			// This is a placeholder for the actual processing logic.
+			return logs, nil
+		},
+	)
+}