[8.19](backport #47693) [beatreceiver] Do not require specifiying otelconsumer output (#47716)

* [beatreceiver] Do not require specifiying otelconsumer output (#47693)

* do not require explicit config

(cherry picked from commit b80d1e9)

# Conflicts:
#	x-pack/filebeat/fbreceiver/receiver_test.go
#	x-pack/filebeat/tests/integration/otel_lsexporter_test.go
#	x-pack/libbeat/cmd/instance/beat.go

* fix conflicts

* fix conflict

* remove new file

* make update

---------

Co-authored-by: Khushi Jain <[email protected]>

Author: mergify[bot]

x-pack/filebeat/fbreceiver/receiver_leak_test.go

@@ -50,9 +50,6 @@ func TestLeak(t *testing.T) {
 					},
 				},
 			},
-			"output": map[string]any{
-				"otelconsumer": map[string]any{},
-			},
 			"logging": map[string]any{
 				"level": "debug",
 				"selectors": []string{

x-pack/filebeat/fbreceiver/receiver_test.go

@@ -60,9 +60,6 @@ func TestNewReceiver(t *testing.T) {
 					},
 				},
 			},
-			"output": map[string]any{
-				"otelconsumer": map[string]any{},
-			},
 			"logging": map[string]any{
 				"level": "debug",
 				"selectors": []string{
@@ -135,9 +132,6 @@ func benchmarkFactoryWithLogLevel(b *testing.B, level zapcore.Level) {
 					},
 				},
 			},
-			"output": map[string]any{
-				"otelconsumer": map[string]any{},
-			},
 			"logging": map[string]any{
 				"level": level.String(),
 				"selectors": []string{
@@ -199,9 +193,6 @@ func TestMultipleReceivers(t *testing.T) {
 						},
 					},
 				},
-				"output": map[string]any{
-					"otelconsumer": map[string]any{},
-				},
 				"logging": map[string]any{
 					"level": "info",
 					"selectors": []string{
@@ -314,9 +305,6 @@ func TestReceiverDegraded(t *testing.T) {
 							},
 						},
 					},
-					"output": map[string]any{
-						"otelconsumer": map[string]any{},
-					},
 					"logging": map[string]any{
 						"level": "debug",
 						"selectors": []string{
@@ -442,9 +430,6 @@ func TestReceiverHook(t *testing.T) {
 					},
 				},
 			},
-			"output": map[string]any{
-				"otelconsumer": map[string]any{},
-			},
 			"management.otel.enabled": true,
 			"path.home":               t.TempDir(),
 		},

x-pack/filebeat/tests/integration/otel_test.go

@@ -65,8 +65,6 @@ func TestFilebeatOTelE2E(t *testing.T) {
             - %s
           prospector.scanner.fingerprint.enabled: false
           file_identity.native: ~
-    output:
-      otelconsumer:
     processors:
       - add_host_metadata: ~
       - add_cloud_metadata: ~
@@ -250,8 +248,6 @@ processors:
         - type: httpjson
           id: httpjson-e2e-otel
           request.url: http://localhost:8090/test
-    output:
-      otelconsumer:
     processors:
       - add_host_metadata: ~
       - add_cloud_metadata: ~
@@ -450,8 +446,6 @@ func TestFilebeatOTelReceiverE2E(t *testing.T) {
             - {{.InputFile}}
           prospector.scanner.fingerprint.enabled: false
           file_identity.native: ~
-    output:
-      otelconsumer:
     logging:
       level: info
       selectors:
@@ -641,8 +635,6 @@ func TestFilebeatOTelMultipleReceiversE2E(t *testing.T) {
             - {{$receiver.InputFile}}
           prospector.scanner.fingerprint.enabled: false
           file_identity.native: ~
-    output:
-      otelconsumer:
     logging:
       level: info
       selectors:
@@ -880,8 +872,6 @@ func TestFilebeatOTelDocumentLevelRetries(t *testing.T) {
             - {{.InputFile}}
           prospector.scanner.fingerprint.enabled: false
           file_identity.native: ~
-    output:
-      otelconsumer:
     logging:
       level: debug
     queue.mem.flush.timeout: 0s
@@ -1063,8 +1053,6 @@ func TestFileBeatKerberos(t *testing.T) {
             - {{.InputFile}}
           prospector.scanner.fingerprint.enabled: false
           file_identity.native: ~
-    output:
-      otelconsumer:
     queue.mem.flush.timeout: 0s
     management.otel.enabled: true
     path.home: {{.PathHome}}	

x-pack/libbeat/cmd/instance/beat.go

@@ -65,6 +65,24 @@ func NewBeatForReceiver(settings instance.Settings, receiverConfig map[string]an
 		ucfg.VarExp,
 	}
 
+	err = setLogger(b, receiverConfig, core)
+	if err != nil {
+		return nil, fmt.Errorf("error configuring beats logger: %w", err)
+	}
+
+	// extracting it here for ease of use
+	logger := b.Info.Logger
+
+	// if output is set and if output is not otelconsumer, inform users
+	if receiverConfig["output"] != nil && receiverConfig["output"].(map[string]any)["otelconsumer"] == nil { //nolint: errcheck // output will always be of map type
+		logger.Debugf("configured output does not work with beatreceiver, please use appropriate exporter instead")
+	}
+
+	// all beatreceivers will use otelconsumer output by default
+	receiverConfig["output"] = map[string]any{
+		"otelconsumer": map[string]any{},
+	}
+
 	tmp, err := ucfg.NewFrom(receiverConfig, cfOpts...)
 	if err != nil {
 		return nil, fmt.Errorf("error converting receiver config to ucfg: %w", err)
@@ -113,37 +131,12 @@ func NewBeatForReceiver(settings instance.Settings, receiverConfig map[string]an
 		return nil, fmt.Errorf("error unpacking config data: %w", err)
 	}
 
-	logpConfig := logp.Config{}
-	logpConfig.AddCaller = true
-	logpConfig.Beat = b.Info.Beat
-	logpConfig.Files.MaxSize = 1
-
-	if b.Config.Logging == nil {
-		b.Config.Logging = config.NewConfig()
-	}
-
-	if err := b.Config.Logging.Unpack(&logpConfig); err != nil {
-		return nil, fmt.Errorf("error unpacking beats logging config: %w\n%v", err, b.Config.Logging)
-	}
-
-	b.Info.Logger, err = logp.ConfigureWithCoreLocal(logpConfig, core)
-	if err != nil {
-		return nil, fmt.Errorf("error configuring beats logp: %w", err)
-	}
-
-	// extracting it here for ease of use
-	logger := b.Info.Logger
-
 	instrumentation, err := instrumentation.New(cfg, b.Info.Beat, b.Info.Version, logger)
 	if err != nil {
 		return nil, fmt.Errorf("error setting up instrumentation: %w", err)
 	}
 	b.Instrumentation = instrumentation
 
-	if err := instance.PromoteOutputQueueSettings(b); err != nil {
-		return nil, fmt.Errorf("could not promote output queue settings: %w", err)
-	}
-
 	if err := features.UpdateFromConfig(b.RawConfig); err != nil {
 		return nil, fmt.Errorf("could not parse features: %w", err)
 	}
@@ -246,17 +239,6 @@ func NewBeatForReceiver(settings instance.Settings, receiverConfig map[string]an
 	}
 	b.SetProcessors(processors)
 
-	// This should be replaced with static config for otel consumer
-	// but need to figure out if we want the Queue settings from here.
-	outputEnabled := b.Config.Output.IsSet() && b.Config.Output.Config().Enabled()
-	if !outputEnabled {
-		if b.Manager.Enabled() {
-			logger.Info("Output is configured through Central Management")
-		} else {
-			return nil, fmt.Errorf("no outputs are defined, please define one under the output section")
-		}
-	}
-
 	reg := b.Monitoring.StatsRegistry().GetOrCreateRegistry("libbeat")
 
 	monitors := pipeline.Monitors{
@@ -283,3 +265,31 @@ func NewBeatForReceiver(settings instance.Settings, receiverConfig map[string]an
 
 	return b, nil
 }
+
+// setLogger configures a logp logger and sets it on b.Info.Logger
+func setLogger(b *instance.Beat, receiverConfig map[string]any, core zapcore.Core) error {
+
+	var err error
+	logpConfig := logp.Config{}
+	logpConfig.AddCaller = true
+	logpConfig.Beat = b.Info.Beat
+	logpConfig.Files.MaxSize = 1
+
+	var logCfg *config.C
+	if _, ok := receiverConfig["logging"]; !ok {
+		logCfg = config.NewConfig()
+	} else {
+		logCfg = config.MustNewConfigFrom(receiverConfig["logging"])
+	}
+
+	if err := logCfg.Unpack(&logpConfig); err != nil {
+		return fmt.Errorf("error unpacking beats logging config: %w\n%v", err, b.Config.Logging)
+	}
+
+	b.Info.Logger, err = logp.ConfigureWithCoreLocal(logpConfig, core)
+	if err != nil {
+		return fmt.Errorf("error configuring beats logp: %w", err)
+	}
+
+	return nil
+}

x-pack/libbeat/cmd/instance/beat_test.go

@@ -30,9 +30,6 @@ func TestManager(t *testing.T) {
 				},
 			},
 		},
-		"output": map[string]any{
-			"otelconsumer": map[string]any{},
-		},
 		"path.home": tmpDir,
 	}
 	t.Run("otel management disabled - key missing", func(t *testing.T) {

x-pack/libbeat/common/otelbeat/oteltestcol/collector_test.go

@@ -21,8 +21,6 @@ func TestNew(t *testing.T) {
           message: "test message"
           count: 1
     processors: ~
-    output:
-      otelconsumer:
     logging:
       level: debug
     queue.mem.flush.timeout: 0s

x-pack/metricbeat/mbreceiver/receiver_leak_test.go

@@ -45,9 +45,6 @@ func TestLeak(t *testing.T) {
 					},
 				},
 			},
-			"output": map[string]any{
-				"otelconsumer": map[string]any{},
-			},
 			"logging": map[string]any{
 				"level": "debug",
 				"selectors": []string{

x-pack/metricbeat/mbreceiver/receiver_test.go

@@ -56,9 +56,6 @@ func TestNewReceiver(t *testing.T) {
 					},
 				},
 			},
-			"output": map[string]any{
-				"otelconsumer": map[string]any{},
-			},
 			"logging": map[string]any{
 				"level": "debug",
 				"selectors": []string{
@@ -139,9 +136,6 @@ func TestMultipleReceivers(t *testing.T) {
 					},
 				},
 			},
-			"output": map[string]any{
-				"otelconsumer": map[string]any{},
-			},
 			"logging": map[string]any{
 				"level": "debug",
 				"selectors": []string{
@@ -167,9 +161,6 @@ func TestMultipleReceivers(t *testing.T) {
 					},
 				},
 			},
-			"output": map[string]any{
-				"otelconsumer": map[string]any{},
-			},
 			"logging": map[string]any{
 				"level": "debug",
 				"selectors": []string{
@@ -336,9 +327,6 @@ func BenchmarkFactory(b *testing.B) {
 					},
 				},
 			},
-			"output": map[string]any{
-				"otelconsumer": map[string]any{},
-			},
 			"logging": map[string]any{
 				"level": "info",
 				"selectors": []string{
@@ -407,9 +395,6 @@ func TestReceiverDegraded(t *testing.T) {
 							},
 						},
 					},
-					"output": map[string]any{
-						"otelconsumer": map[string]any{},
-					},
 					"path.home": t.TempDir(),
 				},
 			}
@@ -444,10 +429,7 @@ func TestReceiverHook(t *testing.T) {
 				},
 			},
 			"management.otel.enabled": true,
-			"output": map[string]any{
-				"otelconsumer": map[string]any{},
-			},
-			"path.home": t.TempDir(),
+			"path.home":               t.TempDir(),
 		},
 	}
 	receiverSettings := receiver.Settings{

x-pack/metricbeat/tests/integration/otel_test.go

@@ -69,8 +69,6 @@ func TestMetricbeatOTelE2E(t *testing.T) {
           - '.*'
          metricsets:
           - cpu
-    output:
-      otelconsumer:
     processors:
       - add_host_metadata: ~
       - add_cloud_metadata: ~
@@ -266,8 +264,6 @@ func TestMetricbeatOTelReceiverE2E(t *testing.T) {
           - '.*'
          metricsets:
           - cpu
-    output:
-      otelconsumer:
     processors:
       - add_host_metadata: ~
       - add_cloud_metadata: ~
@@ -463,8 +459,6 @@ func TestMetricbeatOTelMultipleReceiversE2E(t *testing.T) {
           target: ''
           fields:
             receiverid: "{{$i}}"
-    output:
-      otelconsumer:
     logging:
       level: info
       selectors:

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

@@ -41,8 +41,6 @@ receivers:
           id: host-logs
           paths:
             - /var/log/*.log
-    output:
-      otelconsumer:
 ```
 
 The above Filebeat receiver configuration does not explicitly specify the `processors` option.
@@ -58,8 +56,6 @@ receivers:
           paths:
             - /var/log/*.log
     processors: []
-    output:
-      otelconsumer:
 ```
 
 The above Filebeat receiver configuration specifies an empty list of processors.
@@ -82,8 +78,6 @@ receivers:
       - add_host_metadata:
           netinfo:
             enabled: false
-    output:
-      otelconsumer:
 ```
 
 is functionally equivalent to this one, using the Beat processor:
@@ -98,8 +92,6 @@ receivers:
           paths:
             - /var/log/*.log
     processors: []
-    output:
-      otelconsumer:
 
 processors:
   beat: