[beatreceiver] - add manager for otel mode (#46539)

* iniital commit

* cleanupt

* license

* comments

* comments

* comments

* rename

* comments and unit test case

* go.mod

* lint and notice

* use existing flag

* go.sum

* log input test and UnderAgent test

* test disabled

* stub and cleanup

* comment

* fixes and comments

Author: VihasMakwana

NOTICE.txt

@@ -238,6 +238,7 @@ require (
 	github.com/cilium/ebpf v0.19.0
 	github.com/elastic/opentelemetry-collector-components/extension/beatsauthextension v0.2.0
 	go.opentelemetry.io/collector/client v1.41.0
+	go.opentelemetry.io/collector/consumer/consumertest v0.135.0
 	go.opentelemetry.io/collector/exporter v0.135.0
 	go.opentelemetry.io/collector/exporter/exporterhelper v0.135.0
 	go.opentelemetry.io/collector/exporter/exportertest v0.135.0
@@ -441,7 +442,6 @@ require (
 	go.opentelemetry.io/collector/connector/connectortest v0.135.0 // indirect
 	go.opentelemetry.io/collector/connector/xconnector v0.135.0 // indirect
 	go.opentelemetry.io/collector/consumer/consumererror/xconsumererror v0.135.0 // indirect
-	go.opentelemetry.io/collector/consumer/consumertest v0.135.0 // indirect
 	go.opentelemetry.io/collector/consumer/xconsumer v0.135.0 // indirect
 	go.opentelemetry.io/collector/exporter/exporterhelper/xexporterhelper v0.135.0 // indirect
 	go.opentelemetry.io/collector/exporter/xexporter v0.135.0 // indirect

go.mod

@@ -101,7 +101,7 @@ func NewManager(cfg *config.C, registry *reload.Registry, logger *logp.Logger) (
 			return managerFactory(cfg, registry, logger)
 		}
 	}
-	return &fallbackManager{
+	return &FallbackManager{
 		logger: logger.Named("mgmt"),
 		status: status.Unknown,
 		msg:    "",
@@ -118,8 +118,8 @@ func SetManagerFactory(factory ManagerFactory) {
 	managerFactory = factory
 }
 
-// fallbackManager, fallback when no manager is present
-type fallbackManager struct {
+// FallbackManager, fallback when no manager is present
+type FallbackManager struct {
 	logger   *logp.Logger
 	lock     sync.Mutex
 	status   status.Status
@@ -128,7 +128,7 @@ type fallbackManager struct {
 	stopOnce sync.Once
 }
 
-func (n *fallbackManager) UpdateStatus(status status.Status, msg string) {
+func (n *FallbackManager) UpdateStatus(status status.Status, msg string) {
 	n.lock.Lock()
 	defer n.lock.Unlock()
 	if n.status != status || n.msg != msg {
@@ -138,13 +138,13 @@ func (n *fallbackManager) UpdateStatus(status status.Status, msg string) {
 	}
 }
 
-func (n *fallbackManager) SetStopCallback(f func()) {
+func (n *FallbackManager) SetStopCallback(f func()) {
 	n.lock.Lock()
 	n.stopFunc = f
 	n.lock.Unlock()
 }
 
-func (n *fallbackManager) Stop() {
+func (n *FallbackManager) Stop() {
 	n.lock.Lock()
 	defer n.lock.Unlock()
 	if n.stopFunc != nil {
@@ -162,12 +162,12 @@ func (n *fallbackManager) Stop() {
 // the nilManager is still used for shutdown on some cases,
 // but that does not mean the Beat is being managed externally,
 // hence it will always return false.
-func (n *fallbackManager) Enabled() bool                         { return false }
-func (n *fallbackManager) AgentInfo() client.AgentInfo           { return client.AgentInfo{} }
-func (n *fallbackManager) Start() error                          { return nil }
-func (n *fallbackManager) CheckRawConfig(cfg *config.C) error    { return nil }
-func (n *fallbackManager) RegisterAction(action client.Action)   {}
-func (n *fallbackManager) UnregisterAction(action client.Action) {}
-func (n *fallbackManager) SetPayload(map[string]interface{})     {}
-func (n *fallbackManager) RegisterDiagnosticHook(_ string, _ string, _ string, _ string, _ client.DiagnosticHook) {
+func (n *FallbackManager) Enabled() bool                         { return false }
+func (n *FallbackManager) AgentInfo() client.AgentInfo           { return client.AgentInfo{} }
+func (n *FallbackManager) Start() error                          { return nil }
+func (n *FallbackManager) CheckRawConfig(cfg *config.C) error    { return nil }
+func (n *FallbackManager) RegisterAction(action client.Action)   {}
+func (n *FallbackManager) UnregisterAction(action client.Action) {}
+func (n *FallbackManager) SetPayload(map[string]interface{})     {}
+func (n *FallbackManager) RegisterDiagnosticHook(_ string, _ string, _ string, _ string, _ client.DiagnosticHook) {
 }

libbeat/management/management.go

@@ -24,6 +24,7 @@ import (
 	"github.com/elastic/beats/v7/libbeat/publisher/pipeline"
 	"github.com/elastic/beats/v7/libbeat/publisher/processing"
 	"github.com/elastic/beats/v7/libbeat/version"
+	"github.com/elastic/beats/v7/x-pack/libbeat/common/otelbeat/otelmanager"
 	"github.com/elastic/elastic-agent-libs/config"
 	"github.com/elastic/elastic-agent-libs/keystore"
 	"github.com/elastic/elastic-agent-libs/logp"
@@ -187,8 +188,12 @@ func NewBeatForReceiver(settings instance.Settings, receiverConfig map[string]an
 		b.Info.FQDN = fqdn
 	}
 
+	// register NewOtelManager
+	management.SetManagerFactory(otelmanager.NewOtelManager)
+
 	// initialize config manager
-	m, err := management.NewManager(b.Config.Management, b.Registry, logger)
+	oCfg, _ := cfg.Child("management.otel", -1)
+	m, err := management.NewManager(oCfg, b.Registry, logger)
 	if err != nil {
 		return nil, fmt.Errorf("error creating new manager: %w", err)
 	}

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

@@ -0,0 +1,87 @@
+// 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 instance
+
+import (
+	"maps"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/consumer/consumertest"
+	"go.uber.org/zap/zapcore"
+
+	"github.com/elastic/beats/v7/filebeat/cmd"
+	"github.com/elastic/beats/v7/filebeat/input/log"
+	"github.com/elastic/beats/v7/libbeat/management"
+	"github.com/elastic/beats/v7/x-pack/libbeat/common/otelbeat/otelmanager"
+	conf "github.com/elastic/elastic-agent-libs/config"
+)
+
+func TestManager(t *testing.T) {
+	tmpDir := t.TempDir()
+	cfg := map[string]any{
+		"filebeat": map[string]any{
+			"inputs": []map[string]any{
+				{
+					"type":    "benchmark",
+					"enabled": true,
+					"message": "test",
+					"count":   10,
+				},
+			},
+		},
+		"output": map[string]any{
+			"otelconsumer": map[string]any{},
+		},
+		"path.home": tmpDir,
+	}
+	t.Run("otel management disabled - key missing", func(t *testing.T) {
+		beat, err := NewBeatForReceiver(cmd.FilebeatSettings("filebeat"), cfg, false, consumertest.NewNop(), "testcomponent", zapcore.NewNopCore())
+		assert.NoError(t, err)
+		assert.NotNil(t, beat.Manager)
+		// it should fallback to FallbackManager if key is missing
+		assert.IsType(t, beat.Manager, &management.FallbackManager{})
+		assert.False(t, management.UnderAgent())
+	})
+	t.Run("otel management enabled", func(t *testing.T) {
+		tmpCfg := map[string]any{}
+		maps.Copy(tmpCfg, cfg)
+		tmpCfg["management.otel.enabled"] = true
+		defer func() {
+			management.SetUnderAgent(false) // reset to false
+		}()
+		beat, err := NewBeatForReceiver(cmd.FilebeatSettings("filebeat"), tmpCfg, false, consumertest.NewNop(), "testcomponent", zapcore.NewNopCore())
+		assert.NoError(t, err)
+		assert.NotNil(t, beat.Manager)
+		assert.IsType(t, beat.Manager, &otelmanager.OtelManager{})
+		assert.True(t, management.UnderAgent())
+
+		// test if log input is enabled
+		cfg, err := conf.NewConfigFrom(`
+type: "log"`)
+		require.NoError(t, err)
+		assert.True(t, log.AllowDeprecatedUse(cfg))
+	})
+	t.Run("otel management disabled", func(t *testing.T) {
+		tmpCfg := map[string]any{}
+		maps.Copy(tmpCfg, cfg)
+		tmpCfg["management.otel.enabled"] = false
+		defer func() {
+			management.SetUnderAgent(false) // reset to false
+		}()
+		beat, err := NewBeatForReceiver(cmd.FilebeatSettings("filebeat"), tmpCfg, false, consumertest.NewNop(), "testcomponent", zapcore.NewNopCore())
+		assert.NoError(t, err)
+		assert.NotNil(t, beat.Manager)
+		assert.IsType(t, beat.Manager, &management.FallbackManager{})
+		assert.False(t, management.UnderAgent())
+
+		// test if log input is disabled
+		cfg, err := conf.NewConfigFrom(`
+type: "log"`)
+		require.NoError(t, err)
+		assert.False(t, log.AllowDeprecatedUse(cfg))
+	})
+}

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

@@ -0,0 +1,46 @@
+// 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 otelmanager
+
+import (
+	"github.com/elastic/beats/v7/libbeat/common/reload"
+	"github.com/elastic/beats/v7/libbeat/management"
+	"github.com/elastic/beats/v7/libbeat/management/status"
+	"github.com/elastic/elastic-agent-client/v7/pkg/client"
+	"github.com/elastic/elastic-agent-libs/config"
+	"github.com/elastic/elastic-agent-libs/logp"
+)
+
+var _ management.Manager = (*OtelManager)(nil)
+
+func NewOtelManager(cfg *config.C, registry *reload.Registry, logger *logp.Logger) (management.Manager, error) {
+	management.SetUnderAgent(true)
+	return &OtelManager{}, nil
+}
+
+// OtelManager is the main manager for managing beatreceivers
+type OtelManager struct{}
+
+func (n *OtelManager) UpdateStatus(_ status.Status, _ string) {
+	// a stub implemtation for now.
+	// TODO(@VihasMakwana): Explore the option to tidy and refactor the status reporting for beatsreceivers.
+}
+
+func (n *OtelManager) SetStopCallback(func()) {
+}
+
+func (n *OtelManager) Stop() {}
+
+// Enabled returns false because many places inside beats call manager.Enabled() for various purposes
+// Returning true might lead to side effects.
+func (n *OtelManager) Enabled() bool                         { return false }
+func (n *OtelManager) AgentInfo() client.AgentInfo           { return client.AgentInfo{} }
+func (n *OtelManager) Start() error                          { return nil }
+func (n *OtelManager) CheckRawConfig(cfg *config.C) error    { return nil }
+func (n *OtelManager) RegisterAction(action client.Action)   {}
+func (n *OtelManager) UnregisterAction(action client.Action) {}
+func (n *OtelManager) SetPayload(map[string]interface{})     {}
+func (n *OtelManager) RegisterDiagnosticHook(_ string, _ string, _ string, _ string, _ client.DiagnosticHook) {
+}