Extract ConfigFactory in a ParserProvider interface (#2868)

Signed-off-by: Bogdan Drutu <[email protected]>

Author: bogdandrutu

config/configparser/config.go

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-// Package configparser implements loading of configuration from Viper configuration.
+// Package configparser implements configuration loading from a config.Parser.
 // The implementation relies on registered factories that allow creating
 // default configuration for each type of receiver/exporter/processor.
 package configparser

service/application.go

@@ -18,7 +18,6 @@ package service
 
 import (
 	"context"
-	"errors"
 	"flag"
 	"fmt"
 	"os"
@@ -30,13 +29,13 @@ import (
 	"go.uber.org/zap"
 
 	"go.opentelemetry.io/collector/component"
-	"go.opentelemetry.io/collector/config"
 	"go.opentelemetry.io/collector/config/configcheck"
 	"go.opentelemetry.io/collector/config/configparser"
 	"go.opentelemetry.io/collector/config/configtelemetry"
 	"go.opentelemetry.io/collector/consumer/consumererror"
 	"go.opentelemetry.io/collector/internal/collector/telemetry"
 	"go.opentelemetry.io/collector/service/internal/builder"
+	"go.opentelemetry.io/collector/service/parserprovider"
 )
 
 const (
@@ -66,7 +65,7 @@ type Application struct {
 
 	factories component.Factories
 
-	configFactory ConfigFactory
+	parserProvider parserprovider.ParserProvider
 
 	// stopTestChan is used to terminate the application in end to end tests.
 	stopTestChan chan struct{}
@@ -84,58 +83,25 @@ type Parameters struct {
 	Factories component.Factories
 	// ApplicationStartInfo provides application start information.
 	ApplicationStartInfo component.ApplicationStartInfo
-	// ConfigFactory that creates the configuration.
-	// If it is not provided the default factory (FileLoaderConfigFactory) is used.
-	// The default factory loads the configuration file and overrides component's configuration
+	// ParserProvider provides the configuration's Parser.
+	// If it is not provided a default provider is used. The default provider loads the configuration
+	// from a config file define by the --config command line flag and overrides component's configuration
 	// properties supplied via --set command line flag.
-	ConfigFactory ConfigFactory
+	ParserProvider parserprovider.ParserProvider
 	// LoggingOptions provides a way to change behavior of zap logging.
 	LoggingOptions []zap.Option
 }
 
-// ConfigFactory creates config.
-// The ConfigFactory implementation should call AddSetFlagProperties to enable configuration passed via `--set` flag.
-// Viper and command instances are passed from the Application.
-// The factories also belong to the Application and are equal to the factories passed via Parameters.
-type ConfigFactory func(cmd *cobra.Command, factories component.Factories) (*config.Config, error)
-
-// FileLoaderConfigFactory implements ConfigFactory and it creates configuration from file
-// and from --set command line flag (if the flag is present).
-func FileLoaderConfigFactory(cmd *cobra.Command, factories component.Factories) (*config.Config, error) {
-	file := builder.GetConfigFile()
-	if file == "" {
-		return nil, errors.New("config file not specified")
-	}
-
-	cp, err := config.NewParserFromFile(file)
-	if err != nil {
-		return nil, fmt.Errorf("error loading config file %q: %v", file, err)
-	}
-
-	// next overlay the config file with --set flags
-	if err := AddSetFlagProperties(cp, cmd); err != nil {
-		return nil, fmt.Errorf("failed to process set flag: %v", err)
-	}
-	return configparser.Load(cp, factories)
-}
-
 // New creates and returns a new instance of Application.
 func New(params Parameters) (*Application, error) {
 	if err := configcheck.ValidateConfigFromFactories(params.Factories); err != nil {
 		return nil, err
 	}
 
-	configFactory := params.ConfigFactory
-	if configFactory == nil {
-		// use default factory that loads the configuration file
-		configFactory = FileLoaderConfigFactory
-	}
-
 	app := &Application{
-		info:          params.ApplicationStartInfo,
-		factories:     params.Factories,
-		stateChannel:  make(chan State, Closed+1),
-		configFactory: configFactory,
+		info:         params.ApplicationStartInfo,
+		factories:    params.Factories,
+		stateChannel: make(chan State, Closed+1),
 	}
 
 	rootCmd := &cobra.Command{
@@ -159,6 +125,7 @@ func New(params Parameters) (*Application, error) {
 	flagSet := new(flag.FlagSet)
 	addFlagsFns := []func(*flag.FlagSet){
 		configtelemetry.Flags,
+		parserprovider.Flags,
 		telemetry.Flags,
 		builder.Flags,
 		loggerFlags,
@@ -167,10 +134,15 @@ func New(params Parameters) (*Application, error) {
 		addFlags(flagSet)
 	}
 	rootCmd.Flags().AddGoFlagSet(flagSet)
-	addSetFlag(rootCmd.Flags())
-
 	app.rootCmd = rootCmd
 
+	parserProvider := params.ParserProvider
+	if parserProvider == nil {
+		// use default provider.
+		parserProvider = parserprovider.Default()
+	}
+	app.parserProvider = parserProvider
+
 	return app, nil
 }
 
@@ -249,11 +221,20 @@ func (app *Application) runAndWaitForShutdownEvent() {
 func (app *Application) setupConfigurationComponents(ctx context.Context) error {
 	app.logger.Info("Loading configuration...")
 
-	cfg, err := app.configFactory(app.rootCmd, app.factories)
+	cp, err := app.parserProvider.Get()
+	if err != nil {
+		return fmt.Errorf("cannot load configuration's parser: %w", err)
+	}
+
+	cfg, err := configparser.Load(cp, app.factories)
 	if err != nil {
 		return fmt.Errorf("cannot load configuration: %w", err)
 	}
 
+	if err = cfg.Validate(); err != nil {
+		return fmt.Errorf("invalid configuration: %w", err)
+	}
+
 	app.logger.Info("Applying configuration...")
 
 	service, err := newService(&settings{
@@ -339,7 +320,7 @@ func (app *Application) createMemoryBallast() ([]byte, uint64) {
 }
 
 // updateService shutdowns the current app.service and setups a new one according
-// to the latest configuration. It requires that app.configFactory and app.factories
+// to the latest configuration. It requires that app.parserProvider and app.factories
 // are properly populated to finish successfully.
 func (app *Application) updateService(ctx context.Context) error {
 	if app.service != nil {

service/application_test.go

@@ -19,32 +19,25 @@ import (
 	"bufio"
 	"context"
 	"errors"
-	"flag"
 	"fmt"
 	"net/http"
 	"runtime"
-	"sort"
 	"strconv"
 	"strings"
 	"syscall"
 	"testing"
-	"time"
 
 	"github.com/prometheus/common/expfmt"
-	"github.com/spf13/cobra"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
 
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/config"
-	"go.opentelemetry.io/collector/config/configparser"
-	"go.opentelemetry.io/collector/processor/attributesprocessor"
-	"go.opentelemetry.io/collector/processor/batchprocessor"
-	"go.opentelemetry.io/collector/receiver/jaegerreceiver"
 	"go.opentelemetry.io/collector/service/defaultcomponents"
 	"go.opentelemetry.io/collector/service/internal/builder"
+	"go.opentelemetry.io/collector/service/parserprovider"
 	"go.opentelemetry.io/collector/testutil"
 )
 
@@ -149,10 +142,8 @@ func TestApplication_StartAsGoRoutine(t *testing.T) {
 
 	params := Parameters{
 		ApplicationStartInfo: component.DefaultApplicationStartInfo(),
-		ConfigFactory: func(_ *cobra.Command, factories component.Factories) (*config.Config, error) {
-			return constructMimumalOpConfig(t, factories), nil
-		},
-		Factories: factories,
+		ParserProvider:       new(minimalParserLoader),
+		Factories:            factories,
 	}
 	app, err := New(params)
 	require.NoError(t, err)
@@ -225,105 +216,9 @@ func assertMetrics(t *testing.T, prefix string, metricsPort uint16, mandatoryLab
 	}
 }
 
-func TestSetFlag(t *testing.T) {
-	factories, err := defaultcomponents.Components()
-	require.NoError(t, err)
-	params := Parameters{
-		Factories: factories,
-	}
-	t.Run("unknown_component", func(t *testing.T) {
-		app, err := New(params)
-		require.NoError(t, err)
-		err = app.rootCmd.ParseFlags([]string{
-			"--config=testdata/otelcol-config.yaml",
-			"--set=processors.doesnotexist.timeout=2s",
-		})
-		require.NoError(t, err)
-		cfg, err := FileLoaderConfigFactory(app.rootCmd, factories)
-		require.Error(t, err)
-		require.Nil(t, cfg)
-
-	})
-	t.Run("component_not_added_to_pipeline", func(t *testing.T) {
-		app, err := New(params)
-		require.NoError(t, err)
-		err = app.rootCmd.ParseFlags([]string{
-			"--config=testdata/otelcol-config.yaml",
-			"--set=processors.batch/foo.timeout=2s",
-		})
-		require.NoError(t, err)
-		cfg, err := FileLoaderConfigFactory(app.rootCmd, factories)
-		require.NoError(t, err)
-		assert.NotNil(t, cfg)
-		err = cfg.Validate()
-		require.NoError(t, err)
-
-		var processors []string
-		for k := range cfg.Processors {
-			processors = append(processors, k)
-		}
-		sort.Strings(processors)
-		// batch/foo is not added to the pipeline
-		assert.Equal(t, []string{"attributes", "batch", "batch/foo"}, processors)
-		assert.Equal(t, []string{"attributes", "batch"}, cfg.Service.Pipelines["traces"].Processors)
-	})
-	t.Run("ok", func(t *testing.T) {
-		app, err := New(params)
-		require.NoError(t, err)
-
-		err = app.rootCmd.ParseFlags([]string{
-			"--config=testdata/otelcol-config.yaml",
-			"--set=processors.batch.timeout=2s",
-			// Arrays are overridden and object arrays cannot be indexed
-			// this creates actions array of size 1
-			"--set=processors.attributes.actions.key=foo",
-			"--set=processors.attributes.actions.value=bar",
-			"--set=receivers.jaeger.protocols.grpc.endpoint=localhost:12345",
-			"--set=extensions.health_check.endpoint=localhost:8080",
-		})
-		require.NoError(t, err)
-		cfg, err := FileLoaderConfigFactory(app.rootCmd, factories)
-		require.NoError(t, err)
-		require.NotNil(t, cfg)
-		err = cfg.Validate()
-		require.NoError(t, err)
-
-		assert.Equal(t, 2, len(cfg.Processors))
-		batch := cfg.Processors["batch"].(*batchprocessor.Config)
-		assert.Equal(t, time.Second*2, batch.Timeout)
-		jaeger := cfg.Receivers["jaeger"].(*jaegerreceiver.Config)
-		assert.Equal(t, "localhost:12345", jaeger.GRPC.NetAddr.Endpoint)
-		attributes := cfg.Processors["attributes"].(*attributesprocessor.Config)
-		require.Equal(t, 1, len(attributes.Actions))
-		assert.Equal(t, "foo", attributes.Actions[0].Key)
-		assert.Equal(t, "bar", attributes.Actions[0].Value)
-	})
-}
-
-func TestSetFlag_component_does_not_exist(t *testing.T) {
-	factories, err := defaultcomponents.Components()
-	require.NoError(t, err)
-
-	cmd := &cobra.Command{}
-	addSetFlag(cmd.Flags())
-	fs := &flag.FlagSet{}
-	builder.Flags(fs)
-	cmd.Flags().AddGoFlagSet(fs)
-	cmd.ParseFlags([]string{
-		"--config=testdata/otelcol-config.yaml",
-		"--set=processors.batch.timeout=2s",
-		// Arrays are overridden and object arrays cannot be indexed
-		// this creates actions array of size 1
-		"--set=processors.attributes.actions.key=foo",
-		"--set=processors.attributes.actions.value=bar",
-		"--set=receivers.jaeger.protocols.grpc.endpoint=localhost:12345",
-	})
-	cfg, err := FileLoaderConfigFactory(cmd, factories)
-	require.NoError(t, err)
-	require.NotNil(t, cfg)
-}
+type minimalParserLoader struct{}
 
-func constructMimumalOpConfig(t *testing.T, factories component.Factories) *config.Config {
+func (*minimalParserLoader) Get() (*config.Parser, error) {
 	configStr := `
 receivers:
   otlp:
@@ -347,37 +242,36 @@ service:
 	v := config.NewViper()
 	v.SetConfigType("yaml")
 	v.ReadConfig(strings.NewReader(configStr))
-	cfg, err := configparser.Load(config.ParserFromViper(v), factories)
-	assert.NoError(t, err)
-	err = cfg.Validate()
-	assert.NoError(t, err)
-	return cfg
+	return config.ParserFromViper(v), nil
+}
+
+type errParserLoader struct {
+	err error
+}
+
+func (epl *errParserLoader) Get() (*config.Parser, error) {
+	return nil, epl.err
 }
 
 func TestApplication_updateService(t *testing.T) {
 	factories, err := defaultcomponents.Components()
 	require.NoError(t, err)
 	ctx := context.Background()
 	sentinelError := errors.New("sentinel error")
-	returnConfigFactoryFn := func(cfg *config.Config, err error) ConfigFactory {
-		return func(*cobra.Command, component.Factories) (*config.Config, error) {
-			return cfg, err
-		}
-	}
 
 	tests := []struct {
-		name          string
-		configFactory ConfigFactory
-		service       *service
-		skip          bool
+		name           string
+		parserProvider parserprovider.ParserProvider
+		service        *service
+		skip           bool
 	}{
 		{
-			name:          "first_load_err",
-			configFactory: returnConfigFactoryFn(nil, sentinelError),
+			name:           "first_load_err",
+			parserProvider: &errParserLoader{err: sentinelError},
 		},
 		{
-			name:          "retire_service_ok_load_err",
-			configFactory: returnConfigFactoryFn(nil, sentinelError),
+			name:           "retire_service_ok_load_err",
+			parserProvider: &errParserLoader{err: sentinelError},
 			service: &service{
 				logger:          zap.NewNop(),
 				builtExporters:  builder.Exporters{},
@@ -387,8 +281,8 @@ func TestApplication_updateService(t *testing.T) {
 			},
 		},
 		{
-			name:          "retire_service_ok_load_ok",
-			configFactory: returnConfigFactoryFn(constructMimumalOpConfig(t, factories), nil),
+			name:           "retire_service_ok_load_ok",
+			parserProvider: new(minimalParserLoader),
 			service: &service{
 				logger:          zap.NewNop(),
 				builtExporters:  builder.Exporters{},
@@ -407,10 +301,10 @@ func TestApplication_updateService(t *testing.T) {
 			}
 
 			app := Application{
-				logger:        zap.NewNop(),
-				configFactory: tt.configFactory,
-				factories:     factories,
-				service:       tt.service,
+				logger:         zap.NewNop(),
+				parserProvider: tt.parserProvider,
+				factories:      factories,
+				service:        tt.service,
 			}
 
 			err := app.updateService(ctx)