userspace to switch to use ringbuf

Signed-off-by: Mohamed Mahmoud <[email protected]>

Author: msherif1234

Diff for: examples/packetcapture-dump/README.md

@@ -19,7 +19,8 @@ Start the packetcapture-client using: (in a secondary shell)
 
 Start the agent using:
 ```bash
-sudo TARGET_HOST=localhost TARGET_PORT=9990 ENABLE_PCA="true" FILTER_IP_CIDR="0.0.0.0/0" FILTER_PROTOCOL="TCP" FILTER_PORT=22 FILTER_ACTION="Accept" ./bin/netobserv-ebpf-agent
+sudo TARGET_HOST=localhost TARGET_PORT=9990 ENABLE_PCA="true" FLOW_FILTER_RULES='[{"ip_cidr":"0.0.0.0/0","protocol":"TCP","action":"Accept"}]' ./bin/netobserv-ebpf-agent
+
 ```
 
 You should see output such as:

Diff for: pkg/agent/packets_agent.go

@@ -15,7 +15,7 @@ import (
 	"github.com/netobserv/netobserv-ebpf-agent/pkg/model"
 	"github.com/netobserv/netobserv-ebpf-agent/pkg/tracer"
 
-	"github.com/cilium/ebpf/perf"
+	"github.com/cilium/ebpf/ringbuf"
 	"github.com/sirupsen/logrus"
 )
 
@@ -47,7 +47,7 @@ type ebpfPacketFetcher interface {
 	AttachTCX(iface ifaces.Interface) error
 	DetachTCX(iface ifaces.Interface) error
 	LookupAndDeleteMap(*metrics.Metrics) map[int][]*byte
-	ReadPerf() (perf.Record, error)
+	ReadPerf() (ringbuf.Record, error)
 }
 
 // PacketsAgent instantiates a new agent, given a configuration.

Diff for: pkg/flow/tracer_perf.go

@@ -7,7 +7,7 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/cilium/ebpf/perf"
+	"github.com/cilium/ebpf/ringbuf"
 	"github.com/netobserv/gopipes/pkg/node"
 	"github.com/netobserv/netobserv-ebpf-agent/pkg/model"
 	"github.com/sirupsen/logrus"
@@ -24,7 +24,7 @@ type PerfTracer struct {
 }
 
 type perfReader interface {
-	ReadPerf() (perf.Record, error)
+	ReadPerf() (ringbuf.Record, error)
 }
 
 func NewPerfTracer(
@@ -46,7 +46,7 @@ func (m *PerfTracer) TraceLoop(ctx context.Context) node.StartFunc[*model.Packet
 			default:
 				if err := m.listenAndForwardPerf(out); err != nil {
 
-					if errors.Is(err, perf.ErrClosed) {
+					if errors.Is(err, ringbuf.ErrClosed) {
 						pblog.Debug("Received signal, exiting..")
 						return
 					}

Diff for: pkg/model/packet_record.go

@@ -28,25 +28,25 @@ func NewPacketRecord(
 	return &pr
 }
 
-// ReadRawPacket reads a PacketRecord from a binary source, in LittleEndian order
+// ReadRawPacket reads a PacketRecord from a binary source, in NativeEndian order
 func ReadRawPacket(reader io.Reader) (*PacketRecord, error) {
 	var pr PacketRecord
 	currentTime := time.Now()
 	monotonicTimeNow := monotime.Now()
 	getLen := make([]byte, 4)
 	packetTimestamp := make([]byte, 8)
-	// Read IfIndex and discard it: To be used in other usecases
-	_ = binary.Read(reader, binary.LittleEndian, make([]byte, 4))
-	// Read Length of packet
-	_ = binary.Read(reader, binary.LittleEndian, getLen)
-	pr.Stream = make([]byte, binary.LittleEndian.Uint32(getLen))
-	// Read TimeStamp of packet
-	_ = binary.Read(reader, binary.LittleEndian, packetTimestamp)
+	// Read IfIndex and discard it: To be used in other use cases
+	_ = binary.Read(reader, binary.NativeEndian, make([]byte, 4))
+	// Read Length of a packet
+	_ = binary.Read(reader, binary.NativeEndian, getLen)
+	pr.Stream = make([]byte, binary.NativeEndian.Uint32(getLen))
+	// Read TimeStamp of a packet
+	_ = binary.Read(reader, binary.NativeEndian, packetTimestamp)
 	// The assumption is monotonic time should be as close to time recorded by ebpf.
 	// The difference is considered the delta time from current time.
-	tsDelta := time.Duration(uint64(monotonicTimeNow) - binary.LittleEndian.Uint64(packetTimestamp))
+	tsDelta := time.Duration(uint64(monotonicTimeNow) - binary.NativeEndian.Uint64(packetTimestamp))
 	pr.Time = currentTime.Add(-tsDelta)
 
-	err := binary.Read(reader, binary.LittleEndian, &pr.Stream)
+	err := binary.Read(reader, binary.NativeEndian, &pr.Stream)
 	return &pr, err
 }

Diff for: pkg/model/record.go

@@ -192,10 +192,10 @@ func (m *MacAddr) MarshalJSON() ([]byte, error) {
 	return []byte("\"" + m.String() + "\""), nil
 }
 
-// ReadFrom reads a Record from a binary source, in LittleEndian order
+// ReadFrom reads a Record from a binary source, in NativeEndian order
 func ReadFrom(reader io.Reader) (*RawRecord, error) {
 	var fr RawRecord
-	err := binary.Read(reader, binary.LittleEndian, &fr)
+	err := binary.Read(reader, binary.NativeEndian, &fr)
 	return &fr, err
 }
 

Diff for: pkg/tracer/tracer.go

@@ -19,7 +19,6 @@ import (
 	cilium "github.com/cilium/ebpf"
 	"github.com/cilium/ebpf/btf"
 	"github.com/cilium/ebpf/link"
-	"github.com/cilium/ebpf/perf"
 	"github.com/cilium/ebpf/ringbuf"
 	"github.com/cilium/ebpf/rlimit"
 	"github.com/gavv/monotime"
@@ -1266,7 +1265,7 @@ type PacketFetcher struct {
 	qdiscs                   map[ifaces.Interface]*netlink.GenericQdisc
 	egressFilters            map[ifaces.Interface]*netlink.BpfFilter
 	ingressFilters           map[ifaces.Interface]*netlink.BpfFilter
-	perfReader               *perf.Reader
+	perfReader               *ringbuf.Reader
 	cacheMaxSize             int
 	enableIngress            bool
 	enableEgress             bool
@@ -1379,7 +1378,7 @@ func NewPacketFetcher(cfg *FlowFetcherConfig) (*PacketFetcher, error) {
 	}
 
 	// read packets from igress+egress perf array
-	packets, err := perf.NewReader(objects.PacketRecord, os.Getpagesize())
+	packets, err := ringbuf.NewReader(objects.BpfMaps.PacketRecord)
 	if err != nil {
 		return nil, fmt.Errorf("accessing to perf: %w", err)
 	}
@@ -1739,7 +1738,7 @@ func (p *PacketFetcher) Close() error {
 	return errors.New(`errors: "` + strings.Join(errStrings, `", "`) + `"`)
 }
 
-func (p *PacketFetcher) ReadPerf() (perf.Record, error) {
+func (p *PacketFetcher) ReadPerf() (ringbuf.Record, error) {
 	return p.perfReader.Read()
 }