feat: emit system resource metrics for EDOT subprocess

Author: pkoutsovasilis

internal/pkg/agent/application/application.go

@@ -122,13 +122,16 @@ func New(
 		override(cfg)
 	}
 
+	otelExecMode := otelmanager.GetExecutionModeFromConfig(rawConfig)
+	isOtelSubprocessExecution := otelExecMode == otelmanager.SubprocessExecutionMode
+
 	// monitoring is not supported in bootstrap mode https://github.com/elastic/elastic-agent/issues/1761
 	isMonitoringSupported := !disableMonitoring && cfg.Settings.V1MonitoringEnabled
 	upgrader, err := upgrade.NewUpgrader(log, cfg.Settings.DownloadConfig, agentInfo)
 	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)
+	monitor := monitoring.New(isMonitoringSupported, cfg.Settings.DownloadConfig.OS(), cfg.Settings.MonitoringConfig, agentInfo, isOtelSubprocessExecution)
 
 	runtime, err := runtime.NewManager(
 		log,
@@ -240,7 +243,7 @@ func New(
 		return nil, nil, nil, errors.New(err, "failed to initialize composable controller")
 	}
 
-	otelManager, err := otelmanager.NewOTelManager(log.Named("otel_manager"), logLevel, baseLogger, otelmanager.EmbeddedExecutionMode, agentInfo, monitor.ComponentMonitoringConfig, cfg.Settings.ProcessConfig.StopTimeout)
+	otelManager, err := otelmanager.NewOTelManager(log.Named("otel_manager"), logLevel, baseLogger, otelExecMode, agentInfo, monitor.ComponentMonitoringConfig, cfg.Settings.ProcessConfig.StopTimeout)
 	if err != nil {
 		return nil, nil, nil, fmt.Errorf("failed to create otel manager: %w", err)
 	}

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

@@ -58,6 +58,7 @@ const (
 	monitoringOutput           = "monitoring"
 	defaultMonitoringNamespace = "default"
 	agentName                  = "elastic-agent"
+	edotCollectorName          = "elastic-agent/collector"
 	metricBeatName             = "metricbeat"
 	fileBeatName               = "filebeat"
 
@@ -73,6 +74,9 @@ const (
 	// metricset stream failure threshold before the stream is marked as DEGRADED
 	// to avoid marking the agent degraded for transient errors, we set the default threshold to 5
 	defaultMetricsStreamFailureThreshold = uint(5)
+
+	beatsMonitoringComponentInfoID = "beat/" + monitoringMetricsUnitID
+	httpMonitoringComponentInfoID  = "http/" + monitoringMetricsUnitID
 )
 
 var (
@@ -84,10 +88,11 @@ var (
 // BeatsMonitor provides config values for monitoring of agent clients (beats, endpoint, etc)
 // by injecting the monitoring config into an existing fleet config
 type BeatsMonitor struct {
-	enabled         bool // feature flag disabling whole v1 monitoring story
-	config          *monitoringConfig
-	operatingSystem string
-	agentInfo       info.Agent
+	enabled                 bool // feature flag disabling whole v1 monitoring story
+	config                  *monitoringConfig
+	operatingSystem         string
+	agentInfo               info.Agent
+	isOtelRuntimeSubprocess bool
 }
 
 // componentInfo is the information necessary to generate monitoring configuration for a component. We don't just use
@@ -106,14 +111,15 @@ type monitoringConfig struct {
 }
 
 // New creates a new BeatsMonitor instance.
-func New(enabled bool, operatingSystem string, cfg *monitoringCfg.MonitoringConfig, agentInfo info.Agent) *BeatsMonitor {
+func New(enabled bool, operatingSystem string, cfg *monitoringCfg.MonitoringConfig, agentInfo info.Agent, isOtelRuntimeSubprocess bool) *BeatsMonitor {
 	return &BeatsMonitor{
 		enabled: enabled,
 		config: &monitoringConfig{
 			C: cfg,
 		},
-		operatingSystem: operatingSystem,
-		agentInfo:       agentInfo,
+		operatingSystem:         operatingSystem,
+		agentInfo:               agentInfo,
+		isOtelRuntimeSubprocess: isOtelRuntimeSubprocess,
 	}
 }
 
@@ -418,12 +424,12 @@ func (b *BeatsMonitor) getComponentInfos(components []component.Component, compo
 	if b.config.C.MonitorMetrics {
 		componentInfos = append(componentInfos,
 			componentInfo{
-				ID:             fmt.Sprintf("beat/%s", monitoringMetricsUnitID),
+				ID:             beatsMonitoringComponentInfoID,
 				BinaryName:     metricBeatName,
 				RuntimeManager: component.RuntimeManager(b.config.C.RuntimeManager),
 			},
 			componentInfo{
-				ID:             fmt.Sprintf("http/%s", monitoringMetricsUnitID),
+				ID:             httpMonitoringComponentInfoID,
 				BinaryName:     metricBeatName,
 				RuntimeManager: component.RuntimeManager(b.config.C.RuntimeManager),
 			})
@@ -661,20 +667,28 @@ func (b *BeatsMonitor) getHttpStreams(
 		"namespace":  "agent",
 		"period":     metricsCollectionIntervalString,
 		"index":      indexName,
-		"processors": processorsForAgentHttpStream(monitoringNamespace, dataset, b.agentInfo),
+		"processors": processorsForAgentHttpStream(agentName, agentName, agentName, monitoringNamespace, dataset, b.agentInfo),
 	}
 	if failureThreshold != nil {
 		agentStream[failureThresholdKey] = *failureThreshold
 	}
 	httpStreams = append(httpStreams, agentStream)
 
+	var edotSubprocessEndpoints []interface{}
+
 	for _, compInfo := range componentInfos {
 		binaryName := compInfo.BinaryName
 		if !isSupportedMetricsBinary(binaryName) {
 			continue
 		}
 
 		endpoints := []interface{}{PrefixedEndpoint(BeatsMonitoringEndpoint(compInfo.ID))}
+		if compInfo.RuntimeManager == component.OtelRuntimeManager && compInfo.ID == httpMonitoringComponentInfoID && b.isOtelRuntimeSubprocess {
+			// when Otel runtime is running as subprocess, we need to monitor it as a separate stream, thus we utilise the already
+			// exposed endpoint of httpMonitoringComponentInfoID.
+			edotSubprocessEndpoints = endpoints
+		}
+
 		name := sanitizeName(binaryName)
 
 		// Do not create http streams if runtime-manager is otel and binary is of beat type
@@ -727,6 +741,30 @@ func (b *BeatsMonitor) getHttpStreams(
 		}
 	}
 
+	if edotSubprocessEndpoints != nil {
+		// Otel runtime subprocess metrics are collected using the same processors as the elastic-agent since we want only the
+		// system resources.
+		edotSubprocessStream := map[string]any{
+			idKey: fmt.Sprintf("%s-edot-collector", monitoringMetricsUnitID),
+			"data_stream": map[string]interface{}{
+				"type":      "metrics",
+				"dataset":   dataset,
+				"namespace": monitoringNamespace,
+			},
+			"metricsets": []interface{}{"json"},
+			"path":       "/stats",
+			"hosts":      edotSubprocessEndpoints,
+			"namespace":  "agent",
+			"period":     metricsCollectionIntervalString,
+			"index":      indexName,
+			"processors": processorsForAgentHttpStream(agentName, edotCollectorName, edotCollectorName, monitoringNamespace, dataset, b.agentInfo),
+		}
+		if failureThreshold != nil {
+			edotSubprocessStream[failureThresholdKey] = *failureThreshold
+		}
+		httpStreams = append(httpStreams, edotSubprocessStream)
+	}
+
 	return httpStreams
 }
 
@@ -933,28 +971,28 @@ func processorsForHttpStream(binaryName, unitID, dataset string, agentInfo info.
 }
 
 // processorsForAgentHttpStream returns the processors used for the agent metric stream in the beats input.
-func processorsForAgentHttpStream(namespace, dataset string, agentInfo info.Agent) []any {
+func processorsForAgentHttpStream(binaryName, processName, unitID, namespace, dataset string, agentInfo info.Agent) []any {
 	return []interface{}{
 		addDataStreamFieldsProcessor(dataset, namespace),
 		addEventFieldsProcessor(dataset),
-		addElasticAgentFieldsProcessor(agentName, agentInfo),
+		addElasticAgentFieldsProcessor(processName, agentInfo),
 		addAgentFieldsProcessor(agentInfo.AgentID()),
 		addCopyFieldsProcessor(httpCopyRules(), true, false),
 		dropFieldsProcessor([]any{"http"}, true),
-		addComponentFieldsProcessor(agentName, agentName),
+		addComponentFieldsProcessor(binaryName, unitID),
 	}
 }
 
 // addElasticAgentFieldsProcessor returns a processor definition that adds agent information in an `elastic_agent` field.
-func addElasticAgentFieldsProcessor(binaryName string, agentInfo info.Agent) map[string]any {
+func addElasticAgentFieldsProcessor(processName string, agentInfo info.Agent) map[string]any {
 	return map[string]any{
 		"add_fields": map[string]any{
 			"target": "elastic_agent",
 			"fields": map[string]any{
 				"id":       agentInfo.AgentID(),
 				"version":  agentInfo.Version(),
 				"snapshot": agentInfo.Snapshot(),
-				"process":  binaryName,
+				"process":  processName,
 			},
 		},
 	}

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

@@ -1066,7 +1066,7 @@ func TestMonitorReload(t *testing.T) {
 	monitorcfg.MonitorLogs = false
 	monitorcfg.MonitorMetrics = false
 
-	beatsMonitor := New(true, "", monitorcfg, nil)
+	beatsMonitor := New(true, "", monitorcfg, nil, false)
 	assert.Equal(t, beatsMonitor.config.C.MonitorLogs, false)
 	assert.Equal(t, beatsMonitor.config.C.MonitorLogs, false)
 

internal/pkg/agent/cmd/inspect.go

@@ -29,6 +29,7 @@ import (
 	"github.com/elastic/elastic-agent/internal/pkg/config"
 	"github.com/elastic/elastic-agent/internal/pkg/config/operations"
 	"github.com/elastic/elastic-agent/internal/pkg/diagnostics"
+	otelmanager "github.com/elastic/elastic-agent/internal/pkg/otel/manager"
 	"github.com/elastic/elastic-agent/pkg/component"
 	"github.com/elastic/elastic-agent/pkg/core/logger"
 	"github.com/elastic/elastic-agent/pkg/utils"
@@ -409,7 +410,8 @@ func getMonitoringFn(ctx context.Context, cfg map[string]interface{}) (component
 		return nil, fmt.Errorf("could not load agent info: %w", err)
 	}
 
-	monitor := monitoring.New(agentCfg.Settings.V1MonitoringEnabled, agentCfg.Settings.DownloadConfig.OS(), agentCfg.Settings.MonitoringConfig, agentInfo)
+	isOtelSubprocessExecution := otelmanager.IsSubprocessExecution(config)
+	monitor := monitoring.New(agentCfg.Settings.V1MonitoringEnabled, agentCfg.Settings.DownloadConfig.OS(), agentCfg.Settings.MonitoringConfig, agentInfo, isOtelSubprocessExecution)
 	return monitor.MonitoringConfig, nil
 }
 

internal/pkg/otel/manager/manager.go

@@ -17,9 +17,11 @@ import (
 
 	"github.com/elastic/elastic-agent-client/v7/pkg/client"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/application/info"
+	"github.com/elastic/elastic-agent/internal/pkg/config"
 	"github.com/elastic/elastic-agent/internal/pkg/otel/translate"
 	"github.com/elastic/elastic-agent/pkg/component"
 	"github.com/elastic/elastic-agent/pkg/component/runtime"
+	"github.com/elastic/elastic-agent/pkg/features"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/status"
 	"go.opentelemetry.io/collector/confmap"
@@ -109,6 +111,25 @@ type OTelManager struct {
 	stopTimeout time.Duration
 }
 
+// IsSubprocessExecution returns true if the OTel runtime is running in subprocess mode.
+func IsSubprocessExecution(cfg *config.Config) bool {
+	return GetExecutionModeFromConfig(cfg) == SubprocessExecutionMode
+}
+
+// GetExecutionModeFromConfig sets the execution mode of the OTel runtime based on the config.
+func GetExecutionModeFromConfig(cfg *config.Config) ExecutionMode {
+	flags, err := features.Parse(cfg)
+	if err != nil {
+		return EmbeddedExecutionMode
+	}
+
+	if flags.OtelSubprocessExecution() {
+		return SubprocessExecutionMode
+	} else {
+		return EmbeddedExecutionMode
+	}
+}
+
 // NewOTelManager returns a OTelManager.
 func NewOTelManager(
 	logger *logger.Logger,

pkg/features/features.go

@@ -21,9 +21,13 @@ import (
 // 8.11+ - default is enabled
 const defaultTamperProtection = true
 
+// The default value of otel runtime subprocess execution mode if the flag is missing
+const defaultOtelSubprocessMode = false
+
 var (
 	current = Flags{
-		tamperProtection: defaultTamperProtection,
+		tamperProtection:        defaultTamperProtection,
+		otelSubprocessExecution: defaultOtelSubprocessMode,
 	}
 )
 
@@ -36,7 +40,8 @@ type Flags struct {
 	fqdn          bool
 	fqdnCallbacks map[string]BoolValueOnChangeCallback
 
-	tamperProtection bool
+	tamperProtection        bool
+	otelSubprocessExecution bool
 }
 
 type cfg struct {
@@ -48,6 +53,9 @@ type cfg struct {
 			TamperProtection *struct {
 				Enabled bool `json:"enabled" yaml:"enabled" config:"enabled"`
 			} `json:"tamper_protection,omitempty" yaml:"tamper_protection,omitempty" config:"tamper_protection,omitempty"`
+			Otel *struct {
+				SubprocessExecution bool `json:"subprocess_execution" yaml:"subprocess_execution" config:"subprocess_execution"`
+			} `json:"otel,omitempty" yaml:"otel,omitempty" config:"otel,omitempty"`
 		} `json:"features" yaml:"features" config:"features"`
 	} `json:"agent" yaml:"agent" config:"agent"`
 }
@@ -66,6 +74,13 @@ func (f *Flags) TamperProtection() bool {
 	return f.tamperProtection
 }
 
+func (f *Flags) OtelSubprocessExecution() bool {
+	f.mu.RLock()
+	defer f.mu.RUnlock()
+
+	return f.otelSubprocessExecution
+}
+
 func (f *Flags) AsProto() *proto.Features {
 	return &proto.Features{
 		Fqdn: &proto.FQDNFeature{
@@ -121,6 +136,14 @@ func (f *Flags) setTamperProtection(newValue bool) {
 	f.tamperProtection = newValue
 }
 
+// setOtelSubprocessExecution sets the value of the OtelSubprocessExecution flag in Flags.
+func (f *Flags) setOtelSubprocessExecution(newValue bool) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+
+	f.otelSubprocessExecution = newValue
+}
+
 // setSource sets the source from he given cfg.
 func (f *Flags) setSource(c cfg) error {
 	// Use JSON marshalling-unmarshalling to convert cfg to mapstr
@@ -186,6 +209,12 @@ func Parse(policy any) (*Flags, error) {
 		flags.setTamperProtection(defaultTamperProtection)
 	}
 
+	if parsedFlags.Agent.Features.Otel != nil {
+		flags.setOtelSubprocessExecution(parsedFlags.Agent.Features.Otel.SubprocessExecution)
+	} else {
+		flags.setOtelSubprocessExecution(defaultOtelSubprocessMode)
+	}
+
 	if err := flags.setSource(parsedFlags); err != nil {
 		return nil, fmt.Errorf("error creating feature flags source: %w", err)
 	}
@@ -208,6 +237,7 @@ func Apply(c *config.Config) error {
 
 	current.setFQDN(parsed.FQDN())
 	current.setTamperProtection(parsed.TamperProtection())
+	current.setOtelSubprocessExecution(parsed.OtelSubprocessExecution())
 	return err
 }
 
@@ -220,3 +250,8 @@ func FQDN() bool {
 func TamperProtection() bool {
 	return current.TamperProtection()
 }
+
+// OtelSubprocessExecution reports if otel subprocess execution feature is enabled
+func OtelSubprocessExecution() bool {
+	return current.OtelSubprocessExecution()
+}