filter toGenericMap instead of aggregator

netobserv · Aug 7, 2023 · 74611d2 · 74611d2
1 parent 35a89c6
commit 74611d2
Show file tree

Hide file tree

Showing 4 changed files with 22 additions and 6 deletions.
diff --git a/docs/api.md b/docs/api.md
@@ -241,6 +241,7 @@ Following is the supported API format for specifying connection tracking:
                      last: last
                  splitAB: When true, 2 output fields will be created. One for A->B and one for B->A flows.
                  input: The input field to base the operation on. When omitted, 'name' is used
+                 reportMissing: When true, missing input will produce MissingFieldError metric and error logs
          scheduling: list of timeouts and intervals to apply per selector
                  selector: key-value map to match against connection fields to apply this scheduling
                  endConnectionTimeout: duration of time to wait from the last flow log to end a connection

diff --git a/pkg/pipeline/extract/conntrack/aggregator.go b/pkg/pipeline/extract/conntrack/aggregator.go
@@ -185,7 +185,7 @@ func (agg *aMax) update(conn connection, flowLog config.GenericMap, d direction,
 }
 
 func (cp *aFirst) update(conn connection, flowLog config.GenericMap, d direction, isNew bool) {
-	if isNew && flowLog[cp.inputField] != nil {
+	if isNew {
 		conn.updateAggValue(cp.outputField, flowLog[cp.inputField])
 	}
 }

diff --git a/pkg/pipeline/extract/conntrack/conn.go b/pkg/pipeline/extract/conntrack/conn.go
@@ -19,6 +19,7 @@ package conntrack
 
 import (
 	"fmt"
+	"reflect"
 	"time"
 
 	"github.com/netobserv/flowlogs-pipeline/pkg/utils"
@@ -100,7 +101,9 @@ func (c *connType) getNextHeartbeatTime() time.Time {
 func (c *connType) toGenericMap() config.GenericMap {
 	gm := config.GenericMap{}
 	for k, v := range c.aggFields {
-		gm[k] = v
+		if v != nil && (reflect.TypeOf(v).Kind() != reflect.Float64 || v.(float64) != 0) {
+			gm[k] = v
+		}
 	}
 
 	// In case of a conflict between the keys and the aggFields / cpFields, the keys should prevail.

diff --git a/pkg/pipeline/extract/conntrack/utils_test.go b/pkg/pipeline/extract/conntrack/utils_test.go
@@ -59,14 +59,26 @@ func newMockRecordConnAB(srcIP string, srcPort int, dstIP string, dstPort int, p
 			"DstAddr":            dstIP,
 			"DstPort":            dstPort,
 			"Proto":              protocol,
-			"Bytes_AB":           bytesAB,
-			"Bytes_BA":           bytesBA,
-			"Packets_AB":         packetsAB,
-			"Packets_BA":         packetsBA,
 			"numFlowLogs":        numFlowLogs,
 			api.IsFirstFieldName: false,
 		},
 	}
+
+	if bytesAB != 0 {
+		mock.record["Bytes_AB"] = bytesAB
+	}
+
+	if bytesBA != 0 {
+		mock.record["Bytes_BA"] = bytesBA
+	}
+
+	if bytesAB != 0 {
+		mock.record["Packets_AB"] = packetsAB
+	}
+
+	if bytesBA != 0 {
+		mock.record["Packets_BA"] = packetsBA
+	}
 	return mock
 }