diff --git a/.jules/bolt.md b/.jules/bolt.md new file mode 100644 index 00000000..348ca7dc --- /dev/null +++ b/.jules/bolt.md @@ -0,0 +1,3 @@ +## 2025-01-30 - Binary Deserialization Overhead +**Learning:** `binary.Read` incurs massive overhead (~40,000x slower) for reading C-structs from eBPF buffers compared to manual `binary.LittleEndian` parsing, due to reflection and unnecessary full-struct allocation/copying. +**Action:** Always use manual parsing with `binary.LittleEndian` for high-frequency eBPF event loops. Avoid decoding into large structs with fixed-size arrays; instead, read only the header and slice the payload. diff --git a/ebpftracer/parsing.go b/ebpftracer/parsing.go new file mode 100644 index 00000000..ea11e777 --- /dev/null +++ b/ebpftracer/parsing.go @@ -0,0 +1,193 @@ +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 { + return netaddr.IPPortFrom(netaddr.IPFrom16(ip), port) +} + +func parseL7Event(data []byte) (Event, error) { + if len(data) < 56 { + return Event{}, fmt.Errorf("buffer too short for l7Event: %d", len(data)) + } + + fd := binary.LittleEndian.Uint64(data[0:8]) + timestamp := 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 2 bytes at 34 + statementId := binary.LittleEndian.Uint32(data[36:40]) + payloadSize := binary.LittleEndian.Uint64(data[40:48]) + responseSize := binary.LittleEndian.Uint64(data[48:56]) + + // Payload starts at 56 + // The struct defines Payload [4096]byte and Response [4096]byte + // Total size of payload+response area is 8192 bytes. + // We need to extract payload and response based on payloadSize and responseSize. + + // Bounds check for the rest of the data isn't strictly necessary if we cap at len(data), + // but eBPF should send enough data. + // However, we only care about the actual payload/response bytes. + + offset := 56 + if len(data) < offset { + return Event{}, fmt.Errorf("buffer too short for l7Event payload: %d", len(data)) + } + + // Payload + pSize := int(payloadSize) + // The Payload field is 4096 bytes long in the struct. + // So Response starts at 56 + 4096. + responseOffset := 56 + 4096 + + var payloadData []byte + if pSize > 0 { + // Cap pSize to 4096 (max payload size) just in case + if pSize > 4096 { + pSize = 4096 + } + // Also cap to available data in the payload window + available := len(data) - offset + if available > 4096 { + available = 4096 + } + if pSize > available { + pSize = available + } + + payloadData = make([]byte, pSize) + copy(payloadData, data[offset:offset+pSize]) + } + + // Response + rSize := int(responseSize) + var responseData []byte + + if len(data) >= responseOffset && rSize > 0 { + if rSize > 4096 { + rSize = 4096 + } + available := len(data) - responseOffset + if available > 4096 { + available = 4096 + } + if rSize > available { + rSize = available + } + + responseData = make([]byte, rSize) + copy(responseData, data[responseOffset:responseOffset+rSize]) + } + + 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: payloadData, + Response: responseData, + } + + return Event{ + Type: EventTypeL7Request, + Pid: pid, + Fd: fd, + Timestamp: timestamp, + L7Request: req, + }, nil +} + +func parseFileEvent(data []byte) (Event, error) { + if len(data) < 32 { + return Event{}, fmt.Errorf("buffer too short for fileEvent: %d", len(data)) + } + // type (4), pid (4), fd (8), mnt (8), log (8) + 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]) + logVal := binary.LittleEndian.Uint64(data[24:32]) + + return Event{ + Type: typ, + Pid: pid, + Fd: fd, + Mnt: mnt, + Log: logVal > 0, + }, nil +} + +func parseProcEvent(data []byte) (Event, error) { + if len(data) < 12 { + return Event{}, fmt.Errorf("buffer too short for procEvent: %d", len(data)) + } + // type (4), pid (4), reason (4) + typ := EventType(binary.LittleEndian.Uint32(data[0:4])) + pid := binary.LittleEndian.Uint32(data[4:8]) + reason := binary.LittleEndian.Uint32(data[8:12]) + + return Event{ + Type: typ, + Reason: EventReason(reason), + Pid: pid, + }, nil +} + +func parseTCPEvent(data []byte) (Event, error) { + if len(data) < 102 { + return Event{}, fmt.Errorf("buffer too short for tcpEvent: %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 +} diff --git a/ebpftracer/parsing_test.go b/ebpftracer/parsing_test.go new file mode 100644 index 00000000..7ccedf53 --- /dev/null +++ b/ebpftracer/parsing_test.go @@ -0,0 +1,166 @@ +package ebpftracer + +import ( + "bytes" + "encoding/binary" + "testing" + "time" + + "github.com/stretchr/testify/assert" +) + +// Structs for generating test data (matching original definitions) +type testL7Event 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 + Response [4096]byte +} + +type testTcpEvent 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 testFileEvent struct { + Type EventType + Pid uint32 + Fd uint64 + Mnt uint64 + Log uint64 +} + +type testProcEvent struct { + Type EventType + Pid uint32 + Reason uint32 +} + +func TestParseL7Event(t *testing.T) { + payload := []byte("SELECT * FROM users") + response := []byte("OK") + + v := testL7Event{ + Fd: 12345, + ConnectionTimestamp: 99999, + Pid: 1001, + Status: 200, + Duration: 500000, + Protocol: 1, + Method: 2, + StatementId: 42, + PayloadSize: uint64(len(payload)), + ResponseSize: uint64(len(response)), + } + copy(v.Payload[:], payload) + copy(v.Response[:], response) + + buf := new(bytes.Buffer) + err := binary.Write(buf, binary.LittleEndian, &v) + assert.NoError(t, err) + + event, err := parseL7Event(buf.Bytes()) + assert.NoError(t, err) + + assert.Equal(t, EventTypeL7Request, event.Type) + assert.Equal(t, uint32(1001), event.Pid) + assert.Equal(t, uint64(12345), event.Fd) + assert.Equal(t, uint64(99999), event.Timestamp) + assert.NotNil(t, event.L7Request) + assert.Equal(t, payload, event.L7Request.Payload) + assert.Equal(t, response, event.L7Request.Response) + assert.Equal(t, time.Duration(500000), event.L7Request.Duration) +} + +func TestParseTCPEvent(t *testing.T) { + v := testTcpEvent{ + Fd: 555, + Timestamp: 8888, + Duration: 123, + Type: EventTypeConnectionClose, + Pid: 202, + BytesSent: 100, + BytesReceived: 200, + SPort: 80, + DPort: 12345, + Aport: 54321, + } + // 10.0.0.1 + copy(v.SAddr[:], []byte{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0xff, 0xff, 10, 0, 0, 1}) + + buf := new(bytes.Buffer) + err := binary.Write(buf, binary.LittleEndian, &v) + assert.NoError(t, err) + + // Verify buffer size is 102 + assert.Equal(t, 102, buf.Len()) + + event, err := parseTCPEvent(buf.Bytes()) + assert.NoError(t, err) + + assert.Equal(t, EventTypeConnectionClose, event.Type) + assert.Equal(t, uint32(202), event.Pid) + assert.Equal(t, uint64(555), event.Fd) + assert.Equal(t, "10.0.0.1:80", event.SrcAddr.String()) + assert.Equal(t, uint64(100), event.TrafficStats.BytesSent) + assert.Equal(t, uint64(200), event.TrafficStats.BytesReceived) +} + +func TestParseFileEvent(t *testing.T) { + v := testFileEvent{ + Type: EventTypeFileOpen, + Pid: 303, + Fd: 777, + Mnt: 888, + Log: 1, + } + buf := new(bytes.Buffer) + err := binary.Write(buf, binary.LittleEndian, &v) + assert.NoError(t, err) + + event, err := parseFileEvent(buf.Bytes()) + assert.NoError(t, err) + + assert.Equal(t, EventTypeFileOpen, event.Type) + assert.Equal(t, uint32(303), event.Pid) + assert.Equal(t, uint64(777), event.Fd) + assert.True(t, event.Log) +} + +func TestParseProcEvent(t *testing.T) { + v := testProcEvent{ + Type: EventTypeProcessExit, + Pid: 404, + Reason: uint32(EventReasonOOMKill), + } + buf := new(bytes.Buffer) + err := binary.Write(buf, binary.LittleEndian, &v) + assert.NoError(t, err) + + event, err := parseProcEvent(buf.Bytes()) + assert.NoError(t, err) + + assert.Equal(t, EventTypeProcessExit, event.Type) + assert.Equal(t, uint32(404), event.Pid) + assert.Equal(t, EventReasonOOMKill, event.Reason) +} diff --git a/ebpftracer/tracer.go b/ebpftracer/tracer.go index 1277aee5..f3af0c46 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 @@ -468,80 +421,33 @@ func runEventsReader(name string, r *perf.Reader, ch chan<- Event, typ perfMapTy switch typ { case perfMapTypeL7Events: - v := &l7Event{} - data := rec.RawSample - - if err := binary.Read(bytes.NewBuffer(data), binary.LittleEndian, v); err != nil { + var err error + 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 { + var err error + 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 { + var err error + 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 { + var err error + 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 } @@ -550,10 +456,6 @@ func runEventsReader(name string, r *perf.Reader, ch chan<- Event, typ perfMapTy } } -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"))