added a simple end-to-end pipeline test (#92)

* added a simple end-to-end pipeline test

* fixed data on copyright

* removed extra print statements

* combined pipeline tests into single file

* used data from file in repository

Author: KalmanMeth

pkg/pipeline/decode/decode_json.go

@@ -23,12 +23,13 @@ import (
 	log "github.com/sirupsen/logrus"
 )
 
-type decodeJson struct {
+type DecodeJson struct {
+	PrevRecords []interface{}
 }
 
 // Decode decodes input strings to a list of flow entries
 // All entries should be saved as strings
-func (c *decodeJson) Decode(in []interface{}) []config.GenericMap {
+func (c *DecodeJson) Decode(in []interface{}) []config.GenericMap {
 	out := make([]config.GenericMap, 0)
 	for _, line := range in {
 		log.Debugf("decodeJson: line = %v", line)
@@ -48,11 +49,12 @@ func (c *decodeJson) Decode(in []interface{}) []config.GenericMap {
 		}
 		out = append(out, decodedLine2)
 	}
+	c.PrevRecords = in
 	return out
 }
 
 // NewDecodeJson create a new decode
 func NewDecodeJson() (Decoder, error) {
 	log.Debugf("entering NewDecodeJson")
-	return &decodeJson{}, nil
+	return &DecodeJson{}, nil
 }

pkg/pipeline/decode/decode_json_test.go

@@ -31,7 +31,7 @@ func initNewDecodeJson(t *testing.T) Decoder {
 
 func TestDecodeJson(t *testing.T) {
 	newDecode := initNewDecodeJson(t)
-	decodeJson := newDecode.(*decodeJson)
+	decodeJson := newDecode.(*DecodeJson)
 	inputString1 := "{\"varInt\": 12, \"varString\":\"testString\", \"varBool\":false}"
 	inputString2 := "{\"varInt\": 14, \"varString\":\"testString2\", \"varBool\":true}"
 	inputString3 := "{}"
@@ -55,7 +55,7 @@ func TestDecodeJson(t *testing.T) {
 
 func TestDecodeJsonTimestamps(t *testing.T) {
 	newDecode := initNewDecodeJson(t)
-	decodeJson := newDecode.(*decodeJson)
+	decodeJson := newDecode.(*DecodeJson)
 	inputString1 := "{\"unixTime\": 1645104030 }"
 	var in []interface{}
 	var out []config.GenericMap

pkg/pipeline/ingest/ingest_file.go

@@ -27,15 +27,16 @@ import (
 	"time"
 )
 
-type ingestFile struct {
-	fileName string
-	exitChan chan bool
+type IngestFile struct {
+	fileName    string
+	exitChan    chan bool
+	PrevRecords []interface{}
 }
 
 const delaySeconds = 10
 
 // Ingest ingests entries from a file and resends the same data every delaySeconds seconds
-func (r *ingestFile) Ingest(process ProcessFunction) {
+func (r *IngestFile) Ingest(process ProcessFunction) {
 	lines := make([]interface{}, 0)
 	file, err := os.Open(r.fileName)
 	if err != nil {
@@ -54,6 +55,7 @@ func (r *ingestFile) Ingest(process ProcessFunction) {
 	log.Debugf("Ingesting %d log lines from %s", len(lines), r.fileName)
 	switch config.Opt.PipeLine.Ingest.Type {
 	case "file":
+		r.PrevRecords = lines
 		process(lines)
 	case "file_loop":
 		// loop forever
@@ -65,6 +67,7 @@ func (r *ingestFile) Ingest(process ProcessFunction) {
 				return
 			case <-ticker.C:
 				log.Debugf("ingestFile; for loop; before process")
+				r.PrevRecords = lines
 				process(lines)
 			}
 		}
@@ -82,7 +85,7 @@ func NewIngestFile() (Ingester, error) {
 
 	ch := make(chan bool, 1)
 	utils.RegisterExitChannel(ch)
-	return &ingestFile{
+	return &IngestFile{
 		fileName: config.Opt.PipeLine.Ingest.File.Filename,
 		exitChan: ch,
 	}, nil

pkg/pipeline/pipeline_test.go

@@ -18,9 +18,13 @@
 package pipeline
 
 import (
+	"github.com/json-iterator/go"
 	"github.com/netobserv/flowlogs2metrics/pkg/config"
+	"github.com/netobserv/flowlogs2metrics/pkg/pipeline/decode"
+	"github.com/netobserv/flowlogs2metrics/pkg/pipeline/ingest"
 	"github.com/netobserv/flowlogs2metrics/pkg/pipeline/transform"
 	"github.com/netobserv/flowlogs2metrics/pkg/pipeline/write"
+	"github.com/netobserv/flowlogs2metrics/pkg/test"
 	"github.com/stretchr/testify/require"
 	"testing"
 )
@@ -37,3 +41,67 @@ func Test_transformToLoki(t *testing.T) {
 	loki.Write(transformed)
 	require.NoError(t, err)
 }
+
+const configTemplate = `---
+log-level: debug
+pipeline:
+  ingest:
+    type: file
+    file:
+      filename: ../../hack/examples/ocp-ipfix-flowlogs.json
+  decode:
+    type: json
+  transform:
+    - type: generic
+      generic:
+        - input: Bytes
+          output: fl2m_bytes
+        - input: DstAddr
+          output: fl2m_dstAddr
+        - input: DstPort
+          output: fl2m_dstPort
+        - input: Packets
+          output: fl2m_packets
+        - input: SrcAddr
+          output: fl2m_srcAddr
+        - input: SrcPort
+          output: fl2m_srcPort
+  extract:
+    type: none
+  encode:
+    type: none
+  write:
+    type: none
+`
+
+func Test_SimplePipeline(t *testing.T) {
+	var json = jsoniter.ConfigCompatibleWithStandardLibrary
+	var mainPipeline *Pipeline
+	var err error
+	var b []byte
+	v := test.InitConfig(t, configTemplate)
+	config.Opt.PipeLine.Ingest.Type = "file"
+	config.Opt.PipeLine.Decode.Type = "json"
+	config.Opt.PipeLine.Extract.Type = "none"
+	config.Opt.PipeLine.Encode.Type = "none"
+	config.Opt.PipeLine.Write.Type = "none"
+	config.Opt.PipeLine.Ingest.File.Filename = "../../hack/examples/ocp-ipfix-flowlogs.json"
+
+	val := v.Get("pipeline.transform\n")
+	b, err = json.Marshal(&val)
+	require.NoError(t, err)
+	config.Opt.PipeLine.Transform = string(b)
+
+	mainPipeline, err = NewPipeline()
+	require.NoError(t, err)
+
+	// The file ingester reads the entire file, pushes it down the pipeline, and then exits
+	// So we don't need to run it in a separate go-routine
+	mainPipeline.Run()
+	// What is there left to check? Check length of saved data of each stage in private structure.
+	ingester := mainPipeline.Ingester.(*ingest.IngestFile)
+	decoder := mainPipeline.Decoder.(*decode.DecodeJson)
+	writer := mainPipeline.Writer.(*write.WriteNone)
+	require.Equal(t, len(ingester.PrevRecords), len(decoder.PrevRecords))
+	require.Equal(t, len(ingester.PrevRecords), len(writer.PrevRecords))
+}

pkg/pipeline/write/write.go

@@ -25,18 +25,18 @@ import (
 type Writer interface {
 	Write(in []config.GenericMap) []config.GenericMap
 }
-
-type writeNone struct {
+type WriteNone struct {
+	PrevRecords []config.GenericMap
 }
 
 // Write writes entries
-func (t *writeNone) Write(in []config.GenericMap) []config.GenericMap {
+func (t *WriteNone) Write(in []config.GenericMap) []config.GenericMap {
 	log.Debugf("entering Write none, in = %v", in)
-
+	t.PrevRecords = in
 	return in
 }
 
 // NewWriteNone create a new write
 func NewWriteNone() (Writer, error) {
-	return &writeNone{}, nil
+	return &WriteNone{}, nil
 }

pkg/pipeline/write/write_test.go

@@ -24,13 +24,13 @@ import (
 )
 
 func Test_Write(t *testing.T) {
-	wn := writeNone{}
+	wn := WriteNone{}
 	wn.Write([]config.GenericMap{{"key": "test"}})
 }
 
 func Test_NewWriteNone(t *testing.T) {
 	writer, err := NewWriteNone()
 	require.Nil(t, err)
-	require.Equal(t, writer, &writeNone{})
+	require.Equal(t, writer, &WriteNone{})
 
 }