Ensure beats receivers log the same metadata as beats processes (#8606) (#8711)

* Use the base logger for otel collector

* Add E2E test comparing logs between beats processes and beats receivers

* Clean up tests

* Fix incorrect comment

* Fix conflicts after rebase

* Fix log parsing in integration test

* Use different contexts for controlling agents

* Fix incorrect test ordering

* Fix incorrect test context handling

* Fix bug in testcontext.WithDeadline

(cherry picked from commit aa44f8b)

Co-authored-by: Mikołaj Świątek <mail@mikolajswiatek.com>

Author: mergify[bot]

internal/pkg/agent/application/application.go

@@ -224,7 +224,7 @@ func New(
 		return nil, nil, nil, errors.New(err, "failed to initialize composable controller")
 	}
 
-	otelManager := otelmanager.NewOTelManager(log.Named("otel_manager"))
+	otelManager := otelmanager.NewOTelManager(log.Named("otel_manager"), baseLogger)
 	coord := coordinator.New(log, cfg, logLevel, agentInfo, specs, reexec, upgrader, runtime, configMgr, varsManager, caps, monitor, isManaged, otelManager, actionAcker, compModifiers...)
 	if managed != nil {
 		// the coordinator requires the config manager as well as in managed-mode the config manager requires the

internal/pkg/agent/application/coordinator/coordinator_test.go

@@ -1078,7 +1078,7 @@ func createCoordinator(t testing.TB, ctx context.Context, opts ...CoordinatorOpt
 	cfg.Port = 0
 	rm, err := runtime.NewManager(l, l, ai, apmtest.DiscardTracer, monitoringMgr, cfg)
 	require.NoError(t, err)
-	otelMgr := otelmanager.NewOTelManager(l)
+	otelMgr := otelmanager.NewOTelManager(l, l)
 
 	caps, err := capabilities.LoadFile(paths.AgentCapabilitiesPath(), l)
 	require.NoError(t, err)

internal/pkg/otel/manager/manager.go

@@ -22,8 +22,10 @@ import (
 
 // OTelManager is a manager that manages the lifecycle of the OTel collector inside of the Elastic Agent.
 type OTelManager struct {
-	logger *logger.Logger
-	errCh  chan error
+	// baseLogger is the base logger for the otel collector, and doesn't include any agent-specific fields.
+	baseLogger *logger.Logger
+	logger     *logger.Logger
+	errCh      chan error
 
 	// The current configuration that the OTel collector is using. In the case that
 	// the cfg is nil then the collector is not running.
@@ -42,13 +44,14 @@ type OTelManager struct {
 }
 
 // NewOTelManager returns a OTelManager.
-func NewOTelManager(logger *logger.Logger) *OTelManager {
+func NewOTelManager(logger, baseLogger *logger.Logger) *OTelManager {
 	return &OTelManager{
-		logger:   logger,
-		errCh:    make(chan error, 1), // holds at most one error
-		cfgCh:    make(chan *confmap.Conf),
-		statusCh: make(chan *status.AggregateStatus),
-		doneChan: make(chan struct{}),
+		logger:     logger,
+		baseLogger: baseLogger,
+		errCh:      make(chan error, 1), // holds at most one error
+		cfgCh:      make(chan *confmap.Conf),
+		statusCh:   make(chan *status.AggregateStatus),
+		doneChan:   make(chan struct{}),
 	}
 }
 
@@ -196,7 +199,7 @@ func (m *OTelManager) startCollector(cfg *confmap.Conf, errCh chan error) (conte
 		otel.WithExtensionFactory(NewAgentStatusFactory(m)))
 	settings.DisableGracefulShutdown = true // managed by this manager
 	settings.LoggingOptions = []zap.Option{zap.WrapCore(func(zapcore.Core) zapcore.Core {
-		return m.logger.Core() // use same zap as agent
+		return m.baseLogger.Core() // use the base logger also used for logs from the command runtime
 	})}
 	svc, err := otelcol.NewCollector(*settings)
 	if err != nil {

internal/pkg/otel/manager/manager_test.go

@@ -67,8 +67,9 @@ var (
 func TestOTelManager_Run(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-	l, _ := loggertest.New("otel")
-	m := NewOTelManager(l)
+	base, _ := loggertest.New("otel")
+	l, _ := loggertest.New("otel-manager")
+	m := NewOTelManager(l, base)
 
 	var errMx sync.Mutex
 	var err error
@@ -188,8 +189,9 @@ func TestOTelManager_Run(t *testing.T) {
 func TestOTelManager_ConfigError(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-	l, _ := loggertest.New("otel")
-	m := NewOTelManager(l)
+	base, _ := loggertest.New("otel")
+	l, _ := loggertest.New("otel-manager")
+	m := NewOTelManager(l, base)
 
 	go func() {
 		err := m.Run(ctx)
@@ -240,6 +242,41 @@ outer:
 	assert.Error(t, err, "otel manager should have returned an error")
 }
 
+func TestOTelManager_Logging(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	base, obs := loggertest.New("otel")
+	l, _ := loggertest.New("otel-manager")
+	m := NewOTelManager(l, base)
+
+	go func() {
+		err := m.Run(ctx)
+		assert.ErrorIs(t, err, context.Canceled, "otel manager should be cancelled")
+	}()
+
+	// watch is synchronous, so we need to read from it to avoid blocking the manager
+	go func() {
+		for {
+			select {
+			case <-m.Watch():
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+
+	cfg := confmap.NewFromStringMap(testConfig)
+	m.Update(cfg)
+
+	// the collector should log to the base logger
+	assert.EventuallyWithT(t, func(collect *assert.CollectT) {
+		logs := obs.All()
+		require.NotEmpty(collect, logs, "Logs should not be empty")
+		firstMessage := logs[0].Message
+		assert.Equal(collect, firstMessage, "Setting up own telemetry...")
+	}, time.Second*10, time.Second)
+}
+
 func statusToYaml(s *status.AggregateStatus) string {
 	printable := toSerializableStatus(s)
 	yamlBytes, _ := yaml.Marshal(printable)

pkg/testing/tools/testcontext/testcontext.go

@@ -17,7 +17,9 @@ func WithDeadline(
 	parent context.Context,
 	deadline time.Time) (context.Context, context.CancelFunc) {
 	if d, ok := t.Deadline(); ok {
-		deadline = d
+		if d.Before(deadline) {
+			deadline = d
+		}
 	}
 	ctx, cancel := context.WithDeadline(parent, deadline)
 	return ctx, cancel