Fail on wrong pipeline config with duplicated stage names (#1090)

Fixes #333

Author: jotak

pkg/pipeline/pipeline_builder.go

@@ -161,6 +161,9 @@ func (b *builder) presetIngester(ing ingest.Ingester) {
 func (b *builder) readStages() error {
 	for _, param := range b.configParams {
 		log.Debugf("stage = %v", param.Name)
+		if _, exists := b.pipelineEntryMap[param.Name]; exists {
+			return fmt.Errorf("duplicate stage name '%s' found in pipeline definition", param.Name)
+		}
 		pEntry := pipelineEntry{
 			stageName: param.Name,
 			stageType: findStageType(&param),

pkg/pipeline/pipeline_test.go

@@ -283,3 +283,38 @@ func BenchmarkPipeline(b *testing.B) {
 		p.Run()
 	}
 }
+
+const invalidConfig = `---
+log-level: debug
+pipeline:
+  - name: ingest
+  - name: test
+    follows: ingest
+  - name: test
+    follows: test
+parameters:
+  - name: ingest
+    ingest:
+      type: file
+      file:
+        filename: ../../hack/examples/ocp-ipfix-flowlogs.json
+        decoder:
+          type: json
+  - name: test
+    transform:
+      type: generic
+      generic: {}
+  - name: test
+    write:
+      type: none
+`
+
+func Test_InvalidPipeline_DuplicatedNames(t *testing.T) {
+	var mainPipeline *Pipeline
+	var err error
+	_, cfg := test.InitConfig(t, invalidConfig)
+
+	mainPipeline, err = NewPipeline(cfg)
+	assert.ErrorContains(t, err, "duplicate stage name 'test' found in pipeline definition")
+	assert.Nil(t, mainPipeline)
+}