add ringbuf support

manager.go

@@ -255,6 +255,8 @@ type Manager struct {
 
 	// PerfMaps - List of perf ring buffers handled by the manager
 	PerfMaps []*PerfMap
+
+	RingbufMaps []*RingbufMap
 }
 
 // DumpMaps - Return a string containing human readable info about eBPF maps
@@ -585,6 +587,16 @@ func (m *Manager) Start() error {
 		}
 	}
 
+	// Start ring buf readers
+	for _, ringBuf := range m.RingbufMaps {
+		if err := ringBuf.Start(); err != nil {
+			// Clean up
+			m.Stop(CleanInternal)
+			m.stateLock.Unlock()
+			return err
+		}
+	}
+
 	// Attach eBPF programs
 	for _, probe := range m.Probes {
 		// ignore the error, they are already collected per probes and will be surfaced by the
@@ -647,6 +659,15 @@ func (m *Manager) stop(cleanup MapCleanupType) error {
 		err = ConcatErrors(err, e)
 	}
 
+	// Stop ring buf readers
+	for _, ringBuf := range m.RingbufMaps {
+		e := ringBuf.Stop(cleanup)
+		if e != nil {
+			e = errors.New(fmt.Sprintf("error:%v , ring buf reader %s couldn't gracefully shut down", ringBuf.Stop(cleanup), ringBuf.Name))
+		}
+		err = ConcatErrors(err, e)
+	}
+
 	// Detach eBPF programs
 	for _, probe := range m.Probes {
 		e := probe.Stop()
@@ -1345,6 +1366,13 @@ func (m *Manager) loadCollection() error {
 		}
 	}
 
+	//Initialize ringbufmap
+	for _, ringbufMap := range m.RingbufMaps {
+		if err := ringbufMap.Init(m); err != nil {
+			return err
+		}
+	}
+
 	// Initialize Probes
 	for _, probe := range m.Probes {
 		// Find program

ringbuf.go

@@ -0,0 +1,111 @@
+package manager
+
+import (
+	"errors"
+	"fmt"
+	"github.com/cilium/ebpf/ringbuf"
+)
+
+// RingbufMapOptions - Perf map specific options
+type RingbufMapOptions struct {
+	// Watermark - The reader will start processing samples once their sizes in the perf ring buffer
+	// exceed this value. Must be smaller than PerfRingBufferSize. Defaults to the manager value if not set.
+	Watermark int
+
+	// PerfErrChan - Perf reader error channel
+	PerfErrChan chan error
+
+	// DataHandler - Callback function called when a new sample was retrieved from the perf
+	// ring buffer.
+	DataHandler func(CPU int, data []byte, perfMap *RingbufMap, manager *Manager)
+
+	// PerfMapStats - Perf map statistics event like nr Read errors, lost samples,
+	// RawSamples bytes count. Need to be initialized via manager.NewPerfMapStats()
+	PerfMapStats *PerfMapStats
+
+	// DumpHandler - Callback function called when manager.Dump() is called
+	// and dump the current state (human readable)
+	DumpHandler func(perfMap *PerfMap, manager *Manager) string
+}
+
+// RingbufMap -  ring buffer reader wrapper
+type RingbufMap struct {
+	manager       *Manager
+	ringBufReader *ringbuf.Reader
+
+	// Map - A PerfMap has the same features as a normal Map
+	Map
+	RingbufMapOptions
+}
+
+func (m *RingbufMap) Init(manager *Manager) error {
+	m.manager = manager
+	if m.DataHandler == nil {
+		return fmt.Errorf("no DataHandler set for %s", m.Name)
+	}
+
+	if m.Watermark == 0 {
+		m.Watermark = manager.options.DefaultWatermark
+	}
+
+	// Initialize the underlying map structure
+	if err := m.Map.Init(manager); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (m *RingbufMap) Start() error {
+	m.stateLock.Lock()
+	defer m.stateLock.Unlock()
+
+	if m.state == running {
+		return nil
+	}
+	if m.state < initialized {
+		return ErrMapNotInitialized
+	}
+
+	var err error
+	m.ringBufReader, err = ringbuf.NewReader(m.array)
+	if err != nil {
+		return err
+	}
+	go func() {
+		m.manager.wg.Add(1)
+		for {
+			record, err := m.ringBufReader.Read()
+			if err != nil {
+				if errors.Is(err, ringbuf.ErrClosed) {
+					m.manager.wg.Done()
+					return
+				}
+			}
+			m.DataHandler(0, record.RawSample, m, m.manager)
+		}
+	}()
+	m.state = running
+	return nil
+}
+
+func (m *RingbufMap) Stop(cleanup MapCleanupType) error {
+	m.stateLock.Lock()
+	defer m.stateLock.Unlock()
+
+	if m.state < running {
+		return nil
+	}
+	err := m.ringBufReader.Close()
+	m.state = initialized
+
+	// close underlying map
+	if errTmp := m.Map.close(cleanup); errTmp != nil {
+		if err == nil {
+			err = errTmp
+		} else {
+			err = fmt.Errorf("error%v, %s", errTmp, err.Error())
+		}
+	}
+	return err
+}