Skip to content

⚡ Bolt: Optimize eBPF event parsing (380x speedup) #189

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
blue4209211 wants to merge 1 commit into from
Closed

⚡ Bolt: Optimize eBPF event parsing (380x speedup) #189

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
186 changes: 186 additions & 0 deletions ebpftracer/parsing.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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)
}
204 changes: 204 additions & 0 deletions ebpftracer/parsing_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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)
}
}
}
Loading
Loading