added encode_json support plus test (#60)

* added encode_json support plus test

* added encode json to README

removed default none from options

Author: KalmanMeth

README.md

@@ -36,18 +36,18 @@ Flags:
   -h, --help                                   help for flowlogs2metrics  
       --log-level string                       Log level: debug, info, warning, error (default "error")  
       --pipeline.decode.aws string             aws fields  
-      --pipeline.decode.type string            Decode type: aws, json, none (default "none")  
+      --pipeline.decode.type string            Decode type: aws, json, none  
       --pipeline.encode.prom string            Prometheus encode API  
-      --pipeline.encode.type string            Encode type: prom, none (default "none")  
+      --pipeline.encode.type string            Encode type: prom, json, none  
       --pipeline.extract.aggregates string     Aggregates (see docs)  
-      --pipeline.extract.type string           Extract type: aggregates, none (default "none")  
+      --pipeline.extract.type string           Extract type: aggregates, none  
       --pipeline.ingest.collector string       Ingest collector API  
       --pipeline.ingest.file.filename string   Ingest filename (file)  
       --pipeline.ingest.kafka string           Ingest Kafka API  
       --pipeline.ingest.type string            Ingest type: file, collector,file_loop (required)  
       --pipeline.transform string              Transforms (list) API (default "[{"type": "none"}]")  
       --pipeline.write.loki string             Loki write API  
-      --pipeline.write.type string             Write type: stdout, none (default "none")
+      --pipeline.write.type string             Write type: stdout, none
 ```
 <!---END-AUTO-flowlogs2metrics_help--->
 
@@ -361,6 +361,12 @@ pipeline:
       RecordKey: "value"
 ```
 
+### Json Encoder
+
+The json encoder takes each entry in the internal representation of the data and converts it to a json byte array.
+These byte arrays may then be output by a `write` stage.
+
+
 ### Prometheus encoder
 
 The prometheus encoder specifies which metrics to export to prometheus and which labels should be associated with those metrics.

cmd/flowlogs2metrics/main.go

@@ -132,12 +132,12 @@ func initFlags() {
 	rootCmd.PersistentFlags().StringVar(&config.Opt.PipeLine.Ingest.Collector, "pipeline.ingest.collector", "", "Ingest collector API")
 	rootCmd.PersistentFlags().StringVar(&config.Opt.PipeLine.Ingest.Kafka, "pipeline.ingest.kafka", "", "Ingest Kafka API")
 	rootCmd.PersistentFlags().StringVar(&config.Opt.PipeLine.Decode.Aws, "pipeline.decode.aws", "", "aws fields")
-	rootCmd.PersistentFlags().StringVar(&config.Opt.PipeLine.Decode.Type, "pipeline.decode.type", "none", "Decode type: aws, json, none")
+	rootCmd.PersistentFlags().StringVar(&config.Opt.PipeLine.Decode.Type, "pipeline.decode.type", "", "Decode type: aws, json, none")
 	rootCmd.PersistentFlags().StringVar(&config.Opt.PipeLine.Transform, "pipeline.transform", "[{\"type\": \"none\"}]", "Transforms (list) API")
-	rootCmd.PersistentFlags().StringVar(&config.Opt.PipeLine.Extract.Type, "pipeline.extract.type", "none", "Extract type: aggregates, none")
+	rootCmd.PersistentFlags().StringVar(&config.Opt.PipeLine.Extract.Type, "pipeline.extract.type", "", "Extract type: aggregates, none")
 	rootCmd.PersistentFlags().StringVar(&config.Opt.PipeLine.Extract.Aggregates, "pipeline.extract.aggregates", "", "Aggregates (see docs)")
-	rootCmd.PersistentFlags().StringVar(&config.Opt.PipeLine.Encode.Type, "pipeline.encode.type", "none", "Encode type: prom, none")
-	rootCmd.PersistentFlags().StringVar(&config.Opt.PipeLine.Write.Type, "pipeline.write.type", "none", "Write type: stdout, none")
+	rootCmd.PersistentFlags().StringVar(&config.Opt.PipeLine.Encode.Type, "pipeline.encode.type", "", "Encode type: prom, json, none")
+	rootCmd.PersistentFlags().StringVar(&config.Opt.PipeLine.Write.Type, "pipeline.write.type", "", "Write type: stdout, none")
 	rootCmd.PersistentFlags().StringVar(&config.Opt.PipeLine.Write.Loki, "pipeline.write.loki", "", "Loki write API")
 	rootCmd.PersistentFlags().StringVar(&config.Opt.PipeLine.Encode.Prom, "pipeline.encode.prom", "", "Prometheus encode API")
 

pkg/pipeline/encode/encode_json.go

@@ -0,0 +1,51 @@
+/*
+ * Copyright (C) 2021 IBM, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package encode
+
+import (
+	"encoding/json"
+	"github.com/netobserv/flowlogs2metrics/pkg/config"
+	log "github.com/sirupsen/logrus"
+)
+
+type encodeJson struct {
+}
+
+// Encode encodes json to byte array
+// All entries should be saved as strings
+func (e *encodeJson) Encode(inputMetrics []config.GenericMap) []interface{} {
+	out := make([]interface{}, 0)
+	for _, metric := range inputMetrics {
+		log.Debugf("encodeJson, metric = %v", metric)
+		var line []byte
+		var err error
+		line, err = json.Marshal(metric)
+		if err != nil {
+			log.Errorf("encodeJson Decode: error marshalling a line: %v", err)
+			continue
+		}
+		out = append(out, line)
+	}
+	return out
+}
+
+// NewEndcodeJson create a new encode
+func NewEncodeJson() (Encoder, error) {
+	log.Debugf("entering NewEncodeJson")
+	return &encodeJson{}, nil
+}

pkg/pipeline/encode/encode_json_test.go

@@ -0,0 +1,61 @@
+/*
+ * Copyright (C) 2022 IBM, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package encode
+
+import (
+	"github.com/netobserv/flowlogs2metrics/pkg/config"
+	"github.com/stretchr/testify/require"
+	"testing"
+)
+
+func initNewEncodeJson(t *testing.T) Encoder {
+	newEncode, err := NewEncodeJson()
+	require.NoError(t, err)
+	return newEncode
+}
+
+func TestEncodeJson(t *testing.T) {
+	newEncode := initNewEncodeJson(t)
+	encodeByteArray := newEncode.(*encodeJson)
+	map1 := config.GenericMap{
+		"varInt":    12,
+		"varString": "testString1",
+		"varBool":   true,
+	}
+	map2 := config.GenericMap{
+		"varString": "testString2",
+		"varInt":    14,
+		"varBool":   false,
+	}
+	map3 := config.GenericMap{}
+	var out []interface{}
+	var in []config.GenericMap
+	out = encodeByteArray.Encode(in)
+	require.Equal(t, 0, len(out))
+	in = append(in, map1)
+	in = append(in, map2)
+	in = append(in, map3)
+	out = encodeByteArray.Encode(in)
+	require.Equal(t, len(in), len(out))
+	expected1 := []byte(`{"varInt":12,"varBool":true,"varString":"testString1"}`)
+	expected2 := []byte(`{"varInt":14,"varBool":false,"varString":"testString2"}`)
+	expected3 := []byte(`{}`)
+	require.JSONEq(t, string(expected1), string(out[0].([]byte)))
+	require.JSONEq(t, string(expected2), string(out[1].([]byte)))
+	require.JSONEq(t, string(expected3), string(out[2].([]byte)))
+}

pkg/pipeline/encode/encode_test.go

@@ -0,0 +1,55 @@
+/*
+ * Copyright (C) 2022 IBM, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package encode
+
+import (
+	"github.com/netobserv/flowlogs2metrics/pkg/config"
+	"github.com/stretchr/testify/require"
+	"testing"
+)
+
+func initNewEncodeNone(t *testing.T) Encoder {
+	newEncode, err := NewEncodeNone()
+	require.Equal(t, err, nil)
+	return newEncode
+}
+
+func TestEncodeNone(t *testing.T) {
+	newEncode := initNewEncodeNone(t)
+	encodeNone := newEncode.(*encodeNone)
+	map1 := config.GenericMap{
+		"varInt":    12,
+		"varString": "string1",
+		"varbool":   true,
+	}
+	map2 := config.GenericMap{
+		"varInt":    14,
+		"varString": "string2",
+		"varbool":   false,
+	}
+	map3 := config.GenericMap{}
+	var out []interface{}
+	var in []config.GenericMap
+	out = encodeNone.Encode(in)
+	require.Equal(t, 0, len(out))
+	in = append(in, map1)
+	in = append(in, map2)
+	in = append(in, map3)
+	out = encodeNone.Encode(in)
+	require.Equal(t, len(in), len(out))
+}

pkg/pipeline/pipeline.go

@@ -115,6 +115,8 @@ func getEncoder() (encode.Encoder, error) {
 	switch config.Opt.PipeLine.Encode.Type {
 	case "prom":
 		encoder, _ = encode.NewEncodeProm()
+	case "json":
+		encoder, _ = encode.NewEncodeJson()
 	case "none":
 		encoder, _ = encode.NewEncodeNone()
 	default:

playground/goflow4.yml

@@ -1,38 +1,40 @@
 log-level: debug
 pipeline:
   ingest:
-    type: file
+    type: file_loop
     file:
       filename: playground/goflow2_input.txt
   decode:
     type: json
   transform:
     - type: generic
       generic:
-        - input: Bytes
-          output: v1_bytes
-        - input: DstAddr
-          output: v1_dstAddr
-        - input: Packets
-          output: v1_packets
-        - input: SrcPort
-          output: v1_srcPort
+        rules:
+          - input: Bytes
+            output: v1_bytes
+          - input: DstAddr
+            output: v1_dstAddr
+          - input: Packets
+            output: v1_packets
+          - input: SrcPort
+            output: v1_srcPort
     - type: generic
       generic:
-        - input: v1_srcPort
-          output: v1_srcPort
-        - input: v1_packets
-          output: v1_packets
-        - input: v1_packets
-          output: v2_packets
-        - input: v1_bytes
-          output: v2_bytes
-        - input: v1_dstAddr
-          output: v2_dstAddr
+        rules:
+          - input: v1_srcPort
+            output: v1_srcPort
+          - input: v1_packets
+            output: v1_packets
+          - input: v1_packets
+            output: v2_packets
+          - input: v1_bytes
+            output: v2_bytes
+          - input: v1_dstAddr
+            output: v2_dstAddr
   extract:
     type: none
   encode:
-    type: none
+    type: json
   write:
     type: stdout