Move Otel runtime determination to component modifier

Author: swiatekm

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

@@ -1808,7 +1808,6 @@ 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)
@@ -1840,7 +1839,7 @@ func (c *Coordinator) splitModelBetweenManagers(model *component.Model) (runtime
 // 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) {
+func maybeOverrideRuntimeForComponent(logger *logger.Logger, comp *component.Component) {
 	if comp.RuntimeManager == component.ProcessRuntimeManager {
 		// do nothing, the process runtime can handle any component
 		return
@@ -1849,7 +1848,7 @@ func (c *Coordinator) maybeOverrideRuntimeForComponent(comp *component.Component
 		// 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)
+			logger.Warnf("otel runtime is not supported for component %s, switching to process runtime, reason: %v", comp.ID, err)
 			comp.RuntimeManager = component.ProcessRuntimeManager
 		}
 	}
@@ -1939,8 +1938,15 @@ func (c *Coordinator) generateComponentModel() (err error) {
 		existingCompState[comp.Component.ID] = comp.State.Pid
 	}
 
+	otelRuntimeModifier := func(comps []component.Component, cfg map[string]interface{}) ([]component.Component, error) {
+		for i := range comps {
+			maybeOverrideRuntimeForComponent(c.logger, &comps[i])
+		}
+		return comps, nil
+	}
 	comps, err := c.specs.ToComponents(
 		cfg,
+		append(c.modifiers, otelRuntimeModifier),
 		configInjector,
 		c.state.LogLevel,
 		c.agentInfo,
@@ -1953,13 +1959,6 @@ func (c *Coordinator) generateComponentModel() (err error) {
 	// Filter any disallowed inputs/outputs from the components
 	comps = c.filterByCapabilities(comps)
 
-	for _, modifier := range c.modifiers {
-		comps, err = modifier(comps, cfg)
-		if err != nil {
-			return fmt.Errorf("failed to modify components: %w", err)
-		}
-	}
-
 	// If we made it this far, update our internal derived values and
 	// return with no error
 	c.derivedConfig = cfg

internal/pkg/agent/application/endpoint_component_modifier_test.go

@@ -20,7 +20,6 @@ import (
 	"github.com/elastic/elastic-agent-client/v7/pkg/client"
 	"github.com/elastic/elastic-agent-client/v7/pkg/proto"
 	"github.com/elastic/elastic-agent-libs/testing/certutil"
-	"github.com/elastic/elastic-agent/internal/pkg/agent/application/coordinator"
 	"github.com/elastic/elastic-agent/internal/pkg/testutils/fipsutils"
 	"github.com/elastic/elastic-agent/pkg/core/logger/loggertest"
 
@@ -44,7 +43,7 @@ func TestEndpointComponentModifier(t *testing.T) {
 
 	tests := map[string][]struct {
 		name         string
-		compModifier coordinator.ComponentsModifier
+		compModifier component.ComponentsModifier
 		comps        []component.Component
 		cfg          map[string]interface{}
 		wantComps    []component.Component

internal/pkg/agent/install/componentvalidation/validation.go

@@ -101,7 +101,7 @@ func GetComponentsFromPolicy(ctx context.Context, l *logger.Logger, cfgPath stri
 	}
 
 	// Compute the components from the computed configuration.
-	comps, err := specs.ToComponents(m, monitorFn, lvl, agentInfo, map[string]uint64{})
+	comps, err := specs.ToComponents(m, nil, monitorFn, lvl, agentInfo, map[string]uint64{})
 	if err != nil {
 		return nil, fmt.Errorf("failed to render components: %w", err)
 	}

internal/pkg/agent/install/uninstall.go

@@ -422,7 +422,7 @@ func serviceComponentsFromConfig(specs component.RuntimeSpecs, cfg *config.Confi
 	if err != nil {
 		return nil, aerrors.New("failed to create a map from config", err)
 	}
-	allComps, err := specs.ToComponents(mm, nil, logp.InfoLevel, nil, map[string]uint64{})
+	allComps, err := specs.ToComponents(mm, nil, nil, logp.InfoLevel, nil, map[string]uint64{})
 	if err != nil {
 		return nil, fmt.Errorf("failed to render components: %w", err)
 	}

pkg/component/component.go

@@ -343,6 +343,7 @@ type Model struct {
 // the current runtime specification.
 func (r *RuntimeSpecs) ToComponents(
 	policy map[string]interface{},
+	modifiers []ComponentsModifier,
 	monitoringInjector GenerateMonitoringCfgFn,
 	ll logp.Level,
 	headers HeadersProvider,
@@ -353,6 +354,14 @@ func (r *RuntimeSpecs) ToComponents(
 		return nil, err
 	}
 
+	// Do this here so the monitoring injector has a more accurate view of what components are running
+	for _, modifier := range modifiers {
+		components, err = modifier(components, policy)
+		if err != nil {
+			return nil, err
+		}
+	}
+
 	if monitoringInjector != nil {
 		monitoringCfg, err := monitoringInjector(policy, components, currentServiceCompInts)
 		if err != nil {
@@ -366,6 +375,13 @@ func (r *RuntimeSpecs) ToComponents(
 				return nil, fmt.Errorf("failed to generate monitoring components: %w", err)
 			}
 
+			for _, modifier := range modifiers {
+				monitoringComps, err = modifier(monitoringComps, policy)
+				if err != nil {
+					return nil, err
+				}
+			}
+
 			components = append(components, monitoringComps...)
 		}
 	}

pkg/component/component_test.go

@@ -2503,7 +2503,7 @@ func TestToComponents(t *testing.T) {
 			runtime, err := LoadRuntimeSpecs(filepath.Join("..", "..", "specs"), scenario.Platform, SkipBinaryCheck())
 			require.NoError(t, err)
 
-			result, err := runtime.ToComponents(scenario.Policy, nil, scenario.LogLevel, scenario.headers, map[string]uint64{})
+			result, err := runtime.ToComponents(scenario.Policy, nil, nil, scenario.LogLevel, scenario.headers, map[string]uint64{})
 			if scenario.Err != "" {
 				assert.Equal(t, scenario.Err, err.Error())
 			} else {
@@ -2979,7 +2979,7 @@ func TestFlattenedDataStream(t *testing.T) {
 		t.Fatalf("cannot load runtime specs: %s", err)
 	}
 
-	result, err := runtime.ToComponents(policy, nil, logp.DebugLevel, nil, map[string]uint64{})
+	result, err := runtime.ToComponents(policy, nil, nil, logp.DebugLevel, nil, map[string]uint64{})
 	if err != nil {
 		t.Fatalf("cannot convert policy to component: %s", err)
 	}
@@ -3080,7 +3080,7 @@ func TestFlattenedDataStreamIsolatedUnits(t *testing.T) {
 		t.Fatalf("cannot load runtime specs: %s", err)
 	}
 
-	result, err := runtime.ToComponents(policy, nil, logp.DebugLevel, nil, map[string]uint64{})
+	result, err := runtime.ToComponents(policy, nil, nil, logp.DebugLevel, nil, map[string]uint64{})
 	if err != nil {
 		t.Fatalf("cannot convert policy to component: %s", err)
 	}