[chore] move service pipeline id validation to main config Validate (#6281)

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

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

Author: bogdandrutu

config/moved_config.go

@@ -113,6 +113,10 @@ func (cfg *Config) validateService() error {
 	// Check that all pipelines have at least one receiver and one exporter, and they reference
 	// only configured components.
 	for pipelineID, pipeline := range cfg.Service.Pipelines {
+		if pipelineID.Type() != TracesDataType && pipelineID.Type() != MetricsDataType && pipelineID.Type() != LogsDataType {
+			return fmt.Errorf("unknown pipeline datatype %q for %v", pipelineID.Type(), pipelineID)
+		}
+
 		// Validate pipeline has at least one receiver.
 		if len(pipeline.Receivers) == 0 {
 			return fmt.Errorf("pipeline %q must have at least one receiver", pipelineID)

service/config_test.go

@@ -228,6 +228,19 @@ func TestConfigValidate(t *testing.T) {
 			},
 			expected: fmt.Errorf(`extension "nop" has invalid configuration: %w`, errInvalidExtConfig),
 		},
+		{
+			name: "invalid-service-pipeline-type",
+			cfgFn: func() *Config {
+				cfg := generateConfig()
+				cfg.Service.Pipelines[config.NewComponentID("wrongtype")] = &ConfigServicePipeline{
+					Receivers:  []config.ComponentID{config.NewComponentID("nop")},
+					Processors: []config.ComponentID{config.NewComponentID("nop")},
+					Exporters:  []config.ComponentID{config.NewComponentID("nop")},
+				}
+				return cfg
+			},
+			expected: errors.New(`unknown pipeline datatype "wrongtype" for wrongtype`),
+		},
 	}
 
 	for _, test := range testCases {

service/internal/configunmarshaler/defaultunmarshaler.go

@@ -40,7 +40,6 @@ const (
 	errUnmarshalReceiver
 	errUnmarshalProcessor
 	errUnmarshalExporter
-	errUnmarshalService
 )
 
 type configError struct {
@@ -64,17 +63,14 @@ const (
 
 	// processorsKeyName is the configuration key name for processors section.
 	processorsKeyName = "processors"
-
-	// pipelinesKeyName is the configuration key name for pipelines section.
-	pipelinesKeyName = "pipelines"
 )
 
 type configSettings struct {
 	Receivers  map[config.ComponentID]map[string]interface{} `mapstructure:"receivers"`
 	Processors map[config.ComponentID]map[string]interface{} `mapstructure:"processors"`
 	Exporters  map[config.ComponentID]map[string]interface{} `mapstructure:"exporters"`
 	Extensions map[config.ComponentID]map[string]interface{} `mapstructure:"extensions"`
-	Service    map[string]interface{}                        `mapstructure:"service"`
+	Service    config.Service                                `mapstructure:"service"`
 }
 
 type ConfigUnmarshaler struct{}
@@ -88,17 +84,36 @@ func New() ConfigUnmarshaler {
 // Unmarshal the config.Config from a confmap.Conf.
 // After the config is unmarshalled, `Validate()` must be called to validate.
 func (ConfigUnmarshaler) Unmarshal(v *confmap.Conf, factories component.Factories) (*config.Config, error) {
-	var cfg config.Config
-
 	// Unmarshal top level sections and validate.
-	rawCfg := configSettings{}
+	rawCfg := configSettings{
+		// TODO: Add a component.ServiceFactory to allow this to be defined by the Service.
+		Service: config.Service{
+			Telemetry: telemetry.Config{
+				Logs: telemetry.LogsConfig{
+					Level:             zapcore.InfoLevel,
+					Development:       false,
+					Encoding:          "console",
+					OutputPaths:       []string{"stderr"},
+					ErrorOutputPaths:  []string{"stderr"},
+					DisableCaller:     false,
+					DisableStacktrace: false,
+					InitialFields:     map[string]interface{}(nil),
+				},
+				Metrics: telemetry.MetricsConfig{
+					Level:   configtelemetry.LevelBasic,
+					Address: ":8888",
+				},
+			},
+		},
+	}
 	if err := v.Unmarshal(&rawCfg, confmap.WithErrorUnused()); err != nil {
 		return nil, configError{
 			error: fmt.Errorf("error reading top level configuration sections: %w", err),
 			code:  errUnmarshalTopLevelStructure,
 		}
 	}
 
+	var cfg config.Config
 	var err error
 	if cfg.Extensions, err = unmarshalExtensions(rawCfg.Extensions, factories.Extensions); err != nil {
 		return nil, configError{
@@ -128,12 +143,7 @@ func (ConfigUnmarshaler) Unmarshal(v *confmap.Conf, factories component.Factorie
 		}
 	}
 
-	if cfg.Service, err = unmarshalService(rawCfg.Service); err != nil {
-		return nil, configError{
-			error: err,
-			code:  errUnmarshalService,
-		}
-	}
+	cfg.Service = rawCfg.Service
 
 	return &cfg, nil
 }
@@ -166,44 +176,6 @@ func unmarshalExtensions(exts map[config.ComponentID]map[string]interface{}, fac
 	return extensions, nil
 }
 
-func unmarshalService(srvRaw map[string]interface{}) (config.Service, error) {
-	// Setup default telemetry values as in service/logger.go.
-	// TODO: Add a component.ServiceFactory to allow this to be defined by the Service.
-	srv := config.Service{
-		Telemetry: telemetry.Config{
-			Logs: telemetry.LogsConfig{
-				Level:             zapcore.InfoLevel,
-				Development:       false,
-				Encoding:          "console",
-				OutputPaths:       []string{"stderr"},
-				ErrorOutputPaths:  []string{"stderr"},
-				DisableCaller:     false,
-				DisableStacktrace: false,
-				InitialFields:     map[string]interface{}(nil),
-			},
-			Metrics: defaultServiceTelemetryMetricsSettings(),
-		},
-	}
-
-	if err := confmap.NewFromStringMap(srvRaw).Unmarshal(&srv, confmap.WithErrorUnused()); err != nil {
-		return srv, fmt.Errorf("error reading service configuration: %w", err)
-	}
-
-	for id := range srv.Pipelines {
-		if id.Type() != config.TracesDataType && id.Type() != config.MetricsDataType && id.Type() != config.LogsDataType {
-			return srv, fmt.Errorf("unknown %q datatype %q for %v", pipelinesKeyName, id.Type(), id)
-		}
-	}
-	return srv, nil
-}
-
-func defaultServiceTelemetryMetricsSettings() telemetry.MetricsConfig {
-	return telemetry.MetricsConfig{
-		Level:   configtelemetry.LevelBasic,
-		Address: ":8888",
-	}
-}
-
 // LoadReceiver loads a receiver config from componentConfig using the provided factories.
 func LoadReceiver(componentConfig *confmap.Conf, id config.ComponentID, factory component.ReceiverFactory) (config.Receiver, error) {
 	// Create the default config for this receiver.

service/internal/configunmarshaler/defaultunmarshaler_test.go

@@ -107,44 +107,43 @@ func TestDecodeConfig_Invalid(t *testing.T) {
 		{name: "invalid-receiver-type", expected: errUnmarshalTopLevelStructure},
 		{name: "invalid-processor-type", expected: errUnmarshalTopLevelStructure},
 		{name: "invalid-exporter-type", expected: errUnmarshalTopLevelStructure},
-		{name: "invalid-pipeline-type", expected: errUnmarshalService},
+		{name: "invalid-pipeline-type", expected: errUnmarshalTopLevelStructure},
 
 		{name: "invalid-extension-name-after-slash", expected: errUnmarshalTopLevelStructure},
 		{name: "invalid-receiver-name-after-slash", expected: errUnmarshalTopLevelStructure},
 		{name: "invalid-processor-name-after-slash", expected: errUnmarshalTopLevelStructure},
 		{name: "invalid-exporter-name-after-slash", expected: errUnmarshalTopLevelStructure},
-		{name: "invalid-pipeline-name-after-slash", expected: errUnmarshalService},
+		{name: "invalid-pipeline-name-after-slash", expected: errUnmarshalTopLevelStructure},
 
 		{name: "unknown-extension-type", expected: errUnmarshalExtension, expectedMessage: "extensions"},
 		{name: "unknown-receiver-type", expected: errUnmarshalReceiver, expectedMessage: "receivers"},
 		{name: "unknown-processor-type", expected: errUnmarshalProcessor, expectedMessage: "processors"},
 		{name: "unknown-exporter-type", expected: errUnmarshalExporter, expectedMessage: "exporters"},
-		{name: "unknown-pipeline-type", expected: errUnmarshalService, expectedMessage: "pipelines"},
 
 		{name: "duplicate-extension", expected: errUnmarshalTopLevelStructure, expectedMessage: "duplicate name"},
 		{name: "duplicate-receiver", expected: errUnmarshalTopLevelStructure, expectedMessage: "duplicate name"},
 		{name: "duplicate-processor", expected: errUnmarshalTopLevelStructure, expectedMessage: "duplicate name"},
 		{name: "duplicate-exporter", expected: errUnmarshalTopLevelStructure, expectedMessage: "duplicate name"},
-		{name: "duplicate-pipeline", expected: errUnmarshalService, expectedMessage: "duplicate name"},
+		{name: "duplicate-pipeline", expected: errUnmarshalTopLevelStructure, expectedMessage: "duplicate name"},
 
 		{name: "invalid-top-level-section", expected: errUnmarshalTopLevelStructure, expectedMessage: "top level"},
 		{name: "invalid-extension-section", expected: errUnmarshalExtension, expectedMessage: "extensions"},
 		{name: "invalid-receiver-section", expected: errUnmarshalReceiver, expectedMessage: "receivers"},
 		{name: "invalid-processor-section", expected: errUnmarshalProcessor, expectedMessage: "processors"},
 		{name: "invalid-exporter-section", expected: errUnmarshalExporter, expectedMessage: "exporters"},
-		{name: "invalid-service-section", expected: errUnmarshalService},
-		{name: "invalid-service-extensions-section", expected: errUnmarshalService},
-		{name: "invalid-pipeline-section", expected: errUnmarshalService, expectedMessage: "pipelines"},
-		{name: "invalid-sequence-value", expected: errUnmarshalService, expectedMessage: "pipelines"},
+		{name: "invalid-service-section", expected: errUnmarshalTopLevelStructure},
+		{name: "invalid-service-extensions-section", expected: errUnmarshalTopLevelStructure},
+		{name: "invalid-pipeline-section", expected: errUnmarshalTopLevelStructure, expectedMessage: "pipelines"},
+		{name: "invalid-sequence-value", expected: errUnmarshalTopLevelStructure, expectedMessage: "pipelines"},
 
 		{name: "invalid-extension-sub-config", expected: errUnmarshalTopLevelStructure},
 		{name: "invalid-receiver-sub-config", expected: errUnmarshalTopLevelStructure},
 		{name: "invalid-processor-sub-config", expected: errUnmarshalTopLevelStructure},
 		{name: "invalid-exporter-sub-config", expected: errUnmarshalTopLevelStructure},
-		{name: "invalid-pipeline-sub-config", expected: errUnmarshalService},
+		{name: "invalid-pipeline-sub-config", expected: errUnmarshalTopLevelStructure},
 
-		{name: "invalid-logs-level", expected: errUnmarshalService},
-		{name: "invalid-metrics-level", expected: errUnmarshalService},
+		{name: "invalid-logs-level", expected: errUnmarshalTopLevelStructure},
+		{name: "invalid-metrics-level", expected: errUnmarshalTopLevelStructure},
 	}
 
 	factories, err := componenttest.NopFactories()