bpf: use configurable large buffers for HTTP requests

Signed-off-by: Mattia Meleleo <[email protected]>

Author: mmat11

bpf/common/common.h

@@ -42,6 +42,7 @@
 #define HTTP_HEADER_MAX_LEN 100
 #define HTTP_CONTENT_TYPE_MAX_LEN 16
 
+volatile const u32 http_buffer_size = 0;
 volatile const u32 mysql_buffer_size = 0;
 volatile const u32 postgres_buffer_size = 0;
 
@@ -241,4 +242,4 @@ typedef struct mongo_go_client_req {
     unsigned char coll[32];
     connection_info_t conn;
     tp_info_t tp;
-} mongo_go_client_req_t;
+} mongo_go_client_req_t;

bpf/common/http_info.h

@@ -30,5 +30,6 @@ typedef struct http_info {
     u32 task_tid;
     u16 status;
     unsigned char buf[FULL_BUF_SIZE];
-    u8 _pad[6];
+    u8 has_large_buffers;
+    u8 _pad[5];
 } http_info_t;

bpf/generictracer/iovec_len.h

@@ -3,4 +3,4 @@
 
 #pragma once
 
-enum { k_iovec_max_len = 512 };
+enum { k_iovec_max_len = 8192 };

bpf/generictracer/protocol_http.h

@@ -6,11 +6,13 @@
 #include <bpfcore/vmlinux.h>
 #include <bpfcore/bpf_helpers.h>
 
+#include <common/common.h>
 #include <common/http_types.h>
 #include <common/pin_internal.h>
 #include <common/ringbuf.h>
 #include <common/runtime.h>
 #include <common/trace_common.h>
+#include <common/scratch_mem.h>
 
 #include <generictracer/maps/http_info_mem.h>
 #include <generictracer/protocol_common.h>
@@ -21,6 +23,11 @@
 
 volatile const u32 high_request_volume;
 
+enum {
+    k_http_large_buf_max_size = 1 << 14, // 16K
+    k_http_large_buf_max_size_mask = k_http_large_buf_max_size - 1,
+};
+
 // empty_http_info zeroes and return the unique percpu copy in the map
 // this function assumes that a given thread is not trying to use many
 // instances at the same time
@@ -401,6 +408,52 @@ static __always_inline void handle_http_response(unsigned char *small_buf,
     cleanup_http_request_data(pid_conn, info);
 }
 
+SCRATCH_MEM_SIZED(http_large_buffers, k_http_large_buf_max_size);
+
+static __always_inline int http_send_large_buffer(http_info_t *req,
+                                                  const void *u_buf,
+                                                  u32 bytes_len,
+                                                  u8 packet_type,
+                                                  enum large_buf_action action) {
+    if (http_buffer_size == 0) {
+        return 0;
+    }
+
+    if (!is_pow2(http_buffer_size)) {
+        bpf_dbg_printk("http_send_large_buffer: bug: http_buffer_size is not a power of 2");
+        return -1;
+    }
+    const u32 buf_len_mask = http_buffer_size - 1;
+
+    tcp_large_buffer_t *large_buf = (tcp_large_buffer_t *)http_large_buffers_mem();
+    if (!large_buf) {
+        bpf_dbg_printk("http_send_large_buffer: failed to reserve space for HTTP large buffer");
+        return -1;
+    }
+
+    large_buf->type = EVENT_TCP_LARGE_BUFFER;
+    large_buf->packet_type = packet_type;
+    large_buf->action = action;
+    __builtin_memcpy((void *)&large_buf->tp, (void *)&req->tp, sizeof(tp_info_t));
+
+    large_buf->len = bytes_len;
+    if (large_buf->len >= http_buffer_size) {
+        large_buf->len = http_buffer_size;
+        bpf_dbg_printk("WARN: http_send_large_buffer: buffer is full, truncating data");
+    }
+
+    bpf_probe_read(large_buf->buf, large_buf->len & buf_len_mask, u_buf);
+
+    u32 total_size = sizeof(tcp_large_buffer_t);
+    total_size += large_buf->len > sizeof(void *) ? large_buf->len : sizeof(void *);
+
+    req->has_large_buffers = true;
+
+    bpf_ringbuf_output(
+        &events, large_buf, total_size & k_http_large_buf_max_size_mask, get_flags());
+    return 0;
+}
+
 // k_tail_protocol_http
 SEC("kprobe/http")
 int obi_protocol_http(void *ctx) {
@@ -478,14 +531,31 @@ int obi_protocol_http(void *ctx) {
             bpf_dbg_printk("No META!");
         }
 
-        // we copy some small part of the buffer to the info trace event, so that we can process an event even with
-        // incomplete trace info in user space.
-        bpf_probe_read(info->buf, FULL_BUF_SIZE, (void *)args->u_buf);
+        if (http_buffer_size > 0) {
+            http_send_large_buffer(info,
+                                   (void *)args->u_buf,
+                                   args->bytes_len,
+                                   args->packet_type,
+                                   k_large_buf_action_init);
+        } else {
+            // we copy some small part of the buffer to the info trace event, so that we can process an event even with
+            // incomplete trace info in user space.
+            bpf_probe_read(info->buf, FULL_BUF_SIZE, (void *)args->u_buf);
+        }
+
         process_http_request(info, args->bytes_len, meta, args->direction, args->orig_dport);
     } else if ((args->packet_type == PACKET_TYPE_RESPONSE) && (info->status == 0)) {
+        http_send_large_buffer(
+            info, (void *)args->u_buf, args->bytes_len, args->packet_type, k_large_buf_action_init);
         handle_http_response(
             args->small_buf, &args->pid_conn, info, args->bytes_len, args->direction, args->ssl);
     } else if (still_reading(info)) {
+        http_send_large_buffer(info,
+                               (void *)args->u_buf,
+                               args->bytes_len,
+                               args->packet_type,
+                               k_large_buf_action_append);
+
         info->len += args->bytes_len;
         info->end_monotime_ns = bpf_ktime_get_ns();
     }

bpf/generictracer/protocol_mysql.h

@@ -133,7 +133,7 @@ static __always_inline int mysql_read_fixup_buffer(const connection_info_t *conn
         bpf_dbg_printk("mysql_read_fixup_buffer: bug: mysql_buffer_size is not a power of 2");
         return -1;
     }
-    const u8 buf_len_mask = mysql_buffer_size - 1;
+    const u32 buf_len_mask = mysql_buffer_size - 1;
 
     struct mysql_state_data *state_data = bpf_map_lookup_elem(&mysql_state, conn_info);
     if (state_data != NULL) {

bpf/generictracer/protocol_postgres.h

@@ -62,7 +62,7 @@ static __always_inline int postgres_send_large_buffer(tcp_req_t *req,
         bpf_dbg_printk("postgres_send_large_buffer: bug: postgres_buffer_size is not a power of 2");
         return -1;
     }
-    const u8 buf_len_mask = postgres_buffer_size - 1;
+    const u32 buf_len_mask = postgres_buffer_size - 1;
 
     tcp_large_buffer_t *large_buf = (tcp_large_buffer_t *)postgres_large_buffers_mem();
     if (!large_buf) {

pkg/components/ebpf/common/common.go

@@ -223,7 +223,7 @@ func ReadBPFTraceAsSpan(parseCtx *EBPFParseContext, cfg *config.EBPFTracer, reco
 	case EventTypeSQL:
 		return ReadSQLRequestTraceAsSpan(record)
 	case EventTypeKHTTP:
-		return ReadHTTPInfoIntoSpan(record, filter)
+		return ReadHTTPInfoIntoSpan(parseCtx, record, filter)
 	case EventTypeKHTTP2:
 		return ReadHTTP2InfoIntoSpan(parseCtx, record, filter)
 	case EventTypeTCP:

pkg/components/ebpf/common/httpfltr_test.go

@@ -23,19 +23,19 @@ func TestURL(t *testing.T) {
 	event := BPFHTTPInfo{
 		Buf: [bufSize]byte{'G', 'E', 'T', ' ', '/', 'p', 'a', 't', 'h', '?', 'q', 'u', 'e', 'r', 'y', '=', '1', '2', '3', '4', ' ', 'H', 'T', 'T', 'P', '/', '1', '.', '1'},
 	}
-	assert.Equal(t, "/path?query=1234", event.url())
+	assert.Equal(t, "/path?query=1234", httpURLFromBuf(event.Buf[:]))
 	event = BPFHTTPInfo{}
-	assert.Empty(t, event.url())
+	assert.Empty(t, httpURLFromBuf(event.Buf[:]))
 }
 
 func TestMethod(t *testing.T) {
 	event := BPFHTTPInfo{
 		Buf: [bufSize]byte{'G', 'E', 'T', ' ', '/', 'p', 'a', 't', 'h', ' ', 'H', 'T', 'T', 'P', '/', '1', '.', '1'},
 	}
 
-	assert.Equal(t, "GET", event.method())
+	assert.Equal(t, "GET", httpMethodFromBuf(event.Buf[:]))
 	event = BPFHTTPInfo{}
-	assert.Empty(t, event.method())
+	assert.Empty(t, httpMethodFromBuf(event.Buf[:]))
 }
 
 func TestHostInfo(t *testing.T) {
@@ -108,7 +108,7 @@ func TestToRequestTrace(t *testing.T) {
 	err := binary.Write(buf, binary.LittleEndian, &record)
 	require.NoError(t, err)
 
-	result, _, err := ReadHTTPInfoIntoSpan(&ringbuf.Record{RawSample: buf.Bytes()}, &fltr)
+	result, _, err := ReadHTTPInfoIntoSpan(nil, &ringbuf.Record{RawSample: buf.Bytes()}, &fltr)
 	require.NoError(t, err)
 
 	expected := request.Span{
@@ -145,7 +145,7 @@ func TestToRequestTraceNoConnection(t *testing.T) {
 	err := binary.Write(buf, binary.LittleEndian, &record)
 	require.NoError(t, err)
 
-	result, _, err := ReadHTTPInfoIntoSpan(&ringbuf.Record{RawSample: buf.Bytes()}, &fltr)
+	result, _, err := ReadHTTPInfoIntoSpan(nil, &ringbuf.Record{RawSample: buf.Bytes()}, &fltr)
 	require.NoError(t, err)
 
 	// change the expected port just before testing
@@ -185,7 +185,7 @@ func TestToRequestTrace_BadHost(t *testing.T) {
 	err := binary.Write(buf, binary.LittleEndian, &record)
 	require.NoError(t, err)
 
-	result, _, err := ReadHTTPInfoIntoSpan(&ringbuf.Record{RawSample: buf.Bytes()}, &fltr)
+	result, _, err := ReadHTTPInfoIntoSpan(nil, &ringbuf.Record{RawSample: buf.Bytes()}, &fltr)
 	require.NoError(t, err)
 
 	expected := request.Span{
@@ -204,28 +204,28 @@ func TestToRequestTrace_BadHost(t *testing.T) {
 	}
 	assert.Equal(t, expected, result)
 
-	s, p := record.hostFromBuf()
+	s, p := httpHostFromBuf(record.Buf[:])
 	assert.Equal(t, "example.c", s)
 	assert.Equal(t, -1, p)
 
 	var record1 BPFHTTPInfo
 	copy(record1.Buf[:], "GET /hello HTTP/1.1\r\nHost: example.c:23")
 
-	s, p = record1.hostFromBuf()
+	s, p = httpHostFromBuf(record1.Buf[:])
 	assert.Equal(t, "example.c", s)
 	assert.Equal(t, 23, p)
 
 	var record2 BPFHTTPInfo
 	copy(record2.Buf[:], "GET /hello HTTP/1.1\r\nHost: ")
 
-	s, p = record2.hostFromBuf()
+	s, p = httpHostFromBuf(record2.Buf[:])
 	assert.Empty(t, s)
 	assert.Equal(t, -1, p)
 
 	var record3 BPFHTTPInfo
 	copy(record3.Buf[:], "GET /hello HTTP/1.1\r\nHost")
 
-	s, p = record3.hostFromBuf()
+	s, p = httpHostFromBuf(record3.Buf[:])
 	assert.Empty(t, s)
 	assert.Equal(t, -1, p)
 }

pkg/components/ebpf/common/httpfltr_transform.go

@@ -5,6 +5,7 @@ package ebpfcommon
 
 import (
 	"bytes"
+	"log/slog"
 	"net"
 	"strconv"
 	"strings"
@@ -64,7 +65,7 @@ type HTTPInfo struct {
 	HeaderHost string
 }
 
-func ReadHTTPInfoIntoSpan(record *ringbuf.Record, filter ServiceFilter) (request.Span, bool, error) {
+func ReadHTTPInfoIntoSpan(parseCtx *EBPFParseContext, record *ringbuf.Record, filter ServiceFilter) (request.Span, bool, error) {
 	event, err := ReinterpretCast[BPFHTTPInfo](record.RawSample)
 	if err != nil {
 		return request.Span{}, true, err
@@ -75,15 +76,31 @@ func ReadHTTPInfoIntoSpan(record *ringbuf.Record, filter ServiceFilter) (request
 		return request.Span{}, true, nil
 	}
 
-	return HTTPInfoEventToSpan(event)
+	return HTTPInfoEventToSpan(parseCtx, event)
 }
 
-func HTTPInfoEventToSpan(event *BPFHTTPInfo) (request.Span, bool, error) {
+func HTTPInfoEventToSpan(parseCtx *EBPFParseContext, event *BPFHTTPInfo) (request.Span, bool, error) {
 	result := HTTPInfo{BPFHTTPInfo: *event}
 
-	var bufHost string
-	var bufPort int
-	parsedHost := false
+	var (
+		bufHost       string
+		bufPort       int
+		parsedHost    bool
+		requestBuffer []byte
+	)
+
+	if event.HasLargeBuffers == 1 {
+		b, ok := extractTCPLargeBuffer(parseCtx, event.Tp.TraceId, event.Tp.SpanId, packetTypeRequest)
+		if !ok {
+			slog.Debug("missing large buffer for HTTP request", "traceID", event.Tp.TraceId, "spanID", event.Tp.SpanId)
+			return request.Span{}, false, nil
+		}
+		requestBuffer = b
+
+		// TODO: response?
+	} else {
+		requestBuffer = event.Buf[:]
+	}
 
 	// When we can't find the connection info, we signal that through making the
 	// source and destination ports equal to max short. E.g. async SSL
@@ -92,34 +109,34 @@ func HTTPInfoEventToSpan(event *BPFHTTPInfo) (request.Span, bool, error) {
 		result.Host = target
 		result.Peer = source
 	} else {
-		bufHost, bufPort = event.hostFromBuf()
+		bufHost, bufPort = httpHostFromBuf(requestBuffer)
 		parsedHost = true
 
 		if bufPort >= 0 {
 			result.Host = bufHost
 			result.ConnInfo.D_port = uint16(bufPort)
 		}
 	}
-	result.URL = event.url()
-	result.Method = event.method()
+	result.URL = httpURLFromBuf(requestBuffer)
+	result.Method = httpMethodFromBuf(requestBuffer)
 
 	if request.EventType(result.Type) == request.EventTypeHTTPClient && !parsedHost {
-		bufHost, _ = event.hostFromBuf()
+		bufHost, _ = httpHostFromBuf(requestBuffer)
 	}
 
 	result.HeaderHost = bufHost
 
 	return httpInfoToSpan(&result), false, nil
 }
 
-func (event *BPFHTTPInfo) url() string {
-	buf := string(event.Buf[:])
+func httpURLFromBuf(req []byte) string {
+	buf := string(req)
 	space := strings.Index(buf, " ")
 	if space < 0 {
 		return ""
 	}
 
-	bufEnd := bytes.IndexByte(event.Buf[:], 0) // We assume the buffer was zero initialized in eBPF
+	bufEnd := bytes.IndexByte(req, 0) // We assume the buffer was zero initialized in eBPF
 	if bufEnd < 0 {
 		bufEnd = len(buf)
 	}
@@ -141,8 +158,8 @@ func (event *BPFHTTPInfo) url() string {
 	return buf[space+1 : end]
 }
 
-func (event *BPFHTTPInfo) method() string {
-	buf := string(event.Buf[:])
+func httpMethodFromBuf(req []byte) string {
+	buf := string(req)
 	space := strings.Index(buf, " ")
 	if space < 0 {
 		return ""
@@ -151,8 +168,8 @@ func (event *BPFHTTPInfo) method() string {
 	return buf[:space]
 }
 
-func (event *BPFHTTPInfo) hostFromBuf() (string, int) {
-	buf := cstr(event.Buf[:])
+func httpHostFromBuf(req []byte) (string, int) {
+	buf := cstr(req)
 
 	host := "Host: "
 	idx := strings.Index(buf, host)