fix(service/pipelines): add duplicate validation for receivers and ex…

service/pipelines/config.go

@@ -77,6 +77,16 @@ func (cfg *PipelineConfig) Validate() error {
 		return errMissingServicePipelineExporters
 	}
 
+	// Validate no receivers are duplicated within a pipeline.
+	recvSet := make(map[component.ID]struct{}, len(cfg.Receivers))
+	for _, ref := range cfg.Receivers {
+		// Ensure no receivers are duplicated within the pipeline
+		if _, exists := recvSet[ref]; exists {
+			return fmt.Errorf("references receiver %q multiple times", ref)
+		}
+		recvSet[ref] = struct{}{}
+	}
+
 	// Validate no processors are duplicated within a pipeline.
 	procSet := make(map[component.ID]struct{}, len(cfg.Processors))
 	for _, ref := range cfg.Processors {
@@ -87,5 +97,15 @@ func (cfg *PipelineConfig) Validate() error {
 		procSet[ref] = struct{}{}
 	}
 
+	// Validate no exporters are duplicated within a pipeline.
+	expSet := make(map[component.ID]struct{}, len(cfg.Exporters))
+	for _, ref := range cfg.Exporters {
+		// Ensure no exporters are duplicated within the pipeline
+		if _, exists := expSet[ref]; exists {
+			return fmt.Errorf("references exporter %q multiple times", ref)
+		}
+		expSet[ref] = struct{}{}
+	}
+
 	return nil
 }

service/pipelines/config_test.go

@@ -27,6 +27,16 @@ func TestConfigValidate(t *testing.T) {
 			cfgFn:    generateConfig,
 			expected: nil,
 		},
+		{
+			name: "duplicate-receiver-reference",
+			cfgFn: func(*testing.T) Config {
+				cfg := generateConfig(t)
+				pipe := cfg[pipeline.NewID(pipeline.SignalTraces)]
+				pipe.Receivers = append(pipe.Receivers, pipe.Receivers...)
+				return cfg
+			},
+			expected: errors.New(`references receiver "nop" multiple times`),
+		},
 		{
 			name: "duplicate-processor-reference",
 			cfgFn: func(*testing.T) Config {
@@ -37,6 +47,16 @@ func TestConfigValidate(t *testing.T) {
 			},
 			expected: errors.New(`references processor "nop" multiple times`),
 		},
+		{
+			name: "duplicate-exporter-reference",
+			cfgFn: func(*testing.T) Config {
+				cfg := generateConfig(t)
+				pipe := cfg[pipeline.NewID(pipeline.SignalTraces)]
+				pipe.Exporters = append(pipe.Exporters, pipe.Exporters...)
+				return cfg
+			},
+			expected: errors.New(`references exporter "nop" multiple times`),
+		},
 		{
 			name: "missing-pipeline-receivers",
 			cfgFn: func(*testing.T) Config {