conntrack: handle TCP flags

README.md

@@ -567,6 +567,9 @@ While, in bidirectional setting, they are grouped together.
 
 Bidirectional setting requires defining both `fieldGroupARef` and `fieldGroupBRef` sections to allow the connection
 tracking module to identify which set of fields can swap values and still be considered as the same connection.
+The pairs of fields that can swap are determined by their order in the fieldGroup.
+In the example below, `SrcAddr` and `DstAddr` are first in their fieldGroup, so they are swappable.
+The same is true for `SrcPort` and `DstPort` which are second.
 
 The configuration example below defines a bidirectional setting. So flow-logs that have the values of `SrcAddr` and `SrcPort` 
 swapped with `DstAddr` and `DstPort` are grouped together as long as they have the same `Proto` field.

docs/api.md

@@ -237,6 +237,10 @@ Following is the supported API format for specifying connection tracking:
                  endConnectionTimeout: duration of time to wait from the last flow log to end a connection
                  heartbeatInterval: duration of time to wait between heartbeat reports of a connection
          maxConnectionsTracked: maximum number of connections we keep in our cache (0 means no limit)
+         tcpFlags: settings for handling TCP flags
+             fieldName: name of the field containing TCP flags
+             detectEndConnection: detect end connections by FIN_ACK flag
+             swapAB: swap source and destination when the first flowlog contains the SYN_ACK flag
 </pre>
 ## Time-based Filters API
 Following is the supported API format for specifying metrics time-based filters:

docs/operational-metrics.md

@@ -31,6 +31,14 @@ Each table below provides documentation for an exported flowlogs-pipeline operat
 | **Labels** | type | 
 
 
+### conntrack_tcp_flags
+| **Name** | conntrack_tcp_flags | 
+|:---|:---|
+| **Description** | The total number of actions taken based on TCP flags. | 
+| **Type** | counter | 
+| **Labels** | action | 
+
+
 ### encode_prom_errors
 | **Name** | encode_prom_errors | 
 |:---|:---|

go.mod

@@ -1,6 +1,6 @@
 module github.com/netobserv/flowlogs-pipeline
 
-go 1.17
+go 1.18
 
 require (
 	github.com/Knetic/govaluate v3.0.1-0.20171022003610-9aa49832a739+incompatible
@@ -13,9 +13,9 @@ require (
 	github.com/mariomac/guara v0.0.0-20220523124851-5fc279816f1f
 	github.com/minio/minio-go/v7 v7.0.44
 	github.com/mitchellh/mapstructure v1.4.3
-	github.com/netobserv/gopipes v0.2.0
+	github.com/netobserv/gopipes v0.3.0
 	github.com/netobserv/loki-client-go v0.0.0-20220927092034-f37122a54500
-	github.com/netobserv/netobserv-ebpf-agent v0.2.4-0.20221220155455-aa7838d82f4d
+	github.com/netobserv/netobserv-ebpf-agent v0.2.4
 	github.com/netsampler/goflow2 v1.1.1-0.20220509155230-5300494e4785
 	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.12.1

go.sum

@@ -788,10 +788,14 @@ github.com/nats-io/nkeys v0.3.0/go.mod h1:gvUNGjVcM2IPr5rCsRsC6Wb3Hr2CQAm08dsxtV
 github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
 github.com/netobserv/gopipes v0.2.0 h1:CnJQq32+xNuM85eVYy/HOf+StTJdh2K6RdaEg7NAJDg=
 github.com/netobserv/gopipes v0.2.0/go.mod h1:eGoHZW1ON8Dx/zmDXUhsbVNqatPjtpdO0UZBmGZGmVI=
+github.com/netobserv/gopipes v0.3.0 h1:IYmPnnAVCdSK7VmHmpFhrVBOEm45qpgbZmJz1sSW+60=
+github.com/netobserv/gopipes v0.3.0/go.mod h1:N7/Gz05EOF0CQQSKWsv3eof22Cj2PB08Pbttw98YFYU=
 github.com/netobserv/loki-client-go v0.0.0-20220927092034-f37122a54500 h1:RmnoJe/ci5q+QdM7upFdxiU+D8F3L3qTd5wXCwwHefw=
 github.com/netobserv/loki-client-go v0.0.0-20220927092034-f37122a54500/go.mod h1:LHXpc5tjKvsfZn0pwLKrvlgEhZcCaw3Di9mUEZGAI4E=
 github.com/netobserv/netobserv-ebpf-agent v0.2.4-0.20221220155455-aa7838d82f4d h1:1tn5bJBtv44nglFmMlAbxWYf78Lz3RFxqTxZUsrXKaQ=
 github.com/netobserv/netobserv-ebpf-agent v0.2.4-0.20221220155455-aa7838d82f4d/go.mod h1:sKo7bEgMHchUkD+0c0qTQXZxQDRdgiPzKD9kgdRxgLU=
+github.com/netobserv/netobserv-ebpf-agent v0.2.4 h1:I8dLLF4NJvnO8HPcZIzgEutMh9sP2YHWZcaZa+3osX8=
+github.com/netobserv/netobserv-ebpf-agent v0.2.4/go.mod h1:IRGkUU+tFKr7mbaT/KyQDmuY3Nk4V1IwEZkNezzYLj0=
 github.com/netobserv/prometheus-common v0.31.2-0.20220720134304-43e74fd22881 h1:hx5bi6xBovRjmwUoVJBzhJ3EDo4K4ZUsqqKrJuQ2vMI=
 github.com/netobserv/prometheus-common v0.31.2-0.20220720134304-43e74fd22881/go.mod h1:vu+V0TpY+O6vW9J44gczi3Ap/oXXR10b+M/gUGO4Hls=
 github.com/netsampler/goflow2 v1.1.1-0.20220509155230-5300494e4785 h1:qhDrIMXlk8YV7BxwA6UR/dQVdUzohjLlmrUXymsBx6g=

pkg/api/conntrack.go

@@ -28,33 +28,33 @@ const (
 )
 
 type ConnTrack struct {
-	// TODO: should by a pointer instead?
-	KeyDefinition         KeyDefinition              `yaml:"keyDefinition,omitempty" doc:"fields that are used to identify the connection"`
-	OutputRecordTypes     []string                   `yaml:"outputRecordTypes,omitempty" enum:"ConnTrackOutputRecordTypeEnum" doc:"output record types to emit"`
-	OutputFields          []OutputField              `yaml:"outputFields,omitempty" doc:"list of output fields"`
-	Scheduling            []ConnTrackSchedulingGroup `yaml:"scheduling,omitempty" doc:"list of timeouts and intervals to apply per selector"`
-	MaxConnectionsTracked int                        `yaml:"maxConnectionsTracked,omitempty" doc:"maximum number of connections we keep in our cache (0 means no limit)"`
+	KeyDefinition         KeyDefinition              `yaml:"keyDefinition,omitempty" json:"keyDefinition,omitempty" doc:"fields that are used to identify the connection"`
+	OutputRecordTypes     []string                   `yaml:"outputRecordTypes,omitempty" json:"outputRecordTypes,omitempty" enum:"ConnTrackOutputRecordTypeEnum" doc:"output record types to emit"`
+	OutputFields          []OutputField              `yaml:"outputFields,omitempty" json:"outputFields,omitempty" doc:"list of output fields"`
+	Scheduling            []ConnTrackSchedulingGroup `yaml:"scheduling,omitempty" json:"scheduling,omitempty" doc:"list of timeouts and intervals to apply per selector"`
+	MaxConnectionsTracked int                        `yaml:"maxConnectionsTracked,omitempty" json:"maxConnectionsTracked,omitempty" doc:"maximum number of connections we keep in our cache (0 means no limit)"`
+	TCPFlags              ConnTrackTCPFlags          `yaml:"tcpFlags,omitempty" json:"tcpFlags" doc:"settings for handling TCP flags"`
 }
 
 type ConnTrackOutputRecordTypeEnum struct {
-	NewConnection string `yaml:"newConnection" doc:"New connection"`
-	EndConnection string `yaml:"endConnection" doc:"End connection"`
-	Heartbeat     string `yaml:"heartbeat" doc:"Heartbeat"`
-	FlowLog       string `yaml:"flowLog" doc:"Flow log"`
+	NewConnection string `yaml:"newConnection" json:"newConnection" doc:"New connection"`
+	EndConnection string `yaml:"endConnection" json:"endConnection" doc:"End connection"`
+	Heartbeat     string `yaml:"heartbeat" json:"heartbeat" doc:"Heartbeat"`
+	FlowLog       string `yaml:"flowLog" json:"flowLog" doc:"Flow log"`
 }
 
 func ConnTrackOutputRecordTypeName(operation string) string {
 	return GetEnumName(ConnTrackOutputRecordTypeEnum{}, operation)
 }
 
 type KeyDefinition struct {
-	FieldGroups []FieldGroup  `yaml:"fieldGroups,omitempty" doc:"list of field group definitions"`
-	Hash        ConnTrackHash `yaml:"hash,omitempty" doc:"how to build the connection hash"`
+	FieldGroups []FieldGroup  `yaml:"fieldGroups,omitempty" json:"fieldGroups,omitempty" doc:"list of field group definitions"`
+	Hash        ConnTrackHash `yaml:"hash,omitempty" json:"hash,omitempty" doc:"how to build the connection hash"`
 }
 
 type FieldGroup struct {
-	Name   string   `yaml:"name,omitempty" doc:"field group name"`
-	Fields []string `yaml:"fields" doc:"list of fields in the group"`
+	Name   string   `yaml:"name,omitempty" json:"name,omitempty" doc:"field group name"`
+	Fields []string `yaml:"fields" json:"fields" doc:"list of fields in the group"`
 }
 
 // ConnTrackHash determines how to compute the connection hash.
@@ -64,35 +64,41 @@ type FieldGroup struct {
 // When they are not set, a different hash will be computed for A->B and B->A,
 // and they are tracked as different connections.
 type ConnTrackHash struct {
-	FieldGroupRefs []string `yaml:"fieldGroupRefs,omitempty" doc:"list of field group names to build the hash"`
-	FieldGroupARef string   `yaml:"fieldGroupARef,omitempty" doc:"field group name of endpoint A"`
-	FieldGroupBRef string   `yaml:"fieldGroupBRef,omitempty" doc:"field group name of endpoint B"`
+	FieldGroupRefs []string `yaml:"fieldGroupRefs,omitempty" json:"fieldGroupRefs,omitempty" doc:"list of field group names to build the hash"`
+	FieldGroupARef string   `yaml:"fieldGroupARef,omitempty" json:"fieldGroupARef,omitempty" doc:"field group name of endpoint A"`
+	FieldGroupBRef string   `yaml:"fieldGroupBRef,omitempty" json:"fieldGroupBRef,omitempty" doc:"field group name of endpoint B"`
 }
 
 type OutputField struct {
-	Name      string `yaml:"name,omitempty" doc:"output field name"`
-	Operation string `yaml:"operation,omitempty" enum:"ConnTrackOperationEnum" doc:"aggregate operation on the field value"`
-	SplitAB   bool   `yaml:"splitAB,omitempty" doc:"When true, 2 output fields will be created. One for A->B and one for B->A flows."`
-	Input     string `yaml:"input,omitempty" doc:"The input field to base the operation on. When omitted, 'name' is used"`
+	Name      string `yaml:"name,omitempty" json:"name,omitempty" doc:"output field name"`
+	Operation string `yaml:"operation,omitempty" json:"operation,omitempty" enum:"ConnTrackOperationEnum" doc:"aggregate operation on the field value"`
+	SplitAB   bool   `yaml:"splitAB,omitempty" json:"splitAB,omitempty" doc:"When true, 2 output fields will be created. One for A->B and one for B->A flows."`
+	Input     string `yaml:"input,omitempty" json:"input,omitempty" doc:"The input field to base the operation on. When omitted, 'name' is used"`
 }
 
 type ConnTrackOperationEnum struct {
-	Sum   string `yaml:"sum" doc:"sum"`
-	Count string `yaml:"count" doc:"count"`
-	Min   string `yaml:"min" doc:"min"`
-	Max   string `yaml:"max" doc:"max"`
+	Sum   string `yaml:"sum" json:"sum" doc:"sum"`
+	Count string `yaml:"count" json:"count" doc:"count"`
+	Min   string `yaml:"min" json:"min" doc:"min"`
+	Max   string `yaml:"max" json:"max" doc:"max"`
 }
 
 type ConnTrackSchedulingGroup struct {
-	Selector             map[string]interface{} `yaml:"selector,omitempty" doc:"key-value map to match against connection fields to apply this scheduling"`
-	EndConnectionTimeout Duration               `yaml:"endConnectionTimeout,omitempty" doc:"duration of time to wait from the last flow log to end a connection"`
-	HeartbeatInterval    Duration               `yaml:"heartbeatInterval,omitempty" doc:"duration of time to wait between heartbeat reports of a connection"`
+	Selector             map[string]interface{} `yaml:"selector,omitempty" json:"selector,omitempty" doc:"key-value map to match against connection fields to apply this scheduling"`
+	EndConnectionTimeout Duration               `yaml:"endConnectionTimeout,omitempty" json:"endConnectionTimeout,omitempty" doc:"duration of time to wait from the last flow log to end a connection"`
+	HeartbeatInterval    Duration               `yaml:"heartbeatInterval,omitempty" json:"heartbeatInterval,omitempty" doc:"duration of time to wait between heartbeat reports of a connection"`
 }
 
 func ConnTrackOperationName(operation string) string {
 	return GetEnumName(ConnTrackOperationEnum{}, operation)
 }
 
+type ConnTrackTCPFlags struct {
+	FieldName           string `yaml:"fieldName" json:"fieldName" doc:"name of the field containing TCP flags"`
+	DetectEndConnection bool   `yaml:"detectEndConnection,omitempty" json:"detectEndConnection,omitempty" doc:"detect end connections by FIN_ACK flag"`
+	SwapAB              bool   `yaml:"swapAB,omitempty" json:"swapAB,omitempty" doc:"swap source and destination when the first flowlog contains the SYN_ACK flag"`
+}
+
 func (ct *ConnTrack) Validate() error {
 	isGroupAEmpty := ct.KeyDefinition.Hash.FieldGroupARef == ""
 	isGroupBEmpty := ct.KeyDefinition.Hash.FieldGroupBRef == ""
@@ -201,9 +207,40 @@ func (ct *ConnTrack) Validate() error {
 			msg: fmt.Errorf("found %v default selectors. There should be exactly 1", numOfDefault)}
 	}
 
+	if len(ct.TCPFlags.FieldName) == 0 && (ct.TCPFlags.DetectEndConnection || ct.TCPFlags.SwapAB) {
+		return conntrackInvalidError{emptyTCPFlagsField: true,
+			msg: fmt.Errorf("TCPFlags.FieldName is empty although DetectEndConnection or SwapAB are enabled")}
+	}
+	if ct.TCPFlags.SwapAB && !isBidi {
+		return conntrackInvalidError{swapABWithNoBidi: true,
+			msg: fmt.Errorf("SwapAB is enabled although bidirection is not enabled (fieldGroupARef is empty)")}
+	}
+
+	fieldsA, fieldsB := ct.GetABFields()
+	if len(fieldsA) != len(fieldsB) {
+		return conntrackInvalidError{mismatchABFieldsCount: true,
+			msg: fmt.Errorf("mismatch between the field count of fieldGroupARef %v and fieldGroupBRef %v", len(fieldsA), len(fieldsB))}
+	}
+
 	return nil
 }
 
+func (ct *ConnTrack) GetABFields() ([]string, []string) {
+	endpointAFieldGroupName := ct.KeyDefinition.Hash.FieldGroupARef
+	endpointBFieldGroupName := ct.KeyDefinition.Hash.FieldGroupBRef
+	var endpointAFields []string
+	var endpointBFields []string
+	for _, fg := range ct.KeyDefinition.FieldGroups {
+		if fg.Name == endpointAFieldGroupName {
+			endpointAFields = fg.Fields
+		}
+		if fg.Name == endpointBFieldGroupName {
+			endpointBFields = fg.Fields
+		}
+	}
+	return endpointAFields, endpointBFields
+}
+
 // addToSet adds an item to a set and returns true if it's a new item. Otherwise, it returns false.
 func addToSet(set map[string]struct{}, item string) bool {
 	if _, found := set[item]; found {
@@ -253,6 +290,9 @@ type conntrackInvalidError struct {
 	undefinedSelectorKey      bool
 	defaultGroupAndNotLast    bool
 	exactlyOneDefaultSelector bool
+	swapABWithNoBidi          bool
+	emptyTCPFlagsField        bool
+	mismatchABFieldsCount     bool
 }
 
 func (err conntrackInvalidError) Error() string {

pkg/api/conntrack_test.go

@@ -212,6 +212,45 @@ func TestConnTrackValidate(t *testing.T) {
 			},
 			conntrackInvalidError{defaultGroupAndNotLast: true},
 		},
+		{
+			"Empty TCPFlags field name 1",
+			ConnTrack{
+				Scheduling: []ConnTrackSchedulingGroup{{Selector: map[string]interface{}{}}},
+				TCPFlags:   ConnTrackTCPFlags{DetectEndConnection: true},
+			},
+			conntrackInvalidError{emptyTCPFlagsField: true},
+		},
+		{
+			"Empty TCPFlags field name 2",
+			ConnTrack{
+				Scheduling: []ConnTrackSchedulingGroup{{Selector: map[string]interface{}{}}},
+				TCPFlags:   ConnTrackTCPFlags{SwapAB: true},
+			},
+			conntrackInvalidError{emptyTCPFlagsField: true},
+		},
+		{
+			"Mismatch between field count of FieldGroupARef and FieldGroupBRef",
+			ConnTrack{
+				KeyDefinition: KeyDefinition{
+					FieldGroups: []FieldGroup{
+						{
+							Name:   "src",
+							Fields: []string{"SrcIP", "SrcPort"},
+						},
+						{
+							Name:   "dst",
+							Fields: []string{"DstIP"},
+						},
+					},
+					Hash: ConnTrackHash{
+						FieldGroupARef: "src",
+						FieldGroupBRef: "dst",
+					},
+				},
+				Scheduling: []ConnTrackSchedulingGroup{{Selector: map[string]interface{}{}}},
+			},
+			conntrackInvalidError{mismatchABFieldsCount: true},
+		},
 	}
 
 	for _, tt := range tests {
@@ -221,3 +260,27 @@ func TestConnTrackValidate(t *testing.T) {
 		})
 	}
 }
+
+func TestGetABFields(t *testing.T) {
+	fieldsA := []string{"SrcIP", "SrcPort"}
+	fieldsB := []string{"DstIP", "DstPort"}
+	conf := ConnTrack{
+		Scheduling: []ConnTrackSchedulingGroup{{Selector: map[string]interface{}{}}},
+		KeyDefinition: KeyDefinition{
+			FieldGroups: []FieldGroup{
+				{Name: "src", Fields: fieldsA},
+				{Name: "dst", Fields: fieldsB},
+			},
+			Hash: ConnTrackHash{
+				FieldGroupARef: "src",
+				FieldGroupBRef: "dst",
+			},
+		},
+	}
+
+	require.NoError(t, conf.Validate())
+
+	actualA, actualB := conf.GetABFields()
+	require.Equal(t, fieldsA, actualA)
+	require.Equal(t, fieldsB, actualB)
+}

pkg/config/generic_map.go

@@ -34,7 +34,7 @@ func (m GenericMap) Copy() GenericMap {
 
 func (m GenericMap) IsDuplicate() bool {
 	if duplicate, hasKey := m["Duplicate"]; hasKey {
-		if isDuplicate, err := utils.ConvertToBool(duplicate); err != nil {
+		if isDuplicate, err := utils.ConvertToBool(duplicate); err == nil {
 			return isDuplicate
 		}
 	}

pkg/config/pipeline_builder_test.go

@@ -152,7 +152,7 @@ func TestKafkaPromPipeline(t *testing.T) {
 
 	b, err = json.Marshal(params[2])
 	require.NoError(t, err)
-	require.JSONEq(t, `{"name":"conntrack","extract":{"type":"conntrack","conntrack":{"KeyDefinition":{"FieldGroups":null,"Hash":{"FieldGroupRefs":null,"FieldGroupARef":"","FieldGroupBRef":""}},"OutputRecordTypes":null,"MaxConnectionsTracked":0,"OutputFields":null,"Scheduling":null}}}`, string(b))
+	require.JSONEq(t, `{"name":"conntrack","extract":{"type":"conntrack","conntrack":{"keyDefinition":{"hash":{}},"tcpFlags":{"fieldName":""}}}}`, string(b))
 
 	b, err = json.Marshal(params[3])
 	require.NoError(t, err)

pkg/pipeline/decode/decode_protobuf.go

@@ -52,6 +52,7 @@ func PBFlowToMap(flow *pbflow.Record) config.GenericMap {
 		"TimeReceived":    time.Now().Unix(),
 		"Interface":       flow.Interface,
 		"AgentIP":         ipToStr(flow.AgentIp),
+		"Flags":           flow.Flags,
 	}
 	return out
 }

pkg/pipeline/decode/decode_protobuf_test.go

@@ -44,6 +44,7 @@ func TestDecodeProtobuf(t *testing.T) {
 		AgentIp: &pbflow.IP{
 			IpFamily: &pbflow.IP_Ipv4{Ipv4: 0x0a090807},
 		},
+		Flags: 0x100,
 	}
 	rawPB, err := proto.Marshal(&flow)
 	require.NoError(t, err)
@@ -69,6 +70,7 @@ func TestDecodeProtobuf(t *testing.T) {
 		"TimeFlowEndMs":   someTime.UnixMilli(),
 		"Interface":       "eth0",
 		"AgentIP":         "10.9.8.7",
+		"Flags":           uint32(0x100),
 	}, out)
 }
 
@@ -103,6 +105,7 @@ func TestPBFlowToMap(t *testing.T) {
 		AgentIp: &pbflow.IP{
 			IpFamily: &pbflow.IP_Ipv4{Ipv4: 0x0a090807},
 		},
+		Flags: 0x100,
 	}
 
 	out := PBFlowToMap(flow)
@@ -125,6 +128,7 @@ func TestPBFlowToMap(t *testing.T) {
 		"TimeFlowEndMs":   someTime.UnixMilli(),
 		"Interface":       "eth0",
 		"AgentIP":         "10.9.8.7",
+		"Flags":           uint32(0x100),
 	}, out)
 
 }

pkg/pipeline/encode/encode_kafka_test.go

@@ -117,7 +117,7 @@ func Test_TLSConfigCA(t *testing.T) {
 
 	require.Empty(t, tlsConfig.Certificates)
 	require.NotNil(t, tlsConfig.RootCAs)
-	require.Len(t, tlsConfig.RootCAs.Subjects(), 1)
+	require.Len(t, tlsConfig.RootCAs.Subjects(), 1) //nolint:staticcheck
 }
 
 func Test_MutualTLSConfig(t *testing.T) {
@@ -143,5 +143,5 @@ func Test_MutualTLSConfig(t *testing.T) {
 	require.NotEmpty(t, tlsConfig.Certificates[0].Certificate)
 	require.NotNil(t, tlsConfig.Certificates[0].PrivateKey)
 	require.NotNil(t, tlsConfig.RootCAs)
-	require.Len(t, tlsConfig.RootCAs.Subjects(), 1)
+	require.Len(t, tlsConfig.RootCAs.Subjects(), 1) //nolint:staticcheck
 }

pkg/pipeline/extract/conntrack/aggregator_test.go

@@ -140,7 +140,7 @@ func TestAddField_and_Update(t *testing.T) {
 		},
 	}
 
-	conn := NewConnBuilder().Build()
+	conn := NewConnBuilder(nil).Build()
 	for _, agg := range aggs {
 		agg.addField(conn)
 	}