tetragon: Add buffer between perf reader and events processing code

cmd/tetragon/flags.go

@@ -67,6 +67,7 @@ const (
 
 	keyRBSize      = "rb-size"
 	keyRBSizeTotal = "rb-size-total"
+	keyRBQueueSize = "rb-queue-size"
 
 	keyEventQueueSize = "event-queue-size"
 
@@ -102,6 +103,7 @@ func readAndSetFlags() {
 
 	option.Config.RBSize = viper.GetInt(keyRBSize)
 	option.Config.RBSizeTotal = viper.GetInt(keyRBSizeTotal)
+	option.Config.RBQueueSize = viper.GetInt(keyRBQueueSize)
 
 	option.Config.GopsAddr = viper.GetString(keyGopsAddr)
 

cmd/tetragon/main.go

@@ -768,6 +768,8 @@ func execute() error {
 
 	flags.StringSlice(keyKmods, []string{}, "List of kernel modules to load symbols from")
 
+	flags.Int(keyRBQueueSize, 65535, "Set size of channel between ring buffer and sensor go routines (default 65k)")
+
 	viper.BindPFlags(flags)
 	return rootCmd.Execute()
 }

pkg/metrics/config/initmetrics.go

@@ -14,6 +14,7 @@ import (
 	pfmetrics "github.com/cilium/tetragon/pkg/metrics/policyfilter"
 	"github.com/cilium/tetragon/pkg/metrics/processexecmetrics"
 	"github.com/cilium/tetragon/pkg/metrics/ringbufmetrics"
+	"github.com/cilium/tetragon/pkg/metrics/ringbufqueuemetrics"
 	"github.com/cilium/tetragon/pkg/metrics/syscallmetrics"
 	"github.com/cilium/tetragon/pkg/metrics/watchermetrics"
 	"github.com/cilium/tetragon/pkg/observer"
@@ -30,6 +31,7 @@ func InitAllMetrics(registry *prometheus.Registry) {
 	pfmetrics.InitMetrics(registry)
 	processexecmetrics.InitMetrics(registry)
 	ringbufmetrics.InitMetrics(registry)
+	ringbufqueuemetrics.InitMetrics(registry)
 	syscallmetrics.InitMetrics(registry)
 	watchermetrics.InitMetrics(registry)
 	observer.InitMetrics(registry)

pkg/metrics/ringbufqueuemetrics/ringbufqueuemetrics.go

@@ -0,0 +1,29 @@
+// SPDX-License-Identifier: Apache-2.0
+// Copyright Authors of Tetragon
+
+package ringbufqueuemetrics
+
+import (
+	"github.com/cilium/tetragon/pkg/metrics/consts"
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+var (
+	Received = prometheus.NewCounter(prometheus.CounterOpts{
+		Namespace:   consts.MetricsNamespace,
+		Name:        "ringbuf_queue_received_total",
+		Help:        "The total number of Tetragon events ring buffer queue received.",
+		ConstLabels: nil,
+	})
+	Lost = prometheus.NewCounter(prometheus.CounterOpts{
+		Namespace:   consts.MetricsNamespace,
+		Name:        "ringbuf_queue_lost_total",
+		Help:        "The total number of Tetragon events ring buffer queue lost.",
+		ConstLabels: nil,
+	})
+)
+
+func InitMetrics(registry *prometheus.Registry) {
+	registry.MustRegister(Received)
+	registry.MustRegister(Lost)
+}

pkg/observer/observer.go

@@ -23,6 +23,7 @@ import (
 	"github.com/cilium/tetragon/pkg/metrics/errormetrics"
 	"github.com/cilium/tetragon/pkg/metrics/opcodemetrics"
 	"github.com/cilium/tetragon/pkg/metrics/ringbufmetrics"
+	"github.com/cilium/tetragon/pkg/metrics/ringbufqueuemetrics"
 	"github.com/cilium/tetragon/pkg/option"
 	"github.com/cilium/tetragon/pkg/reader/notify"
 	"github.com/cilium/tetragon/pkg/sensors"
@@ -126,7 +127,6 @@ func (k *Observer) receiveEvent(data []byte) {
 	if option.Config.EnableMsgHandlingLatency {
 		timer = time.Now()
 	}
-	atomic.AddUint64(&k.recvCntr, 1)
 
 	op, events, err := HandlePerfData(data)
 	opcodemetrics.OpTotalInc(int(op))
@@ -186,6 +186,15 @@ func (k *Observer) getRBSize(cpus int) int {
 	return size
 }
 
+func (k *Observer) getRBQueueSize() int {
+	size := option.Config.RBQueueSize
+	if size == 0 {
+		size = 65535
+	}
+	k.log.WithField("size", size).Info("Perf ring buffer events queue size (events)")
+	return size
+}
+
 func (k *Observer) RunEvents(stopCtx context.Context, ready func()) error {
 	pinOpts := ebpf.LoadPinOptions{}
 	perfMap, err := ebpf.LoadPinnedMap(k.PerfConfig.MapName, &pinOpts)
@@ -205,6 +214,10 @@ func (k *Observer) RunEvents(stopCtx context.Context, ready func()) error {
 	k.observerListeners(&readyapi.MsgTetragonReady{})
 	ready()
 
+	// We spawn go routine to read and process perf events,
+	// connected with main app through eventsQueue channel.
+	eventsQueue := make(chan *perf.Record, k.getRBQueueSize())
+
 	// Listeners are ready and about to start reading from perf reader, tell
 	// user everything is ready.
 	k.log.Info("Listening for events...")
@@ -226,7 +239,13 @@ func (k *Observer) RunEvents(stopCtx context.Context, ready func()) error {
 				}
 			} else {
 				if len(record.RawSample) > 0 {
-					k.receiveEvent(record.RawSample)
+					select {
+					case eventsQueue <- &record:
+					default:
+						// eventsQueue channel is full, drop the event
+						ringbufqueuemetrics.Lost.Inc()
+					}
+					k.recvCntr++
 					ringbufmetrics.PerfEventReceived.Inc()
 				}
 
@@ -236,7 +255,23 @@ func (k *Observer) RunEvents(stopCtx context.Context, ready func()) error {
 				}
 			}
 		}
-		k.log.WithError(stopCtx.Err()).Info("Listening for events completed.")
+	}()
+
+	// Start processing records from perf.
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for {
+			select {
+			case event := <-eventsQueue:
+				k.receiveEvent(event.RawSample)
+				ringbufqueuemetrics.Received.Inc()
+			case <-stopCtx.Done():
+				k.log.WithError(stopCtx.Err()).Infof("Listening for events completed.")
+				k.log.Debugf("Unprocessed events in RB queue: %d", len(eventsQueue))
+				return
+			}
+		}
 	}()
 
 	// Loading default program consumes some memory lets kick GC to give

pkg/option/config.go

@@ -42,6 +42,7 @@ type config struct {
 
 	RBSize      int
 	RBSizeTotal int
+	RBQueueSize int
 
 	ProcessCacheSize int
 	DataCacheSize    int