Add component and unit diagnostics for beats receivers (#8991)

* Add diagnostics to otel manager

# Conflicts:
#	internal/pkg/agent/application/coordinator/coordinator_test.go
#	internal/pkg/otel/manager/manager.go

* add integration test

* Move helper function to allow diagnostic tests to run on Windows

* Handle diagnostics errors separately

* Centralize locking for config updates

* Make tests more explicit

* Early exit from diagnostics generation function

* Refactor config generation for easier locking

(cherry picked from commit ce0519f)

Author: swiatekm

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

@@ -148,6 +148,14 @@ type OTelManager interface {
 	// MergedOtelConfig returns the merged Otel collector configuration, containing both the plain config and the
 	// component config.
 	MergedOtelConfig() *confmap.Conf
+
+	// PerformDiagnostics executes the diagnostic action for the provided units. If no units are provided then
+	// it performs diagnostics for all current units.
+	PerformDiagnostics(context.Context, ...runtime.ComponentUnitDiagnosticRequest) []runtime.ComponentUnitDiagnostic
+
+	// PerformComponentDiagnostics executes the diagnostic action for the provided components. If no components are provided,
+	// then it performs the diagnostics for all current units.
+	PerformComponentDiagnostics(ctx context.Context, additionalMetrics []cproto.AdditionalDiagnosticRequest, req ...component.Component) ([]runtime.ComponentDiagnostic, error)
 }
 
 // ConfigChange provides an interface for receiving a new configuration.
@@ -649,12 +657,29 @@ func (c *Coordinator) PerformAction(ctx context.Context, comp component.Componen
 // it performs diagnostics for all current units.
 // Called from external goroutines.
 func (c *Coordinator) PerformDiagnostics(ctx context.Context, req ...runtime.ComponentUnitDiagnosticRequest) []runtime.ComponentUnitDiagnostic {
-	return c.runtimeMgr.PerformDiagnostics(ctx, req...)
+	var diags []runtime.ComponentUnitDiagnostic
+	runtimeDiags := c.runtimeMgr.PerformDiagnostics(ctx, req...)
+	diags = append(diags, runtimeDiags...)
+	otelDiags := c.otelMgr.PerformDiagnostics(ctx, req...)
+	diags = append(diags, otelDiags...)
+	return diags
 }
 
 // PerformComponentDiagnostics executes the diagnostic action for the provided components.
 func (c *Coordinator) PerformComponentDiagnostics(ctx context.Context, additionalMetrics []cproto.AdditionalDiagnosticRequest, req ...component.Component) ([]runtime.ComponentDiagnostic, error) {
-	return c.runtimeMgr.PerformComponentDiagnostics(ctx, additionalMetrics, req...)
+	var diags []runtime.ComponentDiagnostic
+	runtimeDiags, runtimeErr := c.runtimeMgr.PerformComponentDiagnostics(ctx, additionalMetrics, req...)
+	if runtimeErr != nil {
+		runtimeErr = fmt.Errorf("runtime diagnostics failed: %w", runtimeErr)
+	}
+	diags = append(diags, runtimeDiags...)
+	otelDiags, otelErr := c.otelMgr.PerformComponentDiagnostics(ctx, additionalMetrics, req...)
+	if otelErr != nil {
+		otelErr = fmt.Errorf("otel diagnostics failed: %w", otelErr)
+	}
+	diags = append(diags, otelDiags...)
+	err := errors.Join(runtimeErr, otelErr)
+	return diags, err
 }
 
 // SetLogLevel changes the entire log level for the running Elastic Agent.

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

@@ -1341,11 +1341,13 @@ func (f *fakeVarsManager) DefaultProvider() string {
 var _ OTelManager = (*fakeOTelManager)(nil)
 
 type fakeOTelManager struct {
-	updateCollectorCallback func(*confmap.Conf) error
-	updateComponentCallback func([]component.Component) error
-	errChan                 chan error
-	collectorStatusChan     chan *status.AggregateStatus
-	componentStateChan      chan []runtime.ComponentComponentState
+	updateCollectorCallback             func(*confmap.Conf) error
+	updateComponentCallback             func([]component.Component) error
+	performDiagnosticsCallback          func(context.Context, ...runtime.ComponentUnitDiagnosticRequest) []runtime.ComponentUnitDiagnostic
+	performComponentDiagnosticsCallback func(context.Context, []cproto.AdditionalDiagnosticRequest, ...component.Component) ([]runtime.ComponentDiagnostic, error)
+	errChan                             chan error
+	collectorStatusChan                 chan *status.AggregateStatus
+	componentStateChan                  chan []runtime.ComponentComponentState
 }
 
 func (f *fakeOTelManager) Run(ctx context.Context) error {
@@ -1385,12 +1387,28 @@ func (f *fakeOTelManager) WatchComponents() <-chan []runtime.ComponentComponentS
 
 func (f *fakeOTelManager) MergedOtelConfig() *confmap.Conf { return nil }
 
+func (f *fakeOTelManager) PerformDiagnostics(ctx context.Context, reqs ...runtime.ComponentUnitDiagnosticRequest) []runtime.ComponentUnitDiagnostic {
+	if f.performDiagnosticsCallback != nil {
+		return f.performDiagnosticsCallback(ctx, reqs...)
+	}
+	return nil
+}
+
+func (f *fakeOTelManager) PerformComponentDiagnostics(ctx context.Context, additionalMetrics []cproto.AdditionalDiagnosticRequest, req ...component.Component) ([]runtime.ComponentDiagnostic, error) {
+	if f.performComponentDiagnosticsCallback != nil {
+		return f.performComponentDiagnosticsCallback(ctx, additionalMetrics, req...)
+	}
+	return nil, nil
+}
+
 // An implementation of the RuntimeManager interface for use in testing.
 type fakeRuntimeManager struct {
-	state          []runtime.ComponentComponentState
-	updateCallback func([]component.Component) error
-	result         error
-	errChan        chan error
+	state                               []runtime.ComponentComponentState
+	updateCallback                      func([]component.Component) error
+	performDiagnosticsCallback          func(context.Context, ...runtime.ComponentUnitDiagnosticRequest) []runtime.ComponentUnitDiagnostic
+	performComponentDiagnosticsCallback func(context.Context, []cproto.AdditionalDiagnosticRequest, ...component.Component) ([]runtime.ComponentDiagnostic, error)
+	result                              error
+	errChan                             chan error
 }
 
 func (r *fakeRuntimeManager) Run(ctx context.Context) error {
@@ -1428,12 +1446,18 @@ func (r *fakeRuntimeManager) SubscribeAll(context.Context) *runtime.Subscription
 
 // PerformDiagnostics executes the diagnostic action for the provided units. If no units are provided then
 // it performs diagnostics for all current units.
-func (r *fakeRuntimeManager) PerformDiagnostics(context.Context, ...runtime.ComponentUnitDiagnosticRequest) []runtime.ComponentUnitDiagnostic {
+func (r *fakeRuntimeManager) PerformDiagnostics(ctx context.Context, req ...runtime.ComponentUnitDiagnosticRequest) []runtime.ComponentUnitDiagnostic {
+	if r.performDiagnosticsCallback != nil {
+		return r.performDiagnosticsCallback(ctx, req...)
+	}
 	return nil
 }
 
 // PerformComponentDiagnostics  executes the diagnostic action for the provided components.
-func (r *fakeRuntimeManager) PerformComponentDiagnostics(_ context.Context, _ []cproto.AdditionalDiagnosticRequest, _ ...component.Component) ([]runtime.ComponentDiagnostic, error) {
+func (r *fakeRuntimeManager) PerformComponentDiagnostics(ctx context.Context, additionalMetrics []cproto.AdditionalDiagnosticRequest, req ...component.Component) ([]runtime.ComponentDiagnostic, error) {
+	if r.performComponentDiagnosticsCallback != nil {
+		return r.performComponentDiagnosticsCallback(ctx, additionalMetrics, req...)
+	}
 	return nil, nil
 }