⚡ Bolt: Optimize eBPF event parsing

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)
+}

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)
+	}
+}