port timebased fields in bandwidth_per_src_subnet.yaml to new values (#329)

* ported timebased fields in bandwidth_per_src_subnet.yaml to new values

* updated yaml file in contrib/kubernetes

* change Debugf to Errorf for error in config files

Author: KalmanMeth

contrib/kubernetes/flowlogs-pipeline.conf.yaml

@@ -10,8 +10,10 @@ pipeline:
   follows: extract_conntrack
 - name: extract_aggregate
   follows: transform_network
-- name: encode_prom
+- name: extract_timebased
   follows: extract_aggregate
+- name: encode_prom
+  follows: extract_timebased
 - name: write_loki
   follows: transform_network
 parameters:
@@ -235,6 +237,17 @@ parameters:
       - service
       - _RecordType
       operationType: count
+- name: extract_timebased
+  extract:
+    type: timebased
+    timebased:
+      rules:
+      - name: bandwidth_source_subnet
+        indexKey: srcSubnet
+        operationType: sum
+        operationKey: total_value
+        topK: 5
+        timeInterval: 1m0s
 - name: encode_prom
   encode:
     type: prom
@@ -260,6 +273,15 @@ parameters:
         - groupByKeys
         - aggregate
         buckets: []
+      - name: bandwidth_per_source_subnet
+        type: gauge
+        filter:
+          key: name
+          value: bandwidth_source_subnet
+        valueKey: operation_result
+        labels:
+        - srcSubnet
+        buckets: []
       - name: connection_size_histogram
         type: agg_histogram
         filter:

network_definitions/bandwidth_per_src_subnet.yaml

@@ -27,9 +27,9 @@ extract:
   timebased:
     rules:
       - name: bandwidth_source_subnet
-        operation: sum
+        operationType: sum
         operationKey: total_value
-        recordKey: srcSubnet
+        indexKey: srcSubnet
         topK: 5
         timeInterval: 1m
 encode:

pkg/confgen/extract.go

@@ -30,14 +30,14 @@ func (cg *ConfGen) parseExtract(extract *map[string]interface{}) (*aggregate.Def
 	aggregateExtract := (*extract)["aggregates"]
 	b, err := jsoniterJson.Marshal(&aggregateExtract)
 	if err != nil {
-		log.Debugf("jsoniterJson.Marshal err: %v ", err)
+		log.Errorf("jsoniterJson.Marshal err: %v ", err)
 		return nil, nil, err
 	}
 
 	var jsonNetworkAggregate aggregate.Definitions
 	err = config.JsonUnmarshalStrict(b, &jsonNetworkAggregate)
 	if err != nil {
-		log.Debugf("Unmarshal aggregate.Definitions err: %v ", err)
+		log.Errorf("Unmarshal aggregate.Definitions err: %v ", err)
 		return nil, nil, err
 	}
 
@@ -49,14 +49,14 @@ func (cg *ConfGen) parseExtract(extract *map[string]interface{}) (*aggregate.Def
 	}
 	b, err = jsoniterJson.Marshal(&timebasedExtract)
 	if err != nil {
-		log.Debugf("jsoniterJson.Marshal err: %v ", err)
+		log.Errorf("jsoniterJson.Marshal err: %v ", err)
 		return nil, nil, err
 	}
 
 	var jsonTimebasedTopKs api.ExtractTimebased
 	err = config.JsonUnmarshalStrict(b, &jsonTimebasedTopKs)
 	if err != nil {
-		log.Debugf("Unmarshal api.ExtractTimebased err: %v ", err)
+		log.Errorf("Unmarshal api.ExtractTimebased err: %v ", err)
 		return nil, nil, err
 	}