port decode aws to new code structure

Author: KalmanMeth

README.md

@@ -36,12 +36,12 @@ Flags:
   -h, --help                                   help for flowlogs2metrics  
       --log-level string                       Log level: debug, info, warning, error (default "error")  
       --pipeLine.ingest.collector string       Ingest collector API  
+      --pipeline.decode.aws string             aws fields  
       --pipeline.decode.type string            Decode type: aws, json, none (default "none")  
       --pipeline.encode.prom string            Prometheus encode API  
       --pipeline.encode.type string            Encode type: prom, none (default "none")  
       --pipeline.extract.aggregates string     Aggregates (see docs)  
       --pipeline.extract.type string           Extract type: aggregates, none (default "none")  
-      --pipeline.ingest.aws.fields strings     aws fields  
       --pipeline.ingest.file.filename string   Ingest filename (file)  
       --pipeline.ingest.type string            Ingest type: file, collector,file_loop (required)  
       --pipeline.transform string              Transforms (list) API (default "[{"type": "none"}]")  

cmd/flowlogs2metrics/main.go

@@ -128,7 +128,7 @@ func initFlags() {
 	rootCmd.PersistentFlags().StringVar(&config.Opt.PipeLine.Ingest.Type, "pipeline.ingest.type", "", "Ingest type: file, collector,file_loop (required)")
 	rootCmd.PersistentFlags().StringVar(&config.Opt.PipeLine.Ingest.File.Filename, "pipeline.ingest.file.filename", "", "Ingest filename (file)")
 	rootCmd.PersistentFlags().StringVar(&config.Opt.PipeLine.Ingest.Collector, "pipeLine.ingest.collector", "", "Ingest collector API")
-	rootCmd.PersistentFlags().StringSliceVar(&config.Opt.PipeLine.Ingest.Aws.Fields, "pipeline.ingest.aws.fields", nil, "aws fields")
+	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.Transform, "pipeline.transform", "[{\"type\": \"none\"}]", "Transforms (list) API")
 	rootCmd.PersistentFlags().StringVar(&config.Opt.PipeLine.Extract.Type, "pipeline.extract.type", "none", "Extract type: aggregates, none")

pkg/api/decode_aws.go

@@ -0,0 +1,22 @@
+/*
+ * 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 api
+
+type EncodeAwsStruct struct {
+	Fields []string `yaml:"fields" doc:"list of aws flow log fields"`
+}

pkg/config/config.go

@@ -40,7 +40,6 @@ type Ingest struct {
 	Type      string
 	File      File
 	Collector string
-	Aws       Aws
 }
 
 type File struct {
@@ -53,6 +52,7 @@ type Aws struct {
 
 type Decode struct {
 	Type string
+	Aws  string
 }
 
 type Extract struct {

pkg/pipeline/decode/decode_aws.go

@@ -3,21 +3,23 @@
  *
  * 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 ofthe License at
+ * 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 specificlanguage governing permissions and
+ * See the License for the specific language governing permissions and
  * limitations under the License.
  *
  */
 
 package decode
 
 import (
+	"encoding/json"
+	"github.com/netobserv/flowlogs2metrics/pkg/api"
 	"github.com/netobserv/flowlogs2metrics/pkg/config"
 	log "github.com/sirupsen/logrus"
 	"strings"
@@ -51,11 +53,12 @@ func (c *decodeAws) Decode(in []interface{}) []config.GenericMap {
 	out := make([]config.GenericMap, 0)
 	nItems := len(in)
 	log.Debugf("nItems = %d", nItems)
-	for _, line := range in {
+	for i, line := range in {
 		lineSlice := strings.Fields(line.(string))
 		nFields := len(lineSlice)
-		if nFields > len(c.keyTags) {
-			nFields = len(c.keyTags)
+		if nFields != len(c.keyTags) {
+			log.Errorf("decodeAws Decode: wrong number of fields in line %d", i+1)
+			continue
 		}
 		record := make(config.GenericMap)
 		for i := 0; i < nFields; i++ {
@@ -71,12 +74,23 @@ func (c *decodeAws) Decode(in []interface{}) []config.GenericMap {
 // NewDecodeAws create a new decode
 func NewDecodeAws() (Decoder, error) {
 	log.Debugf("entering NewDecodeAws")
-	RecordKeys := config.Opt.PipeLine.Ingest.Aws.Fields
-	if RecordKeys == nil {
-		RecordKeys = defaultKeys
+	var recordKeys []string
+	fieldsString := config.Opt.PipeLine.Decode.Aws
+	log.Debugf("fieldsString = %v", fieldsString)
+	if fieldsString != "" {
+		var awsFields api.EncodeAwsStruct
+		err := json.Unmarshal([]byte(fieldsString), &awsFields)
+		if err != nil {
+			log.Errorf("NewDecodeAws: error in unmarshalling fields: %v", err)
+			return nil, err
+		}
+		recordKeys = awsFields.Fields
+	} else {
+		recordKeys = defaultKeys
 	}
-	log.Debugf("RecordKeys = %v", RecordKeys)
+
+	log.Debugf("recordKeys = %v", recordKeys)
 	return &decodeAws{
-		keyTags: RecordKeys,
+		keyTags: recordKeys,
 	}, nil
 }