Support for validate subcommand for otel configuration (#4023)

* validate subcommand working

* Validate subcommand for otel mode

* fmt

* Update internal/pkg/agent/cmd/validate.go

Co-authored-by: Shaunak Kashyap <ycombinator@gmail.com>

* otel default config

* fix default config file

* cleaner config

* Update internal/pkg/agent/cmd/validate.go

Co-authored-by: Shaunak Kashyap <ycombinator@gmail.com>

* Moved validate under otel subcommand

* revert common defaults to match main

* validate expects at least one config

* validate integration test checking invalid config

* add wait for otel to TestOtelAPMIngestion:

---------

Co-authored-by: Shaunak Kashyap <ycombinator@gmail.com>

Author: michalpristas

internal/pkg/agent/application/application.go

@@ -52,7 +52,8 @@ func New(
 ) (*coordinator.Coordinator, coordinator.ConfigManager, composable.Controller, error) {
 
 	err := version.InitVersionError()
-	if err != nil {
+	if err != nil && !runAsOtel {
+		// ignore this error when running in otel mode
 		// non-fatal error, log a warning and move on
 		log.With("error.message", err).Warnf("Error initializing version information: falling back to %s", release.Version())
 	}

internal/pkg/agent/cmd/otel.go

@@ -23,13 +23,13 @@ const (
 	setFlagName    = "set"
 )
 
-func newOtelCommandWithArgs(_ []string, _ *cli.IOStreams) *cobra.Command {
+func newOtelCommandWithArgs(args []string, streams *cli.IOStreams) *cobra.Command {
 	cmd := &cobra.Command{
 		Use:   "otel",
 		Short: "Start the Elastic Agent in otel mode",
 		Long:  "This command starts the Elastic Agent in otel mode.",
 		RunE: func(cmd *cobra.Command, _ []string) error {
-			cfgFiles, err := getConfigFiles(cmd)
+			cfgFiles, err := getConfigFiles(cmd, true)
 			if err != nil {
 				return err
 			}
@@ -45,14 +45,17 @@ func newOtelCommandWithArgs(_ []string, _ *cli.IOStreams) *cobra.Command {
 
 	cmd.SetHelpFunc(func(c *cobra.Command, s []string) {
 		hideInheritedFlags(c)
-		c.Parent().HelpFunc()(c, s)
+		c.Root().HelpFunc()(c, s)
 	})
 
 	cmd.Flags().StringArray(configFlagName, []string{}, "Locations to the config file(s), note that only a"+
 		" single location can be set per flag entry e.g. `--config=file:/path/to/first --config=file:path/to/second`.")
 
 	cmd.Flags().StringArray(setFlagName, []string{}, "Set arbitrary component config property. The component has to be defined in the config file and the flag"+
 		" has a higher precedence. Array config properties are overridden and maps are joined. Example --set=processors.batch.timeout=2s")
+
+	cmd.AddCommand(newValidateCommandWithArgs(args, streams))
+
 	return cmd
 }
 

internal/pkg/agent/cmd/otel_flags.go

@@ -13,13 +13,16 @@ import (
 	"github.com/elastic/elastic-agent/internal/pkg/agent/application/paths"
 )
 
-func getConfigFiles(cmd *cobra.Command) ([]string, error) {
+func getConfigFiles(cmd *cobra.Command, useDefault bool) ([]string, error) {
 	configFiles, err := cmd.Flags().GetStringArray(configFlagName)
 	if err != nil {
 		return nil, fmt.Errorf("failed to retrieve config flags: %w", err)
 	}
 
 	if len(configFiles) == 0 {
+		if !useDefault {
+			return nil, fmt.Errorf("at least one config flag must be provided")
+		}
 		configFiles = append(configFiles, paths.OtelConfigFile())
 	}
 

internal/pkg/agent/cmd/testdata/otel/elastic-agent.yml

@@ -0,0 +1,25 @@
+outputs:
+  default:
+    type: elasticsearch
+    hosts: [127.0.0.1:9200]
+    api_key: "example-key"  
+    preset: balanced
+inputs:
+  - type: system/metrics
+    id: unique-system-metrics-input
+    data_stream.namespace: default
+    use_output: default
+    streams:
+      - metricsets:
+        - cpu
+        data_stream.dataset: system.cpu
+      - metricsets:
+        - memory
+        data_stream.dataset: system.memory
+      - metricsets:
+        - network
+        data_stream.dataset: system.network
+      - metricsets:
+        - filesystem
+        data_stream.dataset: system.filesystem
+agent.logging.to_stderr: true

internal/pkg/agent/cmd/testdata/otel/otel.yml

@@ -0,0 +1,25 @@
+receivers:
+  filelog:
+    include: [ /var/log/system.log ]
+    start_at: beginning
+
+processors:
+  resource:
+    attributes:
+    - key: service.name
+      action: insert
+      value: elastic-otel-test
+
+exporters:
+  debug:
+    verbosity: detailed
+    sampling_initial: 10000
+    sampling_thereafter: 10000
+
+service:
+  pipelines:
+    logs:
+      receivers: [filelog]
+      processors: [resource]
+      exporters:
+        - debug

internal/pkg/agent/cmd/validate.go

@@ -0,0 +1,47 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package cmd
+
+import (
+	"context"
+
+	"github.com/spf13/cobra"
+
+	"github.com/elastic/elastic-agent/internal/pkg/cli"
+	"github.com/elastic/elastic-agent/internal/pkg/otel"
+)
+
+func newValidateCommandWithArgs(_ []string, _ *cli.IOStreams) *cobra.Command {
+	cmd := &cobra.Command{
+		Use:           "validate",
+		Short:         "Validates the OpenTelemetry collector configuration without running the collector",
+		SilenceUsage:  true, // do not display usage on error
+		SilenceErrors: true,
+		RunE: func(cmd *cobra.Command, _ []string) error {
+			cfgFiles, err := getConfigFiles(cmd, false)
+			if err != nil {
+				return err
+			}
+			return validateOtelConfig(cmd.Context(), cfgFiles)
+		},
+	}
+
+	cmd.Flags().StringArray(configFlagName, []string{}, "Locations to the config file(s), note that only a"+
+		" single location can be set per flag entry e.g. `--config=file:/path/to/first --config=file:path/to/second`.")
+
+	cmd.Flags().StringArray(setFlagName, []string{}, "Set arbitrary component config property. The component has to be defined in the config file and the flag"+
+		" has a higher precedence. Array config properties are overridden and maps are joined. Example --set=processors.batch.timeout=2s")
+
+	cmd.SetHelpFunc(func(c *cobra.Command, s []string) {
+		hideInheritedFlags(c)
+		c.Root().HelpFunc()(c, s)
+	})
+
+	return cmd
+}
+
+func validateOtelConfig(ctx context.Context, cfgFiles []string) error {
+	return otel.Validate(ctx, cfgFiles)
+}

internal/pkg/agent/cmd/validate_test.go

@@ -0,0 +1,39 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package cmd
+
+import (
+	"context"
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestValidateCommand(t *testing.T) {
+	tt := []struct {
+		Name         string
+		ConfigPaths  []string
+		ExpectingErr bool
+	}{
+		{
+			"otel config",
+			[]string{filepath.Join("testdata", "otel", "otel.yml")},
+			false,
+		},
+		{
+			"agent config",
+			[]string{filepath.Join("testdata", "otel", "elastic-agent.yml")},
+			true,
+		},
+	}
+
+	for _, tc := range tt {
+		t.Run(tc.Name, func(t *testing.T) {
+			err := validateOtelConfig(context.Background(), tc.ConfigPaths)
+			require.Equal(t, tc.ExpectingErr, err != nil)
+		})
+	}
+}

internal/pkg/otel/validate.go

@@ -0,0 +1,27 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package otel
+
+import (
+	"context"
+
+	"go.opentelemetry.io/collector/otelcol"
+
+	"github.com/elastic/elastic-agent/internal/pkg/release"
+)
+
+func Validate(ctx context.Context, configPaths []string) error {
+	settings, err := newSettings(release.Version(), configPaths)
+	if err != nil {
+		return err
+	}
+
+	col, err := otelcol.NewCollector(*settings)
+	if err != nil {
+		return err
+	}
+	return col.DryRun(ctx)
+
+}

testing/integration/otel_test.go

@@ -44,6 +44,22 @@ service:
       exporters:
         - file`)
 
+var fileInvalidOtelConfig = []byte(`receivers:
+  filelog:
+    include: [ "/var/log/system.log", "/var/log/syslog"  ]
+    start_at: beginning
+
+exporters:
+  file:
+    path: ` + fileProcessingFilename + `
+service:
+  pipelines:
+    logs:
+      receivers: [filelog]
+      processors: [nonexistingprocessor]
+      exporters:
+        - file`)
+
 const apmProcessingContent = `2023-06-19 05:20:50 ERROR This is a test error message
 2023-06-20 12:50:00 DEBUG This is a test debug message 2
 2023-06-20 12:51:00 DEBUG This is a test debug message 3
@@ -133,6 +149,8 @@ func TestOtelFileProcessing(t *testing.T) {
 		`"stringValue":"system.log"`, // system.log is being processed
 	})
 
+	validateCommandIsWorking(t, ctx, fixture, tempDir)
+
 	// check `elastic-agent status` returns successfully
 	require.Eventuallyf(t, func() bool {
 		// This will return errors until it connects to the agent,
@@ -176,6 +194,25 @@ func TestOtelFileProcessing(t *testing.T) {
 	require.True(t, err == nil || err == context.Canceled || err == context.DeadlineExceeded, "Retrieved unexpected error: %s", err.Error())
 }
 
+func validateCommandIsWorking(t *testing.T, ctx context.Context, fixture *aTesting.Fixture, tempDir string) {
+	cfgFilePath := filepath.Join(tempDir, "otel-valid.yml")
+	require.NoError(t, os.WriteFile(cfgFilePath, []byte(fileProcessingConfig), 0600))
+
+	// check `elastic-agent otel validate` command works for otel config
+	out, err := fixture.Exec(ctx, []string{"otel", "validate", "--config", cfgFilePath})
+	require.NoError(t, err)
+	require.Equal(t, 0, len(out)) // no error printed out
+
+	// check `elastic-agent otel validate` command works for invalid otel config
+	cfgFilePath = filepath.Join(tempDir, "otel-invalid.yml")
+	require.NoError(t, os.WriteFile(cfgFilePath, []byte(fileInvalidOtelConfig), 0600))
+
+	out, err = fixture.Exec(ctx, []string{"otel", "validate", "--config", cfgFilePath})
+	require.Error(t, err)
+	require.False(t, len(out) == 0)
+	require.Contains(t, string(out), `service::pipelines::logs: references processor "nonexistingprocessor" which is not configured`)
+}
+
 func TestOtelAPMIngestion(t *testing.T) {
 	info := define.Require(t, define.Requirements{
 		Group: Default,
@@ -261,12 +298,30 @@ func TestOtelAPMIngestion(t *testing.T) {
 		fixtureWg.Done()
 	}()
 
+	// wait for apm to start
 	err = logWatcher.WaitForKeys(context.Background(),
 		10*time.Minute,
 		500*time.Millisecond,
 		apmReadyLog,
 	)
 	require.NoError(t, err, "APM not initialized")
+
+	// wait for otel collector to start
+	require.Eventuallyf(t, func() bool {
+		// This will return errors until it connects to the agent,
+		// they're mostly noise because until the agent starts running
+		// we will get connection errors. If the test fails
+		// the agent logs will be present in the error message
+		// which should help to explain why the agent was not
+		// healthy.
+		err = fixture.IsHealthy(ctx)
+		return err == nil
+	},
+		2*time.Minute, time.Second,
+		"Elastic-Agent did not report healthy. Agent status error: \"%v\"",
+		err,
+	)
+
 	require.NoError(t, os.WriteFile(filepath.Join(tempDir, fileName), []byte(apmProcessingContent), 0600))
 
 	// check index