oteltest: make CheckReceivers environment more similar to the collector (#43302)

* oteltest: make CheckReceivers environment more similar to the collector

The CheckReceivers function is designed to start receivers in the same
process to assert conditions. This simulated environment should closely
resemble the real OTel collector startup.

This PR fixes some inconsistencies in the test environment.

First, when instantiating receivers, the collector first creates them using
the factory and only then starts each component. Previously, we were
creating and starting each receiver sequentially, which is incorrect and
masked issues with global state when multiple receivers were present.

Second, a Beats receiver logger inherits from the zap.Core of the
collector logger. This core includes certain fields—specifically
data_type, kind, and name. In the tests, these fields were previously
missing, so this PR ensures they are included.

* remove i++

* use EventuallyWithT for assertions

* fix imports

* specify factory per receiver

(cherry picked from commit 8d31036)

Author: mauri870

Diff for: libbeat/otelbeat/oteltest/oteltest.go

@@ -39,19 +39,18 @@ import (
 )
 
 type ReceiverConfig struct {
-	Name   string
-	Config component.Config
+	Name    string
+	Config  component.Config
+	Factory receiver.Factory
 }
 
 type CheckReceiversParams struct {
 	T *testing.T
-	// Factory that allows to create a receiver.
-	Factory receiver.Factory
 	// Receivers is a list of receiver configurations to create.
 	Receivers []ReceiverConfig
 	// AssertFunc is a function that asserts the test conditions.
-	// The function is called periodically until it returns true which ends the test.
-	AssertFunc func(t *testing.T, logs map[string][]mapstr.M, zapLogs *observer.ObservedLogs) bool
+	// The function is called periodically until the assertions are met or the timeout is reached.
+	AssertFunc func(t *assert.CollectT, logs map[string][]mapstr.M, zapLogs *observer.ObservedLogs)
 }
 
 // CheckReceivers creates receivers using the provided configuration.
@@ -69,11 +68,22 @@ func CheckReceivers(params CheckReceiversParams) {
 	)
 	observed, zapLogs := observer.New(zapcore.DebugLevel)
 
-	createReceiver := func(t *testing.T, name string, cfg component.Config) receiver.Logs {
+	core := zapcore.NewTee(zapCore, observed)
+
+	createReceiver := func(t *testing.T, rc ReceiverConfig) receiver.Logs {
 		t.Helper()
 
 		var receiverSettings receiver.Settings
-		receiverSettings.Logger = zap.New(zapcore.NewTee(zapCore, observed)).Named(name)
+
+		// Replicate the behavior of the collector logger
+		receiverCore := core.
+			With([]zapcore.Field{
+				zap.String("name", rc.Name),
+				zap.String("kind", "receiver"),
+				zap.String("data_type", "logs"),
+			})
+
+		receiverSettings.Logger = zap.New(receiverCore)
 
 		logConsumer, err := consumer.NewLogs(func(ctx context.Context, ld plog.Logs) error {
 			for i := 0; i < ld.ResourceLogs().Len(); i++ {
@@ -83,33 +93,46 @@ func CheckReceivers(params CheckReceiversParams) {
 					for k := 0; k < sl.LogRecords().Len(); k++ {
 						log := sl.LogRecords().At(k)
 						logsMu.Lock()
-						logs[name] = append(logs[name], log.Body().Map().AsRaw())
+						logs[rc.Name] = append(logs[rc.Name], log.Body().Map().AsRaw())
 						logsMu.Unlock()
 					}
 				}
 			}
 			return nil
 		})
-		assert.NoErrorf(t, err, "Error creating log consumer for %q", name)
+		assert.NoErrorf(t, err, "Error creating log consumer for %q", rc.Name)
 
-		r, err := params.Factory.CreateLogs(ctx, receiverSettings, cfg, logConsumer)
-		assert.NoErrorf(t, err, "Error creating receiver %q", name)
+		r, err := rc.Factory.CreateLogs(ctx, receiverSettings, rc.Config, logConsumer)
+		assert.NoErrorf(t, err, "Error creating receiver %q", rc.Name)
 		return r
 	}
 
+	// Replicate the collector behavior to instantiate components first and then start them.
+	var receivers []receiver.Logs
 	for _, rec := range params.Receivers {
-		r := createReceiver(t, rec.Name, rec.Config)
+		receivers = append(receivers, createReceiver(t, rec))
+	}
+
+	for i, r := range receivers {
 		err := r.Start(ctx, nil)
-		require.NoErrorf(t, err, "Error starting receiver %q", rec.Name)
+		require.NoErrorf(t, err, "Error starting receiver %d", i)
 		defer func() {
-			require.NoErrorf(t, r.Shutdown(ctx), "Error shutting down receiver %q", rec.Name)
+			require.NoErrorf(t, r.Shutdown(ctx), "Error shutting down receiver %d", i)
 		}()
 	}
 
-	require.Eventually(t, func() bool {
+	require.EventuallyWithT(t, func(ct *assert.CollectT) {
 		logsMu.Lock()
 		defer logsMu.Unlock()
 
-		return params.AssertFunc(t, logs, zapLogs)
+		// Ensure the logger fields from the otel collector are present in the logs.
+		for _, zl := range zapLogs.All() {
+			require.Contains(t, zl.ContextMap(), "name")
+			require.Equal(t, zl.ContextMap()["kind"], "receiver")
+			require.Equal(t, zl.ContextMap()["data_type"], "logs")
+			break
+		}
+
+		params.AssertFunc(ct, logs, zapLogs)
 	}, time.Minute, 100*time.Millisecond, "timeout waiting for assertion to pass")
 }

Diff for: x-pack/filebeat/fbreceiver/receiver_test.go

@@ -12,6 +12,7 @@ import (
 	"github.com/elastic/beats/v7/libbeat/otelbeat/oteltest"
 	"github.com/elastic/elastic-agent-libs/mapstr"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/receiver"
 	"go.uber.org/zap"
@@ -46,17 +47,17 @@ func TestNewReceiver(t *testing.T) {
 	}
 
 	oteltest.CheckReceivers(oteltest.CheckReceiversParams{
-		T:       t,
-		Factory: NewFactory(),
+		T: t,
 		Receivers: []oteltest.ReceiverConfig{
 			{
-				Name:   "r1",
-				Config: &config,
+				Name:    "r1",
+				Config:  &config,
+				Factory: NewFactory(),
 			},
 		},
-		AssertFunc: func(t *testing.T, logs map[string][]mapstr.M, zapLogs *observer.ObservedLogs) bool {
+		AssertFunc: func(t *assert.CollectT, logs map[string][]mapstr.M, zapLogs *observer.ObservedLogs) {
 			_ = zapLogs
-			return len(logs["r1"]) == 1
+			assert.Len(t, logs["r1"], 1)
 		},
 	})
 }
@@ -88,18 +89,16 @@ func TestReceiverDefaultProcessors(t *testing.T) {
 	}
 
 	oteltest.CheckReceivers(oteltest.CheckReceiversParams{
-		T:       t,
-		Factory: NewFactory(),
+		T: t,
 		Receivers: []oteltest.ReceiverConfig{
 			{
-				Name:   "r1",
-				Config: &config,
+				Name:    "r1",
+				Config:  &config,
+				Factory: NewFactory(),
 			},
 		},
-		AssertFunc: func(t *testing.T, logs map[string][]mapstr.M, zapLogs *observer.ObservedLogs) bool {
-			if len(logs["r1"]) == 0 {
-				return false
-			}
+		AssertFunc: func(t *assert.CollectT, logs map[string][]mapstr.M, zapLogs *observer.ObservedLogs) {
+			require.Len(t, logs["r1"], 1)
 
 			processorsLoaded := zapLogs.FilterMessageSnippet("Generated new processors").
 				FilterMessageSnippet("add_host_metadata").
@@ -110,8 +109,6 @@ func TestReceiverDefaultProcessors(t *testing.T) {
 			require.True(t, processorsLoaded, "processors not loaded")
 			// Check that add_host_metadata works, other processors are not guaranteed to add fields in all environments
 			require.Contains(t, logs["r1"][0].Flatten(), "host.architecture")
-
-			return true
 		},
 	})
 }
@@ -187,22 +184,25 @@ func TestMultipleReceivers(t *testing.T) {
 		},
 	}
 
+	factory := NewFactory()
 	oteltest.CheckReceivers(oteltest.CheckReceiversParams{
-		T:       t,
-		Factory: NewFactory(),
+		T: t,
 		Receivers: []oteltest.ReceiverConfig{
 			{
-				Name:   "r1",
-				Config: &config,
+				Name:    "r1",
+				Config:  &config,
+				Factory: factory,
 			},
 			{
-				Name:   "r2",
-				Config: &config,
+				Name:    "r2",
+				Config:  &config,
+				Factory: factory,
 			},
 		},
-		AssertFunc: func(t *testing.T, logs map[string][]mapstr.M, zapLogs *observer.ObservedLogs) bool {
+		AssertFunc: func(t *assert.CollectT, logs map[string][]mapstr.M, zapLogs *observer.ObservedLogs) {
 			_ = zapLogs
-			return len(logs["r1"]) == 1 && len(logs["r2"]) == 1
+			require.Len(t, logs["r1"], 1)
+			require.Len(t, logs["r2"], 1)
 		},
 	})
 }

Diff for: x-pack/metricbeat/mbreceiver/receiver_test.go

@@ -10,6 +10,8 @@ import (
 	"github.com/elastic/beats/v7/libbeat/otelbeat/oteltest"
 	"github.com/elastic/elastic-agent-libs/mapstr"
 
+	"github.com/stretchr/testify/assert"
+
 	"go.uber.org/zap/zaptest/observer"
 )
 
@@ -41,17 +43,19 @@ func TestNewReceiver(t *testing.T) {
 	}
 
 	oteltest.CheckReceivers(oteltest.CheckReceiversParams{
-		T:       t,
-		Factory: NewFactory(),
+		T: t,
 		Receivers: []oteltest.ReceiverConfig{
 			{
-				Name:   "r1",
-				Config: &config,
+				Name:    "r1",
+				Config:  &config,
+				Factory: NewFactory(),
 			},
 		},
-		AssertFunc: func(t *testing.T, logs map[string][]mapstr.M, zapLogs *observer.ObservedLogs) bool {
+		AssertFunc: func(t *assert.CollectT, logs map[string][]mapstr.M, zapLogs *observer.ObservedLogs) {
 			_ = zapLogs
-			return len(logs["r1"]) > 0
+			assert.Conditionf(t, func() bool {
+				return len(logs["r1"]) > 0
+			}, "expected at least one ingest log, got logs: %v", logs["r1"])
 		},
 	})
 }
@@ -83,22 +87,26 @@ func TestMultipleReceivers(t *testing.T) {
 		},
 	}
 
+	factory := NewFactory()
 	oteltest.CheckReceivers(oteltest.CheckReceiversParams{
-		T:       t,
-		Factory: NewFactory(),
+		T: t,
 		Receivers: []oteltest.ReceiverConfig{
 			{
-				Name:   "r1",
-				Config: &config,
+				Name:    "r1",
+				Config:  &config,
+				Factory: factory,
 			},
 			{
-				Name:   "r2",
-				Config: &config,
+				Name:    "r2",
+				Config:  &config,
+				Factory: factory,
 			},
 		},
-		AssertFunc: func(t *testing.T, logs map[string][]mapstr.M, zapLogs *observer.ObservedLogs) bool {
+		AssertFunc: func(t *assert.CollectT, logs map[string][]mapstr.M, zapLogs *observer.ObservedLogs) {
 			_ = zapLogs
-			return len(logs["r1"]) > 0 && len(logs["r2"]) > 0
+			assert.Conditionf(t, func() bool {
+				return len(logs["r1"]) > 0 && len(logs["r2"]) > 0
+			}, "expected at least one ingest log for each receiver, got logs: %v", logs)
 		},
 	})
 }