Fix goroutine leak when using expirable.LRU cache (#236)

The expirable.LRU cache implementation from the hashicorp cache lib does
not offer any way to stop the spawned goroutine. Hence the library has
been replaced by elastic/go-freelru.

cmd/agent/daemon/enrichment/enrichers.go

@@ -15,16 +15,22 @@ import (
 	"github.com/castai/kvisor/pkg/ebpftracer/types"
 	"github.com/castai/kvisor/pkg/logging"
 	"github.com/castai/kvisor/pkg/proc"
-	lru "github.com/hashicorp/golang-lru/v2"
+	"github.com/cespare/xxhash"
+	"github.com/elastic/go-freelru"
 	"github.com/minio/sha256-simd"
 )
 
 type fileHashCacheKey string
 type ContainerForCgroupGetter func(cgroup uint64) (*containers.Container, bool, error)
 type PIDsInNamespaceGetter func(ns uint32) []uint32
 
+// more hash function in https://github.com/elastic/go-freelru/blob/main/bench/hash.go
+func hashStringXXHASH(s fileHashCacheKey) uint32 {
+	return uint32(xxhash.Sum64String(string(s)))
+}
+
 func EnrichWithFileHash(log *logging.Logger, mountNamespacePIDStore *types.PIDsPerNamespace, procFS proc.ProcFS) EventEnricher {
-	cache, err := lru.New[fileHashCacheKey, []byte](1024)
+	cache, err := freelru.NewSynced[fileHashCacheKey, []byte](1024, hashStringXXHASH)
 	if err != nil {
 		// This case can never happen, as err is only thrown if cache size is <0, which it isn't.
 		panic(err)
@@ -42,7 +48,7 @@ type fileHashEnricher struct {
 	log                    *logging.Logger
 	mountNamespacePIDStore *types.PIDsPerNamespace
 	procFS                 fs.StatFS
-	cache                  *lru.Cache[fileHashCacheKey, []byte]
+	cache                  freelru.Cache[fileHashCacheKey, []byte]
 }
 
 func (enricher *fileHashEnricher) EventTypes() []castpb.EventType {

go.mod

@@ -6,9 +6,11 @@ require (
 	github.com/aquasecurity/trivy v0.37.1
 	github.com/castai/image-analyzer v0.3.0
 	github.com/cenkalti/backoff/v4 v4.2.1
+	github.com/cespare/xxhash v1.1.0
 	github.com/cilium/ebpf v0.12.3
 	github.com/containerd/containerd v1.7.13
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc
+	github.com/elastic/go-freelru v0.13.0
 	github.com/fatih/color v1.16.0
 	github.com/florianl/go-conntrack v0.4.0
 	github.com/go-playground/validator/v10 v10.17.0
@@ -18,7 +20,6 @@ require (
 	github.com/google/uuid v1.6.0
 	github.com/grafana/pyroscope-go v1.1.1
 	github.com/hashicorp/golang-lru v1.0.2
-	github.com/hashicorp/golang-lru/v2 v2.0.7
 	github.com/iancoleman/strcase v0.3.0
 	github.com/json-iterator/go v1.1.12
 	github.com/kelseyhightower/envconfig v1.4.0
@@ -51,7 +52,10 @@ require (
 	k8s.io/klog/v2 v2.110.1
 )
 
-require github.com/klauspost/cpuid/v2 v2.2.3 // indirect
+require (
+	github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect
+	github.com/klauspost/cpuid/v2 v2.2.3 // indirect
+)
 
 require (
 	cloud.google.com/go v0.112.0 // indirect

go.sum

@@ -315,6 +315,7 @@ github.com/castai/image-analyzer v0.3.0/go.mod h1:CmHR7wL7P+z3JYvkFUP7WOsPsJ8hrj
 github.com/cenkalti/backoff/v4 v4.2.1 h1:y4OZtCnogmCPw98Zjyt5a6+QwPLGkiQsYW5oUqylYbM=
 github.com/cenkalti/backoff/v4 v4.2.1/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE=
 github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
+github.com/cespare/xxhash v1.1.0 h1:a6HrQnmkObjyL+Gs60czilIUGqrzKutQD6XZog3p+ko=
 github.com/cespare/xxhash v1.1.0/go.mod h1:XrSqR1VqqWfGrhpAt58auRo0WTKS1nRRg3ghfAqPWnc=
 github.com/cespare/xxhash/v2 v2.1.1/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/cespare/xxhash/v2 v2.2.0 h1:DC2CZ1Ep5Y4k3ZQ899DldepgrayRUGE6BBZ/cd9Cj44=
@@ -405,6 +406,8 @@ github.com/docker/libtrust v0.0.0-20150114040149-fa567046d9b1/go.mod h1:cyGadeNE
 github.com/dustin/go-humanize v1.0.0/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk=
 github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY=
 github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto=
+github.com/elastic/go-freelru v0.13.0 h1:TKKY6yCfNNNky7Pj9xZAOEpBcdNgZJfihEftOb55omg=
+github.com/elastic/go-freelru v0.13.0/go.mod h1:bSdWT4M0lW79K8QbX6XY2heQYSCqD7THoYf82pT/H3I=
 github.com/emicklei/go-restful/v3 v3.11.2 h1:1onLa9DcsMYO9P+CXaL0dStDqQ2EHHXLiz+BtnqkLAU=
 github.com/emicklei/go-restful/v3 v3.11.2/go.mod h1:6n3XBCmQQb25CM2LCACGz8ukIrRry+4bhvbpWn3mrbc=
 github.com/emirpasic/gods v1.12.0 h1:QAUIPSaCu4G+POclxeqb3F+WPpdKqFGlw36+yOzGlrg=
@@ -1008,6 +1011,7 @@ github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ
 github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ=
 github.com/sourcegraph/conc v0.3.0 h1:OQTbbt6P72L20UqAkXXuLOj79LfEanQ+YQFNpLA9ySo=
 github.com/sourcegraph/conc v0.3.0/go.mod h1:Sdozi7LEKbFPqYX2/J+iBAM6HpqSLTASQIKqDmF7Mt0=
+github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72 h1:qLC7fQah7D6K1B0ujays3HV9gkFtllcxhzImRR7ArPQ=
 github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA=
 github.com/spdx/tools-golang v0.3.0 h1:rtm+DHk3aAt74Fh0Wgucb4pCxjXV8SqHCPEb2iBd30k=
 github.com/spdx/tools-golang v0.3.0/go.mod h1:RO4Y3IFROJnz+43JKm1YOrbtgQNljW4gAPpA/sY2eqo=

pkg/bucketcache/bucketcache.go

@@ -1,23 +1,23 @@
 package bucketcache
 
 import (
-	lru "github.com/hashicorp/golang-lru/v2"
+	"github.com/elastic/go-freelru"
 )
 
 type BucketCache[K comparable, V any] struct {
-	cache         *lru.Cache[K, []V]
+	cache         freelru.Cache[K, []V]
 	maxBucketSize int
 }
 
-func New[K comparable, V any](cacheSize int, maxBucketSize int) (*BucketCache[K, V], error) {
-	cache, err := lru.New[K, []V](cacheSize)
+func New[K comparable, V any](cacheSize uint32, maxBucketSize uint32, hash freelru.HashKeyCallback[K]) (*BucketCache[K, V], error) {
+	cache, err := freelru.NewSynced[K, []V](cacheSize, hash)
 	if err != nil {
 		return nil, err
 	}
 
 	return &BucketCache[K, V]{
 		cache:         cache,
-		maxBucketSize: maxBucketSize,
+		maxBucketSize: int(maxBucketSize),
 	}, nil
 }
 

pkg/bucketcache/bucketcache_test.go

@@ -6,14 +6,18 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+func intHash(n int) uint32 {
+	return uint32(n)
+}
+
 func TestBucketCache(t *testing.T) {
 	t.Run("add single value", func(t *testing.T) {
 		r := require.New(t)
 
 		key := 10
 		val := 20
 
-		cache, err := New[int, int](10, 5)
+		cache, err := New[int, int](10, 5, intHash)
 		r.NoError(err)
 
 		added := cache.AddToBucket(key, val)
@@ -29,7 +33,7 @@ func TestBucketCache(t *testing.T) {
 		key := 10
 		vals := []int{20, 30, 40}
 
-		cache, err := New[int, int](10, 5)
+		cache, err := New[int, int](10, 5, intHash)
 		r.NoError(err)
 
 		for _, v := range vals {
@@ -50,7 +54,7 @@ func TestBucketCache(t *testing.T) {
 		key2 := 90
 		val2 := 99
 
-		cache, err := New[int, int](10, 5)
+		cache, err := New[int, int](10, 5, intHash)
 		r.NoError(err)
 
 		added := cache.AddToBucket(key1, val1)
@@ -72,7 +76,7 @@ func TestBucketCache(t *testing.T) {
 		key := 10
 		vals := []int{1, 2, 3, 4, 5, 6, 7, 8}
 
-		cache, err := New[int, int](2, 2)
+		cache, err := New[int, int](2, 2, intHash)
 		r.NoError(err)
 
 		for i, val := range vals {
@@ -95,7 +99,7 @@ func TestBucketCache(t *testing.T) {
 		key2 := 12
 		key3 := 13
 
-		cache, err := New[int, int](2, 2)
+		cache, err := New[int, int](2, 2, intHash)
 		r.NoError(err)
 
 		cache.AddToBucket(key1, 1)

pkg/ebpftracer/policy_filters.go

@@ -8,7 +8,8 @@ import (
 	castpb "github.com/castai/kvisor/api/v1/runtime"
 	"github.com/castai/kvisor/pkg/ebpftracer/types"
 	"github.com/castai/kvisor/pkg/logging"
-	"github.com/hashicorp/golang-lru/v2/expirable"
+	"github.com/cespare/xxhash"
+	"github.com/elastic/go-freelru"
 	"github.com/samber/lo"
 	"golang.org/x/time/rate"
 )
@@ -125,12 +126,24 @@ func FilterEmptyDnsAnswers(l *logging.Logger) EventFilterGenerator {
 	}
 }
 
+// more hash function in https://github.com/elastic/go-freelru/blob/main/bench/hash.go
+func hashStringXXHASH(s string) uint32 {
+	return uint32(xxhash.Sum64String(s))
+}
+
 // DeduplicateDnsEvents creates a filter that will drop any DNS event with questions already seen in `ttl` time
-func DeduplicateDnsEvents(l *logging.Logger, size int, ttl time.Duration) EventFilterGenerator {
+func DeduplicateDnsEvents(l *logging.Logger, size uint32, ttl time.Duration) EventFilterGenerator {
 	type cacheValue struct{}
 
 	return func() EventFilter {
-		cache := expirable.NewLRU[string, cacheValue](size, nil, ttl)
+		cache, err := freelru.New[string, cacheValue](size, hashStringXXHASH)
+		// err is only ever returned on configuration issues. There is nothing we can really do here, besides
+		// panicing and surfacing the error to the user.
+		if err != nil {
+			panic(err)
+		}
+
+    cache.SetLifetime(ttl)
 
 		return func(event *castpb.Event) error {
 			if event.GetEventType() != castpb.EventType_EVENT_DNS {

pkg/ebpftracer/types/types.go

@@ -21,8 +21,12 @@ type AddrTuple struct {
 
 type PIDsPerNamespace = bucketcache.BucketCache[proc.NamespaceID, proc.PID]
 
-func NewPIDsPerNamespaceCache(size, maxBucketSize int) (*PIDsPerNamespace, error) {
-	result, err := bucketcache.New[proc.NamespaceID, proc.PID](size, maxBucketSize)
+func namespaceHash(ns proc.NamespaceID) uint32 {
+	return uint32(ns)
+}
+
+func NewPIDsPerNamespaceCache(size, maxBucketSize uint32) (*PIDsPerNamespace, error) {
+	result, err := bucketcache.New[proc.NamespaceID, proc.PID](size, maxBucketSize, namespaceHash)
 	if err != nil {
 		return nil, err
 	}