add support for connection tracking

eranra · eranra · commit d462138f387f · 2022-02-06T12:56:53.000+02:00
diff --git a/Makefile b/Makefile
@@ -81,7 +81,7 @@ clean: ## Clean
 # note: to review coverage execute: go tool cover -html=/tmp/coverage.out
 .PHONY: test
 test: validate_go ## Test
-	go test -race -covermode=atomic -coverprofile=/tmp/coverage.out ./...
+	go test -p 1 -race -covermode=atomic -coverprofile=/tmp/coverage.out ./...
 
 # note: to review profile execute: go tool pprof -web /tmp/flowlogs2metrics-cpu-profile.out (make sure graphviz is installed)
 .PHONY: benchmarks
diff --git a/README.md b/README.md
@@ -275,6 +275,10 @@ pipeline:
           output: match-10.0
           type: add_regex_if
           parameters: 10.0.*
+        - input: "{{.srcIP}},{{.srcPort}},{{.dstIP}},{{.dstPort}},{{.protocol}}"
+          output: isNewFlow
+          type: conn_tracking
+          parameters: "1"
 ```
 
 The first rule `add_subnet` generates a new field named `srcSubnet` with the 
@@ -306,11 +310,15 @@ All the kubernetes fields will be named by appending `output` value
 > 2. using `KUBECONFIG` environment variable
 > 3. using local `~/.kube/config`
 
-The sixth `add_regex_if` generates a new field named `match-10.0` that contains
-the contents of the `srcSubnet` field for entries that match regex expression specified
+The sixth rule `add_regex_if` generates a new field named `match-10.0` that contains
+the contents of the `srcSubnet` field for entries that match regex expression specified 
 in the `parameters` variable. In addition, the field `match-10.0_Matched` with 
 value `true` is added to all matched entries
 
+The seventh rule `conn_tracking` generates a new field named `isNewFlow` that contains
+the contents of the `parameters` variable **only for new entries** (first seen in 120 seconds) 
+that match hash of template fields from the `input` variable. 
+
 
 > Note: above example describes all available transform network `Type` options
 > Note: above transform is essential for the `aggregation` phase  
diff --git a/contrib/kubernetes/flowlogs2metrics.conf.yaml b/contrib/kubernetes/flowlogs2metrics.conf.yaml
@@ -135,7 +135,7 @@ pipeline:
     - name: dest_connection_subnet_count
       by:
       - dstSubnet
-      operation: count
+      operation: sum
       recordkey: ""
     - name: src_connection_count
       by:
@@ -203,16 +203,18 @@ pipeline:
   transform:
   - generic:
       rules:
-      - input: DstAddr
-        output: dstIP
       - input: SrcAddr
         output: srcIP
-      - input: Bytes
-        output: bytes
+      - input: SrcPort
+        output: srcPort
+      - input: DstAddr
+        output: dstIP
       - input: DstPort
         output: dstPort
       - input: Proto
         output: proto
+      - input: Bytes
+        output: bytes
       - input: TCPFlags
         output: TCPFlags
       - input: SrcAS
@@ -238,6 +240,10 @@ pipeline:
         output: srcSubnet
         type: add_subnet
         parameters: /16
+      - input: '{{.srcIP}},{{.srcPort}},{{.dstIP}},{{.dstPort}},{{.proto}}'
+        output: isNewFlow
+        type: conn_tracking
+        parameters: "1"
       - input: dstIP
         output: dstSubnet
         type: add_subnet
diff --git a/docs/api.md b/docs/api.md
@@ -43,6 +43,7 @@ Following is the supported API format for network transformations:
                  input: entry input field
                  output: entry output field
                  type: (enum) one of the following:
+                     conn_tracking: set output field to value of parameters field only for new flows by matching template in input field
                      add_regex_if: add output field if input field satisfies regex pattern from parameters field
                      add_if: add output field if input field satisfies criteria from parameters field
                      add_subnet: add output subnet field from input field and prefix length from parameters field
diff --git a/docs/metrics.md b/docs/metrics.md
@@ -51,10 +51,10 @@ and the transformation to generate the exported metric.
 ### connection rate per dest subnet
 | **Description** | This metric observes network connections rate per destination subnet | 
 |:---|:---|
-| **Details** | Counts the number of connections per subnet with network prefix length /16 | 
+| **Details** | Counts the number of connections per subnet with network prefix length /16 (using conn_tracking sum isNewFlow field) | 
 | **Usage** | Evaluate network connections per subnet | 
 | **Labels** | rate, subnet |
-| **Operation** | aggregate by `dstSubnet` and `count`  |
+| **Operation** | aggregate by `dstSubnet` and `sum`  |
 | **Exposed as** | `fl2m_connections_per_destination_subnet` of type `gauge` |
 | **Visualized as** | "Connections rate per destinationIP /16 subnets" on dashboard `details` |
 |||  
diff --git a/network_definitions/bandwidth_per_network_service.yaml b/network_definitions/bandwidth_per_network_service.yaml
@@ -22,7 +22,7 @@ extract:
       by:
         - service
       operation: sum
-      RecordKey: bytes
+      recordKey: bytes
 encode:
   type: prom
   prom:
diff --git a/network_definitions/bandwidth_per_src_dest_subnet.yaml b/network_definitions/bandwidth_per_src_dest_subnet.yaml
@@ -27,7 +27,7 @@ extract:
         - dstSubnet24
         - srcSubnet24
       operation: sum
-      RecordKey: bytes
+      recordKey: bytes
 encode:
   type: prom
   prom:
diff --git a/network_definitions/bandwidth_per_src_subnet.yaml b/network_definitions/bandwidth_per_src_subnet.yaml
@@ -22,7 +22,7 @@ extract:
       by:
         - srcSubnet
       operation: sum
-      RecordKey: bytes
+      recordKey: bytes
 encode:
   type: prom
   prom:
diff --git a/network_definitions/config.yaml b/network_definitions/config.yaml
@@ -10,16 +10,18 @@ ingest:
 transform:
   generic:
     rules:
-      - input: DstAddr
-        output: dstIP
       - input: SrcAddr
         output: srcIP
-      - input: Bytes
-        output: bytes
+      - input: SrcPort
+        output: srcPort
+      - input: DstAddr
+        output: dstIP
       - input: DstPort
         output: dstPort
       - input: Proto
         output: proto
+      - input: Bytes
+        output: bytes
       - input: TCPFlags
         output: TCPFlags
       - input: SrcAS
diff --git a/network_definitions/connection_rate_per_dest_subnet.yaml b/network_definitions/connection_rate_per_dest_subnet.yaml
@@ -2,14 +2,18 @@
 description:
   This metric observes network connections rate per destination subnet
 details:
-  Counts the number of connections per subnet with network prefix length /16
+  Counts the number of connections per subnet with network prefix length /16 (using conn_tracking sum isNewFlow field)
 usage:
   Evaluate network connections per subnet
 labels:
   - rate
   - subnet
 transform:
   rules:
+    - input: "{{.srcIP}},{{.srcPort}},{{.dstIP}},{{.dstPort}},{{.proto}}"
+      output: isNewFlow
+      type: conn_tracking
+      parameters: "1"
     - input: dstIP
       output: dstSubnet
       type: add_subnet
@@ -20,7 +24,8 @@ extract:
     - name: dest_connection_subnet_count
       by:
         - dstSubnet
-      operation: count
+      operation: sum
+      recordKey: isNewFlow
 encode:
   type: prom
   prom:
diff --git a/network_definitions/egress_bandwidth_per_dest_subnet.yaml b/network_definitions/egress_bandwidth_per_dest_subnet.yaml
@@ -22,7 +22,7 @@ extract:
       by:
         - dstSubnet
       operation: sum
-      RecordKey: bytes
+      recordKey: bytes
 encode:
   type: prom
   prom:
diff --git a/network_definitions/egress_bandwidth_per_namespace.yaml b/network_definitions/egress_bandwidth_per_namespace.yaml
@@ -21,7 +21,7 @@ extract:
         - srcK8S_Namespace
         - srcK8S_Type
       operation: sum
-      RecordKey: bytes
+      recordKey: bytes
 encode:
   type: prom
   prom:
diff --git a/pkg/api/transform_network.go b/pkg/api/transform_network.go
@@ -23,6 +23,7 @@ type TransformNetwork struct {
 }
 
 type TransformNetworkOperationEnum struct {
+	ConnTracking  string `yaml:"conn_tracking" doc:"set output field to value of parameters field only for new flows by matching template in input field"`
 	AddRegExIf    string `yaml:"add_regex_if" doc:"add output field if input field satisfies regex pattern from parameters field"`
 	AddIf         string `yaml:"add_if" doc:"add output field if input field satisfies criteria from parameters field"`
 	AddSubnet     string `yaml:"add_subnet" doc:"add output subnet field from input field and prefix length from parameters field"`
diff --git a/pkg/pipeline/transform/connection_tracking/connection_tracking.go b/pkg/pipeline/transform/connection_tracking/connection_tracking.go
@@ -0,0 +1,151 @@
+/*
+ * 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 connection_tracking
+
+import (
+	"container/list"
+	"fmt"
+	"hash/fnv"
+	"sync"
+	"time"
+)
+
+type cacheKey uint32
+
+type cacheInfo struct {
+	key             cacheKey
+	flowlogsCount   int
+	lastUpdatedTime int64
+	listElement     *list.Element
+}
+
+const defaultExpiryTime = 120
+
+type ConnectionTracking struct {
+	cacheMap  map[cacheKey]cacheInfo
+	cacheList *list.List
+
+	expiryTime int64
+	mutex      *sync.Mutex
+}
+
+var (
+	CT         ConnectionTracking
+	expiryTime = int64(defaultExpiryTime)
+)
+
+func getHashKey(flowIDFields string) cacheKey {
+	h := fnv.New32a()
+	h.Write([]byte(flowIDFields))
+	return cacheKey(h.Sum32())
+}
+
+func (ct ConnectionTracking) getCacheEntry(flowIDFields string) (cacheInfo, error) {
+	ct.mutex.Lock()
+	defer ct.mutex.Unlock()
+
+	key := getHashKey(flowIDFields)
+	if entry, ok := ct.cacheMap[key]; ok {
+		return entry, nil
+	}
+
+	return cacheInfo{}, fmt.Errorf("not found")
+}
+
+func (ct ConnectionTracking) updateCacheEntry(flowIDFields string, entry cacheInfo) {
+	ct.mutex.Lock()
+	defer ct.mutex.Unlock()
+
+	key := getHashKey(flowIDFields)
+	entry.key = key
+	entry.lastUpdatedTime = time.Now().Unix()
+	// move element to back of the cache list, or add new entry in the back if needed
+	if _, ok := ct.cacheMap[key]; ok {
+		ct.cacheList.MoveToBack(entry.listElement)
+	} else {
+		entry.listElement = ct.cacheList.PushBack(&entry)
+	}
+	ct.cacheMap[key] = entry
+}
+
+func (ct ConnectionTracking) IsFlowKnown(flowIDFields string) bool {
+	if _, err := ct.getCacheEntry(flowIDFields); err == nil {
+		return true
+	}
+	return false
+}
+
+func (ct ConnectionTracking) AddFlow(flowIDFields string) bool {
+	entry := cacheInfo{}
+	isNew := true
+	if cacheEntry, err := ct.getCacheEntry(flowIDFields); err == nil {
+		entry = cacheEntry
+		isNew = false
+	}
+	entry.flowlogsCount += 1
+
+	ct.updateCacheEntry(flowIDFields, entry)
+
+	return isNew
+}
+
+func (ct ConnectionTracking) cleanupExpiredEntriesLoop() {
+	ticker := time.NewTicker(time.Duration(ct.expiryTime) * time.Second)
+	done := make(chan bool)
+	go func() {
+		for {
+			select {
+			case <-done:
+				return
+			case <-ticker.C:
+				ct.cleanupExpiredEntries()
+			}
+		}
+	}()
+}
+
+func (ct ConnectionTracking) cleanupExpiredEntries() {
+	ct.mutex.Lock()
+	defer ct.mutex.Unlock()
+
+	nowInSecs := time.Now().Unix()
+	expireTime := nowInSecs - ct.expiryTime
+	for {
+		listEntry := ct.cacheList.Front()
+		if listEntry == nil {
+			return
+		}
+		pCacheInfo := listEntry.Value.(*cacheInfo)
+		if pCacheInfo.lastUpdatedTime > expireTime {
+			return
+		}
+		delete(ct.cacheMap, pCacheInfo.key)
+		ct.cacheList.Remove(listEntry)
+	}
+}
+
+func InitConnectionTracking() {
+	CT = ConnectionTracking{
+		cacheMap:   map[cacheKey]cacheInfo{},
+		cacheList:  list.New(),
+		expiryTime: expiryTime,
+		mutex:      &sync.Mutex{},
+	}
+
+	CT.cleanupExpiredEntriesLoop()
+}
diff --git a/pkg/pipeline/transform/connection_tracking/connection_tracking_test.go b/pkg/pipeline/transform/connection_tracking/connection_tracking_test.go
diff --git a/pkg/pipeline/transform/location/location.go b/pkg/pipeline/transform/location/location.go
diff --git a/pkg/pipeline/transform/location/location_test.go b/pkg/pipeline/transform/location/location_test.go
diff --git a/pkg/pipeline/transform/transform_network.go b/pkg/pipeline/transform/transform_network.go
diff --git a/pkg/pipeline/transform/transform_network_test.go b/pkg/pipeline/transform/transform_network_test.go
