Skip to content

⚡ Bolt: optimize eBPF event parsing #185

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
google-labs-jules wants to merge 1 commit into from
Closed

⚡ Bolt: optimize eBPF event parsing #185

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
67 changes: 67 additions & 0 deletions ebpftracer/parsing.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,67 @@
package ebpftracer

import (
"encoding/binary"
)

func parseL7Event(data []byte, v *l7Event) bool {
if len(data) < 56 {
return false
}
v.Fd = binary.LittleEndian.Uint64(data[0:8])
v.ConnectionTimestamp = binary.LittleEndian.Uint64(data[8:16])
v.Pid = binary.LittleEndian.Uint32(data[16:20])
v.Status = int32(binary.LittleEndian.Uint32(data[20:24]))
v.Duration = binary.LittleEndian.Uint64(data[24:32])
v.Protocol = data[32]
v.Method = data[33]
v.Padding = binary.LittleEndian.Uint16(data[34:36])
v.StatementId = binary.LittleEndian.Uint32(data[36:40])
v.PayloadSize = binary.LittleEndian.Uint64(data[40:48])
v.ResponseSize = binary.LittleEndian.Uint64(data[48:56])
// Payload and Response arrays are intentionally not copied here for performance.
// They should be extracted directly from data using offsets 56 and 4152.
return true
}

func parseTCPEvent(data []byte, v *tcpEvent) bool {
if len(data) < 104 {
return false
}
v.Fd = binary.LittleEndian.Uint64(data[0:8])
v.Timestamp = binary.LittleEndian.Uint64(data[8:16])
v.Duration = binary.LittleEndian.Uint64(data[16:24])
v.Type = EventType(binary.LittleEndian.Uint32(data[24:28]))
v.Pid = binary.LittleEndian.Uint32(data[28:32])
v.BytesSent = binary.LittleEndian.Uint64(data[32:40])
v.BytesReceived = binary.LittleEndian.Uint64(data[40:48])
v.SPort = binary.LittleEndian.Uint16(data[48:50])
v.DPort = binary.LittleEndian.Uint16(data[50:52])
v.Aport = binary.LittleEndian.Uint16(data[52:54])
copy(v.SAddr[:], data[54:70])
copy(v.DAddr[:], data[70:86])
copy(v.AAddr[:], data[86:102])
return true
}

func parseFileEvent(data []byte, v *fileEvent) bool {
if len(data) < 32 {
return false
}
v.Type = EventType(binary.LittleEndian.Uint32(data[0:4]))
v.Pid = binary.LittleEndian.Uint32(data[4:8])
v.Fd = binary.LittleEndian.Uint64(data[8:16])
v.Mnt = binary.LittleEndian.Uint64(data[16:24])
v.Log = binary.LittleEndian.Uint64(data[24:32])
return true
}

func parseProcEvent(data []byte, v *procEvent) bool {
if len(data) < 12 {
return false
}
v.Type = EventType(binary.LittleEndian.Uint32(data[0:4]))
v.Pid = binary.LittleEndian.Uint32(data[4:8])
v.Reason = uint32(binary.LittleEndian.Uint32(data[8:12]))
return true
}
128 changes: 128 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,128 @@
package ebpftracer

import (
"bytes"
"encoding/binary"
"testing"

"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)

func TestParseL7Event(t *testing.T) {
expected := l7Event{
Fd: 123,
ConnectionTimestamp: 456,
Pid: 789,
Status: 200,
Duration: 1000,
Protocol: 1,
Method: 2,
Padding: 0,
StatementId: 3,
PayloadSize: 100,
ResponseSize: 200,
}

buf := new(bytes.Buffer)
require.NoError(t, binary.Write(buf, binary.LittleEndian, expected))

data := buf.Bytes()
var actual l7Event
require.True(t, parseL7Event(data, &actual))

// Check fields
assert.Equal(t, expected.Fd, actual.Fd)
assert.Equal(t, expected.ConnectionTimestamp, actual.ConnectionTimestamp)
assert.Equal(t, expected.Pid, actual.Pid)
assert.Equal(t, expected.Status, actual.Status)
assert.Equal(t, expected.Duration, actual.Duration)
assert.Equal(t, expected.Protocol, actual.Protocol)
assert.Equal(t, expected.Method, actual.Method)
assert.Equal(t, expected.StatementId, actual.StatementId)
assert.Equal(t, expected.PayloadSize, actual.PayloadSize)
assert.Equal(t, expected.ResponseSize, actual.ResponseSize)
// Payload/Response are not copied by parseL7Event, so we don't check them here
}

func TestParseTCPEvent(t *testing.T) {
expected := tcpEvent{
Fd: 1,
Timestamp: 2,
Duration: 3,
Type: EventTypeConnectionOpen,
Pid: 4,
BytesSent: 5,
BytesReceived: 6,
SPort: 8080,
DPort: 9090,
Aport: 10000,
}
copy(expected.SAddr[:], []byte{1, 2, 3, 4})
copy(expected.DAddr[:], []byte{5, 6, 7, 8})
copy(expected.AAddr[:], []byte{9, 10, 11, 12})

buf := new(bytes.Buffer)
require.NoError(t, binary.Write(buf, binary.LittleEndian, expected))
buf.Write([]byte{0, 0}) // Add padding to match C struct size (104 bytes)

data := buf.Bytes()
var actual tcpEvent
require.True(t, parseTCPEvent(data, &actual))

assert.Equal(t, expected.Fd, actual.Fd)
assert.Equal(t, expected.Timestamp, actual.Timestamp)
assert.Equal(t, expected.Duration, actual.Duration)
assert.Equal(t, expected.Type, actual.Type)
assert.Equal(t, expected.Pid, actual.Pid)
assert.Equal(t, expected.BytesSent, actual.BytesSent)
assert.Equal(t, expected.BytesReceived, actual.BytesReceived)
assert.Equal(t, expected.SPort, actual.SPort)
assert.Equal(t, expected.DPort, actual.DPort)
assert.Equal(t, expected.Aport, actual.Aport)
assert.Equal(t, expected.SAddr, actual.SAddr)
assert.Equal(t, expected.DAddr, actual.DAddr)
assert.Equal(t, expected.AAddr, actual.AAddr)
}

func TestParseFileEvent(t *testing.T) {
expected := fileEvent{
Type: EventTypeFileOpen,
Pid: 123,
Fd: 456,
Mnt: 789,
Log: 1,
}

buf := new(bytes.Buffer)
require.NoError(t, binary.Write(buf, binary.LittleEndian, expected))

data := buf.Bytes()
var actual fileEvent
require.True(t, parseFileEvent(data, &actual))

assert.Equal(t, expected.Type, actual.Type)
assert.Equal(t, expected.Pid, actual.Pid)
assert.Equal(t, expected.Fd, actual.Fd)
assert.Equal(t, expected.Mnt, actual.Mnt)
assert.Equal(t, expected.Log, actual.Log)
}

func TestParseProcEvent(t *testing.T) {
expected := procEvent{
Type: EventTypeProcessStart,
Pid: 123,
Reason: 1,
}

buf := new(bytes.Buffer)
require.NoError(t, binary.Write(buf, binary.LittleEndian, expected))

data := buf.Bytes()
var actual procEvent
require.True(t, parseProcEvent(data, &actual))

assert.Equal(t, expected.Type, actual.Type)
assert.Equal(t, expected.Pid, actual.Pid)
assert.Equal(t, expected.Reason, actual.Reason)
}
38 changes: 23 additions & 15 deletions ebpftracer/tracer.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,6 @@ import (
"bytes"
"compress/gzip"
"encoding/base64"
"encoding/binary"
"errors"
"fmt"
"io"
Expand Down Expand Up @@ -471,21 +470,30 @@ func runEventsReader(name string, r *perf.Reader, ch chan<- Event, typ perfMapTy
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)
if !parseL7Event(data, v) {
klog.Warningln("failed to read l7 event: invalid size")
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))
// Extract payload data directly from the raw data (v.Payload/v.Response are empty)
payloadSize := int(v.PayloadSize)
if payloadSize > 4096 {
payloadSize = 4096
}
responseSize := int(v.ResponseSize)
if responseSize > 4096 {
responseSize = 4096
}

// Copy the actual data (preventing garbage from unused buffer space)
payloadData := make([]byte, payloadSize)
copy(payloadData, v.Payload[:payloadSize])
if payloadSize > 0 && len(data) >= 56+payloadSize {
copy(payloadData, data[56:56+payloadSize])
}

responseData := make([]byte, responseSize)
copy(responseData, v.Response[:responseSize])
if responseSize > 0 && len(data) >= 4152+responseSize {
copy(responseData, data[4152:4152+responseSize])
}

req := &l7.RequestData{
Protocol: l7.Protocol(v.Protocol),
Expand All @@ -508,22 +516,22 @@ func runEventsReader(name string, r *perf.Reader, ch chan<- Event, typ perfMapTy
}
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)
if !parseFileEvent(rec.RawSample, v) {
klog.Warningln("failed to read file event: invalid size")
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 {
klog.Warningln("failed to read proc event:", err)
if !parseProcEvent(rec.RawSample, v) {
klog.Warningln("failed to read proc event: invalid size")
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 {
klog.Warningln("failed to read tcp event:", err)
if !parseTCPEvent(rec.RawSample, v) {
klog.Warningln("failed to read tcp event: invalid size")
continue
}
event = Event{
Expand Down
Loading