Warn about deprecated API usage

Author: jotak

contrib/kubernetes/flowlogs-pipeline.conf.yaml

@@ -20,8 +20,8 @@ pipeline:
 parameters:
 - name: ingest_collector
   ingest:
-    type: collector
-    collector:
+    type: ipfix
+    ipfix:
       hostName: 0.0.0.0
       port: 2055
       portLegacy: 2056

hack/examples/docker-ipfix-config.yaml

@@ -19,8 +19,8 @@ pipeline:
 parameters:
   - name: ingest
     ingest: # use nflow generator to simulate flows: ./nflow-generator -t localhost -p 2055
-      type: collector
-      collector:
+      type: ipfix
+      ipfix:
         hostName: localhost
         port: 4739 # Use this for IPFIX / netflow v9
         portLegacy: 2055 # Use this for legacy v5 netflow

pkg/pipeline/pipeline_builder.go

@@ -396,7 +396,10 @@ func getIngester(opMetrics *operational.Metrics, params config.StageParam) (inge
 		ingester, err = ingest.NewIngestFile(params)
 	case api.SyntheticType:
 		ingester, err = ingest.NewIngestSynthetic(opMetrics, params)
-	case api.CollectorType, api.IpfixType:
+	case api.CollectorType:
+		log.Warnf("Stage %s uses deprecated 'collector' API. It should be renamed 'ipfix', as 'collector' will be removed in a future release.", params.Name)
+		ingester, err = ingest.NewIngestIPFIX(opMetrics, params)
+	case api.IpfixType:
 		ingester, err = ingest.NewIngestIPFIX(opMetrics, params)
 	case api.StdinType:
 		ingester, err = ingest.NewIngestStdin(opMetrics, params)

pkg/pipeline/pipeline_builder_test.go

@@ -2,6 +2,7 @@ package pipeline
 
 import (
 	"errors"
+	"reflect"
 	"testing"
 
 	"github.com/netobserv/flowlogs-pipeline/pkg/test"
@@ -30,6 +31,27 @@ func TestConnectionVerification_Pass(t *testing.T) {
 	assert.NoError(t, err)
 }
 
+func TestPipelineIPFIXBackwardCompatible(t *testing.T) {
+	_, cfg := test.InitConfig(t, `
+pipeline:
+  - follows: ingest1
+    name: write1
+parameters:
+  - name: ingest1
+    ingest: 
+      type: collector
+  - name: write1
+    write:
+      type: none
+`)
+	p, err := NewPipeline(cfg)
+	assert.NoError(t, err)
+
+	assert.NotNil(t, p.pipelineStages[0].Ingester)
+	ty := reflect.TypeOf(p.pipelineStages[0].Ingester)
+	assert.Equal(t, "*ingest.ingestIPFIX", ty.String())
+}
+
 func TestConnectionVerification(t *testing.T) {
 	type testCase struct {
 		description     string