feat: add standalone monitoring server in supervised EDOT

Author: pkoutsovasilis

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

@@ -30,6 +30,7 @@ import (
 	"github.com/elastic/elastic-agent/internal/pkg/agent/errors"
 	"github.com/elastic/elastic-agent/internal/pkg/config"
 	monitoringCfg "github.com/elastic/elastic-agent/internal/pkg/core/monitoring/config"
+	otelMonitoring "github.com/elastic/elastic-agent/internal/pkg/otel/monitoring"
 )
 
 const (
@@ -58,7 +59,6 @@ const (
 	monitoringOutput           = "monitoring"
 	defaultMonitoringNamespace = "default"
 	agentName                  = "elastic-agent"
-	edotCollectorName          = "elastic-agent/collector"
 	metricBeatName             = "metricbeat"
 	fileBeatName               = "filebeat"
 
@@ -74,9 +74,6 @@ 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 (
@@ -355,6 +352,16 @@ func (b *BeatsMonitor) Prepare(unit string) error {
 	return nil
 }
 
+// Returns true if any component in the list uses the otel runtime.
+func usingOtelRuntime(componentInfos []componentInfo) bool {
+	for _, ci := range componentInfos {
+		if ci.RuntimeManager == component.OtelRuntimeManager {
+			return true
+		}
+	}
+	return false
+}
+
 // Cleanup removes files that were created for monitoring.
 func (b *BeatsMonitor) Cleanup(unit string) error {
 	if !b.Enabled() {
@@ -424,12 +431,12 @@ func (b *BeatsMonitor) getComponentInfos(components []component.Component, compo
 	if b.config.C.MonitorMetrics {
 		componentInfos = append(componentInfos,
 			componentInfo{
-				ID:             beatsMonitoringComponentInfoID,
+				ID:             fmt.Sprintf("beat/%s", monitoringMetricsUnitID),
 				BinaryName:     metricBeatName,
 				RuntimeManager: component.RuntimeManager(b.config.C.RuntimeManager),
 			},
 			componentInfo{
-				ID:             httpMonitoringComponentInfoID,
+				ID:             fmt.Sprintf("http/%s", monitoringMetricsUnitID),
 				BinaryName:     metricBeatName,
 				RuntimeManager: component.RuntimeManager(b.config.C.RuntimeManager),
 			})
@@ -674,7 +681,27 @@ func (b *BeatsMonitor) getHttpStreams(
 	}
 	httpStreams = append(httpStreams, agentStream)
 
-	var edotSubprocessEndpoints []interface{}
+	if usingOtelRuntime(componentInfos) && b.isOtelRuntimeSubprocess {
+		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":      []interface{}{PrefixedEndpoint(otelMonitoring.EDOTMonitoringEndpoint())},
+			"namespace":  "agent",
+			"period":     metricsCollectionIntervalString,
+			"index":      indexName,
+			"processors": processorsForAgentHttpStream(agentName, otelMonitoring.EDOTComponentID, otelMonitoring.EDOTComponentID, monitoringNamespace, dataset, b.agentInfo),
+		}
+		if failureThreshold != nil {
+			edotSubprocessStream[failureThresholdKey] = *failureThreshold
+		}
+		httpStreams = append(httpStreams, edotSubprocessStream)
+	}
 
 	for _, compInfo := range componentInfos {
 		binaryName := compInfo.BinaryName
@@ -683,12 +710,6 @@ func (b *BeatsMonitor) getHttpStreams(
 		}
 
 		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
@@ -741,30 +762,6 @@ 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
 }
 

internal/pkg/agent/cmd/otel.go

@@ -23,6 +23,7 @@ import (
 	"github.com/elastic/elastic-agent/internal/pkg/otel"
 	"github.com/elastic/elastic-agent/internal/pkg/otel/agentprovider"
 	"github.com/elastic/elastic-agent/internal/pkg/otel/manager"
+	"github.com/elastic/elastic-agent/internal/pkg/otel/monitoring"
 	"github.com/elastic/elastic-agent/internal/pkg/release"
 	"github.com/elastic/elastic-agent/pkg/core/logger"
 )
@@ -45,10 +46,14 @@ func newOtelCommandWithArgs(args []string, streams *cli.IOStreams) *cobra.Comman
 			if err != nil {
 				return err
 			}
+			supervisedMonitoringURL, err := cmd.Flags().GetString(manager.OtelSupervisedMonitoringURLFlagName)
+			if err != nil {
+				return err
+			}
 			if err := prepareEnv(); err != nil {
 				return err
 			}
-			return RunCollector(cmd.Context(), cfgFiles, supervised, supervisedLoggingLevel)
+			return RunCollector(cmd.Context(), cfgFiles, supervised, supervisedLoggingLevel, supervisedMonitoringURL)
 		},
 		PreRun: func(c *cobra.Command, args []string) {
 			// hide inherited flags not to bloat help with flags not related to otel
@@ -76,7 +81,7 @@ func hideInheritedFlags(c *cobra.Command) {
 	})
 }
 
-func RunCollector(cmdCtx context.Context, configFiles []string, supervised bool, supervisedLoggingLevel string) error {
+func RunCollector(cmdCtx context.Context, configFiles []string, supervised bool, supervisedLoggingLevel string, supervisedMonitoringURL string) error {
 	settings, err := prepareCollectorSettings(configFiles, supervised, supervisedLoggingLevel)
 	if err != nil {
 		return fmt.Errorf("failed to prepare collector settings: %w", err)
@@ -89,6 +94,15 @@ func RunCollector(cmdCtx context.Context, configFiles []string, supervised bool,
 		service.WaitExecutionDone()
 	}()
 
+	if supervisedMonitoringURL != "" {
+		server, err := monitoring.NewServer(settings.log, supervisedMonitoringURL)
+		if err != nil {
+			return fmt.Errorf("error create monitoring server: %w", err)
+		}
+		server.Start()
+		defer server.Stop()
+	}
+
 	service.BeforeRun()
 	defer service.Cleanup()
 
@@ -102,18 +116,23 @@ func RunCollector(cmdCtx context.Context, configFiles []string, supervised bool,
 	defer cancel()
 	go service.ProcessWindowsControlEvents(stopCollector)
 
-	return otel.Run(ctx, stop, settings)
+	return otel.Run(ctx, stop, settings.otelSettings)
+}
+
+type edotSettings struct {
+	log          *logger.Logger
+	otelSettings *otelcol.CollectorSettings
 }
 
-func prepareCollectorSettings(configFiles []string, supervised bool, supervisedLoggingLevel string) (*otelcol.CollectorSettings, error) {
-	var settings *otelcol.CollectorSettings
+func prepareCollectorSettings(configFiles []string, supervised bool, supervisedLoggingLevel string) (edotSettings, error) {
+	var settings edotSettings
 	if supervised {
 		// add stdin config provider
 		configProvider, err := agentprovider.NewBufferProvider(os.Stdin)
 		if err != nil {
-			return nil, fmt.Errorf("failed to create config provider: %w", err)
+			return settings, fmt.Errorf("failed to create config provider: %w", err)
 		}
-		settings = otel.NewSettings(release.Version(), []string{configProvider.URI()},
+		settings.otelSettings = otel.NewSettings(release.Version(), []string{configProvider.URI()},
 			otel.WithConfigProviderFactory(configProvider.NewFactory()),
 		)
 
@@ -138,20 +157,21 @@ func prepareCollectorSettings(configFiles []string, supervised bool, supervisedL
 
 		l, err := logger.NewFromConfig("edot", defaultCfg, defaultEventLogCfg, false)
 		if err != nil {
-			return nil, fmt.Errorf("failed to create logger: %w", err)
+			return settings, fmt.Errorf("failed to create logger: %w", err)
 		}
+		settings.log = l
 
 		if logLevelSettingErr != nil {
 			l.Warnf("Fallback to default logging level due to: %v", logLevelSettingErr)
 		}
 
-		settings.LoggingOptions = []zap.Option{zap.WrapCore(func(zapcore.Core) zapcore.Core {
+		settings.otelSettings.LoggingOptions = []zap.Option{zap.WrapCore(func(zapcore.Core) zapcore.Core {
 			return l.Core()
 		})}
 
-		settings.DisableGracefulShutdown = false
+		settings.otelSettings.DisableGracefulShutdown = false
 	} else {
-		settings = otel.NewSettings(release.Version(), configFiles)
+		settings.otelSettings = otel.NewSettings(release.Version(), configFiles)
 	}
 	return settings, nil
 }

internal/pkg/agent/cmd/otel_flags.go

@@ -39,6 +39,11 @@ func setupOtelFlags(flags *pflag.FlagSet) {
 	// but look above, so we explicitly ignore it
 	_ = flags.MarkHidden(manager.OtelSupervisedLoggingLevelFlagName)
 
+	flags.String(manager.OtelSupervisedMonitoringURLFlagName, "", "Set the monitoring path")
+	// the only error we can get here is that the flag does not exist
+	// but look above, so we explicitly ignore it
+	_ = flags.MarkHidden(manager.OtelSupervisedMonitoringURLFlagName)
+
 	goFlags := new(flag.FlagSet)
 	featuregate.GlobalRegistry().RegisterFlags(goFlags)
 

internal/pkg/agent/cmd/otel_test.go

@@ -30,23 +30,23 @@ func TestPrepareCollectorSettings(t *testing.T) {
 		settings, err := prepareCollectorSettings(nil, true, "info")
 		require.NoError(t, err, "failed to prepare collector settings")
 		require.NotNil(t, settings, "settings should not be nil")
-		require.NotNil(t, settings.ConfigProviderSettings.ResolverSettings.URIs, "URIs should not be nil")
+		require.NotNil(t, settings.otelSettings.ConfigProviderSettings.ResolverSettings.URIs, "URIs should not be nil")
 		agentProviderURIFound := false
-		for _, uri := range settings.ConfigProviderSettings.ResolverSettings.URIs {
+		for _, uri := range settings.otelSettings.ConfigProviderSettings.ResolverSettings.URIs {
 			agentProviderURIFound = strings.Contains(uri, agentprovider.AgentConfigProviderSchemeName)
 			if agentProviderURIFound {
 				break
 			}
 		}
 		require.True(t, agentProviderURIFound, "agentprovider Scheme not found in the URIS of ConfigProviderSettings")
-		require.NotNil(t, settings.LoggingOptions, "loggingOptions should not be nil for supervised mode")
+		require.NotNil(t, settings.otelSettings.LoggingOptions, "loggingOptions should not be nil for supervised mode")
 	})
 
 	t.Run("returns valid settings in standalone mode", func(t *testing.T) {
 		settings, err := prepareCollectorSettings([]string{"fake-config.yaml"}, false, "info")
 		require.NoError(t, err, "failed to prepare collector settings")
 		require.NotNil(t, settings, "settings should not be nil")
-		require.Contains(t, settings.ConfigProviderSettings.ResolverSettings.URIs, "fake-config.yaml", "fake-config.yaml not found in the URIS of ConfigProviderSettings")
+		require.Contains(t, settings.otelSettings.ConfigProviderSettings.ResolverSettings.URIs, "fake-config.yaml", "fake-config.yaml not found in the URIS of ConfigProviderSettings")
 	})
 
 	t.Run("fails when supervised mode has invalid config from stdin", func(t *testing.T) {
@@ -61,7 +61,7 @@ func TestPrepareCollectorSettings(t *testing.T) {
 
 		settings, err := prepareCollectorSettings(nil, true, "info")
 		require.Error(t, err)
-		require.Nil(t, settings)
+		require.Nil(t, settings.otelSettings)
 	})
 
 	t.Run("doesn't fail when unsupervised mode has invalid config from stdin", func(t *testing.T) {

internal/pkg/otel/manager/execution_subprocess.go

@@ -24,14 +24,16 @@ import (
 
 	"github.com/elastic/elastic-agent-libs/logp"
 
+	"github.com/elastic/elastic-agent/internal/pkg/otel/monitoring"
 	runtimeLogger "github.com/elastic/elastic-agent/pkg/component/runtime"
 	"github.com/elastic/elastic-agent/pkg/core/logger"
 	"github.com/elastic/elastic-agent/pkg/core/process"
 )
 
 const (
-	OtelSetSupervisedFlagName          = "supervised"
-	OtelSupervisedLoggingLevelFlagName = "supervised.logging.level"
+	OtelSetSupervisedFlagName           = "supervised"
+	OtelSupervisedLoggingLevelFlagName  = "supervised.logging.level"
+	OtelSupervisedMonitoringURLFlagName = "supervised.monitoring.url"
 )
 
 func newSubprocessExecution(logLevel logp.Level, collectorPath string) (*subprocessExecution, error) {
@@ -51,6 +53,7 @@ func newSubprocessExecution(logLevel logp.Level, collectorPath string) (*subproc
 			"otel",
 			fmt.Sprintf("--%s", OtelSetSupervisedFlagName),
 			fmt.Sprintf("--%s=%s", OtelSupervisedLoggingLevelFlagName, logLevel.String()),
+			fmt.Sprintf("--%s=%s", OtelSupervisedMonitoringURLFlagName, monitoring.EDOTMonitoringEndpoint()),
 		},
 		logLevel:               logLevel,
 		healthCheckExtensionID: healthCheckExtensionID,

internal/pkg/otel/manager/testing/testing.go

@@ -45,8 +45,10 @@ func main() {
 		})
 	}
 
+	monitoringURL := os.Getenv("TEST_SUPERVISED_COLLECTOR_MONITORING_URL")
+
 	exitCode := 0
-	err = cmd.RunCollector(ctx, nil, true, "debug")
+	err = cmd.RunCollector(ctx, nil, true, "debug", monitoringURL)
 	if err != nil && !errors.Is(err, context.Canceled) {
 		exitCode = 1
 	}