Xdp and Events UT

Makefile

@@ -46,6 +46,10 @@ EBPF_TEST_RECOVERY_SOURCE := test-data/recoverydata.bpf.c
 EBPF_TEST_RECOVERY_BINARY := test-data/recoverydata.bpf.elf
 EBPF_TEST_KPROBE_SOURCE := test-data/test-kprobe.bpf.c
 EBPF_TEST_KPROBE_BINARY := test-data/test-kprobe.bpf.elf    
+EBPF_TEST_XDP_SOURCE := test-data/xdp.bpf.c
+EBPF_TEST_XDP_BINARY := test-data/xdp.bpf.elf  
+EBPF_TEST_RINGBUFFER_SOURCE := test-data/ring_buffer.bpf.c
+EBPF_TEST_RINGBUFFER_BINARY := test-data/ring_buffer.bpf.elf  
 build-bpf: ## Build BPF
 	$(CLANG) $(CLANG_INCLUDE) -g -O2 -Wall -fpie -target bpf -DCORE -D__BPF_TRACING__ -march=bpf -D__TARGET_ARCH_$(ARCH) -c $(EBPF_SOURCE) -o $(EBPF_BINARY)
 	$(CLANG) $(CLANG_INCLUDE) -g -O2 -Wall -fpie -target bpf -DCORE -D__BPF_TRACING__ -march=bpf -D__TARGET_ARCH_$(ARCH) -c $(EBPF_TEST_SOURCE) -o $(EBPF_TEST_BINARY)
@@ -54,6 +58,8 @@ build-bpf: ## Build BPF
 	$(CLANG) $(CLANG_INCLUDE) -g -O2 -Wall -fpie -target bpf -DCORE -D__BPF_TRACING__ -march=bpf -D__TARGET_ARCH_$(ARCH) -c $(EBPF_TEST_INV_MAP_SOURCE) -o $(EBPF_TEST_INV_MAP_BINARY)
 	$(CLANG) $(CLANG_INCLUDE) -g -O2 -Wall -fpie -target bpf -DCORE -D__BPF_TRACING__ -march=bpf -D__TARGET_ARCH_$(ARCH) -c $(EBPF_TEST_RECOVERY_SOURCE) -o $(EBPF_TEST_RECOVERY_BINARY)
 	$(CLANG) $(CLANG_INCLUDE) -g -O2 -Wall -fpie -target bpf -DCORE -D__BPF_TRACING__ -march=bpf -D__TARGET_ARCH_$(ARCH) -c $(EBPF_TEST_KPROBE_SOURCE) -o $(EBPF_TEST_KPROBE_BINARY)
+	$(CLANG) $(CLANG_INCLUDE) -g -O2 -Wall -fpie -target bpf -DCORE -D__BPF_TRACING__ -march=bpf -D__TARGET_ARCH_$(ARCH) -c $(EBPF_TEST_XDP_SOURCE) -o $(EBPF_TEST_XDP_BINARY)
+	$(CLANG) $(CLANG_INCLUDE) -g -O2 -Wall -fpie -target bpf -DCORE -D__BPF_TRACING__ -march=bpf -D__TARGET_ARCH_$(ARCH) -c $(EBPF_TEST_RINGBUFFER_SOURCE) -o $(EBPF_TEST_RINGBUFFER_BINARY)
 
 vmlinuxh:
 	bpftool btf dump file /sys/kernel/btf/vmlinux format c > $(abspath ./test-data/vmlinux.h)

pkg/elfparser/elf_test.go

@@ -526,13 +526,6 @@ func TestRecovery(t *testing.T) {
 			wantProg:    3,
 			wantErr:     nil,
 		},
-		{
-			name:         "Missing BPF mount",
-			elfFileName:  "../../test-data/recoverydata.bpf.elf",
-			wantProg:     0,
-			forceUnMount: true,
-			wantErr:      errors.New("error checking BPF FS, please make sure it is mounted"),
-		},
 	}
 
 	for _, tt := range progtests {
@@ -560,9 +553,6 @@ func TestRecovery(t *testing.T) {
 					assert.NoError(t, err)
 				}
 
-				if tt.forceUnMount {
-					utils.Unmount_bpf_fs()
-				}
 				recoveredData, err := bpfSDKclient.RecoverAllBpfProgramsAndMaps()
 				if tt.wantErr != nil {
 					assert.EqualError(t, err, tt.wantErr.Error())

pkg/events/events.go

@@ -30,15 +30,27 @@ import (
 
 var log = logger.Get()
 
-type Events struct {
-	RingBuffers          []*RingBuffer
-	PageSize             int
-	RingCnt              int
-	stopRingBufferChan   chan struct{}
-	updateRingBufferChan chan *RingBuffer
-	eventsStopChannel    chan struct{}
-	wg                   sync.WaitGroup
-	eventsDataChannel    chan []byte
+type Events interface {
+	InitRingBuffer(mapFDlist []int) (map[int]chan []byte, error)
+}
+
+var _ Events = &events{}
+
+func New() Events {
+	return &events{
+		PageSize: os.Getpagesize(),
+		RingCnt:  0,
+	}
+
+}
+
+type events struct {
+	RingBuffers       []*RingBuffer
+	PageSize          int
+	RingCnt           int
+	eventsStopChannel chan struct{}
+	wg                sync.WaitGroup
+	eventsDataChannel chan []byte
 
 	epoller *poller.EventPoller
 }
@@ -62,18 +74,13 @@ func isValidMapFDList(mapFDlist []int) bool {
 	return true
 }
 
-func InitRingBuffer(mapFDlist []int) (map[int]chan []byte, error) {
+func (ev *events) InitRingBuffer(mapFDlist []int) (map[int]chan []byte, error) {
 
 	// Validate mapFD
 	if !isValidMapFDList(mapFDlist) {
 		return nil, fmt.Errorf("mapFDs passed to InitRingBuffer is invalid")
 	}
 
-	ev := &Events{
-		PageSize: os.Getpagesize(),
-		RingCnt:  0,
-	}
-
 	epoll, err := poller.NewEventPoller()
 	if err != nil {
 		return nil, fmt.Errorf("failed to create epoll instance: %s", err)
@@ -89,9 +96,9 @@ func InitRingBuffer(mapFDlist []int) (map[int]chan []byte, error) {
 			return nil, fmt.Errorf("failed to map info")
 		}
 
-		eventsChan, err := ev.SetupRingBuffer(mapFD, mapInfo.MaxEntries)
+		eventsChan, err := ev.setupRingBuffer(mapFD, mapInfo.MaxEntries)
 		if err != nil {
-			ev.CleanupRingBuffer()
+			ev.cleanupRingBuffer()
 			return nil, fmt.Errorf("failed to add ring buffer: %s", err)
 		}
 
@@ -101,7 +108,7 @@ func InitRingBuffer(mapFDlist []int) (map[int]chan []byte, error) {
 	return ringBufferChanList, nil
 }
 
-func (ev *Events) SetupRingBuffer(mapFD int, maxEntries uint32) (chan []byte, error) {
+func (ev *events) setupRingBuffer(mapFD int, maxEntries uint32) (chan []byte, error) {
 	ringbuffer := &RingBuffer{
 		RingBufferMapFD: mapFD,
 		Mask:            uint64(maxEntries - 1),
@@ -148,7 +155,7 @@ func (ev *Events) SetupRingBuffer(mapFD int, maxEntries uint32) (chan []byte, er
 	return ev.eventsDataChannel, nil
 }
 
-func (ev *Events) CleanupRingBuffer() {
+func (ev *events) cleanupRingBuffer() {
 
 	for i := 0; i < ev.RingCnt; i++ {
 		_ = unix.Munmap(ev.RingBuffers[i].Producer)
@@ -165,38 +172,41 @@ func (ev *Events) CleanupRingBuffer() {
 	return
 }
 
-func (ev *Events) reconcileEventsDataChannel() {
-
-	pollerCh := ev.epoller.EpollStart()
-	defer func() {
-		ev.wg.Done()
-	}()
-
+func (ev *events) reconcileEventsDataChannelHandler(pollerCh <-chan int) {
 	for {
 		select {
 		case bufferPtr, ok := <-pollerCh:
-
 			if !ok {
 				return
 			}
-			ev.ReadRingBuffer(ev.RingBuffers[bufferPtr])
+			ev.readRingBuffer(ev.RingBuffers[bufferPtr])
 
 		case <-ev.eventsStopChannel:
 			return
 		}
 	}
 }
 
+func (ev *events) reconcileEventsDataChannel() {
+
+	pollerCh := ev.epoller.EpollStart()
+	defer ev.wg.Done()
+
+	go ev.reconcileEventsDataChannelHandler(pollerCh)
+
+	<-ev.eventsStopChannel
+}
+
 // Similar to libbpf poll ring
-func (ev *Events) ReadRingBuffer(eventRing *RingBuffer) {
+func (ev *events) readRingBuffer(eventRing *RingBuffer) {
 	readDone := true
 	consPosition := eventRing.getConsumerPosition()
 	for !readDone {
 		readDone = ev.parseBuffer(consPosition, eventRing)
 	}
 }
 
-func (ev *Events) parseBuffer(consumerPosition uint64, eventRing *RingBuffer) bool {
+func (ev *events) parseBuffer(consumerPosition uint64, eventRing *RingBuffer) bool {
 	readDone := true
 	producerPosition := eventRing.getProducerPosition()
 	for consumerPosition < producerPosition {

pkg/events/generate_mocks.go

@@ -0,0 +1,15 @@
+// Copyright Amazon.com Inc. or its affiliates. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License"). You may
+// not use this file except in compliance with the License. A copy of the
+// License is located at
+//
+//	http://aws.amazon.com/apache2.0/
+//
+// or in the "license" file accompanying this file. This file is distributed
+// on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+// express or implied. See the License for the specific language governing
+// permissions and limitations under the License.
+package events
+
+//go:generate go run github.com/golang/mock/mockgen -destination mocks/events_mocks.go . Events

pkg/events/poll/epoll.go

@@ -17,17 +17,19 @@ type EventPoller struct {
 	epollEvent []unix.EpollEvent
 	bufferCnt  int
 
-	stopEventPollerChan   chan struct{}
-	updateEventPollerChan chan int
+	stopEventPollerChan chan struct{}
+	fdEventPollerChan   chan int
 }
 
 func NewEventPoller() (*EventPoller, error) {
 	epollFD, err := unix.EpollCreate1(unix.EPOLL_CLOEXEC)
 	if err != nil {
-		return nil, fmt.Errorf("Failed to create epoll instance: %s", err)
+		return nil, fmt.Errorf("failed to create epoll instance: %s", err)
 	}
 	e := &EventPoller{
-		epollFd: epollFD,
+		epollFd:             epollFD,
+		stopEventPollerChan: make(chan struct{}),
+		fdEventPollerChan:   make(chan int),
 	}
 	return e, nil
 }
@@ -44,22 +46,28 @@ func (e *EventPoller) AddEpollCtl(mapFD, eventFD int) error {
 
 	err := unix.EpollCtl(e.epollFd, unix.EPOLL_CTL_ADD, mapFD, &epollEvent)
 	if err != nil {
-		return fmt.Errorf("Failed to Epoll event: %s", err)
+		return fmt.Errorf("failed to Epoll event: %s", err)
 	}
 	e.epollEvent = append(e.epollEvent, epollEvent)
 	return nil
 }
 
 func (e *EventPoller) EpollStart() <-chan int {
-
-	e.stopEventPollerChan = make(chan struct{})
-	e.updateEventPollerChan = make(chan int)
 	e.wg.Add(1)
 	go e.eventsPoller()
-
-	return e.updateEventPollerChan
+	return e.fdEventPollerChan
 }
 
+func (e *EventPoller) getEventFDs(totalEvents int) {
+	for _, event := range e.epollEvent[:totalEvents] {
+		select {
+		case e.fdEventPollerChan <- int(event.Fd):
+
+		case <-e.stopEventPollerChan:
+			return
+		}
+	}
+}
 func (e *EventPoller) eventsPoller() {
 	defer e.wg.Done()
 	for {
@@ -69,20 +77,12 @@ func (e *EventPoller) eventsPoller() {
 		default:
 			break
 		}
-		numEvents := e.poll(e.epollEvent[:e.bufferCnt])
-		for _, event := range e.epollEvent[:numEvents] {
-			select {
-			case e.updateEventPollerChan <- int(event.Fd):
-
-			case <-e.stopEventPollerChan:
-				return
-			}
-		}
+		totalEvents := e.poll(e.epollEvent[:e.bufferCnt])
+		e.getEventFDs(totalEvents)
 	}
 }
 
 func (e *EventPoller) poll(events []unix.EpollEvent) int {
-
 	timeoutMs := 150
 	n, err := unix.EpollWait(e.epollFd, events, timeoutMs)
 	if err != nil {

pkg/tc/tc.go

@@ -37,7 +37,6 @@ type BpfTc interface {
 	TCEgressAttach(interfaceName string, progFD int, funcName string) error
 	TCEgressDetach(interfaceName string) error
 	CleanupQdiscs(ingressCleanup bool, egressCleanup bool) error
-	mismatchedInterfacePrefix(interfaceName string) error
 }
 
 var _ BpfTc = &bpfTc{}
@@ -76,17 +75,17 @@ func enableQdisc(link netlink.Link) bool {
 
 }
 
-func (m *bpfTc) mismatchedInterfacePrefix(interfaceName string) error {
-	if !strings.HasPrefix(interfaceName, m.InterfacePrefix) {
-		log.Errorf("expected prefix - %s but got %s", m.InterfacePrefix, interfaceName)
+func mismatchedInterfacePrefix(interfaceName string, interfacePrefix string) error {
+	if !strings.HasPrefix(interfaceName, interfacePrefix) {
+		log.Errorf("expected prefix - %s but got %s", interfacePrefix, interfaceName)
 		return errors.New("Mismatched initialized prefix name and passed interface name")
 	}
 	return nil
 }
 
 func (m *bpfTc) TCIngressAttach(interfaceName string, progFD int, funcName string) error {
 
-	if err := m.mismatchedInterfacePrefix(interfaceName); err != nil {
+	if err := mismatchedInterfacePrefix(interfaceName, m.InterfacePrefix); err != nil {
 		return err
 	}
 
@@ -138,7 +137,7 @@ func (m *bpfTc) TCIngressAttach(interfaceName string, progFD int, funcName strin
 
 func (m *bpfTc) TCIngressDetach(interfaceName string) error {
 
-	if err := m.mismatchedInterfacePrefix(interfaceName); err != nil {
+	if err := mismatchedInterfacePrefix(interfaceName, m.InterfacePrefix); err != nil {
 		return err
 	}
 
@@ -174,7 +173,7 @@ func (m *bpfTc) TCIngressDetach(interfaceName string) error {
 
 func (m *bpfTc) TCEgressAttach(interfaceName string, progFD int, funcName string) error {
 
-	if err := m.mismatchedInterfacePrefix(interfaceName); err != nil {
+	if err := mismatchedInterfacePrefix(interfaceName, m.InterfacePrefix); err != nil {
 		return err
 	}
 
@@ -226,7 +225,7 @@ func (m *bpfTc) TCEgressAttach(interfaceName string, progFD int, funcName string
 
 func (m *bpfTc) TCEgressDetach(interfaceName string) error {
 
-	if err := m.mismatchedInterfacePrefix(interfaceName); err != nil {
+	if err := mismatchedInterfacePrefix(interfaceName, m.InterfacePrefix); err != nil {
 		return err
 	}
 

pkg/tc/tc_test.go

@@ -120,8 +120,7 @@ func TestMismatchedPrefixName(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 
-			testTcClient := New("eni")
-			err := testTcClient.mismatchedInterfacePrefix(tt.interfaceName)
+			err := mismatchedInterfacePrefix(tt.interfaceName, "eni")
 			if tt.wantErr != nil {
 				assert.EqualError(t, err, tt.wantErr.Error())
 			} else {