Propagate context in http2 (#585)

bpf/go_nethttp.c

@@ -19,6 +19,7 @@
 #include "go_nethttp.h"
 #include "go_traceparent.h"
 #include "tracing.h"
+#include "hpack.h"
 
 typedef struct http_func_invocation {
     u64 start_monotime_ns;
@@ -210,9 +211,7 @@ struct {
 #endif
 
 /* HTTP Client. We expect to see HTTP client in both HTTP server and gRPC server calls.*/
-
-SEC("uprobe/roundTrip")
-int uprobe_roundTrip(struct pt_regs *ctx) {
+static __always_inline void roundTripStartHelper(struct pt_regs *ctx) {
     bpf_dbg_printk("=== uprobe/proc http roundTrip === ");
 
     void *goroutine_addr = GOROUTINE_PTR(ctx);
@@ -243,8 +242,12 @@ int uprobe_roundTrip(struct pt_regs *ctx) {
             bpf_map_update_elem(&header_req_map, &headers_ptr, &goroutine_addr, BPF_ANY);
         }
     }
-#endif    
+#endif
+}
 
+SEC("uprobe/roundTrip")
+int uprobe_roundTrip(struct pt_regs *ctx) {
+    roundTripStartHelper(ctx);
     return 0;
 }
 
@@ -391,3 +394,181 @@ int uprobe_http2ResponseWriterStateWriteHeader(struct pt_regs *ctx) {
 
     return writeHeaderHelper(ctx, rws_req_pos);
 }
+
+// HTTP 2.0 client support
+#ifndef NO_HEADER_PROPAGATION
+struct {
+    __uint(type, BPF_MAP_TYPE_LRU_HASH);
+    __type(key, u32); // key: stream id
+    __type(value, u64); // the goroutine of the round trip request, which is the key for our traceparent info
+    __uint(max_entries, MAX_CONCURRENT_REQUESTS);
+} http2_req_map SEC(".maps");
+#endif
+
+SEC("uprobe/http2RoundTrip")
+int uprobe_http2RoundTrip(struct pt_regs *ctx) {
+    // we use the usual start helper, just like for normal http calls, but we later save
+    // more context, like the streamID
+    roundTripStartHelper(ctx);
+
+#ifndef NO_HEADER_PROPAGATION
+    void *cc_ptr = GO_PARAM1(ctx);
+
+    if (cc_ptr) {
+        u32 stream_id = 0;
+        bpf_probe_read(&stream_id, sizeof(stream_id), (void *)(cc_ptr + cc_next_stream_id_pos));
+
+        bpf_dbg_printk("cc_ptr = %llx, nextStreamID=%d", cc_ptr, stream_id);
+        if (stream_id) {
+            void *goroutine_addr = GOROUTINE_PTR(ctx);
+
+            bpf_map_update_elem(&http2_req_map, &stream_id, &goroutine_addr, BPF_ANY);
+        }
+    }
+#endif    
+
+    return 0;
+}
+
+#ifndef NO_HEADER_PROPAGATION
+typedef struct framer_func_invocation {
+    u64 framer_ptr;
+    tp_info_t tp;
+} framer_func_invocation_t;
+
+struct {
+    __uint(type, BPF_MAP_TYPE_HASH);
+    __type(key, void*); // key: go routine doing framer write headers
+    __type(value, framer_func_invocation_t); // the goroutine of the round trip request, which is the key for our traceparent info
+    __uint(max_entries, MAX_CONCURRENT_REQUESTS);
+} framer_invocation_map SEC(".maps");
+
+SEC("uprobe/http2FramerWriteHeaders")
+int uprobe_http2FramerWriteHeaders(struct pt_regs *ctx) {
+    bpf_dbg_printk("=== uprobe/proc http2 Framer writeHeaders === ");
+
+    void *framer = GO_PARAM1(ctx);
+    u64 stream_id = (u64)GO_PARAM2(ctx);
+
+    bpf_printk("framer=%llx, stream_id=%lld", framer, ((u64)stream_id));
+
+    u32 stream_lookup = (u32)stream_id;
+
+    void **go_ptr = bpf_map_lookup_elem(&http2_req_map, &stream_lookup);
+    bpf_map_delete_elem(&http2_req_map, &stream_lookup);
+
+    if (go_ptr) {
+        void *go_addr = *go_ptr;
+        bpf_dbg_printk("Found existing stream data goaddr = %llx", go_addr);
+
+        http_func_invocation_t *info = bpf_map_lookup_elem(&ongoing_http_client_requests, &go_addr);
+
+        if (info) {
+            bpf_dbg_printk("Found func info %llx", info);
+            void *goroutine_addr = GOROUTINE_PTR(ctx);
+
+            framer_func_invocation_t f_info = {
+                .tp = info->tp,
+                .framer_ptr = (u64)framer,
+            };
+
+            bpf_map_update_elem(&framer_invocation_map, &goroutine_addr, &f_info, BPF_ANY);
+        }
+    }
+
+    return 0;
+}
+#else
+SEC("uprobe/http2FramerWriteHeaders")
+int uprobe_http2FramerWriteHeaders(struct pt_regs *ctx) {
+    return 0;
+}
+#endif
+
+#ifndef NO_HEADER_PROPAGATION
+#define HTTP2_ENCODED_HEADER_LEN 66 // 1 + 1 + 8 + 1 + 55 = type byte + hpack_len_as_byte("traceparent") + strlen(hpack("traceparent")) + len_as_byte(55) + generated traceparent id
+
+SEC("uprobe/http2FramerWriteHeaders_returns")
+int uprobe_http2FramerWriteHeaders_returns(struct pt_regs *ctx) {
+    bpf_dbg_printk("=== uprobe/proc http2 Framer writeHeaders returns === ");
+
+    void *goroutine_addr = GOROUTINE_PTR(ctx);
+
+    framer_func_invocation_t *f_info = bpf_map_lookup_elem(&framer_invocation_map, &goroutine_addr);
+    bpf_map_delete_elem(&framer_invocation_map, &goroutine_addr);
+
+    if (f_info) {
+        void *w_ptr = 0;
+        bpf_probe_read(&w_ptr, sizeof(w_ptr), (void *)(f_info->framer_ptr + framer_w_pos + 8));
+
+        if (w_ptr) {
+            void *buf_arr = 0;
+            s64 n = 0;
+            s64 cap = 0;
+
+            bpf_probe_read(&buf_arr, sizeof(buf_arr), (void *)(w_ptr + 16));
+            bpf_probe_read(&n, sizeof(n), (void *)(w_ptr + 40));
+            bpf_probe_read(&cap, sizeof(cap), (void *)(w_ptr + 24));
+
+            bpf_dbg_printk("Found f_info, this is the place to write to w = %llx, buf=%llx, n=%d, size=%d", w_ptr, buf_arr, n, cap);
+            if (buf_arr && n < (cap - HTTP2_ENCODED_HEADER_LEN)) {
+                uint8_t tp_str[TP_MAX_VAL_LENGTH];
+
+                u8 type_byte = 0;
+                u8 key_len = TP_ENCODED_LEN | 0x80; // high tagged to signify hpack encoded value
+                u8 val_len = TP_MAX_VAL_LENGTH;
+
+                // We don't hpack encode the value of the traceparent field, because that will require that 
+                // we use bpf_loop, which in turn increases the kernel requirement to 5.17+.
+                make_tp_string(tp_str, &f_info->tp);
+                bpf_dbg_printk("Will write %s, type = %d, key_len = %d, val_len = %d", tp_str, type_byte, key_len, val_len);
+
+                bpf_probe_write_user(buf_arr + (n & 0x0ffff), &type_byte, sizeof(type_byte));                        
+                n++;
+                // Write the length of the key = 8
+                bpf_probe_write_user(buf_arr + (n & 0x0ffff), &key_len, sizeof(key_len));
+                n++;
+                // Write 'traceparent' encoded as hpack
+                bpf_probe_write_user(buf_arr + (n & 0x0ffff), tp_encoded, sizeof(tp_encoded));;
+                n += TP_ENCODED_LEN;
+                // Write the length of the hpack encoded traceparent field 
+                bpf_probe_write_user(buf_arr + (n & 0x0ffff), &val_len, sizeof(val_len));
+                n++;
+                bpf_probe_write_user(buf_arr + (n & 0x0ffff), tp_str, sizeof(tp_str));
+                n += TP_MAX_VAL_LENGTH;
+                // Update the value of n in w to reflect the new size
+                bpf_probe_write_user((void *)(w_ptr + 40), &n, sizeof(n));
+
+                // http2 encodes the length of the headers in the first 3 bytes of buf, we need to update those
+                s8 size_1 = 0;
+                s8 size_2 = 0;
+                s8 size_3 = 0;
+
+                bpf_probe_read(&size_1, sizeof(size_1), (void *)(buf_arr));
+                bpf_probe_read(&size_2, sizeof(size_2), (void *)(buf_arr + 1));
+                bpf_probe_read(&size_3, sizeof(size_3), (void *)(buf_arr + 2));
+
+                s32 original_size = ((s32)(size_1) << 16) | ((s32)(size_2) << 8) | size_3;
+                s32 new_size = original_size + HTTP2_ENCODED_HEADER_LEN;
+
+                bpf_dbg_printk("Changing size from %d to %d", original_size, new_size);
+                size_1 = (s8)(new_size >> 16);
+                size_2 = (s8)(new_size >> 8);
+                size_3 = (s8)(new_size);
+
+                bpf_probe_write_user((void *)(buf_arr), &size_1, sizeof(size_1));
+                bpf_probe_write_user((void *)(buf_arr+1), &size_2, sizeof(size_2));
+                bpf_probe_write_user((void *)(buf_arr+2), &size_3, sizeof(size_3));
+            }
+        }
+    }
+
+    return 0;
+}
+#else
+SEC("uprobe/http2FramerWriteHeaders_returns")
+int uprobe_http2FramerWriteHeaders_returns(struct pt_regs *ctx) {
+    return 0;
+}
+#endif 
+

bpf/go_nethttp.h

@@ -30,5 +30,7 @@ volatile const u64 req_header_ptr_pos;
 volatile const u64 io_writer_buf_ptr_pos;
 volatile const u64 io_writer_n_pos;
 volatile const u64 rws_req_pos;
+volatile const u64 cc_next_stream_id_pos;
+volatile const u64 framer_w_pos;
 
 #endif

bpf/sockaddr.h

@@ -68,7 +68,7 @@ static __always_inline u16 get_sockaddr_port(struct sockaddr *addr) {
     BPF_CORE_READ_INTO(&sa_family, addr, sa_family);
     u16 bport = 0;
 
-    bpf_dbg_printk("addr = %llx, sa_family %d", addr, sa_family);
+    //bpf_dbg_printk("addr = %llx, sa_family %d", addr, sa_family);
 
     if (sa_family == AF_INET) {
         struct sockaddr_in *baddr = (struct sockaddr_in *)addr;

configs/offsets/http2/inspect.go

@@ -1,6 +1,8 @@
 package main
 
 import (
+	"context"
+	"crypto/tls"
 	"fmt"
 	"net/http"
 	"os"
@@ -16,6 +18,22 @@ func checkErr(err error, msg string) {
 	os.Exit(1)
 }
 
+func roundTripExample() {
+	req, err := http.NewRequestWithContext(context.Background(), "GET", os.Getenv("TARGET_URL")+"/pingrt", nil)
+	checkErr(err, "during new request")
+
+	tr := &http2.Transport{
+		TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
+	}
+
+	resp, err := tr.RoundTrip(req)
+	checkErr(err, "during roundtrip")
+
+	if err == nil {
+		fmt.Printf("RoundTrip Proto: %d\n", resp.ProtoMajor)
+	}
+}
+
 func main() {
 	handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		fmt.Fprintf(w, "Hello, %v, http: %v\n", r.URL.Path, r.TLS == nil)
@@ -27,6 +45,7 @@ func main() {
 	}
 	http2.ConfigureServer(server, nil)
 
+	roundTripExample()
 	fmt.Printf("Listening [0.0.0.0:8080]...\n")
 	checkErr(server.ListenAndServeTLS("cert.pem", "key.pem"), "while listening")
 }

configs/offsets/tracker_input.json

@@ -46,6 +46,12 @@
       ],
       "golang.org/x/net/http2/hpack.Encoder": [
         "w"
+      ],
+      "golang.org/x/net/http2.ClientConn": [
+        "nextStreamID"
+      ],
+      "golang.org/x/net/http2.Framer": [
+        "w"
       ]
     }
   },