Network metrics: report client and server port

bpf/flow.h

@@ -46,6 +46,8 @@ typedef struct flow_metrics_t {
     u16 flags;
     // direction of the flow EGRESS / INGRESS
     u8 direction;
+    // who initiated of the connection: INITIATOR_SRC or INITIATOR_DST
+    u8 initiator;
     // The positive errno of a failed map insertion that caused a flow
     // to be sent via ringbuffer.
     // 0 otherwise

bpf/flows.c

@@ -195,6 +195,7 @@ static inline int flow_monitor(struct __sk_buff *skb) {
             .end_mono_time_ns = current_time,
             .flags = flags,
             .direction = UNKNOWN,
+            .initiator = INITIATOR_UNKNOWN,
         };
 
         u8 *direction = (u8 *)bpf_map_lookup_elem(&flow_directions, &id);
@@ -225,6 +226,8 @@ static inline int flow_monitor(struct __sk_buff *skb) {
             new_flow.direction = *direction;
         }
 
+        new_flow.initiator = get_connection_initiator(&id, flags);
+
         // even if we know that the entry is new, another CPU might be concurrently inserting a flow
         // so we need to specify BPF_ANY
         long ret = bpf_map_update_elem(&aggregated_flows, &id, &new_flow, BPF_ANY);

bpf/flows_common.h

@@ -31,6 +31,16 @@
 #define FIN_ACK_FLAG 0x200
 #define RST_ACK_FLAG 0x400
 
+// In conn_initiator_key, which sorted ip:port inititated the connection
+#define INITIATOR_LOW 1
+#define INITIATOR_HIGH 2
+
+// In flow_metrics, who initiated the connection
+#define INITIATOR_SRC 1
+#define INITIATOR_DST 2
+
+#define INITIATOR_UNKNOWN 0
+
 // Common Ringbuffer as a conduit for ingress/egress flows to userspace
 struct {
     __uint(type, BPF_MAP_TYPE_RINGBUF);
@@ -47,16 +57,138 @@ struct {
 
 // Key: the flow identifier. Value: the flow direction.
 struct {
-	__uint(type, BPF_MAP_TYPE_LRU_HASH);
-	__type(key, flow_id);
-	__type(value, u8);
+    __uint(type, BPF_MAP_TYPE_LRU_HASH);
+    __type(key, flow_id);
+    __type(value, u8);
 } flow_directions SEC(".maps");
 
+// To know who initiated each connection, we store the src/dst ip:ports but ordered
+// by numeric value of the IP (and port as secondary criteria), so the key is consistent
+// for either client and server flows.
+typedef struct conn_initiator_key_t {
+    struct in6_addr low_ip;
+    struct in6_addr high_ip;
+    u16 low_ip_port;
+    u16 high_ip_port;
+} __attribute__((packed)) conn_initiator_key;
+
+// Key: the flow identifier.
+// Value: the connection initiator index (INITIATOR_LOW, INITIATOR_HIGH).
+struct {
+    __uint(type, BPF_MAP_TYPE_LRU_HASH);
+    __type(key, conn_initiator_key);
+    __type(value, u8);
+} conn_initiators SEC(".maps");
+
 const u8 ip4in6[] = {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0xff, 0xff};
 
 // Constant definitions, to be overridden by the invoker
 volatile const u32 sampling = 0;
 volatile const u8 trace_messages = 0;
 
+// we can safely assume that the passed address is IPv6 as long as we encode IPv4
+// as IPv6 during the creation of the flow_id.
+static inline s32 compare_ipv6(flow_id *fid) {
+    for (int i = 0; i < 4; i++) {
+        s32 diff = fid->src_ip.in6_u.u6_addr32[i] - fid->dst_ip.in6_u.u6_addr32[i];
+        if (diff != 0) {
+            return diff;
+        }
+    }
+    return 0;
+}
+
+// creates a key that is consistent for both requests and responses, by
+// ordering endpoints (ip:port) numerically into a lower and a higher endpoint.
+// returns true if the lower address corresponds to the source address
+// (false if the lower address corresponds to the destination address)
+static inline u8 fill_conn_initiator_key(flow_id *id, conn_initiator_key *key) {
+    s32 cmp = compare_ipv6(id);
+    if (cmp < 0) {
+        __builtin_memcpy(&key->low_ip, &id->src_ip, sizeof(struct in6_addr));
+        key->low_ip_port = id->src_port;
+        __builtin_memcpy(&key->high_ip, &id->dst_ip, sizeof(struct in6_addr));
+        key->high_ip_port = id->dst_port;
+        return 1;
+    }
+    // if the IPs are equal (cmp == 0) we will use the ports as secondary order criteria
+    __builtin_memcpy(&key->high_ip, &id->src_ip, sizeof(struct in6_addr));
+    __builtin_memcpy(&key->low_ip, &id->dst_ip, sizeof(struct in6_addr));
+    if (cmp > 0 || id->src_port > id->dst_port) {
+        key->high_ip_port = id->src_port;
+        key->low_ip_port = id->dst_port;
+        return 0;
+    }
+    key->low_ip_port = id->src_port;
+    key->high_ip_port = id->dst_port;
+    return 1;
+}
+
+// returns INITIATOR_SRC or INITIATOR_DST, but might return INITIATOR_UNKNOWN
+// if the connection initiator couldn't be found. The user-space Beyla pipeline
+// will handle this last case heuristically
+static inline u8 get_connection_initiator(flow_id *id, u16 flags) {
+    conn_initiator_key initiator_key;
+    // from the initiator_key with sorted ip/ports, know the index of the
+    // endpoint that that initiated the connection, which might be the low or the high address
+    u8 low_is_src = fill_conn_initiator_key(id, &initiator_key);
+    u8 *initiator = (u8 *)bpf_map_lookup_elem(&conn_initiators, &initiator_key);
+    u8 initiator_index = INITIATOR_UNKNOWN;
+    if (initiator == NULL) {
+        // SYN and ACK is sent from the server to the client
+        // The initiator is the destination address
+        if ((flags & (SYN_FLAG | ACK_FLAG)) == (SYN_FLAG | ACK_FLAG)) {
+            if (low_is_src) {
+                initiator_index = INITIATOR_HIGH;
+            } else {
+                initiator_index = INITIATOR_LOW;
+            }
+        }
+        // SYN is sent from the client to the server.
+        // The initiator is the source address
+        else if (flags & SYN_FLAG) {
+            if (low_is_src) {
+                initiator_index = INITIATOR_LOW;
+            } else {
+                initiator_index = INITIATOR_HIGH;
+            }
+        }
+
+        if (initiator_index != INITIATOR_UNKNOWN) {
+            bpf_map_update_elem(&conn_initiators, &initiator_key, &initiator_index, BPF_NOEXIST);
+        }
+    } else {
+        initiator_index = *initiator;
+    }
+
+    // when flow receives FIN or RST, clean flow_directions
+    if (flags & FIN_FLAG || flags & RST_FLAG || flags & FIN_ACK_FLAG || flags & RST_ACK_FLAG) {
+        bpf_map_delete_elem(&conn_initiators, &initiator_key);
+    }
+
+    u8 flow_initiator = INITIATOR_UNKNOWN;
+    // at this point, we should know the index of the endpoint that initiated the connection.
+    // Then we accordingly set whether the initiator is the source or the destination address.
+    // If not, we forward the unknown status and the userspace will take
+    // heuristic actions to guess who is
+    switch (initiator_index) {
+    case INITIATOR_LOW:
+        if (low_is_src) {
+            flow_initiator = INITIATOR_SRC;
+        } else {
+            flow_initiator = INITIATOR_DST;
+        }
+        break;
+    case INITIATOR_HIGH:
+        if (low_is_src) {
+            flow_initiator = INITIATOR_DST;
+        } else {
+            flow_initiator = INITIATOR_SRC;
+        }
+        break;
+    }
+
+    return flow_initiator;
+}
 
 #endif //__FLOW_HELPERS_H__

bpf/flows_sock.c

@@ -241,6 +241,8 @@ int socket__http_filter(struct __sk_buff *skb) {
             new_flow.direction = *direction;
         }
 
+        new_flow.initiator = get_connection_initiator(&id, flags);
+
         // even if we know that the entry is new, another CPU might be concurrently inserting a flow
         // so we need to specify BPF_ANY
         long ret = bpf_map_update_elem(&aggregated_flows, &id, &new_flow, BPF_ANY);

pkg/internal/export/attributes/attr_defs.go

@@ -204,6 +204,8 @@ func getDefinitions(groups AttrGroups) map[Section]AttrReportGroup {
 				attr.DstPort:    false,
 				attr.SrcName:    false,
 				attr.DstName:    false,
+				attr.ServerPort: false,
+				attr.ClientPort: false,
 				attr.Direction:  Default(ifaceDirEnabled),
 				attr.Iface:      Default(ifaceDirEnabled),
 			},

pkg/internal/export/attributes/attr_selector_test.go

@@ -108,7 +108,9 @@ func TestFor_GlobEntries_Order(t *testing.T) {
 	require.NoError(t, err)
 	assert.Equal(t, []attr.Name{
 		"beyla.ip",
+		"client.port",
 		"dst.name",
+		"server.port",
 		"src.address",
 		"src.name",
 		"src.port",

pkg/internal/export/attributes/names/attrs.go

@@ -81,6 +81,8 @@ const (
 	SrcCIDR    = Name("src.cidr")
 	DstCIDR    = Name("dst.cidr")
 
+	ClientPort = Name("client.port")
+
 	K8sSrcOwnerName = Name("k8s.src.owner.name")
 	K8sSrcNamespace = Name("k8s.src.namespace")
 	K8sDstOwnerName = Name("k8s.dst.owner.name")