diff --git a/ebpftracer/parsing.go b/ebpftracer/parsing.go new file mode 100644 index 00000000..007ed4e8 --- /dev/null +++ b/ebpftracer/parsing.go @@ -0,0 +1,186 @@ +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) < 8248 { // Size of l7Event struct + return nil, fmt.Errorf("buffer too short for l7Event: %d", len(data)) + } + + // Manual parsing of l7Event + // type l7Event struct { + // Fd uint64 // 0 + // ConnectionTimestamp uint64 // 8 + // Pid uint32 // 16 + // Status int32 // 20 + // Duration uint64 // 24 + // Protocol uint8 // 32 + // Method uint8 // 33 + // Padding uint16 // 34 + // StatementId uint32 // 36 + // PayloadSize uint64 // 40 + // ResponseSize uint64 // 48 + // Payload [4096]byte // 56 + // Response [4096]byte // 4152 + // } // Total: 8248 + + 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 skipped + statementId := binary.LittleEndian.Uint32(data[36:40]) + payloadSize := binary.LittleEndian.Uint64(data[40:48]) + responseSize := binary.LittleEndian.Uint64(data[48:56]) + + pSize := int(payloadSize) + if pSize > 4096 { + pSize = 4096 + } + // Payload starts at 56 + payloadData := make([]byte, pSize) + copy(payloadData, data[56:56+pSize]) + + rSize := int(responseSize) + if rSize > 4096 { + rSize = 4096 + } + // Response starts at 56 + 4096 = 4152 + responseData := make([]byte, rSize) + copy(responseData, data[4152:4152+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: connectionTimestamp, + L7Request: req, + }, nil +} + +func parseTcpEvent(data []byte) (*Event, error) { + if len(data) < 102 { + return nil, fmt.Errorf("buffer too short for tcpEvent: %d", len(data)) + } + // type tcpEvent struct { + // Fd uint64 // 0 + // Timestamp uint64 // 8 + // Duration uint64 // 16 + // Type EventType // 24 (uint32) + // Pid uint32 // 28 + // BytesSent uint64 // 32 + // BytesReceived uint64 // 40 + // SPort uint16 // 48 + // DPort uint16 // 50 + // Aport uint16 // 52 + // SAddr [16]byte // 54 + // DAddr [16]byte // 70 + // AAddr [16]byte // 86 + // } + + typ := EventType(binary.LittleEndian.Uint32(data[24:28])) + pid := binary.LittleEndian.Uint32(data[28:32]) + sPort := binary.LittleEndian.Uint16(data[48:50]) + dPort := binary.LittleEndian.Uint16(data[50:52]) + aPort := binary.LittleEndian.Uint16(data[52:54]) + + var sAddr, dAddr, aAddr [16]byte + copy(sAddr[:], data[54:70]) + copy(dAddr[:], data[70:86]) + copy(aAddr[:], data[86:102]) + + event := &Event{ + Type: typ, + Pid: pid, + SrcAddr: ipPort(sAddr, sPort), + DstAddr: ipPort(dAddr, dPort), + ActualDstAddr: ipPort(aAddr, aPort), + Fd: binary.LittleEndian.Uint64(data[0:8]), + Timestamp: binary.LittleEndian.Uint64(data[8:16]), + Duration: time.Duration(binary.LittleEndian.Uint64(data[16:24])), + } + + if typ == EventTypeConnectionClose { + event.TrafficStats = &TrafficStats{ + BytesSent: binary.LittleEndian.Uint64(data[32:40]), + BytesReceived: binary.LittleEndian.Uint64(data[40:48]), + } + } + + return event, nil +} + +func parseFileEvent(data []byte) (*Event, error) { + if len(data) < 32 { + return nil, fmt.Errorf("buffer too short for fileEvent: %d", len(data)) + } + // type fileEvent struct { + // Type EventType // 0 (uint32) + // Pid uint32 // 4 + // Fd uint64 // 8 + // Mnt uint64 // 16 + // Log uint64 // 24 + // } + + 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 nil, fmt.Errorf("buffer too short for procEvent: %d", len(data)) + } + // type procEvent struct { + // Type EventType // 0 (uint32) + // Pid uint32 // 4 + // Reason uint32 // 8 + // } + + 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, + Pid: pid, + Reason: EventReason(reason), + }, 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..781ece21 --- /dev/null +++ b/ebpftracer/parsing_test.go @@ -0,0 +1,204 @@ +package ebpftracer + +import ( + "bytes" + "encoding/binary" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// Structs for generating test data (must match parsing logic expectations) +type l7EventStruct 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 tcpEventStruct 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 fileEventStruct struct { + Type EventType + Pid uint32 + Fd uint64 + Mnt uint64 + Log uint64 +} + +type procEventStruct struct { + Type EventType + Pid uint32 + Reason uint32 +} + +func TestParseL7Event(t *testing.T) { + event := l7EventStruct{ + Fd: 123, + ConnectionTimestamp: 456, + Pid: 789, + Status: 200, + Duration: 1000, + Protocol: 1, + Method: 2, + StatementId: 999, + PayloadSize: 5, + ResponseSize: 3, + Payload: [4096]byte{1, 2, 3, 4, 5}, + Response: [4096]byte{6, 7, 8}, + } + buf := new(bytes.Buffer) + require.NoError(t, binary.Write(buf, binary.LittleEndian, event)) + data := buf.Bytes() + + parsed, err := parseL7Event(data) + require.NoError(t, err) + + assert.Equal(t, EventTypeL7Request, parsed.Type) + assert.Equal(t, event.Fd, parsed.Fd) + assert.Equal(t, event.ConnectionTimestamp, parsed.Timestamp) + assert.Equal(t, event.Pid, parsed.Pid) + assert.Equal(t, int(event.Status), int(parsed.L7Request.Status)) + assert.Equal(t, time.Duration(event.Duration), parsed.L7Request.Duration) + assert.Equal(t, int(event.Protocol), int(parsed.L7Request.Protocol)) + assert.Equal(t, int(event.Method), int(parsed.L7Request.Method)) + assert.Equal(t, event.StatementId, parsed.L7Request.StatementId) + assert.Equal(t, event.PayloadSize, parsed.L7Request.PayloadSize) + assert.Equal(t, event.ResponseSize, parsed.L7Request.ResponseSize) + assert.Equal(t, []byte{1, 2, 3, 4, 5}, parsed.L7Request.Payload) + assert.Equal(t, []byte{6, 7, 8}, parsed.L7Request.Response) + + // Test short buffer + _, err = parseL7Event(data[:len(data)-1]) + assert.Error(t, err) +} + +func TestParseTcpEvent(t *testing.T) { + event := tcpEventStruct{ + Fd: 123, + Timestamp: 456, + Duration: 789, + Type: EventTypeConnectionClose, + Pid: 101, + BytesSent: 1000, + SPort: 80, + DPort: 443, + SAddr: [16]byte{10, 0, 0, 1}, + DAddr: [16]byte{192, 168, 1, 1}, + } + buf := new(bytes.Buffer) + require.NoError(t, binary.Write(buf, binary.LittleEndian, event)) + data := buf.Bytes() + + parsed, err := parseTcpEvent(data) + require.NoError(t, err) + + assert.Equal(t, event.Type, parsed.Type) + assert.Equal(t, event.Pid, parsed.Pid) + assert.Equal(t, event.Fd, parsed.Fd) + assert.Equal(t, event.Timestamp, parsed.Timestamp) + assert.Equal(t, time.Duration(event.Duration), parsed.Duration) + assert.Equal(t, uint64(1000), parsed.TrafficStats.BytesSent) + assert.Equal(t, "[a00:1::]:80", parsed.SrcAddr.String()) +} + +func TestParseFileEvent(t *testing.T) { + event := fileEventStruct{ + Type: EventTypeFileOpen, + Pid: 123, + Fd: 456, + Mnt: 789, + Log: 1, + } + buf := new(bytes.Buffer) + require.NoError(t, binary.Write(buf, binary.LittleEndian, event)) + data := buf.Bytes() + + parsed, err := parseFileEvent(data) + require.NoError(t, err) + + assert.Equal(t, event.Type, parsed.Type) + assert.Equal(t, event.Pid, parsed.Pid) + assert.Equal(t, event.Fd, parsed.Fd) + assert.Equal(t, event.Mnt, parsed.Mnt) + assert.True(t, parsed.Log) +} + +func TestParseProcEvent(t *testing.T) { + event := procEventStruct{ + Type: EventTypeProcessExit, + Pid: 123, + Reason: uint32(EventReasonOOMKill), + } + buf := new(bytes.Buffer) + require.NoError(t, binary.Write(buf, binary.LittleEndian, event)) + data := buf.Bytes() + + parsed, err := parseProcEvent(data) + require.NoError(t, err) + + assert.Equal(t, event.Type, parsed.Type) + assert.Equal(t, event.Pid, parsed.Pid) + assert.Equal(t, EventReasonOOMKill, parsed.Reason) +} + +func BenchmarkParseL7Event_Manual(b *testing.B) { + event := l7EventStruct{ + Fd: 123, + Pid: 456, + PayloadSize: 100, + Payload: [4096]byte{1, 2, 3}, + } + buf := new(bytes.Buffer) + binary.Write(buf, binary.LittleEndian, event) + data := buf.Bytes() + + b.ResetTimer() + for i := 0; i < b.N; i++ { + if _, err := parseL7Event(data); err != nil { + b.Fatal(err) + } + } +} + +func BenchmarkParseTcpEvent_Manual(b *testing.B) { + event := tcpEventStruct{ + Fd: 123, + Pid: 456, + } + buf := new(bytes.Buffer) + binary.Write(buf, binary.LittleEndian, event) + data := buf.Bytes() + + b.ResetTimer() + for i := 0; i < b.N; i++ { + if _, err := parseTcpEvent(data); err != nil { + b.Fatal(err) + } + } +} diff --git a/ebpftracer/tracer.go b/ebpftracer/tracer.go index 1277aee5..890e498e 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 { + e, 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, - } + event = *e case perfMapTypeFileEvents: - v := &fileEvent{} - if err := binary.Read(bytes.NewBuffer(rec.RawSample), binary.LittleEndian, v); err != nil { + e, 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} + event = *e case perfMapTypeProcEvents: - v := &procEvent{} - if err := binary.Read(bytes.NewBuffer(rec.RawSample), binary.LittleEndian, v); err != nil { + e, 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} + event = *e case perfMapTypeTCPEvents: - v := &tcpEvent{} - if err := binary.Read(bytes.NewBuffer(rec.RawSample), binary.LittleEndian, v); err != nil { + e, 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, - } - } + 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 {