Check supported outputs in monitoring config generation

Author: swiatekm

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

@@ -1812,12 +1812,6 @@ func (c *Coordinator) splitModelBetweenManagers(model *component.Model) (runtime
 		case component.OtelRuntimeManager:
 			otelComponents = append(otelComponents, comp)
 		case component.ProcessRuntimeManager:
-			// Hack to fix https://github.com/elastic/elastic-agent/issues/11169
-			// TODO: Remove this after https://github.com/elastic/elastic-agent/issues/10220 is resolved
-			if comp.ID == "prometheus/metrics-monitoring" {
-				c.logger.Warnf("The Otel prometheus metrics monitoring input can't run in a beats process, skipping")
-				continue
-			}
 			runtimeComponents = append(runtimeComponents, comp)
 		default:
 			// this should be impossible if we parse the configuration correctly

internal/pkg/agent/application/monitoring/component/testdata/monitoring_config_full_otel.yaml

@@ -915,4 +915,7 @@ inputs:
   type: system/metrics
   use_output: monitoring
 outputs:
-  monitoring: {}
+  monitoring:
+    hosts:
+    - localhost:9200
+    type: elasticsearch

internal/pkg/agent/application/monitoring/component/testdata/monitoring_config_full_process.yaml

@@ -357,6 +357,14 @@ inputs:
           binary: metricbeat
           id: prometheus/metrics-monitoring
         target: component
+    - drop_fields:
+        fields:
+        - beat.stats.cgroup
+        - beat.stats.cpu
+        - beat.stats.handles
+        - beat.stats.memstats
+        - beat.stats.runtime
+        ignore_missing: true
   type: beat/metrics
   use_output: monitoring
 - data_stream:
@@ -875,61 +883,6 @@ inputs:
           binary: metricbeat
           id: http/metrics-monitoring
         target: component
-  - data_stream:
-      dataset: elastic_agent.elastic_agent
-      namespace: default
-      type: metrics
-    failure_threshold: 5
-    hosts:
-    - placeholder
-    id: metrics-monitoring-metricbeat-1
-    index: metrics-elastic_agent.elastic_agent-default
-    metricsets:
-    - json
-    namespace: agent
-    path: /stats
-    period: 1m0s
-    processors:
-    - add_fields:
-        fields:
-          dataset: elastic_agent.elastic_agent
-        target: event
-    - add_fields:
-        fields:
-          id: ""
-          process: metricbeat
-          snapshot: false
-          version: placeholder
-        target: elastic_agent
-    - add_fields:
-        fields:
-          id: ""
-        target: agent
-    - copy_fields:
-        fail_on_error: false
-        fields:
-        - from: http.agent.beat.cpu
-          to: system.process.cpu
-        - from: http.agent.beat.memstats.memory_sys
-          to: system.process.memory.size
-        - from: http.agent.beat.handles
-          to: system.process.fd
-        - from: http.agent.beat.cgroup
-          to: system.process.cgroup
-        - from: http.agent.apm-server
-          to: apm-server
-        - from: http.filebeat_input
-          to: filebeat_input
-        ignore_missing: true
-    - drop_fields:
-        fields:
-        - http
-        ignore_missing: true
-    - add_fields:
-        fields:
-          binary: metricbeat
-          id: prometheus/metrics-monitoring
-        target: component
   type: http/metrics
   use_output: monitoring
 - _runtime_experimental: otel
@@ -1099,4 +1052,7 @@ inputs:
   type: system/metrics
   use_output: monitoring
 outputs:
-  monitoring: {}
+  monitoring:
+    hosts:
+    - localhost:9200
+    type: elasticsearch

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

@@ -23,6 +23,11 @@ import (
 
 	koanfmaps "github.com/knadh/koanf/maps"
 
+	"github.com/elastic/elastic-agent/internal/pkg/agent/application/monitoring/monitoringhelpers"
+
+	"github.com/elastic/elastic-agent-libs/logp"
+	"github.com/elastic/elastic-agent/internal/pkg/otel/translate"
+
 	"github.com/elastic/elastic-agent/pkg/component"
 	"github.com/elastic/elastic-agent/pkg/utils"
 
@@ -71,8 +76,9 @@ const (
 	fileBeatName               = "filebeat"
 	collectorName              = "collector"
 
-	monitoringMetricsUnitID = "metrics-monitoring"
-	monitoringFilesUnitsID  = "filestream-monitoring"
+	monitoringMetricsUnitID         = "metrics-monitoring"
+	monitoringFilesUnitsID          = "filestream-monitoring"
+	prometheusMonitoringComponentId = "prometheus/" + monitoringMetricsUnitID
 
 	windowsOS = "windows"
 
@@ -228,15 +234,17 @@ func (b *BeatsMonitor) MonitoringConfig(
 		}
 	}
 
-	componentInfos := b.getComponentInfos(components, componentIDPidMap)
-
-	if err := b.injectMonitoringOutput(policy, cfg, monitoringOutputName); err != nil && !errors.Is(err, errNoOutputPresent) {
+	outputCfg, err := b.injectMonitoringOutput(policy, cfg, monitoringOutputName)
+	if err != nil && !errors.Is(err, errNoOutputPresent) {
 		return nil, errors.New(err, "failed to inject monitoring output")
 	} else if errors.Is(err, errNoOutputPresent) {
 		// nothing to inject, no monitoring output
 		return nil, nil
 	}
 
+	outputOtelSupported := isOutputOtelSupported(outputCfg)
+	componentInfos := b.getComponentInfos(components, outputOtelSupported, componentIDPidMap)
+
 	// initializes inputs collection so injectors don't have to deal with it
 	b.initInputs(cfg)
 
@@ -287,7 +295,7 @@ func (b *BeatsMonitor) ComponentMonitoringConfig(unitID, binary string) map[stri
 	}
 
 	configMap := make(map[string]any)
-	endpoint := BeatsMonitoringEndpoint(unitID)
+	endpoint := monitoringhelpers.BeatsMonitoringEndpoint(unitID)
 	if endpoint != "" {
 		httpConfigMap := map[string]any{
 			"enabled": true,
@@ -393,20 +401,27 @@ func (b *BeatsMonitor) initInputs(cfg map[string]interface{}) {
 	cfg[inputsKey] = inputsCollection
 }
 
-func (b *BeatsMonitor) injectMonitoringOutput(source, dest map[string]interface{}, monitoringOutputName string) error {
+// injectMonitoringOutput injects the monitoring output into the configuration. It takes an existing output named
+// `monitoringOutputName` and makes a copy of it named `monitoring`. It returns the output configuration.
+func (b *BeatsMonitor) injectMonitoringOutput(source, dest map[string]interface{}, monitoringOutputName string) (map[string]any, error) {
 	outputsNode, found := source[outputsKey]
 	if !found {
-		return errNoOutputPresent
+		return nil, errNoOutputPresent
 	}
 
 	outputs, ok := outputsNode.(map[string]interface{})
 	if !ok {
-		return fmt.Errorf("outputs not a map")
+		return nil, fmt.Errorf("outputs not a map")
 	}
 
 	outputNode, found := outputs[monitoringOutputName]
 	if !found {
-		return fmt.Errorf("output %q used for monitoring not found", monitoringOutputName)
+		return nil, fmt.Errorf("output %q used for monitoring not found", monitoringOutputName)
+	}
+
+	outputMap, ok := outputNode.(map[string]any)
+	if !ok {
+		return nil, fmt.Errorf("output %q used for monitoring not a map", monitoringOutputName)
 	}
 
 	monitoringOutputs := map[string]interface{}{
@@ -415,13 +430,17 @@ func (b *BeatsMonitor) injectMonitoringOutput(source, dest map[string]interface{
 
 	dest[outputsKey] = monitoringOutputs
 
-	return nil
+	return outputMap, nil
 }
 
 // getComponentInfos returns a slice of componentInfo structs based on the provided components. This slice contains
 // all the information needed to generate the monitoring configuration for these components, as well as configuration
 // for new components which are going to be doing the monitoring.
-func (b *BeatsMonitor) getComponentInfos(components []component.Component, componentIDPidMap map[string]uint64) []componentInfo {
+func (b *BeatsMonitor) getComponentInfos(components []component.Component, outputOtelSupported bool, componentIDPidMap map[string]uint64) []componentInfo {
+	monitoringRuntime := component.RuntimeManager(b.config.C.RuntimeManager)
+	if !outputOtelSupported {
+		monitoringRuntime = monitoringCfg.ProcessRuntimeManager
+	}
 	componentInfos := make([]componentInfo, 0, len(components))
 	for _, comp := range components {
 		compInfo := componentInfo{
@@ -440,29 +459,29 @@ func (b *BeatsMonitor) getComponentInfos(components []component.Component, compo
 			componentInfo{
 				ID:             fmt.Sprintf("beat/%s", monitoringMetricsUnitID),
 				BinaryName:     metricBeatName,
-				RuntimeManager: component.RuntimeManager(b.config.C.RuntimeManager),
+				RuntimeManager: monitoringRuntime,
 			},
 			componentInfo{
 				ID:             fmt.Sprintf("http/%s", monitoringMetricsUnitID),
 				BinaryName:     metricBeatName,
-				RuntimeManager: component.RuntimeManager(b.config.C.RuntimeManager),
+				RuntimeManager: monitoringRuntime,
 			})
 	}
 	if b.config.C.MonitorLogs {
 		componentInfos = append(componentInfos, componentInfo{
 			ID:             monitoringFilesUnitsID,
 			BinaryName:     fileBeatName,
-			RuntimeManager: component.RuntimeManager(b.config.C.RuntimeManager),
+			RuntimeManager: monitoringRuntime,
 		})
 	}
-	// If any other component uses the Otel runtime, also add a component to monitor
-	// its telemetry.
-	if b.config.C.MonitorMetrics && usingOtelRuntime(componentInfos) {
+	// If any other component uses the Otel runtime, also add a component to monitor its telemetry.
+	// This component only works in the Otel runtime, so we can't add it if the output doesn't support it.
+	if b.config.C.MonitorMetrics && usingOtelRuntime(componentInfos) && outputOtelSupported {
 		componentInfos = append(componentInfos,
 			componentInfo{
-				ID:             fmt.Sprintf("prometheus/%s", monitoringMetricsUnitID),
+				ID:             prometheusMonitoringComponentId,
 				BinaryName:     metricBeatName,
-				RuntimeManager: component.RuntimeManager(b.config.C.RuntimeManager),
+				RuntimeManager: component.OtelRuntimeManager,
 			})
 	}
 	// sort the components to ensure a consistent order of inputs in the configuration
@@ -564,7 +583,12 @@ func (b *BeatsMonitor) injectMetricsInput(
 		},
 	}
 
-	if usingOtelRuntime(componentInfos) {
+	// We only add this stream if the Otel manager is enabled and the respective component info exists. This is a
+	// special case where this input shouldn't exists if the output doesn't support otel, which we check while
+	// creating the component infos.
+	if usingOtelRuntime(componentInfos) && slices.ContainsFunc(componentInfos, func(ci componentInfo) bool {
+		return ci.ID == prometheusMonitoringComponentId
+	}) {
 		prometheusStream := b.getPrometheusStream(failureThreshold, metricsCollectionIntervalString)
 		inputs = append(inputs, map[string]interface{}{
 			idKey:        fmt.Sprintf("%s-collector", monitoringMetricsUnitID),
@@ -750,7 +774,7 @@ func (b *BeatsMonitor) getHttpStreams(
 			continue
 		}
 
-		endpoints := []interface{}{PrefixedEndpoint(BeatsMonitoringEndpoint(compInfo.ID))}
+		endpoints := []interface{}{PrefixedEndpoint(monitoringhelpers.BeatsMonitoringEndpoint(compInfo.ID))}
 		name := sanitizeName(binaryName)
 
 		// Do not create http streams if runtime-manager is otel and binary is of beat type
@@ -1440,10 +1464,6 @@ func AgentMonitoringEndpoint(cfg *monitoringCfg.MonitoringConfig) string {
 	return fmt.Sprintf(`unix:///tmp/elastic-agent/%x.sock`, sha256.Sum256([]byte(path)))
 }
 
-func BeatsMonitoringEndpoint(componentID string) string {
-	return utils.SocketURLWithFallback(componentID, paths.TempDir())
-}
-
 func httpCopyRules() []interface{} {
 	fromToMap := []interface{}{
 		// I should be able to see the CPU Usage on the running machine. Am using too much CPU?
@@ -1504,6 +1524,16 @@ func isSupportedBeatsBinary(binaryName string) bool {
 	return false
 }
 
+func isOutputOtelSupported(outputCfg map[string]any) bool {
+	parsed, err := component.ParseOutput(monitoringOutput, outputCfg, logp.InfoLevel, nil)
+	if err != nil {
+		return false
+	}
+
+	err = translate.VerifyOutputIsOtelSupported(parsed.OutputType, outputCfg)
+	return err == nil
+}
+
 func monitoringDrop(path string) (drop string) {
 	defer func() {
 		if drop != "" {