NETOBSERV-1275: Introduce new "INNER" direction for inner-node traffic (#483)

* Introduce new "INNER" direction for inner-node traffic

The flows (and duplicates) generated for inner-node traffic differs
compared to node-to-node traffic, and reinterpret direction isn't able
to decide between ingress or egress. This is causing discrepancies with
the dedup mechanism that filters out flows where Duplicate=true and also
favors ingress over egress.

To fix that, the proposed solution is to create this new INNER direction
specifically for this kind of traffic. Deduping this INNER traffic can
then rely solely on the Duplicate flag, since that flag was set from a
single Agent (single node) there will always be only one
Duplicate=false.

* update doc

* Enable reinterpret on conversations

Author: jotak

docs/api.md

@@ -158,7 +158,7 @@ Following is the supported API format for network transformations:
                      add_location: add output location fields from input
                      add_service: add output network service field from input port and parameters protocol field
                      add_kubernetes: add output kubernetes fields from input
-                     reinterpret_direction: reinterpret flow direction at a higher level than the interface
+                     reinterpret_direction: reinterpret flow direction at the node level (instead of net interface), to ease the deduplication process
                      add_ip_category: categorize IPs based on known subnets configuration
                  parameters: parameters specific to type
                  assignee: value needs to assign to output field

pkg/api/transform_network.go

@@ -52,7 +52,7 @@ type TransformNetworkOperationEnum struct {
 	AddLocation          string `yaml:"add_location" json:"add_location" doc:"add output location fields from input"`
 	AddService           string `yaml:"add_service" json:"add_service" doc:"add output network service field from input port and parameters protocol field"`
 	AddKubernetes        string `yaml:"add_kubernetes" json:"add_kubernetes" doc:"add output kubernetes fields from input"`
-	ReinterpretDirection string `yaml:"reinterpret_direction" json:"reinterpret_direction" doc:"reinterpret flow direction at a higher level than the interface"`
+	ReinterpretDirection string `yaml:"reinterpret_direction" json:"reinterpret_direction" doc:"reinterpret flow direction at the node level (instead of net interface), to ease the deduplication process"`
 	AddIPCategory        string `yaml:"add_ip_category" json:"add_ip_category" doc:"categorize IPs based on known subnets configuration"`
 }
 

pkg/pipeline/transform/transform_network.go

@@ -124,10 +124,7 @@ func (n *Network) Transform(inputEntry config.GenericMap) (config.GenericMap, bo
 				}
 			}
 		case api.OpReinterpretDirection:
-			// only reinterpret direction on flowlogs
-			if rt, ok := outputEntry["_RecordType"]; !ok || rt == "flowLog" {
-				reinterpretDirection(outputEntry, &n.DirectionInfo)
-			}
+			reinterpretDirection(outputEntry, &n.DirectionInfo)
 		case api.OpAddIPCategory:
 			if strIP, ok := outputEntry[rule.Input].(string); ok {
 				cat, ok := n.ipCatCache.GetCacheEntry(strIP)

pkg/pipeline/transform/transform_network_direction.go

@@ -10,6 +10,7 @@ import (
 const (
 	ingress = 0
 	egress  = 1
+	inner   = 2
 )
 
 func validateReinterpretDirectionConfig(info *api.NetworkTransformDirectionInfo) error {
@@ -57,5 +58,7 @@ func reinterpretDirection(output config.GenericMap, info *api.NetworkTransformDi
 		} else if dstNode == reporter {
 			output[info.FlowDirectionField] = ingress
 		}
+	} else if srcNode != "" {
+		output[info.FlowDirectionField] = inner
 	}
 }

pkg/pipeline/transform/transform_network_test.go

@@ -370,6 +370,22 @@ func Test_ReinterpretDirection(t *testing.T) {
 		"FlowDirection": 0,
 	}, output)
 
+	output, ok = tr.Transform(config.GenericMap{
+		"ReporterIP":    "10.1.2.3",
+		"SrcHostIP":     "10.1.2.3",
+		"DstHostIP":     "10.1.2.3",
+		"FlowDirection": "whatever",
+	})
+	require.True(t, ok)
+	// Inner node => inner (2)
+	require.Equal(t, config.GenericMap{
+		"ReporterIP":    "10.1.2.3",
+		"SrcHostIP":     "10.1.2.3",
+		"DstHostIP":     "10.1.2.3",
+		"IfDirection":   "whatever",
+		"FlowDirection": 2,
+	}, output)
+
 	output, ok = tr.Transform(config.GenericMap{
 		"ReporterIP":    "10.1.2.100",
 		"SrcHostIP":     "10.1.2.3",