Fall back to process runtime if otel runtime is unsupported (#10087)

* Move component monitoring to its own package

Agent monitoring has two separate functions - implementing the control
plane monitoring server and self-monitoring for components. Having both
in the same packages caused a dependency cycle involving the otel
translation package. Resolve this by putting component monitoring in a
subpackage

* Fall back to process runtime if otel runtime is unsupported

* Fix integration test

* Normalize import names

* Add logstash unit test

* Use indices instead of allow_older_version

* Add log line for skipped components

* Change argument order

* Fix linter warning

(cherry picked from commit 2f0ba69)

# Conflicts:
#	internal/pkg/agent/application/application.go
#	internal/pkg/agent/application/coordinator/coordinator.go
#	internal/pkg/otel/translate/otelconfig.go
#	internal/pkg/otel/translate/otelconfig_test.go

Author: swiatekm

internal/pkg/agent/application/application.go

@@ -11,12 +11,18 @@ import (
 
 	"go.elastic.co/apm/v2"
 
+<<<<<<< HEAD
+=======
+	componentmonitoring "github.com/elastic/elastic-agent/internal/pkg/agent/application/monitoring/component"
+
+	"github.com/elastic/go-ucfg"
+
+>>>>>>> 2f0ba69f2 (Fall back to process runtime if otel runtime is unsupported (#10087))
 	"github.com/elastic/elastic-agent-libs/logp"
 
 	"github.com/elastic/elastic-agent/internal/pkg/agent/application/coordinator"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/application/dispatcher"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/application/info"
-	"github.com/elastic/elastic-agent/internal/pkg/agent/application/monitoring"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/application/paths"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/application/upgrade"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/application/upgrade/details"
@@ -130,7 +136,7 @@ func New(
 	if err != nil {
 		return nil, nil, nil, fmt.Errorf("failed to create upgrader: %w", err)
 	}
-	monitor := monitoring.New(isMonitoringSupported, cfg.Settings.DownloadConfig.OS(), cfg.Settings.MonitoringConfig, agentInfo, isOtelExecModeSubprocess)
+	monitor := componentmonitoring.New(isMonitoringSupported, cfg.Settings.DownloadConfig.OS(), cfg.Settings.MonitoringConfig, agentInfo, isOtelExecModeSubprocess)
 
 	runtime, err := runtime.NewManager(
 		log,

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

@@ -13,6 +13,12 @@ import (
 	"sync/atomic"
 	"time"
 
+<<<<<<< HEAD
+=======
+	"github.com/elastic/elastic-agent/internal/pkg/core/backoff"
+	"github.com/elastic/elastic-agent/internal/pkg/otel/translate"
+
+>>>>>>> 2f0ba69f2 (Fall back to process runtime if otel runtime is unsupported (#10087))
 	"go.opentelemetry.io/collector/component/componentstatus"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/status"
@@ -1635,6 +1641,7 @@ func (c *Coordinator) updateManagersWithConfig(model *component.Model) {
 func (c *Coordinator) splitModelBetweenManagers(model *component.Model) (runtimeModel *component.Model, otelModel *component.Model) {
 	var otelComponents, runtimeComponents []component.Component
 	for _, comp := range model.Components {
+		c.maybeOverrideRuntimeForComponent(&comp)
 		switch comp.RuntimeManager {
 		case component.OtelRuntimeManager:
 			otelComponents = append(otelComponents, comp)
@@ -1656,6 +1663,59 @@ func (c *Coordinator) splitModelBetweenManagers(model *component.Model) (runtime
 	return
 }
 
+<<<<<<< HEAD
+=======
+// maybeOverrideRuntimeForComponent sets the correct runtime for the given component.
+// Normally, we use the runtime set in the component itself via the configuration, but
+// we may also fall back to the process runtime if the otel runtime is unsupported for
+// some reason. One example is the output using unsupported config options.
+func (c *Coordinator) maybeOverrideRuntimeForComponent(comp *component.Component) {
+	if comp.RuntimeManager == component.ProcessRuntimeManager {
+		// do nothing, the process runtime can handle any component
+		return
+	}
+	if comp.RuntimeManager == component.OtelRuntimeManager {
+		// check if the component is actually supported
+		err := translate.VerifyComponentIsOtelSupported(comp)
+		if err != nil {
+			c.logger.Warnf("otel runtime is not supported for component %s, switching to process runtime, reason: %v", comp.ID, err)
+			comp.RuntimeManager = component.ProcessRuntimeManager
+		}
+	}
+}
+
+func (c *Coordinator) isFleetServer() bool {
+	for _, s := range c.state.Components {
+		if s.Component.InputType == fleetServer {
+			return true
+		}
+	}
+	return false
+}
+
+func (c *Coordinator) HasEndpoint() bool {
+	for _, component := range c.state.Components {
+		if component.Component.InputType == endpoint {
+			return true
+		}
+	}
+
+	return false
+}
+
+func (c *Coordinator) ackMigration(ctx context.Context, action *fleetapi.ActionMigrate, acker acker.Acker) error {
+	if err := acker.Ack(ctx, action); err != nil {
+		return fmt.Errorf("failed to ack migrate action: %w", err)
+	}
+
+	if err := acker.Commit(ctx); err != nil {
+		return fmt.Errorf("failed to commit migrate action: %w", err)
+	}
+
+	return nil
+}
+
+>>>>>>> 2f0ba69f2 (Fall back to process runtime if otel runtime is unsupported (#10087))
 // generateComponentModel regenerates the configuration tree and
 // components from the current AST and vars and returns the result.
 // Called from both the main Coordinator goroutine and from external

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

@@ -1023,8 +1023,9 @@ func TestCoordinatorPolicyChangeUpdatesRuntimeAndOTelManagerWithOtelComponents(t
 		secretMarkerFunc:   testSecretMarkerFunc,
 	}
 
-	// Create a policy with one input and one output (no otel configuration)
-	cfg := config.MustNewConfigFrom(`
+	t.Run("mixed policy", func(t *testing.T) {
+		// Create a policy with one input and one output (no otel configuration)
+		cfg := config.MustNewConfigFrom(`
 outputs:
   default:
     type: elasticsearch
@@ -1051,38 +1052,87 @@ service:
         - nop
 `)
 
-	// Send the policy change and make sure it was acknowledged.
-	cfgChange := &configChange{cfg: cfg}
-	configChan <- cfgChange
-	coord.runLoopIteration(ctx)
-	assert.True(t, cfgChange.acked, "Coordinator should ACK a successful policy change")
+		// Send the policy change and make sure it was acknowledged.
+		cfgChange := &configChange{cfg: cfg}
+		configChan <- cfgChange
+		coord.runLoopIteration(ctx)
+		assert.True(t, cfgChange.acked, "Coordinator should ACK a successful policy change")
+
+		// Make sure the runtime manager received the expected component update.
+		// An assert.Equal on the full component model doesn't play nice with
+		// the embedded proto structs, so instead we verify the important fields
+		// manually (sorry).
+		assert.True(t, updated, "Runtime manager should be updated after a policy change")
+		require.Equal(t, 1, len(components), "Test policy should generate one component")
+		assert.True(t, otelUpdated, "OTel manager should be updated after a policy change")
+		require.NotNil(t, otelConfig, "OTel manager should have config")
+
+		runtimeComponent := components[0]
+		assert.Equal(t, "system/metrics-default", runtimeComponent.ID)
+		require.NotNil(t, runtimeComponent.Err, "Input with no spec should produce a component error")
+		assert.Equal(t, "input not supported", runtimeComponent.Err.Error(), "Input with no spec should report 'input not supported'")
+		require.Equal(t, 2, len(runtimeComponent.Units))
+
+		units := runtimeComponent.Units
+		// Verify the input unit
+		assert.Equal(t, "system/metrics-default-test-other-input", units[0].ID)
+		assert.Equal(t, client.UnitTypeInput, units[0].Type)
+		assert.Equal(t, "test-other-input", units[0].Config.Id)
+		assert.Equal(t, "system/metrics", units[0].Config.Type)
+
+		// Verify the output unit
+		assert.Equal(t, "system/metrics-default", units[1].ID)
+		assert.Equal(t, client.UnitTypeOutput, units[1].Type)
+		assert.Equal(t, "elasticsearch", units[1].Config.Type)
+	})
+
+	t.Run("unsupported otel output option", func(t *testing.T) {
+		// Create a policy with one input and one output (no otel configuration)
+		cfg := config.MustNewConfigFrom(`
+outputs:
+  default:
+    type: elasticsearch
+    hosts:
+      - localhost:9200
+    indices: [] # not supported by the elasticsearch exporter
+inputs:
+  - id: test-input
+    type: filestream
+    use_output: default
+    _runtime_experimental: otel
+  - id: test-other-input
+    type: system/metrics
+    use_output: default
+receivers:
+  nop:
+exporters:
+  nop:
+service:
+  pipelines:
+    traces:
+      receivers:
+        - nop
+      exporters:
+        - nop
+`)
 
-	// Make sure the runtime manager received the expected component update.
-	// An assert.Equal on the full component model doesn't play nice with
-	// the embedded proto structs, so instead we verify the important fields
-	// manually (sorry).
-	assert.True(t, updated, "Runtime manager should be updated after a policy change")
-	require.Equal(t, 1, len(components), "Test policy should generate one component")
-	assert.True(t, otelUpdated, "OTel manager should be updated after a policy change")
-	require.NotNil(t, otelConfig, "OTel manager should have config")
+		// Send the policy change and make sure it was acknowledged.
+		cfgChange := &configChange{cfg: cfg}
+		configChan <- cfgChange
+		coord.runLoopIteration(ctx)
+		assert.True(t, cfgChange.acked, "Coordinator should ACK a successful policy change")
 
-	runtimeComponent := components[0]
-	assert.Equal(t, "system/metrics-default", runtimeComponent.ID)
-	require.NotNil(t, runtimeComponent.Err, "Input with no spec should produce a component error")
-	assert.Equal(t, "input not supported", runtimeComponent.Err.Error(), "Input with no spec should report 'input not supported'")
-	require.Equal(t, 2, len(runtimeComponent.Units))
+		// Make sure the runtime manager received the expected component update.
+		// An assert.Equal on the full component model doesn't play nice with
+		// the embedded proto structs, so instead we verify the important fields
+		// manually (sorry).
+		assert.True(t, updated, "Runtime manager should be updated after a policy change")
+		assert.True(t, otelUpdated, "OTel manager should be updated after a policy change")
+		require.NotNil(t, otelConfig, "OTel manager should have config")
 
-	units := runtimeComponent.Units
-	// Verify the input unit
-	assert.Equal(t, "system/metrics-default-test-other-input", units[0].ID)
-	assert.Equal(t, client.UnitTypeInput, units[0].Type)
-	assert.Equal(t, "test-other-input", units[0].Config.Id)
-	assert.Equal(t, "system/metrics", units[0].Config.Type)
+		assert.Len(t, components, 2, "both components should be assigned to the runtime manager")
+	})
 
-	// Verify the output unit
-	assert.Equal(t, "system/metrics-default", units[1].ID)
-	assert.Equal(t, client.UnitTypeOutput, units[1].Type)
-	assert.Equal(t, "elasticsearch", units[1].Config.Type)
 }
 
 func TestCoordinatorReportsRuntimeManagerUpdateFailure(t *testing.T) {

internal/pkg/agent/application/monitoring/testdata/monitoring_config_full.yaml renamed to internal/pkg/agent/application/monitoring/component/testdata/monitoring_config_full.yaml

@@ -2,7 +2,7 @@
 // or more contributor license agreements. Licensed under the Elastic License 2.0;
 // you may not use this file except in compliance with the Elastic License 2.0.
 
-package monitoring
+package component
 
 import (
 	"crypto/sha256"

internal/pkg/agent/application/monitoring/v1_monitor.go renamed to internal/pkg/agent/application/monitoring/component/v1_monitor.go

@@ -2,7 +2,7 @@
 // or more contributor license agreements. Licensed under the Elastic License 2.0;
 // you may not use this file except in compliance with the Elastic License 2.0.
 
-package monitoring
+package component
 
 import (
 	"context"

internal/pkg/agent/application/monitoring/v1_monitor_test.go renamed to internal/pkg/agent/application/monitoring/component/v1_monitor_test.go

@@ -16,6 +16,8 @@ import (
 
 	"github.com/gorilla/mux"
 
+	componentmonitoring "github.com/elastic/elastic-agent/internal/pkg/agent/application/monitoring/component"
+
 	"github.com/elastic/elastic-agent/internal/pkg/agent/application/paths"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/errors"
 	"github.com/elastic/elastic-agent/pkg/utils"
@@ -120,7 +122,7 @@ func isProcessRedirectable(componentID string) bool {
 }
 
 func redirectToPath(w http.ResponseWriter, r *http.Request, id, path string) error {
-	endpoint := PrefixedEndpoint(utils.SocketURLWithFallback(id, paths.TempDir()))
+	endpoint := componentmonitoring.PrefixedEndpoint(utils.SocketURLWithFallback(id, paths.TempDir()))
 	metricsBytes, statusCode, metricsErr := GetProcessMetrics(r.Context(), endpoint, path)
 	if metricsErr != nil {
 		return metricsErr

internal/pkg/agent/application/monitoring/process.go

@@ -18,6 +18,8 @@ import (
 	"go.elastic.co/apm/module/apmgorilla/v2"
 	"go.elastic.co/apm/v2"
 
+	componentmonitoring "github.com/elastic/elastic-agent/internal/pkg/agent/application/monitoring/component"
+
 	"github.com/elastic/elastic-agent-libs/api"
 	"github.com/elastic/elastic-agent-libs/monitoring"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/application/monitoring/reload"
@@ -74,7 +76,7 @@ func NewServer(
 
 		srvCfg := api.DefaultConfig()
 		srvCfg.Enabled = cfg.Enabled
-		srvCfg.Host = AgentMonitoringEndpoint(cfg)
+		srvCfg.Host = componentmonitoring.AgentMonitoringEndpoint(cfg)
 		srvCfg.Port = cfg.HTTP.Port
 		log.Infof("creating monitoring API with cfg %#v", srvCfg)
 		if err := createAgentMonitoringDrop(srvCfg.Host); err != nil {

internal/pkg/agent/application/monitoring/server.go

@@ -14,11 +14,12 @@ import (
 	"github.com/spf13/cobra"
 	"gopkg.in/yaml.v2"
 
+	componentmonitoring "github.com/elastic/elastic-agent/internal/pkg/agent/application/monitoring/component"
+
 	"github.com/elastic/elastic-agent-libs/logp"
 	"github.com/elastic/elastic-agent-libs/service"
 
 	"github.com/elastic/elastic-agent/internal/pkg/agent/application/info"
-	"github.com/elastic/elastic-agent/internal/pkg/agent/application/monitoring"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/application/paths"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/configuration"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/errors"
@@ -412,7 +413,7 @@ func getMonitoringFn(ctx context.Context, logger *logger.Logger, cfg map[string]
 	}
 	otelExecMode := otelconfig.GetExecutionModeFromConfig(logger, config)
 	isOtelExecModeSubprocess := otelExecMode == manager.SubprocessExecutionMode
-	monitor := monitoring.New(agentCfg.Settings.V1MonitoringEnabled, agentCfg.Settings.DownloadConfig.OS(), agentCfg.Settings.MonitoringConfig, agentInfo, isOtelExecModeSubprocess)
+	monitor := componentmonitoring.New(agentCfg.Settings.V1MonitoringEnabled, agentCfg.Settings.DownloadConfig.OS(), agentCfg.Settings.MonitoringConfig, agentInfo, isOtelExecModeSubprocess)
 	return monitor.MonitoringConfig, nil
 }
 

internal/pkg/agent/cmd/inspect.go

@@ -20,6 +20,7 @@ import (
 	"strings"
 
 	"github.com/elastic/elastic-agent/internal/pkg/agent/application/monitoring"
+	componentmonitoring "github.com/elastic/elastic-agent/internal/pkg/agent/application/monitoring/component"
 
 	"google.golang.org/protobuf/types/known/timestamppb"
 
@@ -208,7 +209,7 @@ func GetBeatInputMetricsDiagnostics(ctx context.Context, componentID string) (*p
 }
 
 func GetBeatMetricsPayload(ctx context.Context, componentID string, path string) ([]byte, error) {
-	endpoint := monitoring.PrefixedEndpoint(monitoring.BeatsMonitoringEndpoint(componentID))
+	endpoint := componentmonitoring.PrefixedEndpoint(componentmonitoring.BeatsMonitoringEndpoint(componentID))
 	metricBytes, statusCode, err := monitoring.GetProcessMetrics(ctx, endpoint, path)
 	if err != nil {
 		return nil, err