From 3550e3d8796bb7c84ae59c43fef3bc54bd52f962 Mon Sep 17 00:00:00 2001 From: "google-labs-jules[bot]" <161369871+google-labs-jules[bot]@users.noreply.github.com> Date: Mon, 2 Feb 2026 06:06:58 +0000 Subject: [PATCH] =?UTF-8?q?=E2=9A=A1=20Bolt:=20Optimize=20eBPF=20event=20p?= =?UTF-8?q?arsing?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replaced `binary.Read` with manual parsing using `binary.LittleEndian` to eliminate reflection overhead and reduce heap allocations, particularly for the large `l7Event` struct (8KB). - Created `ebpftracer/parsing.go` with manual parsing functions. - Added `ebpftracer/parsing_test.go` for verification. - Updated `ebpftracer/tracer.go` to use the new parsing logic. - Removed unused struct definitions from `tracer.go`. Co-authored-by: blue4209211 <3078106+blue4209211@users.noreply.github.com> --- .jules/bolt.md | 3 + ebpftracer/parsing.go | 157 +++++++++++++++++++++++++++++++++++++ ebpftracer/parsing_test.go | 118 ++++++++++++++++++++++++++++ ebpftracer/tracer.go | 125 +++-------------------------- 4 files changed, 290 insertions(+), 113 deletions(-) create mode 100644 .jules/bolt.md create mode 100644 ebpftracer/parsing.go create mode 100644 ebpftracer/parsing_test.go diff --git a/.jules/bolt.md b/.jules/bolt.md new file mode 100644 index 00000000..b3e35c49 --- /dev/null +++ b/.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. diff --git a/ebpftracer/parsing.go b/ebpftracer/parsing.go new file mode 100644 index 00000000..f6cfb286 --- /dev/null +++ b/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 +} diff --git a/ebpftracer/parsing_test.go b/ebpftracer/parsing_test.go new file mode 100644 index 00000000..04a5671c --- /dev/null +++ b/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) +} diff --git a/ebpftracer/tracer.go b/ebpftracer/tracer.go index 1277aee5..cbf4e8b8 100644 --- a/ebpftracer/tracer.go +++ b/ebpftracer/tracer.go @@ -4,7 +4,6 @@ import ( "bytes" "compress/gzip" "encoding/base64" - "encoding/binary" "errors" "fmt" "io" @@ -355,52 +354,6 @@ func (t EventReason) String() string { return "unknown: " + strconv.Itoa(int(t)) } -type procEvent struct { - Type EventType - Pid uint32 - Reason uint32 -} - -type tcpEvent struct { - Fd uint64 - Timestamp uint64 - Duration uint64 - Type EventType - Pid uint32 - BytesSent uint64 - BytesReceived uint64 - SPort uint16 - DPort uint16 - Aport uint16 - SAddr [16]byte - DAddr [16]byte - AAddr [16]byte -} - -type fileEvent struct { - Type EventType - Pid uint32 - Fd uint64 - Mnt uint64 - Log uint64 -} - -type l7Event struct { - Fd uint64 - ConnectionTimestamp uint64 - Pid uint32 - Status int32 - Duration uint64 - Protocol uint8 - Method uint8 - Padding uint16 - StatementId uint32 - PayloadSize uint64 - ResponseSize uint64 - Payload [4096]byte // Must match MAX_PAYLOAD_SIZE in eBPF - Response [4096]byte // Must match MAX_PAYLOAD_SIZE in eBPF -} - // HTTP response fragment event (must match eBPF struct) type httpResponseFragment struct { Fd uint64 @@ -464,97 +417,43 @@ func runEventsReader(name string, r *perf.Reader, ch chan<- Event, typ perfMapTy tracker.recordLostSamples(name, rec.LostSamples, rec.CPU) continue } - var event Event + var ( + event *Event + ) switch typ { case perfMapTypeL7Events: - v := &l7Event{} - data := rec.RawSample - - if err := binary.Read(bytes.NewBuffer(data), binary.LittleEndian, v); err != nil { + event, err = parseL7Event(rec.RawSample) + if err != nil { klog.Warningln("failed to read l7 event:", err) continue } - - // Extract payload data directly from the struct arrays - payloadSize := min(int(v.PayloadSize), len(v.Payload)) - responseSize := min(int(v.ResponseSize), len(v.Response)) - - // Copy the actual data (preventing garbage from unused buffer space) - payloadData := make([]byte, payloadSize) - copy(payloadData, v.Payload[:payloadSize]) - - responseData := make([]byte, responseSize) - copy(responseData, v.Response[:responseSize]) - - req := &l7.RequestData{ - Protocol: l7.Protocol(v.Protocol), - Status: l7.Status(v.Status), - Duration: time.Duration(v.Duration), - Method: l7.Method(v.Method), - StatementId: v.StatementId, - PayloadSize: v.PayloadSize, - ResponseSize: v.ResponseSize, - Payload: payloadData, - Response: responseData, - } - - event = Event{ - Type: EventTypeL7Request, - Pid: v.Pid, - Fd: v.Fd, - Timestamp: v.ConnectionTimestamp, - L7Request: req, - } case perfMapTypeFileEvents: - v := &fileEvent{} - if err := binary.Read(bytes.NewBuffer(rec.RawSample), binary.LittleEndian, v); err != nil { + event, err = parseFileEvent(rec.RawSample) + if err != nil { klog.Warningln("failed to read file event:", err) continue } - event = Event{Type: v.Type, Pid: v.Pid, Fd: v.Fd, Mnt: v.Mnt, Log: v.Log > 0} case perfMapTypeProcEvents: - v := &procEvent{} - if err := binary.Read(bytes.NewBuffer(rec.RawSample), binary.LittleEndian, v); err != nil { + event, err = parseProcEvent(rec.RawSample) + if err != nil { klog.Warningln("failed to read proc event:", err) continue } - event = Event{Type: v.Type, Reason: EventReason(v.Reason), Pid: v.Pid} case perfMapTypeTCPEvents: - v := &tcpEvent{} - if err := binary.Read(bytes.NewBuffer(rec.RawSample), binary.LittleEndian, v); err != nil { + event, err = parseTCPEvent(rec.RawSample) + if err != nil { klog.Warningln("failed to read tcp event:", err) continue } - event = Event{ - Type: v.Type, - Pid: v.Pid, - SrcAddr: ipPort(v.SAddr, v.SPort), - DstAddr: ipPort(v.DAddr, v.DPort), - ActualDstAddr: ipPort(v.AAddr, v.Aport), - Fd: v.Fd, - Timestamp: v.Timestamp, - Duration: time.Duration(v.Duration), - } - if v.Type == EventTypeConnectionClose { - event.TrafficStats = &TrafficStats{ - BytesSent: v.BytesSent, - BytesReceived: v.BytesReceived, - } - } default: continue } - ch <- event + ch <- *event } } -func ipPort(ip [16]byte, port uint16) netaddr.IPPort { - i, _ := netaddr.FromStdIP(ip[:]) - return netaddr.IPPortFrom(i, port) -} - func isCtxExtraPaddingRequired(traceFsPath string) bool { f, err := os.Open(path.Join(traceFsPath, "events/task/task_newtask/format")) if err != nil {