fix: re-introduce beats receivers log the same metadata as beats processes (#8717)

Author: pkoutsovasilis

internal/pkg/otel/manager/manager.go

@@ -141,7 +141,7 @@ func (m *OTelManager) Run(ctx context.Context) error {
 
 			newRetries := m.recoveryRetries.Add(1)
 			m.logger.Infof("collector recovery restarting, total retries: %d", newRetries)
-			proc, err = m.execution.startCollector(ctx, m.logger, m.cfg, collectorRunErr, m.statusCh)
+			proc, err = m.execution.startCollector(ctx, m.baseLogger, m.cfg, collectorRunErr, m.statusCh)
 			if err != nil {
 				reportErr(ctx, m.errCh, err)
 				// reset the restart timer to the next backoff
@@ -174,7 +174,7 @@ func (m *OTelManager) Run(ctx context.Context) error {
 
 				// in this rare case the collector stopped running but a configuration was
 				// provided and the collector stopped with a clean exit
-				proc, err = m.execution.startCollector(ctx, m.logger, m.cfg, collectorRunErr, m.statusCh)
+				proc, err = m.execution.startCollector(ctx, m.baseLogger, m.cfg, collectorRunErr, m.statusCh)
 				if err != nil {
 					// failed to create the collector (this is different then
 					// it's failing to run). we do not retry creation on failure
@@ -239,7 +239,7 @@ func (m *OTelManager) Run(ctx context.Context) error {
 			} else {
 				// either a new configuration or the first configuration
 				// that results in the collector being started
-				proc, err = m.execution.startCollector(ctx, m.logger, m.cfg, collectorRunErr, m.statusCh)
+				proc, err = m.execution.startCollector(ctx, m.baseLogger, m.cfg, collectorRunErr, m.statusCh)
 				if err != nil {
 					// failed to create the collector (this is different then
 					// it's failing to run). we do not retry creation on failure

internal/pkg/otel/manager/manager_test.go

@@ -516,6 +516,42 @@ func TestOTelManager_Run(t *testing.T) {
 	}
 }
 
+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, err := NewOTelManager(l, logp.DebugLevel, base, EmbeddedExecutionMode)
+	require.NoError(t, err, "could not create otel manager")
+
+	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)
+}
+
 // statusToYaml converts the status.AggregateStatus to a YAML string representation.
 func statusToYaml(s *status.AggregateStatus) string {
 	printable := toSerializableStatus(s)