⚡ Bolt: Optimize eBPF event parsing

.jules/bolt.md

@@ -0,0 +1,3 @@
+## 2026-02-02 - Avoid binary.Read for eBPF events
+**Learning:** `binary.Read` uses reflection and causes significant heap allocations (allocating the target struct) and CPU overhead, especially for large structs like `l7Event` (8KB).
+**Action:** Use manual parsing with `binary.LittleEndian` and slice operations. For large payloads, copy only the needed bytes to small slices instead of allocating the full struct.

ebpftracer/parsing.go

@@ -0,0 +1,157 @@
+package ebpftracer
+
+import (
+	"encoding/binary"
+	"fmt"
+	"time"
+
+	"github.com/coroot/coroot-node-agent/ebpftracer/l7"
+	"inet.af/netaddr"
+)
+
+func ipPort(ip [16]byte, port uint16) netaddr.IPPort {
+	i, _ := netaddr.FromStdIP(ip[:])
+	return netaddr.IPPortFrom(i, port)
+}
+
+func parseL7Event(data []byte) (*Event, error) {
+	if len(data) < 56 {
+		return nil, fmt.Errorf("buffer too small for l7_event header: %d", len(data))
+	}
+
+	fd := binary.LittleEndian.Uint64(data[0:8])
+	connectionTimestamp := binary.LittleEndian.Uint64(data[8:16])
+	pid := binary.LittleEndian.Uint32(data[16:20])
+	status := int32(binary.LittleEndian.Uint32(data[20:24]))
+	duration := binary.LittleEndian.Uint64(data[24:32])
+	protocol := data[32]
+	method := data[33]
+	// padding 34-36
+	statementId := binary.LittleEndian.Uint32(data[36:40])
+	payloadSize := binary.LittleEndian.Uint64(data[40:48])
+	responseSize := binary.LittleEndian.Uint64(data[48:56])
+
+	var payload []byte
+	if payloadSize > 0 {
+		start := 56
+		end := start + int(payloadSize)
+		if end > len(data) {
+			end = len(data)
+		}
+		if start < end {
+			payload = make([]byte, end-start)
+			copy(payload, data[start:end])
+		}
+	}
+
+	var response []byte
+	if responseSize > 0 {
+		start := 56 + 4096
+		end := start + int(responseSize)
+		if end > len(data) {
+			end = len(data)
+		}
+		if start < end {
+			response = make([]byte, end-start)
+			copy(response, data[start:end])
+		}
+	}
+
+	req := &l7.RequestData{
+		Protocol:     l7.Protocol(protocol),
+		Status:       l7.Status(status),
+		Duration:     time.Duration(duration),
+		Method:       l7.Method(method),
+		StatementId:  statementId,
+		PayloadSize:  payloadSize,
+		ResponseSize: responseSize,
+		Payload:      payload,
+		Response:     response,
+	}
+
+	return &Event{
+		Type:      EventTypeL7Request,
+		Pid:       pid,
+		Fd:        fd,
+		Timestamp: connectionTimestamp,
+		L7Request: req,
+	}, nil
+}
+
+func parseTCPEvent(data []byte) (*Event, error) {
+	if len(data) < 102 {
+		return nil, fmt.Errorf("buffer too small for tcp_event: %d", len(data))
+	}
+
+	fd := binary.LittleEndian.Uint64(data[0:8])
+	timestamp := binary.LittleEndian.Uint64(data[8:16])
+	duration := binary.LittleEndian.Uint64(data[16:24])
+	typ := EventType(binary.LittleEndian.Uint32(data[24:28]))
+	pid := binary.LittleEndian.Uint32(data[28:32])
+	bytesSent := binary.LittleEndian.Uint64(data[32:40])
+	bytesReceived := binary.LittleEndian.Uint64(data[40:48])
+	sPort := binary.LittleEndian.Uint16(data[48:50])
+	dPort := binary.LittleEndian.Uint16(data[50:52])
+	aPort := binary.LittleEndian.Uint16(data[52:54])
+
+	var sAddr [16]byte
+	copy(sAddr[:], data[54:70])
+	var dAddr [16]byte
+	copy(dAddr[:], data[70:86])
+	var aAddr [16]byte
+	copy(aAddr[:], data[86:102])
+
+	event := &Event{
+		Type:          typ,
+		Pid:           pid,
+		SrcAddr:       ipPort(sAddr, sPort),
+		DstAddr:       ipPort(dAddr, dPort),
+		ActualDstAddr: ipPort(aAddr, aPort),
+		Fd:            fd,
+		Timestamp:     timestamp,
+		Duration:      time.Duration(duration),
+	}
+	if typ == EventTypeConnectionClose {
+		event.TrafficStats = &TrafficStats{
+			BytesSent:     bytesSent,
+			BytesReceived: bytesReceived,
+		}
+	}
+	return event, nil
+}
+
+func parseFileEvent(data []byte) (*Event, error) {
+	if len(data) < 32 {
+		return nil, fmt.Errorf("buffer too small for file_event: %d", len(data))
+	}
+	// struct file_event {
+	// 	__u32 type;
+	// 	__u32 pid;
+	// 	__u64 fd;
+	// 	__u64 mnt;
+	// 	__u64 log;
+	// };
+	typ := EventType(binary.LittleEndian.Uint32(data[0:4]))
+	pid := binary.LittleEndian.Uint32(data[4:8])
+	fd := binary.LittleEndian.Uint64(data[8:16])
+	mnt := binary.LittleEndian.Uint64(data[16:24])
+	log := binary.LittleEndian.Uint64(data[24:32])
+
+	return &Event{Type: typ, Pid: pid, Fd: fd, Mnt: mnt, Log: log > 0}, nil
+}
+
+func parseProcEvent(data []byte) (*Event, error) {
+	if len(data) < 12 {
+		return nil, fmt.Errorf("buffer too small for proc_event: %d", len(data))
+	}
+	// struct proc_event {
+	//     __u32 type;
+	//     __u32 pid;
+	//     __u32 reason;
+	// };
+	typ := EventType(binary.LittleEndian.Uint32(data[0:4]))
+	pid := binary.LittleEndian.Uint32(data[4:8])
+	reason := EventReason(binary.LittleEndian.Uint32(data[8:12]))
+
+	return &Event{Type: typ, Reason: reason, Pid: pid}, nil
+}

ebpftracer/parsing_test.go

@@ -0,0 +1,118 @@
+package ebpftracer
+
+import (
+	"encoding/binary"
+	"testing"
+	"time"
+
+	"github.com/coroot/coroot-node-agent/ebpftracer/l7"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestParseL7Event(t *testing.T) {
+	// struct l7_event size is 8248
+	data := make([]byte, 8248)
+
+	// Fd = 123
+	binary.LittleEndian.PutUint64(data[0:8], 123)
+	// ConnectionTimestamp = 1000
+	binary.LittleEndian.PutUint64(data[8:16], 1000)
+	// Pid = 456
+	binary.LittleEndian.PutUint32(data[16:20], 456)
+	// Status = 200
+	binary.LittleEndian.PutUint32(data[20:24], 200)
+	// Duration = 50ms
+	binary.LittleEndian.PutUint64(data[24:32], uint64(50*time.Millisecond))
+	// Protocol = HTTP (1)
+	data[32] = 1
+	// Method = GET (1)
+	data[33] = 1
+	// StatementId = 789
+	binary.LittleEndian.PutUint32(data[36:40], 789)
+	// PayloadSize = 5
+	binary.LittleEndian.PutUint64(data[40:48], 5)
+	// ResponseSize = 3
+	binary.LittleEndian.PutUint64(data[48:56], 3)
+
+	// Payload "hello" at offset 56
+	copy(data[56:], []byte("hello"))
+
+	// Response "bye" at offset 56 + 4096 = 4152
+	copy(data[4152:], []byte("bye"))
+
+	event, err := parseL7Event(data)
+	assert.NoError(t, err)
+	assert.NotNil(t, event)
+	assert.Equal(t, EventTypeL7Request, event.Type)
+	assert.Equal(t, uint64(123), event.Fd)
+	assert.Equal(t, uint64(1000), event.Timestamp)
+	assert.Equal(t, uint32(456), event.Pid)
+
+	req := event.L7Request
+	assert.NotNil(t, req)
+	assert.Equal(t, l7.Protocol(1), req.Protocol)
+	assert.Equal(t, l7.Status(200), req.Status)
+	assert.Equal(t, 50*time.Millisecond, req.Duration)
+	assert.Equal(t, l7.Method(1), req.Method)
+	assert.Equal(t, uint32(789), req.StatementId)
+	assert.Equal(t, uint64(5), req.PayloadSize)
+	assert.Equal(t, uint64(3), req.ResponseSize)
+	assert.Equal(t, []byte("hello"), req.Payload)
+	assert.Equal(t, []byte("bye"), req.Response)
+}
+
+func TestParseTCPEvent(t *testing.T) {
+	data := make([]byte, 104) // 102 bytes data + padding
+	// Fd = 1
+	binary.LittleEndian.PutUint64(data[0:8], 1)
+	// Timestamp
+	binary.LittleEndian.PutUint64(data[8:16], 2000)
+	// Type = EventTypeConnectionOpen (3)
+	binary.LittleEndian.PutUint32(data[24:28], 3)
+	// Pid = 2
+	binary.LittleEndian.PutUint32(data[28:32], 2)
+	// SPort = 80
+	binary.LittleEndian.PutUint16(data[48:50], 80)
+	// DPort = 443
+	binary.LittleEndian.PutUint16(data[50:52], 443)
+
+	// SAddr 127.0.0.1 (IPv4 mapped IPv6)
+	// ::ffff:127.0.0.1
+	sAddr := [16]byte{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0xff, 0xff, 127, 0, 0, 1}
+	copy(data[54:70], sAddr[:])
+
+	event, err := parseTCPEvent(data)
+	assert.NoError(t, err)
+	assert.Equal(t, EventTypeConnectionOpen, event.Type)
+	assert.Equal(t, "127.0.0.1:80", event.SrcAddr.String())
+}
+
+func TestParseFileEvent(t *testing.T) {
+	data := make([]byte, 32)
+	// Type = EventTypeFileOpen (8)
+	binary.LittleEndian.PutUint32(data[0:4], 8)
+	// Pid = 10
+	binary.LittleEndian.PutUint32(data[4:8], 10)
+	// Fd = 5
+	binary.LittleEndian.PutUint64(data[8:16], 5)
+
+	event, err := parseFileEvent(data)
+	assert.NoError(t, err)
+	assert.Equal(t, EventTypeFileOpen, event.Type)
+	assert.Equal(t, uint32(10), event.Pid)
+	assert.Equal(t, uint64(5), event.Fd)
+}
+
+func TestParseProcEvent(t *testing.T) {
+	data := make([]byte, 12)
+	// Type = EventTypeProcessStart (1)
+	binary.LittleEndian.PutUint32(data[0:4], 1)
+	// Pid = 20
+	binary.LittleEndian.PutUint32(data[4:8], 20)
+	// Reason = EventReasonNone (0)
+
+	event, err := parseProcEvent(data)
+	assert.NoError(t, err)
+	assert.Equal(t, EventTypeProcessStart, event.Type)
+	assert.Equal(t, uint32(20), event.Pid)
+}