fix: prevent OOM crash from L7 protocol misidentification

containers/container.go

@@ -102,6 +102,9 @@ type ActiveConnection struct {
 	http2Parser    *l7.Http2Parser
 	postgresParser *l7.PostgresParser
 	mysqlParser    *l7.MysqlParser
+
+	parseFailCount   int
+	protocolOverride l7.Protocol // non-zero = override eBPF-detected protocol
 }
 
 type ListenDetails struct {
@@ -1088,8 +1091,16 @@ func (c *Container) onL7RequestWithResult(pid uint32, fd uint64, timestamp uint6
 		trace = c.tracer.NewTrace(conn.DestinationKey.ActualDestinationIfKnown(), conn.srcWorkload, destWorkload, actualDestWorkload)
 	}
 
+	// Protocol reclassification: if previous parse attempts failed repeatedly,
+	// the connection was likely misidentified by eBPF heuristics. Use the
+	// override to skip further parsing for this connection.
+	protocol := r.Protocol
+	if conn.protocolOverride != 0 {
+		protocol = conn.protocolOverride
+	}
+
 	// Process L7 requests and update metrics
-	switch r.Protocol {
+	switch protocol {
 	case l7.ProtocolDNS:
 		status := r.Status.DNS()
 		if status == "" {
@@ -1282,6 +1293,11 @@ func (c *Container) onL7RequestWithResult(pid uint32, fd uint64, timestamp uint6
 			conn.postgresParser = l7.NewPostgresParser()
 		}
 		query := conn.postgresParser.Parse(r.Payload)
+		if query == "" && r.Method != l7.MethodStatementClose {
+			c.trackParseFail(conn, pid, fd, r.Protocol)
+		} else {
+			conn.parseFailCount = 0
+		}
 		if trace != nil {
 			trace.PostgresQuery(query, r.Status.Error(), r.Duration)
 		}
@@ -1331,13 +1347,23 @@ func (c *Container) onL7RequestWithResult(pid uint32, fd uint64, timestamp uint6
 		// Update stats for Clickhouse
 		c.l7Stats.observe(r.Protocol, r.Status.String(), "", "", r.Duration, conn.DestinationKey, conn.srcWorkload, r, "")
 		query := l7.ParseClickhouse(r.Payload)
+		if query == "" {
+			c.trackParseFail(conn, pid, fd, r.Protocol)
+		} else {
+			conn.parseFailCount = 0
+		}
 		if trace != nil {
 			trace.ClickhouseQuery(query, r.Status.Error(), r.Duration)
 		}
 	case l7.ProtocolZookeeper:
 		// Update stats for Zookeeper
 		c.l7Stats.observe(r.Protocol, r.Status.Zookeeper(), "", "", r.Duration, conn.DestinationKey, conn.srcWorkload, r, "")
 		op, arg := l7.ParseZookeeper(r.Payload)
+		if op == "" {
+			c.trackParseFail(conn, pid, fd, r.Protocol)
+		} else {
+			conn.parseFailCount = 0
+		}
 		if trace != nil {
 			trace.ZookeeperRequest(op, arg, r.Status, r.Duration)
 		}
@@ -1351,6 +1377,25 @@ func (c *Container) onL7RequestWithResult(pid uint32, fd uint64, timestamp uint6
 	return nil, L7RequestProcessed
 }
 
+const (
+	parseFailThreshold   = 3
+	protocolReclassified = l7.Protocol(0xFF) // sentinel: no protocol matches, hits default case
+)
+
+// trackParseFail tracks consecutive parse failures on a connection. After
+// parseFailThreshold failures, the connection's protocol is overridden to
+// skip further parsing. This handles eBPF protocol misidentification where
+// weak heuristics (e.g., 3-byte ClickHouse check) tag a non-matching
+// connection permanently.
+func (c *Container) trackParseFail(conn *ActiveConnection, pid uint32, fd uint64, proto l7.Protocol) {
+	conn.parseFailCount++
+	if conn.parseFailCount == parseFailThreshold {
+		conn.protocolOverride = protocolReclassified
+		klog.Warningf("reclassified connection pid=%d fd=%d from %s to unknown after %d consecutive parse failures",
+			pid, fd, proto, conn.parseFailCount)
+	}
+}
+
 // processHTTP2WithoutConnection handles HTTP/2 events when TCP connection tracking failed.
 // This is common for Go TLS connections where goroutines switch threads between
 // TCP connect and TLS write, causing fd_by_pid_tgid lookup to fail in eBPF.

ebpftracer/ebpf/l7/l7.c

@@ -408,8 +408,6 @@ int trace_enter_write(void *ctx, __u64 fd, __u16 is_tls, char *buf, __u64 size,
             req->protocol = PROTOCOL_RABBITMQ;
         } else if (is_amqp_method_frame(payload, size)) {
             req->protocol = PROTOCOL_RABBITMQ;
-        } else if (is_clickhouse_query(payload, size)) {
-            req->protocol = PROTOCOL_CLICKHOUSE;
         } else if (is_zk_request(payload, total_size)) {
             req->protocol = PROTOCOL_ZOOKEEPER;
         }  else if (is_kafka_request(payload, size, &req->request_id)) {

ebpftracer/l7/clickhouse.go

@@ -2,10 +2,13 @@ package l7
 
 import (
 	"bytes"
+	"io"
 
 	"github.com/ClickHouse/ch-go/proto"
 )
 
+const clickhouseClientCodeQuery = 1
+
 func ParseClickhouse(payload []byte) (result string) {
 	// Recover from panics caused by malformed/incomplete packets
 	defer func() {
@@ -14,7 +17,28 @@ func ParseClickhouse(payload []byte) (result string) {
 		}
 	}()
 
-	r := proto.NewReader(bytes.NewReader(payload))
+	// Layer 2: Structural validation before invoking ch-go.
+	// Reject misidentified payloads early, before the library can
+	// decode garbage varint lengths and attempt unbounded allocations.
+	if len(payload) < 3 {
+		return ""
+	}
+	if payload[0] != clickhouseClientCodeQuery {
+		return ""
+	}
+	// Query ID length is varint-encoded. Single-byte varints (0-127) cover
+	// all practical query IDs. Multi-byte varints (>127) in this position
+	// indicate garbage data from a misidentified connection.
+	if payload[1] > 127 {
+		return ""
+	}
+
+	// Layer 1: Bound the reader to the actual payload size.
+	// ch-go's proto.Reader decodes varint string lengths and pre-allocates
+	// that many bytes. With corrupted data, the varint can decode to TB-scale
+	// values, causing a fatal OOM that bypasses defer/recover.
+	// LimitReader caps reads at len(payload), turning the OOM into io.EOF.
+	r := proto.NewReader(io.LimitReader(bytes.NewReader(payload), int64(len(payload))))
 	var err error
 	if _, err = r.Byte(); err != nil {
 		return ""

ebpftracer/l7/postgres.go

@@ -26,6 +26,15 @@ func (p *PostgresParser) Parse(payload []byte) string {
 		return ""
 	}
 	cmd := payload[0]
+	// Reject unknown frame types early — Postgres detection in eBPF is weak
+	// (single-byte check), so misidentified connections can reach here.
+	// Accept all valid frontend message types per the Postgres wire protocol.
+	switch cmd {
+	case PostgresFrameQuery, PostgresFrameBind, PostgresFrameParse, PostgresFrameClose,
+		'E', 'D', 'S', 'H', 'X': // Execute, Describe, Sync, Flush, Terminate
+	default:
+		return ""
+	}
 	switch cmd {
 	case PostgresFrameQuery:
 		var query string

ebpftracer/l7/zookeeper.go

@@ -108,11 +108,25 @@ func zkReadString(r io.Reader) string {
 	return string(res[:n])
 }
 
+func isValidZkOp(op int32) bool {
+	switch op {
+	case zkOpCreate, zkOpDelete, zkOpExists, zkOpGetData, zkOpSetData,
+		zkOpGetAcl, zkOpSetAcl, zkOpGetChildren, zkOpSync, zkOpPing,
+		zkOpGetChildren2, zkOpCheck, zkOpMulti, zkOpReconfig,
+		zkOpCreateContainer, zkOpCreateTTL, zkOpClose, zkOpSetAuth, zkOpSetWatches:
+		return true
+	}
+	return false
+}
+
 func ParseZookeeper(payload []byte) (string, string) {
 	r := bytes.NewReader(payload)
 	h := zkRequestHeader{}
 	if err := binary.Read(r, binary.BigEndian, &h); err != nil {
 		return "", ""
 	}
+	if !isValidZkOp(h.OpType) {
+		return "", ""
+	}
 	return zkParse(r, h.OpType)
 }