Added TCP Flags to Flow Record metrics (#68)

* Test commit

* Adding TCP flags

* Adding TCP flags to record metrics.TODO: Remove log stmts

* Updated tcp flags u32-> u16

* Updated indentation in C file

* Update flow.h

* Indent fixed. Tab -> 4 space

fixing indents

Indent fixes for flows.c

Indent fixes for flows.c-II

Still fixing indents

* Removed extra file

* Added clang-format config to bpf/. To use clang-format -i <file.c>

* Fixed message record member sequences

* Added TCP Flags combinations Read from flw

Updated changes according to comments on PR

* Removed log statements.

Rebase merges

* Fixing merge conflits

* Revert the renaming of grpc_test file

* Simplify setting flags for v4,v6

* Add flags to ipfix exporter

* Update protobuf

* Remove type conversions due to conflicts

* Remove .gitignore change

* Fix indentation

* Remove commented line

* Fixed typecast errors since move to go 1.18

* Reverting commit d1e9c44. Added changes as separate PR

* set flags for v6

Co-authored-by: Pravein Govindan Kannan <[email protected]>

Author: shach33

bpf/.clang-format

@@ -0,0 +1,9 @@
+{
+  BasedOnStyle: LLVM,
+  AllowShortFunctionsOnASingleLine: InlineOnly,
+  ColumnLimit: 100,
+  IndentWidth: 4,
+  SortIncludes: false,
+  ReflowComments: false,
+  TabWidth: 4,
+}

bpf/flow.h

@@ -17,6 +17,8 @@ typedef struct flow_metrics_t {
     // as output from bpf_ktime_get_ns()
     u64 start_mono_time_ts;
     u64 end_mono_time_ts;
+    // TCP Flags from https://www.ietf.org/rfc/rfc793.txt
+    u16 flags;
     // The positive errno of a failed map insertion that caused a flow
     // to be sent via ringbuffer.
     // 0 otherwise

bpf/flows.c

@@ -26,7 +26,6 @@
 #include <linux/udp.h>
 #include <linux/tcp.h>
 #include <string.h>
-
 #include <stdbool.h>
 #include <linux/if_ether.h>
 
@@ -42,6 +41,20 @@
 #define INGRESS 0
 #define EGRESS 1
 
+// Flags according to RFC 9293 & https://www.iana.org/assignments/ipfix/ipfix.xhtml
+#define FIN_FLAG 0x01
+#define SYN_FLAG 0x02
+#define RST_FLAG 0x04
+#define PSH_FLAG 0x08
+#define ACK_FLAG 0x10
+#define URG_FLAG 0x20
+#define ECE_FLAG 0x40
+#define CWR_FLAG 0x80
+// Custom flags exported
+#define SYN_ACK_FLAG 0x100
+#define FIN_ACK_FLAG 0x200
+#define RST_ACK_FLAG 0x400
+
 // Common Ringbuffer as a conduit for ingress/egress flows to userspace
 struct {
     __uint(type, BPF_MAP_TYPE_RINGBUF);
@@ -62,8 +75,35 @@ volatile const u8 trace_messages = 0;
 
 const u8 ip4in6[] = {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0xff, 0xff};
 
+// sets the TCP header flags for connection information
+static inline void set_flags(struct tcphdr *th, u16 *flags) {
+    //If both ACK and SYN are set, then it is server -> client communication during 3-way handshake. 
+    if (th->ack && th->syn) {
+        *flags |= SYN_ACK_FLAG;
+    } else if (th->ack && th->fin ) {
+        // If both ACK and FIN are set, then it is graceful termination from server.
+        *flags |= FIN_ACK_FLAG;
+    } else if (th->ack && th->rst ) {
+        // If both ACK and RST are set, then it is abrupt connection termination. 
+        *flags |= RST_ACK_FLAG;
+    } else if (th->fin) {
+        *flags |= FIN_FLAG;
+    } else if (th->syn) {
+        *flags |= SYN_FLAG;
+    } else if (th->rst) {
+        *flags |= RST_FLAG;
+    } else if (th->psh) {
+        *flags |= PSH_FLAG;
+    } else if (th->urg) {
+        *flags |= URG_FLAG;
+    } else if (th->ece) {
+        *flags |= ECE_FLAG;
+    } else if (th->cwr) {
+        *flags |= CWR_FLAG;
+    }
+}
 // sets flow fields from IPv4 header information
-static inline int fill_iphdr(struct iphdr *ip, void *data_end, flow_id *id) {
+static inline int fill_iphdr(struct iphdr *ip, void *data_end, flow_id *id, u16 *flags) {
     if ((void *)ip + sizeof(*ip) > data_end) {
         return DISCARD;
     }
@@ -81,6 +121,7 @@ static inline int fill_iphdr(struct iphdr *ip, void *data_end, flow_id *id) {
         if ((void *)tcp + sizeof(*tcp) <= data_end) {
             id->src_port = __bpf_ntohs(tcp->source);
             id->dst_port = __bpf_ntohs(tcp->dest);
+            set_flags(tcp, flags);
         }
     } break;
     case IPPROTO_UDP: {
@@ -97,7 +138,7 @@ static inline int fill_iphdr(struct iphdr *ip, void *data_end, flow_id *id) {
 }
 
 // sets flow fields from IPv6 header information
-static inline int fill_ip6hdr(struct ipv6hdr *ip, void *data_end, flow_id *id) {
+static inline int fill_ip6hdr(struct ipv6hdr *ip, void *data_end, flow_id *id, u16 *flags) {
     if ((void *)ip + sizeof(*ip) > data_end) {
         return DISCARD;
     }
@@ -113,6 +154,7 @@ static inline int fill_ip6hdr(struct ipv6hdr *ip, void *data_end, flow_id *id) {
         if ((void *)tcp + sizeof(*tcp) <= data_end) {
             id->src_port = __bpf_ntohs(tcp->source);
             id->dst_port = __bpf_ntohs(tcp->dest);
+            set_flags(tcp, flags);
         }
     } break;
     case IPPROTO_UDP: {
@@ -128,7 +170,7 @@ static inline int fill_ip6hdr(struct ipv6hdr *ip, void *data_end, flow_id *id) {
     return SUBMIT;
 }
 // sets flow fields from Ethernet header information
-static inline int fill_ethhdr(struct ethhdr *eth, void *data_end, flow_id *id) {
+static inline int fill_ethhdr(struct ethhdr *eth, void *data_end, flow_id *id, u16 *flags) {
     if ((void *)eth + sizeof(*eth) > data_end) {
         return DISCARD;
     }
@@ -138,23 +180,22 @@ static inline int fill_ethhdr(struct ethhdr *eth, void *data_end, flow_id *id) {
 
     if (id->eth_protocol == ETH_P_IP) {
         struct iphdr *ip = (void *)eth + sizeof(*eth);
-        return fill_iphdr(ip, data_end, id);
+        return fill_iphdr(ip, data_end, id, flags);
     } else if (id->eth_protocol == ETH_P_IPV6) {
         struct ipv6hdr *ip6 = (void *)eth + sizeof(*eth);
-        return fill_ip6hdr(ip6, data_end, id);
+        return fill_ip6hdr(ip6, data_end, id, flags);
     } else {
         // TODO : Need to implement other specific ethertypes if needed
         // For now other parts of flow id remain zero
-        memset (&(id->src_ip),0, sizeof(struct in6_addr));
-        memset (&(id->dst_ip),0, sizeof(struct in6_addr));
+        memset(&(id->src_ip), 0, sizeof(struct in6_addr));
+        memset(&(id->dst_ip), 0, sizeof(struct in6_addr));
         id->transport_protocol = 0;
         id->src_port = 0;
         id->dst_port = 0;
     }
     return SUBMIT;
 }
 
-
 static inline int flow_monitor(struct __sk_buff *skb, u8 direction) {
     // If sampling is defined, will only parse 1 out of "sampling" flows
     if (sampling != 0 && (bpf_get_prandom_u32() % sampling) != 0) {
@@ -166,7 +207,8 @@ static inline int flow_monitor(struct __sk_buff *skb, u8 direction) {
     flow_id id;
     u64 current_time = bpf_ktime_get_ns();
     struct ethhdr *eth = data;
-    if (fill_ethhdr(eth, data_end, &id) == DISCARD) {
+    u16 flags = 0;
+    if (fill_ethhdr(eth, data_end, &id, &flags) == DISCARD) {
         return TC_ACT_OK;
     }
     id.if_index = skb->ifindex;
@@ -184,7 +226,7 @@ static inline int flow_monitor(struct __sk_buff *skb, u8 direction) {
         if (aggregate_flow->start_mono_time_ts == 0) {
             aggregate_flow->start_mono_time_ts = current_time;
         }
-
+        aggregate_flow->flags |= flags;
         long ret = bpf_map_update_elem(&aggregated_flows, &id, aggregate_flow, BPF_ANY);
         if (trace_messages && ret != 0) {
             // usually error -16 (-EBUSY) is printed here.
@@ -198,9 +240,10 @@ static inline int flow_monitor(struct __sk_buff *skb, u8 direction) {
         // Key does not exist in the map, and will need to create a new entry.
         flow_metrics new_flow = {
             .packets = 1,
-            .bytes=skb->len,
+            .bytes = skb->len,
             .start_mono_time_ts = current_time,
             .end_mono_time_ts = current_time,
+            .flags = flags, 
         };
 
         // even if we know that the entry is new, another CPU might be concurrently inserting a flow
@@ -230,15 +273,14 @@ static inline int flow_monitor(struct __sk_buff *skb, u8 direction) {
         }
     }
     return TC_ACT_OK;
-
 }
 SEC("tc_ingress")
-int ingress_flow_parse (struct __sk_buff *skb) {
+int ingress_flow_parse(struct __sk_buff *skb) {
     return flow_monitor(skb, INGRESS);
 }
 
 SEC("tc_egress")
-int egress_flow_parse (struct __sk_buff *skb) {
+int egress_flow_parse(struct __sk_buff *skb) {
     return flow_monitor(skb, EGRESS);
 }
 char _license[] SEC("license") = "GPL";

examples/flowlogs-dump/server/flowlogs-dump-collector.go

@@ -61,7 +61,7 @@ func main() {
 	log.SetFlags(0)
 	flag.Parse()
 
-	receivedRecords := make(chan *pbflow.Records, 100)
+	receivedRecords := make(chan *pbflow.Records, 1000)
 	log.Println("starting flowlogs-dump-collector on port", *port)
 	go func() {
 		_, err := grpc.StartCollector(*port, receivedRecords)
@@ -72,7 +72,7 @@ func main() {
 	for records := range receivedRecords {
 		for _, record := range records.Entries {
 			if record.EthProtocol == ipv6 {
-				log.Printf("%s: %v %s IP %s:%d > %s:%d: protocol:%s dir:%d bytes:%d packets:%d ends: %v\n",
+				log.Printf("%s: %v %s IP %s:%d > %s:%d: protocol:%s dir:%d bytes:%d packets:%d flags:%d ends: %v\n",
 					ipProto[record.EthProtocol],
 					record.TimeFlowStart.AsTime().Local().Format("15:04:05.000000"),
 					record.Interface,
@@ -84,10 +84,11 @@ func main() {
 					record.Direction,
 					record.Bytes,
 					record.Packets,
+					record.Flags,
 					record.TimeFlowEnd.AsTime().Local().Format("15:04:05.000000"),
 				)
 			} else {
-				log.Printf("%s: %v %s IP %s:%d > %s:%d: protocol:%s dir:%d bytes:%d packets:%d ends: %v\n",
+				log.Printf("%s: %v %s IP %s:%d > %s:%d: protocol:%s dir:%d bytes:%d packets:%d flags:%d ends: %v\n",
 					ipProto[record.EthProtocol],
 					record.TimeFlowStart.AsTime().Local().Format("15:04:05.000000"),
 					record.Interface,
@@ -99,6 +100,7 @@ func main() {
 					record.Direction,
 					record.Bytes,
 					record.Packets,
+					record.Flags,
 					record.TimeFlowEnd.AsTime().Local().Format("15:04:05.000000"),
 				)
 			}

go.mod

@@ -6,6 +6,7 @@ require (
 	github.com/caarlos0/env/v6 v6.9.1
 	github.com/cilium/ebpf v0.8.1
 	github.com/gavv/monotime v0.0.0-20190418164738-30dba4353424
+	github.com/golang/protobuf v1.5.2
 	github.com/mariomac/guara v0.0.0-20220523124851-5fc279816f1f
 	github.com/netobserv/gopipes v0.3.0
 	github.com/paulbellamy/ratecounter v0.2.0
@@ -35,7 +36,6 @@ require (
 	github.com/go-openapi/jsonreference v0.19.5 // indirect
 	github.com/go-openapi/swag v0.19.14 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
-	github.com/golang/protobuf v1.5.2 // indirect
 	github.com/google/gnostic v0.5.7-v3refs // indirect
 	github.com/google/go-cmp v0.5.7 // indirect
 	github.com/google/gofuzz v1.1.0 // indirect