Make event cache tunable

install/kubernetes/tetragon/templates/tetragon_configmap.yaml

@@ -67,4 +67,6 @@ data:
 {{- if .Values.tetragon.pprof.enabled }}
   pprof-address: {{ .Values.tetragon.pprof.address }}:{{ .Values.tetragon.pprof.port }}
 {{- end }}
+  event-cache-retries: {{ .Values.tetragon.eventCacheRetries | quote }}
+  event-cache-retry-delay: {{ .Values.tetragon.eventCacheRetryDelay | quote }}
   {{- include "configmap.extra" . | nindent 2 }}

install/kubernetes/tetragon/values.yaml

@@ -222,6 +222,10 @@ tetragon:
     extraVolumeMounts: []
     # -- resources for the the oci-hook-setup init container
     resources: {}
+  # -- Configure the number of retries in tetragon's event cache.
+  eventCacheRetries: 15
+  # -- Configure the delay (in seconds) between retires in tetragon's event cache.
+  eventCacheRetryDelay: 2
 # Tetragon Operator settings
 tetragonOperator:
   # -- Enables the Tetragon Operator.

pkg/defaults/defaults.go

@@ -45,6 +45,10 @@ const (
 
 	// Pid file where to write tetragon main PID
 	DefaultPidFile = DefaultRunDir + "tetragon.pid"
+
+	// defaults for the event cache
+	DefaultEventCacheNumRetries = 15
+	DefaultEventCacheRetryDelay = 2
 )
 
 var (

pkg/eventcache/eventcache.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/cilium/tetragon/api/v1/tetragon"
 	"github.com/cilium/tetragon/pkg/ktime"
+	"github.com/cilium/tetragon/pkg/logger"
 	"github.com/cilium/tetragon/pkg/option"
 	"github.com/cilium/tetragon/pkg/process"
 	"github.com/cilium/tetragon/pkg/reader/node"
@@ -23,13 +24,6 @@ const (
 	FROM_EV_CACHE
 )
 
-const (
-	// garbage collection retries
-	CacheStrikes = 15
-	// garbage collection run interval
-	EventRetryTimer = time.Second * 2
-)
-
 var (
 	cache    *Cache
 	nodeName string
@@ -135,7 +129,7 @@ func (ec *Cache) handleEvents() {
 		}
 		if err != nil {
 			event.color++
-			if event.color < CacheStrikes {
+			if event.color < option.Config.EventCacheNumRetries {
 				tmp = append(tmp, event)
 				continue
 			}
@@ -169,9 +163,9 @@ func (ec *Cache) loop() {
 	for {
 		select {
 		case <-ticker.C:
-			/* Every 'EventRetryTimer' walk the slice of events pending pod info. If
-			 * an event hasn't completed its podInfo after two iterations send the
-			 * event anyways.
+			/* Every 'option.Config.EventCacheRetryDelay' seconds walk the slice of events
+			 * pending pod info. If an event hasn't completed its podInfo after two iterations
+			 * send the event anyways.
 			 */
 			ec.handleEvents()
 
@@ -228,6 +222,8 @@ func NewWithTimer(s *server.Server, dur time.Duration) *Cache {
 		cache.done <- true
 	}
 
+	logger.GetLogger().WithField("retries", option.Config.EventCacheNumRetries).WithField("delay", dur).Info("Creating new EventCache")
+
 	cache = &Cache{
 		objsChan: make(chan CacheObj),
 		done:     make(chan bool),
@@ -241,7 +237,7 @@ func NewWithTimer(s *server.Server, dur time.Duration) *Cache {
 }
 
 func New(s *server.Server) *Cache {
-	return NewWithTimer(s, EventRetryTimer)
+	return NewWithTimer(s, time.Second*time.Duration(option.Config.EventCacheRetryDelay))
 }
 
 func Get() *Cache {

pkg/grpc/exec/exec_test_helper.go

@@ -27,7 +27,7 @@ import (
 )
 
 const (
-	CacheTimerMs = 100
+	CacheTimerMs = 1
 )
 
 var (
@@ -415,7 +415,7 @@ func GrpcExecOutOfOrder[EXEC notify.Message, EXIT notify.Message](t *testing.T)
 		AllEvents = append(AllEvents, e)
 	}
 
-	time.Sleep(time.Millisecond * ((eventcache.CacheStrikes + 4) * CacheTimerMs)) // wait for cache to do it's work
+	time.Sleep(time.Millisecond * time.Duration((option.Config.EventCacheNumRetries+4)*CacheTimerMs)) // wait for cache to do it's work
 	CheckExecEvents(t, AllEvents, parentPid, currentPid)
 }
 
@@ -474,7 +474,7 @@ func GrpcExecMisingParent[EXEC notify.Message, EXIT notify.Message](t *testing.T
 		AllEvents = append(AllEvents, e)
 	}
 
-	time.Sleep(time.Millisecond * ((eventcache.CacheStrikes + 4) * CacheTimerMs)) // wait for cache to do it's work
+	time.Sleep(time.Millisecond * time.Duration((option.Config.EventCacheNumRetries+4)*CacheTimerMs)) // wait for cache to do it's work
 
 	assert.Equal(t, len(AllEvents), 1)
 	execEv := AllEvents[0].GetProcessExec()
@@ -503,7 +503,7 @@ func GrpcMissingExec[EXEC notify.Message, EXIT notify.Message](t *testing.T) {
 		AllEvents = append(AllEvents, e)
 	}
 
-	time.Sleep(time.Millisecond * ((eventcache.CacheStrikes + 4) * CacheTimerMs)) // wait for cache to do it's work
+	time.Sleep(time.Millisecond * time.Duration((option.Config.EventCacheNumRetries+4)*CacheTimerMs)) // wait for cache to do it's work
 
 	assert.Equal(t, len(AllEvents), 1)
 	ev := AllEvents[0]
@@ -659,7 +659,7 @@ func GrpcExecCloneOutOfOrder[EXEC notify.Message, CLONE notify.Message, EXIT not
 		AllEvents = append(AllEvents, e)
 	}
 
-	time.Sleep(time.Millisecond * ((eventcache.CacheStrikes + 4) * CacheTimerMs)) // wait for cache to do it's work
+	time.Sleep(time.Millisecond * time.Duration((option.Config.EventCacheNumRetries+4)*CacheTimerMs)) // wait for cache to do it's work
 
 	CheckCloneEvents(t, AllEvents, currentPid, clonePid)
 }
@@ -793,8 +793,8 @@ func GrpcExecPodInfoInOrder[EXEC notify.Message, EXIT notify.Message](t *testing
 		AllEvents = append(AllEvents, e)
 	}
 
-	fakeWatcher.AddPod(dummyPod)                                                  // setup some dummy pod to return
-	time.Sleep(time.Millisecond * ((eventcache.CacheStrikes + 4) * CacheTimerMs)) // wait for cache to do it's work
+	fakeWatcher.AddPod(dummyPod)                                                                      // setup some dummy pod to return
+	time.Sleep(time.Millisecond * time.Duration((option.Config.EventCacheNumRetries+4)*CacheTimerMs)) // wait for cache to do it's work
 	CheckPodEvents(t, AllEvents)
 }
 
@@ -838,7 +838,7 @@ func GrpcExecPodInfoOutOfOrder[EXEC notify.Message, EXIT notify.Message](t *test
 	}
 
 	fakeWatcher.AddPod(dummyPod)
-	time.Sleep(time.Millisecond * ((eventcache.CacheStrikes + 4) * CacheTimerMs)) // wait for cache to do it's work
+	time.Sleep(time.Millisecond * time.Duration((option.Config.EventCacheNumRetries+4)*CacheTimerMs)) // wait for cache to do it's work
 	CheckPodEvents(t, AllEvents)
 }
 
@@ -886,7 +886,7 @@ func GrpcExecPodInfoInOrderAfter[EXEC notify.Message, EXIT notify.Message](t *te
 		AllEvents = append(AllEvents, e)
 	}
 
-	time.Sleep(time.Millisecond * ((eventcache.CacheStrikes + 4) * CacheTimerMs)) // wait for cache to do it's work
+	time.Sleep(time.Millisecond * time.Duration((option.Config.EventCacheNumRetries+4)*CacheTimerMs)) // wait for cache to do it's work
 	CheckPodEvents(t, AllEvents)
 }
 
@@ -933,7 +933,7 @@ func GrpcExecPodInfoOutOfOrderAfter[EXEC notify.Message, EXIT notify.Message](t
 		AllEvents = append(AllEvents, e)
 	}
 
-	time.Sleep(time.Millisecond * ((eventcache.CacheStrikes + 4) * CacheTimerMs)) // wait for cache to do it's work
+	time.Sleep(time.Millisecond * time.Duration((option.Config.EventCacheNumRetries+4)*CacheTimerMs)) // wait for cache to do it's work
 	CheckPodEvents(t, AllEvents)
 }
 
@@ -983,7 +983,7 @@ func GrpcExecPodInfoDelayedOutOfOrder[EXEC notify.Message, EXIT notify.Message](
 
 	fakeWatcher.AddPod(dummyPod) // setup some dummy pod to return
 
-	time.Sleep(time.Millisecond * ((eventcache.CacheStrikes + 4) * CacheTimerMs)) // wait for cache to do it's work
+	time.Sleep(time.Millisecond * time.Duration((option.Config.EventCacheNumRetries+4)*CacheTimerMs)) // wait for cache to do it's work
 
 	CheckPodEvents(t, AllEvents)
 }
@@ -1032,7 +1032,7 @@ func GrpcExecPodInfoDelayedInOrder[EXEC notify.Message, EXIT notify.Message](t *
 
 	fakeWatcher.AddPod(dummyPod) // setup some dummy pod to return
 
-	time.Sleep(time.Millisecond * ((eventcache.CacheStrikes + 4) * CacheTimerMs)) // wait for cache to do it's work
+	time.Sleep(time.Millisecond * time.Duration((option.Config.EventCacheNumRetries+4)*CacheTimerMs)) // wait for cache to do it's work
 
 	CheckPodEvents(t, AllEvents)
 }
@@ -1079,7 +1079,7 @@ func GrpcDelayedExecK8sOutOfOrder[EXEC notify.Message, EXIT notify.Message](t *t
 		AllEvents = append(AllEvents, e)
 	}
 
-	time.Sleep(time.Millisecond * ((eventcache.CacheStrikes + 4) * CacheTimerMs)) // wait for cache to do it's work
+	time.Sleep(time.Millisecond * time.Duration((option.Config.EventCacheNumRetries+4)*CacheTimerMs)) // wait for cache to do it's work
 
 	CheckPodEvents(t, AllEvents)
 }

pkg/jsonchecker/jsonchecker.go

@@ -15,15 +15,14 @@ import (
 	"github.com/cilium/tetragon/api/v1/tetragon"
 	ec "github.com/cilium/tetragon/api/v1/tetragon/codegen/eventchecker"
 	"github.com/cilium/tetragon/api/v1/tetragon/codegen/helpers"
-	"github.com/cilium/tetragon/pkg/eventcache"
 	"github.com/cilium/tetragon/pkg/logger"
 	"github.com/cilium/tetragon/pkg/testutils"
 	"github.com/sirupsen/logrus"
 )
 
-var (
+const (
 	Retries    = 13
-	RetryDelay = eventcache.EventRetryTimer + (1 * time.Second)
+	RetryDelay = 3 * time.Second
 )
 
 // DebugError is an error that will create a debug output message

pkg/option/config.go

@@ -10,6 +10,7 @@ import (
 	"path/filepath"
 	"time"
 
+	"github.com/cilium/tetragon/pkg/defaults"
 	"github.com/cilium/tetragon/pkg/logger"
 	"github.com/cilium/tetragon/pkg/metrics"
 	"github.com/spf13/viper"
@@ -100,6 +101,9 @@ type config struct {
 
 	EnableCgIDmap      bool
 	EnableCgIDmapDebug bool
+
+	EventCacheNumRetries int
+	EventCacheRetryDelay int
 }
 
 var (
@@ -117,6 +121,11 @@ var (
 
 		// Enable all metrics labels by default
 		MetricsLabelFilter: DefaultLabelFilter(),
+
+		// set default valus for the event cache
+		// mainly used in the case of testing
+		EventCacheNumRetries: defaults.DefaultEventCacheNumRetries,
+		EventCacheRetryDelay: defaults.DefaultEventCacheRetryDelay,
 	}
 )
 

pkg/option/flags.go

@@ -112,6 +112,9 @@ const (
 
 	KeyEnableCgIDmap      = "enable-cgidmap"
 	KeyEnableCgIDmapDebug = "enable-cgidmap-debug"
+
+	KeyEventCacheRetries    = "event-cache-retries"
+	KeyEventCacheRetryDelay = "event-cache-retry-delay"
 )
 
 type UsernameMetadaCode int
@@ -238,6 +241,10 @@ func ReadAndSetFlags() error {
 
 	Config.EnableCgIDmap = viper.GetBool(KeyEnableCgIDmap)
 	Config.EnableCgIDmapDebug = viper.GetBool(KeyEnableCgIDmapDebug)
+
+	Config.EventCacheNumRetries = viper.GetInt(KeyEventCacheRetries)
+	Config.EventCacheRetryDelay = viper.GetInt(KeyEventCacheRetryDelay)
+
 	return nil
 }
 
@@ -401,4 +408,7 @@ func AddFlags(flags *pflag.FlagSet) {
 
 	flags.Bool(KeyEnableCgIDmap, false, "enable pod resolution via cgroup ids")
 	flags.Bool(KeyEnableCgIDmapDebug, false, "enable cgidmap debugging info")
+
+	flags.Int(KeyEventCacheRetries, defaults.DefaultEventCacheNumRetries, "Number of retries for event cache")
+	flags.Int(KeyEventCacheRetryDelay, defaults.DefaultEventCacheRetryDelay, "Delay in seconds between event cache retries")
 }