diff --git a/ebpftracer/parsing.go b/ebpftracer/parsing.go new file mode 100644 index 00000000..1ec4276c --- /dev/null +++ b/ebpftracer/parsing.go @@ -0,0 +1,166 @@ +package ebpftracer + +import ( + "encoding/binary" + "fmt" + "time" + + "github.com/coroot/coroot-node-agent/ebpftracer/l7" + "inet.af/netaddr" +) + +func parseL7Event(data []byte) (Event, error) { + if len(data) < 56 { + return Event{}, fmt.Errorf("buffer too short for l7 event header: %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 + statementId := binary.LittleEndian.Uint32(data[36:40]) + payloadSize := binary.LittleEndian.Uint64(data[40:48]) + responseSize := binary.LittleEndian.Uint64(data[48:56]) + + const maxPayloadSize = 4096 + + // Payload is at offset 56 + pSize := int(payloadSize) + if pSize > maxPayloadSize { + pSize = maxPayloadSize + } + + payloadOffset := 56 + if len(data) < payloadOffset+pSize { + return Event{}, fmt.Errorf("buffer too short for l7 payload: %d", len(data)) + } + payload := make([]byte, pSize) + copy(payload, data[payloadOffset:payloadOffset+pSize]) + + // Response is at offset 56 + 4096 + rSize := int(responseSize) + if rSize > maxPayloadSize { + rSize = maxPayloadSize + } + + responseOffset := 56 + maxPayloadSize + if len(data) < responseOffset+rSize { + // It is possible the buffer is truncated if BPF program didn't send full struct? + // But usually it sends fixed size. + // If data is smaller, we truncate or error? + // For safety, let's limit rSize to available data + if len(data) > responseOffset { + rSize = len(data) - responseOffset + } else { + rSize = 0 + } + } + response := make([]byte, rSize) + if rSize > 0 { + copy(response, 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: payload, + Response: response, + } + + return Event{ + Type: EventTypeL7Request, + Pid: pid, + Fd: fd, + Timestamp: timestamp, + L7Request: req, + }, nil +} + +func parseTcpEvent(data []byte) (Event, error) { + // TCP event size check + // Fd(8)+Ts(8)+Dur(8)+Type(4)+Pid(4)+Sent(8)+Recv(8)+SPort(2)+DPort(2)+APort(2) = 54 + // SAddr(16) + DAddr(16) + AAddr(16) = 48 + // Total 102 bytes. + if len(data) < 102 { + return Event{}, fmt.Errorf("buffer too short 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) { + // Type(4)+Pid(4)+Fd(8)+Mnt(8)+Log(8) = 32 bytes + // The struct in tracer.go uses Log uint64, but checks > 0. + if len(data) < 32 { + return Event{}, fmt.Errorf("buffer too short for file event: %d", len(data)) + } + + 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) { + // Type(4)+Pid(4)+Reason(4) = 12 bytes + if len(data) < 12 { + return Event{}, fmt.Errorf("buffer too short for proc event: %d", len(data)) + } + + 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 ipPort(ip [16]byte, port uint16) netaddr.IPPort { + i, _ := netaddr.FromStdIP(ip[:]) + return netaddr.IPPortFrom(i, port) +} diff --git a/ebpftracer/parsing_test.go b/ebpftracer/parsing_test.go new file mode 100644 index 00000000..adf3f2c3 --- /dev/null +++ b/ebpftracer/parsing_test.go @@ -0,0 +1,117 @@ +package ebpftracer + +import ( + "bytes" + "encoding/binary" + "testing" + "time" + "unsafe" + + "github.com/stretchr/testify/assert" +) + +// Legacy struct for benchmark comparison +type l7EventLegacy 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 +} + +func TestParseL7Event(t *testing.T) { + // Construct a sample byte slice + header := l7EventLegacy{ + Fd: 123, + ConnectionTimestamp: 456, + Pid: 789, + Status: 200, + Duration: 1000, + Protocol: 1, + Method: 2, + StatementId: 999, + PayloadSize: 5, + ResponseSize: 5, + } + copy(header.Payload[:], []byte("hello")) + copy(header.Response[:], []byte("world")) + + buf := new(bytes.Buffer) + err := binary.Write(buf, binary.LittleEndian, &header) + assert.NoError(t, err) + + data := buf.Bytes() + + // Test parsing + event, err := parseL7Event(data) + assert.NoError(t, err) + + assert.Equal(t, EventTypeL7Request, event.Type) + assert.Equal(t, uint64(123), event.Fd) + assert.Equal(t, uint64(456), event.Timestamp) + assert.Equal(t, uint32(789), event.Pid) + assert.NotNil(t, event.L7Request) + assert.Equal(t, time.Duration(1000), event.L7Request.Duration) + assert.Equal(t, []byte("hello"), event.L7Request.Payload) + assert.Equal(t, []byte("world"), event.L7Request.Response) +} + +func TestParseTcpEvent(t *testing.T) { + // 102 bytes + data := make([]byte, 102) + binary.LittleEndian.PutUint64(data[0:], 1001) // Fd + binary.LittleEndian.PutUint64(data[8:], 2002) // Timestamp + binary.LittleEndian.PutUint64(data[16:], 50) // Duration + binary.LittleEndian.PutUint32(data[24:], uint32(EventTypeConnectionClose)) + binary.LittleEndian.PutUint32(data[28:], 123) // Pid + binary.LittleEndian.PutUint64(data[32:], 100) // BytesSent + binary.LittleEndian.PutUint64(data[40:], 200) // BytesReceived + binary.LittleEndian.PutUint16(data[48:], 8080) // SPort + binary.LittleEndian.PutUint16(data[50:], 9090) // DPort + binary.LittleEndian.PutUint16(data[52:], 9091) // APort + + // SAddr, DAddr, AAddr are zero + + event, err := parseTcpEvent(data) + assert.NoError(t, err) + assert.Equal(t, EventTypeConnectionClose, event.Type) + assert.Equal(t, uint64(1001), event.Fd) + assert.Equal(t, uint32(123), event.Pid) + assert.NotNil(t, event.TrafficStats) + assert.Equal(t, uint64(100), event.TrafficStats.BytesSent) + assert.Equal(t, uint64(200), event.TrafficStats.BytesReceived) +} + +func BenchmarkBinaryReadL7(b *testing.B) { + var event l7EventLegacy + data := make([]byte, unsafe.Sizeof(event)) + binary.LittleEndian.PutUint64(data[0:], 12345) // Fd + + b.ResetTimer() + for i := 0; i < b.N; i++ { + var v l7EventLegacy + _ = binary.Read(bytes.NewBuffer(data), binary.LittleEndian, &v) + } +} + +func BenchmarkManualParseL7(b *testing.B) { + var event l7EventLegacy + data := make([]byte, unsafe.Sizeof(event)) + // Set payload sizes to something non-zero to test copy overhead + // But small enough to be realistic + binary.LittleEndian.PutUint64(data[40:], 100) // PayloadSize + binary.LittleEndian.PutUint64(data[48:], 100) // ResponseSize + + b.ResetTimer() + for i := 0; i < b.N; i++ { + _, _ = parseL7Event(data) + } +} diff --git a/ebpftracer/tracer.go b/ebpftracer/tracer.go index 1277aee5..2a186ae8 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 { - klog.Warningln("failed to read l7 event:", err) + e, err := parseL7Event(rec.RawSample) + if err != nil { + klog.Warningln("failed to parse 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, - } + event = e case perfMapTypeFileEvents: - v := &fileEvent{} - if err := binary.Read(bytes.NewBuffer(rec.RawSample), binary.LittleEndian, v); err != nil { - klog.Warningln("failed to read file event:", err) + e, err := parseFileEvent(rec.RawSample) + if err != nil { + klog.Warningln("failed to parse file event:", err) continue } - event = Event{Type: v.Type, Pid: v.Pid, Fd: v.Fd, Mnt: v.Mnt, Log: v.Log > 0} + event = e case perfMapTypeProcEvents: - v := &procEvent{} - if err := binary.Read(bytes.NewBuffer(rec.RawSample), binary.LittleEndian, v); err != nil { - klog.Warningln("failed to read proc event:", err) + e, err := parseProcEvent(rec.RawSample) + if err != nil { + klog.Warningln("failed to parse proc event:", err) continue } - event = Event{Type: v.Type, Reason: EventReason(v.Reason), Pid: v.Pid} + event = e case perfMapTypeTCPEvents: - v := &tcpEvent{} - if err := binary.Read(bytes.NewBuffer(rec.RawSample), binary.LittleEndian, v); err != nil { - klog.Warningln("failed to read tcp event:", err) + e, err := parseTcpEvent(rec.RawSample) + if err != nil { + klog.Warningln("failed to parse 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, - } - } + event = e default: continue } @@ -550,11 +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")) if err != nil {