Fix missing gRPC/HTTP2 events

bpf/http_sock.h

@@ -384,7 +384,7 @@ static __always_inline void handle_http_response(unsigned char *small_buf, pid_c
     }
 }
 
-static __always_inline void http2_grpc_start(http2_conn_stream_t *s_key, void *u_buf, int len, u8 direction) {
+static __always_inline void http2_grpc_start(http2_conn_stream_t *s_key, void *u_buf, int len, u8 direction, u8 ssl) {
     http2_grpc_request_t *h2g_info = empty_http2_info();
     if (h2g_info) {
         http_connection_metadata_t *meta = connection_meta(&s_key->pid_conn, direction, PACKET_TYPE_REQUEST);
@@ -396,6 +396,7 @@ static __always_inline void http2_grpc_start(http2_conn_stream_t *s_key, void *u
         h2g_info->flags = EVENT_K_HTTP2_REQUEST;
         h2g_info->start_monotime_ns = bpf_ktime_get_ns();
         h2g_info->len = len;
+        h2g_info->ssl = ssl;
         h2g_info->conn_info = s_key->pid_conn.conn;
         if (meta) { // keep verifier happy
             h2g_info->pid = meta->pid;
@@ -423,7 +424,7 @@ 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, u8 direction) {
+static __always_inline void process_http2_grpc_frames(pid_connection_info_t *pid_conn, void *u_buf, int bytes_len, u8 direction, u8 ssl) {
     int pos = 0;
     u8 found_start_frame = 0;
     u8 found_end_frame = 0;
@@ -488,7 +489,7 @@ static __always_inline void process_http2_grpc_frames(pid_connection_info_t *pid
     }
 
     if (found_start_frame) {
-        http2_grpc_start(&stream, (void *)((u8 *)u_buf + pos), bytes_len, direction);
+        http2_grpc_start(&stream, (void *)((u8 *)u_buf + pos), bytes_len, direction, ssl);
     } else {
         // We only loop 6 times looking for the stream termination. If the data packed is large we'll miss the
         // frame saying the stream closed. In that case we try this backup path.
@@ -628,7 +629,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, direction);
+            process_http2_grpc_frames(pid_conn, u_buf, bytes_len, direction, ssl);
         } else { // large request tracking
             http_info_t *info = bpf_map_lookup_elem(&ongoing_http, pid_conn);
 

bpf/http_types.h

@@ -139,6 +139,7 @@ typedef struct http2_grpc_request {
     // also to filter traces from unsolicited processes that share the executable
     // with other instrumented processes
     pid_info pid;
+    u8 ssl;
     tp_info_t tp;
 } http2_grpc_request_t;
 

pkg/internal/ebpf/common/common.go

@@ -25,6 +25,7 @@ type HTTPRequestTrace bpfHttpRequestTrace
 type SQLRequestTrace bpfSqlRequestTrace
 type BPFHTTPInfo bpfHttpInfoT
 type BPFConnInfo bpfConnectionInfoT
+type TCPRequestInfo bpfTcpReqT
 
 const EventTypeSQL = 5    // EVENT_SQL_CLIENT
 const EventTypeKHTTP = 6  // HTTP Events generated by kprobes
@@ -87,6 +88,13 @@ type Filter struct {
 	Fd int
 }
 
+type MisclassifiedEvent struct {
+	EventType int
+	TCPInfo   *TCPRequestInfo
+}
+
+var MisclassifiedEvents = make(chan MisclassifiedEvent)
+
 func ptlog() *slog.Logger { return slog.With("component", "ebpf.ProcessTracer") }
 
 func ReadHTTPRequestTraceAsSpan(record *ringbuf.Record) (request.Span, bool, error) {

pkg/internal/ebpf/common/http2grpc_transform.go

@@ -29,6 +29,7 @@ const (
 )
 
 var hdec = hpack.NewDecoder(0, nil)
+var hdecRet = hpack.NewDecoder(0, nil)
 
 // not all requests for a given stream specify the protocol, but one must
 // we remember if we see grpc mentioned and tag the rest of the streams for
@@ -109,7 +110,7 @@ func readRetMetaFrame(conn *BPFConnInfo, fr *http2.Framer, hf *http2.HeadersFram
 	status := 0
 	proto := defaultProtocol(conn)
 
-	hdec.SetEmitFunc(func(hf hpack.HeaderField) {
+	hdecRet.SetEmitFunc(func(hf hpack.HeaderField) {
 		hfKey := strings.ToLower(hf.Name)
 		// grpc requests may have :status and grpc-status. :status will be HTTP code.
 		// we prefer the grpc one if it exists, it's always later since : tagged headers
@@ -125,11 +126,11 @@ func readRetMetaFrame(conn *BPFConnInfo, fr *http2.Framer, hf *http2.HeadersFram
 		}
 	})
 	// Lose reference to MetaHeadersFrame:
-	defer hdec.SetEmitFunc(func(_ hpack.HeaderField) {})
+	defer hdecRet.SetEmitFunc(func(_ hpack.HeaderField) {})
 
 	for {
 		frag := hf.HeaderBlockFragment()
-		if _, err := hdec.Write(frag); err != nil {
+		if _, err := hdecRet.Write(frag); err != nil {
 			return status, proto
 		}
 
@@ -220,22 +221,23 @@ func ReadHTTP2InfoIntoSpan(record *ringbuf.Record) (request.Span, bool, error) {
 	// we can and terminate without an error when things fail to decode because of
 	// partial buffers.
 
-	retF, _ := retFramer.ReadFrame()
-
 	status := 0
 	eventType := HTTP2
 
-	if ff, ok := retF.(*http2.HeadersFrame); ok {
-		status, eventType = readRetMetaFrame((*BPFConnInfo)(&event.ConnInfo), retFramer, ff)
-	}
-
 	f, _ := framer.ReadFrame()
 
 	if ff, ok := f.(*http2.HeadersFrame); ok {
 		method, path, proto := readMetaFrame((*BPFConnInfo)(&event.ConnInfo), framer, ff)
 
-		if eventType != GRPC && proto == GRPC {
-			eventType = proto
+		retF, _ := retFramer.ReadFrame()
+
+		if ff, ok := retF.(*http2.HeadersFrame); ok {
+			status, eventType = readRetMetaFrame((*BPFConnInfo)(&event.ConnInfo), retFramer, ff)
+		}
+
+		// if we don't have a path or much else, assume gRPC if it's not ssl. HTTP2 is almost always SSL.
+		if eventType != GRPC && (proto == GRPC || (path == "" && event.Ssl == 0)) {
+			eventType = GRPC
 			status = http2grpcStatus(status)
 		}
 

pkg/internal/ebpf/common/tcp_detect_transform.go

@@ -8,13 +8,12 @@ import (
 
 	"github.com/cilium/ebpf/ringbuf"
 	trace2 "go.opentelemetry.io/otel/trace"
+	"golang.org/x/net/http2"
 
 	"github.com/grafana/beyla/pkg/internal/request"
 	"github.com/grafana/beyla/pkg/internal/sqlprune"
 )
 
-type TCPRequestInfo bpfTcpReqT
-
 func ReadTCPRequestIntoSpan(record *ringbuf.Record) (request.Span, bool, error) {
 	var event TCPRequestInfo
 
@@ -36,6 +35,8 @@ func ReadTCPRequestIntoSpan(record *ringbuf.Record) (request.Span, bool, error)
 	sqlIndex := isSQL(buf)
 	if sqlIndex >= 0 {
 		return TCPToSQLToSpan(&event, buf[sqlIndex:]), false, nil
+	} else if isHTTP2(b, &event) {
+		MisclassifiedEvents <- MisclassifiedEvent{EventType: EventTypeKHTTP2, TCPInfo: &event}
 	}
 
 	return request.Span{}, true, nil // ignore if we couldn't parse it
@@ -115,3 +116,16 @@ func TCPToSQLToSpan(trace *TCPRequestInfo, s string) request.Span {
 		Statement: sql,
 	}
 }
+
+func isHTTP2(data []uint8, event *TCPRequestInfo) bool {
+	framer := byteFramer(data)
+
+	f, _ := framer.ReadFrame()
+
+	if ff, ok := f.(*http2.HeadersFrame); ok {
+		method, path, _ := readMetaFrame((*BPFConnInfo)(&event.ConnInfo), framer, ff)
+		return method != "" || path != ""
+	}
+
+	return false
+}

pkg/internal/ebpf/httpfltr/httpfltr.go

@@ -247,6 +247,7 @@ func (p *Tracer) Run(ctx context.Context, eventsChan chan<- []request.Span) {
 
 	timeoutTicker := time.NewTicker(2 * time.Second)
 
+	go p.watchForMisclassifedEvents()
 	go p.lookForTimeouts(timeoutTicker, eventsChan)
 	defer timeoutTicker.Stop()
 
@@ -309,3 +310,19 @@ func (p *Tracer) lookForTimeouts(ticker *time.Ticker, eventsChan chan<- []reques
 		}
 	}
 }
+
+func (p *Tracer) watchForMisclassifedEvents() {
+	for e := range ebpfcommon.MisclassifiedEvents {
+		if e.EventType == ebpfcommon.EventTypeKHTTP2 {
+			if p.bpfObjects.OngoingHttp2Connections != nil {
+				err := p.bpfObjects.OngoingHttp2Connections.Put(
+					&bpfPidConnectionInfoT{Conn: bpfConnectionInfoT(e.TCPInfo.ConnInfo), Pid: e.TCPInfo.Pid.HostPid},
+					uint8(e.TCPInfo.Ssl),
+				)
+				if err != nil {
+					p.log.Debug("error writing HTTP2/gRPC connection info", "error", err)
+				}
+			}
+		}
+	}
+}