Better client detection for kprobes

bpf/http_sock.c

@@ -235,7 +235,7 @@ int BPF_KPROBE(kprobe_tcp_sendmsg, struct sock *sk, struct msghdr *msg, size_t s
         if (size > 0) {
             void *iovec_ptr = find_msghdr_buf(msg);
             if (iovec_ptr) {
-                handle_buf_with_connection(&info, iovec_ptr, size, 0);
+                handle_buf_with_connection(&info, iovec_ptr, size, 0, TCP_SEND);
             } else {
                 bpf_dbg_printk("can't find iovec ptr in msghdr, not tracking sendmsg");
             }
@@ -315,7 +315,7 @@ int BPF_KRETPROBE(kretprobe_tcp_recvmsg, int copied_len) {
         sort_connection_info(&info.conn);
         //dbg_print_http_connection_info(&info.conn);
         info.pid = pid_from_pid_tgid(id);
-        handle_buf_with_connection(&info, (void *)args->iovec_ptr, copied_len, 0);
+        handle_buf_with_connection(&info, (void *)args->iovec_ptr, copied_len, 0, TCP_RECV);
     }
 
 done:

bpf/http_sock.h

@@ -278,13 +278,20 @@ static __always_inline void process_http_response(http_info_t *info, unsigned ch
     info->status += (buf[RESPONSE_STATUS_POS + 2] - '0');
 }
 
-static __always_inline void handle_http_response(unsigned char *small_buf, pid_connection_info_t *pid_conn, http_info_t *info, int orig_len) {
+static __always_inline void handle_http_response(unsigned char *small_buf, pid_connection_info_t *pid_conn, http_info_t *info, int orig_len, u8 direction) {
     http_connection_metadata_t *meta = bpf_map_lookup_elem(&filtered_connections, pid_conn);
-    http_connection_metadata_t dummy_meta = {
-        .type = EVENT_HTTP_REQUEST
-    };
+    http_connection_metadata_t dummy_meta = {};
 
     if (!meta) {
+        // In case we can't find metadata stored by accept4 or sys_connect, we guess the
+        // HTTP type. If we are receiving a response, we should be a client, otherwise 
+        // guess a server.
+        if (direction == TCP_RECV) {
+            dummy_meta.type = EVENT_HTTP_CLIENT;
+        } else {
+            dummy_meta.type = EVENT_HTTP_REQUEST;
+        }
+
         task_pid(&dummy_meta.pid);
         meta = &dummy_meta;
     }
@@ -293,15 +300,22 @@ static __always_inline void handle_http_response(unsigned char *small_buf, pid_c
     finish_http(info);
 }
 
-static __always_inline void http2_grpc_start(http2_conn_stream_t *s_key, void *u_buf, int len) {
+static __always_inline void http2_grpc_start(http2_conn_stream_t *s_key, void *u_buf, int len, u8 direction) {
     http2_grpc_request_t *h2g_info = empty_http2_info();
     if (h2g_info) {
         http_connection_metadata_t *meta = bpf_map_lookup_elem(&filtered_connections, &s_key->pid_conn);
-        http_connection_metadata_t dummy_meta = {
-            .type = EVENT_HTTP_REQUEST
-        };
+        http_connection_metadata_t dummy_meta = {};
 
         if (!meta) {
+            // In case we can't find metadata stored by accept4 or sys_connect, we guess the
+            // HTTP type. If we are making a request, we should be a client, otherwise 
+            // guess a server.
+            if (direction == TCP_SEND) {
+                dummy_meta.type = EVENT_HTTP_CLIENT;
+            } else {
+                dummy_meta.type = EVENT_HTTP_REQUEST;
+            }
+
             task_pid(&dummy_meta.pid);
             meta = &dummy_meta;
         }
@@ -335,16 +349,15 @@ static __always_inline void http2_grpc_end(http2_conn_stream_t *stream, http2_gr
     bpf_map_delete_elem(&ongoing_http2_grpc, stream);
 }
 
-static __always_inline void process_http2_grpc_frames(pid_connection_info_t *pid_conn, void *u_buf, int bytes_len) {
+static __always_inline void process_http2_grpc_frames(pid_connection_info_t *pid_conn, void *u_buf, int bytes_len, u8 direction) {
     int pos = 0;
     u8 found_frame = 0;
     http2_grpc_request_t *prev_info = 0;
     u32 saved_stream_id = 0;
     int saved_buf_pos = 0;
     u8 found_data_frame = 0;
 
-    for (int i = 0; i < 8; i++) {
-        u8 found = 0;
+    for (int i = 0; i < 4; i++) {
         unsigned char frame_buf[FRAME_HEADER_LEN];
         frame_header_t frame = {0};
 
@@ -355,7 +368,7 @@ static __always_inline void process_http2_grpc_frames(pid_connection_info_t *pid
         bpf_probe_read(&frame_buf, FRAME_HEADER_LEN, (void *)((u8 *)u_buf + pos));
         read_http2_grpc_frame_header(&frame, frame_buf, FRAME_HEADER_LEN);
 
-        bpf_dbg_printk("http2 frame type = %d, len = %d, stream_id = %d, flags = %d", frame.type, frame.length, frame.stream_id, frame.flags);
+        //bpf_dbg_printk("http2 frame type = %d, len = %d, stream_id = %d, flags = %d", frame.type, frame.length, frame.stream_id, frame.flags);
 
         if (is_headers_frame(&frame)) {
             http2_conn_stream_t stream = {0};
@@ -373,12 +386,12 @@ static __always_inline void process_http2_grpc_frames(pid_connection_info_t *pid
                     found_frame = 1;
                 } 
             } else {
-                http2_grpc_start(&stream, (void *)((u8 *)u_buf + pos), bytes_len);
+                http2_grpc_start(&stream, (void *)((u8 *)u_buf + pos), bytes_len, direction);
                 found_frame = 1;
             }
         } 
 
-        if (found) {
+        if (found_frame) {
             break;
         }
 
@@ -387,18 +400,18 @@ static __always_inline void process_http2_grpc_frames(pid_connection_info_t *pid
         }
 
         if (is_invalid_frame(&frame)) {
-            bpf_dbg_printk("Invalid frame, terminating search");
+            //bpf_dbg_printk("Invalid frame, terminating search");
             break;
         }
 
         if (frame.length + FRAME_HEADER_LEN >= bytes_len) {
-            bpf_dbg_printk("Frame length bigger than bytes len");
+            //bpf_dbg_printk("Frame length bigger than bytes len");
             break;
         }
 
         if (pos < (bytes_len - frame.length + FRAME_HEADER_LEN)) {
             pos += (frame.length + FRAME_HEADER_LEN);
-            bpf_dbg_printk("New buf read pos = %d", pos);
+            //bpf_dbg_printk("New buf read pos = %d", pos);
         }
     }
 
@@ -413,7 +426,7 @@ static __always_inline void process_http2_grpc_frames(pid_connection_info_t *pid
     }
 }
 
-static __always_inline void handle_buf_with_connection(pid_connection_info_t *pid_conn, void *u_buf, int bytes_len, u8 ssl) {
+static __always_inline void handle_buf_with_connection(pid_connection_info_t *pid_conn, void *u_buf, int bytes_len, u8 ssl, u8 direction) {
     unsigned char small_buf[MIN_HTTP2_SIZE] = {0};   // MIN_HTTP2_SIZE > MIN_HTTP_SIZE
     bpf_probe_read(small_buf, MIN_HTTP2_SIZE, u_buf);
 
@@ -474,7 +487,7 @@ static __always_inline void handle_buf_with_connection(pid_connection_info_t *pi
             bpf_probe_read(info->buf, FULL_BUF_SIZE, u_buf);
             process_http_request(info, bytes_len);
         } else if (packet_type == PACKET_TYPE_RESPONSE) {
-            handle_http_response(small_buf, pid_conn, info, bytes_len);
+            handle_http_response(small_buf, pid_conn, info, bytes_len, direction);
         } else if (still_reading(info)) {
             info->len += bytes_len;
         }     
@@ -488,7 +501,7 @@ static __always_inline void handle_buf_with_connection(pid_connection_info_t *pi
     } else {
         u8 *h2g = bpf_map_lookup_elem(&ongoing_http2_connections, pid_conn);
         if (h2g && *h2g == ssl) {
-            process_http2_grpc_frames(pid_conn, u_buf, bytes_len);
+            process_http2_grpc_frames(pid_conn, u_buf, bytes_len, direction);
         }
     }
 }

bpf/http_ssl.c

@@ -79,7 +79,7 @@ int BPF_URETPROBE(uretprobe_ssl_read, int ret) {
 
     ssl_args_t *args = bpf_map_lookup_elem(&active_ssl_read_args, &id);
 
-    handle_ssl_buf(id, args, ret);
+    handle_ssl_buf(id, args, ret, TCP_RECV);
     bpf_map_delete_elem(&active_ssl_read_args, &id);
     return 0;
 }
@@ -124,7 +124,7 @@ int BPF_URETPROBE(uretprobe_ssl_read_ex, int ret) {
     size_t read_len = 0;
     bpf_probe_read(&read_len, sizeof(read_len), (void *)args->len_ptr);
 
-    handle_ssl_buf(id, args, read_len);
+    handle_ssl_buf(id, args, read_len, TCP_RECV);
 done:
     bpf_map_delete_elem(&active_ssl_read_args, &id);
     return 0;
@@ -165,7 +165,7 @@ int BPF_URETPROBE(uretprobe_ssl_write, int ret) {
 
     ssl_args_t *args = bpf_map_lookup_elem(&active_ssl_write_args, &id);
 
-    handle_ssl_buf(id, args, ret);
+    handle_ssl_buf(id, args, ret, TCP_SEND);
     bpf_map_delete_elem(&active_ssl_write_args, &id);
     return 0;
 }
@@ -209,7 +209,7 @@ int BPF_URETPROBE(uretprobe_ssl_write_ex, int ret) {
     size_t wrote_len = 0;
     bpf_probe_read(&wrote_len, sizeof(wrote_len), (void *)args->len_ptr);
 
-    handle_ssl_buf(id, args, wrote_len);
+    handle_ssl_buf(id, args, wrote_len, TCP_SEND);
 done:
     bpf_map_delete_elem(&active_ssl_write_args, &id);
     return 0;

bpf/http_ssl.h

@@ -75,7 +75,7 @@ struct {
     __uint(pinning, LIBBPF_PIN_BY_NAME);
 } active_ssl_write_args SEC(".maps");
 
-static __always_inline void handle_ssl_buf(u64 id, ssl_args_t *args, int bytes_len) {
+static __always_inline void handle_ssl_buf(u64 id, ssl_args_t *args, int bytes_len, u8 direction) {
     if (args && bytes_len > 0) {
         void *ssl = ((void *)args->ssl);
         u64 ssl_ptr = (u64)ssl;
@@ -147,7 +147,7 @@ static __always_inline void handle_ssl_buf(u64 id, ssl_args_t *args, int bytes_l
             //     bpf_dbg_printk("%x ", buf[i]);
             // }
 
-            handle_buf_with_connection(&pid_conn, (void *)args->buf, bytes_len, 1);
+            handle_buf_with_connection(&pid_conn, (void *)args->buf, bytes_len, 1, direction);
         } else {
             bpf_dbg_printk("No connection info! This is a bug.");
         }

bpf/http_types.h

@@ -23,6 +23,9 @@
 #define TP_MAX_VAL_LENGTH   55
 #define TP_MAX_KEY_LENGTH   11
 
+#define TCP_SEND 1
+#define TCP_RECV 0
+
 // Struct to keep information on the connections in flight 
 // s = source, d = destination
 // h = high word, l = low word

pkg/internal/export/debug/debug.go

@@ -21,10 +21,11 @@ func PrinterNode(_ PrintEnabled) (node.TerminalFunc[[]request.Span], error) {
 		for spans := range input {
 			for i := range spans {
 				t := spans[i].Timings()
-				fmt.Printf("%s (%s[%s]) %v %s %s [%s]->[%s:%d] size:%dB svc=[%s %s] traceparent=[%s]\n",
+				fmt.Printf("%s (%s[%s]) %s %v %s %s [%s]->[%s:%d] size:%dB svc=[%s %s] traceparent=[%s]\n",
 					t.Start.Format("2006-01-02 15:04:05.12345"),
 					t.End.Sub(t.RequestStart),
 					t.End.Sub(t.Start),
+					spanType(&spans[i]),
 					spans[i].Status,
 					spans[i].Method,
 					spans[i].Path,
@@ -48,6 +49,23 @@ func traceparent(span *request.Span) string {
 	return fmt.Sprintf("00-%s-%s-%02x", trace.TraceID(span.TraceID).String(), trace.SpanID(span.ParentSpanID).String(), span.Flags)
 }
 
+func spanType(span *request.Span) string {
+	switch span.Type {
+	case request.EventTypeHTTP:
+		return "SRV"
+	case request.EventTypeHTTPClient:
+		return "CLNT"
+	case request.EventTypeGRPC:
+		return "GRPC_SRV"
+	case request.EventTypeGRPCClient:
+		return "GRPC_CLNT"
+	case request.EventTypeSQLClient:
+		return "SQL"
+	}
+
+	return ""
+}
+
 type NoopEnabled bool
 
 func (n NoopEnabled) Enabled() bool {