Minor refactors

Makefile

@@ -48,6 +48,8 @@ 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)
@@ -57,6 +59,7 @@ 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_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/events/events.go

@@ -179,17 +179,10 @@ 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
 			}
@@ -201,6 +194,16 @@ func (ev *events) reconcileEventsDataChannel() {
 	}
 }
 
+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) {
 	readDone := true

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 {

test-data/ring_buffer.bpf.c

@@ -0,0 +1,32 @@
+#include "vmlinux.h"
+#include <bpf/bpf_helpers.h>
+#include <bpf/bpf_tracing.h>
+#include <bpf/bpf_core_read.h>
+
+#define BPF_MAP_TYPE_RINGBUF 27
+#define PIN_GLOBAL_NS 2
+
+struct bpf_map_def_pvt {
+	__u32 type;
+	__u32 key_size;
+	__u32 value_size;
+	__u32 max_entries;
+	__u32 map_flags;
+	__u32 pinning;
+	__u32 inner_map_fd;
+};
+
+struct bpf_map_def_pvt SEC("maps") test_events = {
+    .type = BPF_MAP_TYPE_RINGBUF,
+    .max_entries = 256 * 1024,
+    .pinning = PIN_GLOBAL_NS,
+};
+
+SEC("kprobe/__nf_conntrack_hash_insert")
+int conn_insert(struct pt_regs *ctx) {
+	__u32 evt_test = 20;
+	bpf_ringbuf_output(&test_events, &evt_test, sizeof(evt_test), 2);
+	return 0;
+}
+
+char _license[] SEC("license") = "GPL";