From defd5c844c93a8779d0d68e0a41345b4582c0c97 Mon Sep 17 00:00:00 2001 From: Matthias Bertschy Date: Tue, 15 Sep 2026 12:27:20 +0200 Subject: [PATCH 1/2] fix: replace SafeMap with standard library synchronization Signed-off-by: Matthias Bertschy --- admission/rulebinding/cache/cache.go | 61 ++++++++++++++----- admission/rulebinding/cache/cache_test.go | 72 ++++++++++++++++++++--- go.mod | 2 +- watcher/podwatcher.go | 4 +- watcher/podwatcher_test.go | 10 ++-- watcher/sbomwatcher.go | 14 +++-- watcher/sbomwatcher_test.go | 40 ++++++++----- watcher/watchhandler.go | 11 ++-- 8 files changed, 162 insertions(+), 52 deletions(-) diff --git a/admission/rulebinding/cache/cache.go b/admission/rulebinding/cache/cache.go index 9f2cafe..980d3aa 100644 --- a/admission/rulebinding/cache/cache.go +++ b/admission/rulebinding/cache/cache.go @@ -2,8 +2,8 @@ package cache import ( "context" + "sync" - "github.com/goradd/maps" "github.com/kubescape/go-logger" "github.com/kubescape/go-logger/helpers" "github.com/kubescape/node-agent/pkg/k8sclient" @@ -27,8 +27,9 @@ var _ watcher.Adaptor = (*RBCache)(nil) type RBCache struct { k8sClient k8sclient.K8sClientInterface - rbNameToRB maps.SafeMap[string, typesv1.RuntimeAlertRuleBinding] // rule binding name -> rule binding - rbNameToRules maps.SafeMap[string, []rules.RuleEvaluator] // rule binding name -> []created rules + mu sync.RWMutex // protects rbNameToRB and rbNameToRules + rbNameToRB map[string]typesv1.RuntimeAlertRuleBinding // rule binding name -> rule binding + rbNameToRules map[string][]rules.RuleEvaluator // rule binding name -> []created rules ruleCreator rules.RuleCreator watchResources []watcher.WatchResource notifiers []*chan rulebindingmanager.RuleBindingNotify @@ -40,17 +41,35 @@ func NewCache(k8sClient k8sclient.K8sClientInterface, ruleCreator rules.RuleCrea return &RBCache{ k8sClient: k8sClient, ruleCreator: ruleCreator, - rbNameToRB: maps.SafeMap[string, typesv1.RuntimeAlertRuleBinding]{}, + rbNameToRB: make(map[string]typesv1.RuntimeAlertRuleBinding), + rbNameToRules: make(map[string][]rules.RuleEvaluator), watchResources: resourcesToWatch(), ignoreRuleBindings: ignoreRuleBindings, } } func (c *RBCache) RefreshRules() { - for _, rb := range c.rbNameToRB.Values() { + for _, rb := range c.ruleBindings() { rbName := uniqueName(&rb) - c.rbNameToRules.Set(rbName, c.createRules(rb.Spec.Rules)) + createdRules := c.createRules(rb.Spec.Rules) + c.mu.Lock() + if c.rbNameToRules == nil { + c.rbNameToRules = make(map[string][]rules.RuleEvaluator) + } + c.rbNameToRules[rbName] = createdRules + c.mu.Unlock() + } +} + +// ruleBindings returns a shallow snapshot so callers can process bindings without holding the lock. +func (c *RBCache) ruleBindings() []typesv1.RuntimeAlertRuleBinding { + c.mu.RLock() + defer c.mu.RUnlock() + bindings := make([]typesv1.RuntimeAlertRuleBinding, 0, len(c.rbNameToRB)) + for _, binding := range c.rbNameToRB { + bindings = append(bindings, binding) } + return bindings } // ----------------- watcher.WatchResources methods ----------------- @@ -77,7 +96,7 @@ func (c *RBCache) ListRulesForObject(ctx context.Context, object *unstructured.U var rulesSlice []rules.RuleEvaluator var rbNames []string - for _, rb := range c.rbNameToRB.Values() { + for _, rb := range c.ruleBindings() { rbName := uniqueName(&rb) // check if the object is cluster object if object.GetNamespace() == "" { @@ -128,11 +147,11 @@ func (c *RBCache) ListRulesForObject(ctx context.Context, object *unstructured.U rbNames = append(rbNames, rbName) } + c.mu.RLock() for _, ruleName := range rbNames { - if c.rbNameToRules.Has(ruleName) { - rulesSlice = append(rulesSlice, c.rbNameToRules.Get(ruleName)...) - } + rulesSlice = append(rulesSlice, c.rbNameToRules[ruleName]...) } + c.mu.RUnlock() return rulesSlice } @@ -201,8 +220,20 @@ func (c *RBCache) addRuleBinding(ruleBinding *typesv1.RuntimeAlertRuleBinding) [ logger.L().Info("RuleBinding added/modified", helpers.String("name", rbName)) // add the rule binding to the cache - c.rbNameToRB.Set(rbName, *ruleBinding) - c.rbNameToRules.Set(rbName, c.createRules(ruleBinding.Spec.Rules)) + c.mu.Lock() + if c.rbNameToRB == nil { + c.rbNameToRB = make(map[string]typesv1.RuntimeAlertRuleBinding) + } + c.rbNameToRB[rbName] = *ruleBinding + c.mu.Unlock() + + createdRules := c.createRules(ruleBinding.Spec.Rules) + c.mu.Lock() + if c.rbNameToRules == nil { + c.rbNameToRules = make(map[string][]rules.RuleEvaluator) + } + c.rbNameToRules[rbName] = createdRules + c.mu.Unlock() return rbs } @@ -211,8 +242,10 @@ func (c *RBCache) deleteRuleBinding(uniqueName string) []rulebindingmanager.Rule var rbs []rulebindingmanager.RuleBindingNotify // remove the rule binding from the cache - c.rbNameToRB.Delete(uniqueName) - c.rbNameToRules.Delete(uniqueName) + c.mu.Lock() + delete(c.rbNameToRB, uniqueName) + delete(c.rbNameToRules, uniqueName) + c.mu.Unlock() logger.L().Info("DeleteRuleBinding", helpers.String("name", uniqueName)) return rbs diff --git a/admission/rulebinding/cache/cache_test.go b/admission/rulebinding/cache/cache_test.go index b1fee7b..42b9346 100644 --- a/admission/rulebinding/cache/cache_test.go +++ b/admission/rulebinding/cache/cache_test.go @@ -2,9 +2,9 @@ package cache import ( "context" + "sync" "testing" - "github.com/goradd/maps" "github.com/kubescape/k8s-interface/k8sinterface" typesv1 "github.com/kubescape/node-agent/pkg/rulebindingmanager/types/v1" "github.com/kubescape/operator/admission/rules" @@ -18,7 +18,6 @@ func NewCacheMock() *RBCache { return &RBCache{ k8sClient: k8sinterface.NewKubernetesApiMock(), ruleCreator: &rules.RuleCreatorMock{}, - rbNameToRules: maps.SafeMap[string, []rules.RuleEvaluator]{}, // rule binding name -> []created rules ignoreRuleBindings: false, } } @@ -45,6 +44,66 @@ func TestNewCache(t *testing.T) { } } +func TestCacheConcurrentAccess(t *testing.T) { + for _, newCache := range []struct { + name string + new func() *RBCache + }{ + {name: "constructor", new: func() *RBCache { + return NewCache(nil, &rules.RuleCreatorMock{}, false) + }}, + {name: "partial literal", new: func() *RBCache { + return &RBCache{ruleCreator: &rules.RuleCreatorMock{}} + }}, + } { + t.Run(newCache.name, func(t *testing.T) { + c := newCache.new() + binding := &typesv1.RuntimeAlertRuleBinding{ + ObjectMeta: metav1.ObjectMeta{Name: "binding", Namespace: "test"}, + Spec: typesv1.RuntimeAlertRuleBindingSpec{ + Rules: []typesv1.RuntimeAlertRuleBindingRule{{RuleID: "R2000"}}, + }, + } + object := &unstructured.Unstructured{} + object.SetNamespace("test") + ctx := t.Context() + start := make(chan struct{}) + var wg sync.WaitGroup + for _, operation := range []func(){ + func() { c.addRuleBinding(binding) }, + func() { c.deleteRuleBinding(uniqueName(binding)) }, + func() { c.ListRulesForObject(ctx, object) }, + c.RefreshRules, + } { + wg.Go(func() { + <-start + for range 64 { + operation() + } + }) + } + close(start) + wg.Wait() + + // Assert final behavior after the concurrent operations have finished. + c.addRuleBinding(binding) + beforeRefresh := c.ListRulesForObject(ctx, object) + if !assert.Len(t, beforeRefresh, 1) { + return + } + assert.Equal(t, "R2000", beforeRefresh[0].ID()) + c.RefreshRules() + afterRefresh := c.ListRulesForObject(ctx, object) + if assert.Len(t, afterRefresh, 1) { + assert.Equal(t, "R2000", afterRefresh[0].ID()) + assert.NotSame(t, beforeRefresh[0], afterRefresh[0], "refresh should recreate the rule") + } + c.deleteRuleBinding(uniqueName(binding)) + assert.Empty(t, c.ListRulesForObject(ctx, object)) + }) + } +} + func TestRuntimeObjAddHandler(t *testing.T) { type rules struct { ruleID string @@ -232,21 +291,21 @@ func TestHandlersIgnoreNonRuleBindingKinds(t *testing.T) { t.Run("AddHandler ignores Rules CRD", func(t *testing.T) { c := NewCacheMock() c.AddHandler(context.Background(), rulesEvent) - assert.Equal(t, 0, c.rbNameToRB.Len(), "no rule binding should be stored") + assert.Len(t, c.rbNameToRB, 0, "no rule binding should be stored") }) t.Run("ModifyHandler ignores Rules CRD", func(t *testing.T) { c := NewCacheMock() c.ModifyHandler(context.Background(), rulesEvent) - assert.Equal(t, 0, c.rbNameToRB.Len()) + assert.Len(t, c.rbNameToRB, 0) }) t.Run("DeleteHandler ignores Rules CRD", func(t *testing.T) { c := NewCacheMock() // Seed a binding so we can detect spurious deletes. - c.rbNameToRB.Set("kubescape/admission-test-rules", typesv1.RuntimeAlertRuleBinding{}) + c.rbNameToRB = map[string]typesv1.RuntimeAlertRuleBinding{"kubescape/admission-test-rules": {}} c.DeleteHandler(context.Background(), rulesEvent) - assert.Equal(t, 1, c.rbNameToRB.Len(), "the seeded binding must not be deleted by a Rules CRD event") + assert.Len(t, c.rbNameToRB, 1, "the seeded binding must not be deleted by a Rules CRD event") }) } @@ -254,7 +313,6 @@ func TestListRulesForObjectIgnoreBindings(t *testing.T) { c := &RBCache{ k8sClient: k8sinterface.NewKubernetesApiMock(), ruleCreator: &rules.RuleCreatorMock{}, - rbNameToRules: maps.SafeMap[string, []rules.RuleEvaluator]{}, ignoreRuleBindings: true, } diff --git a/go.mod b/go.mod index bab1d72..f29fcc6 100644 --- a/go.mod +++ b/go.mod @@ -20,7 +20,6 @@ require ( github.com/go-openapi/runtime v0.28.0 github.com/google/cel-go v0.29.0 github.com/google/uuid v1.6.0 - github.com/goradd/maps v1.3.0 github.com/kubescape/backend v0.0.37 github.com/kubescape/go-logger v0.0.28 github.com/kubescape/k8s-interface v0.0.214 @@ -197,6 +196,7 @@ require ( github.com/google/s2a-go v0.1.9 // indirect github.com/googleapis/enterprise-certificate-proxy v0.3.11 // indirect github.com/googleapis/gax-go/v2 v2.17.0 // indirect + github.com/goradd/maps v1.3.0 // indirect github.com/gorilla/mux v1.8.1 // indirect github.com/gregjones/httpcache v0.0.0-20190611155906-901d90724c79 // indirect github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0 // indirect diff --git a/watcher/podwatcher.go b/watcher/podwatcher.go index de15740..4eaac24 100644 --- a/watcher/podwatcher.go +++ b/watcher/podwatcher.go @@ -114,7 +114,7 @@ func (wh *WatchHandler) handlePodWatcher(ctx context.Context, pod *corev1.Pod, w wh.scanImage(ctx, pod, containerData, workerPool) } - wh.SlugToImageID.Set(containerData.Slug, containerData.ImageID) + wh.SlugToImageID.Store(containerData.Slug, containerData.ImageID) wh.WlidAndImageID.Add(getWlidAndImageID(containerData)) } } else { @@ -127,7 +127,7 @@ func (wh *WatchHandler) handlePodWatcher(ctx context.Context, pod *corev1.Pod, w } // cache the new slug - wh.SlugToImageID.Set(containerData.Slug, containerData.ImageID) + wh.SlugToImageID.Store(containerData.Slug, containerData.ImageID) if wh.WlidAndImageID.Contains(getWlidAndImageID(containerData)) { // wlid+imageID already exists, ignoring event diff --git a/watcher/podwatcher_test.go b/watcher/podwatcher_test.go index 4b65214..8be0b6e 100644 --- a/watcher/podwatcher_test.go +++ b/watcher/podwatcher_test.go @@ -390,10 +390,12 @@ func Test_handlePodWatcher(t *testing.T) { resourcesCreatedWg.Wait() // test slug to image ID map - assert.Equal(t, len(tc.expectedSlugToImageIDMap), wh.SlugToImageID.Len(), "Slug to image ID map doesn’t match") - for k, v := range tc.expectedSlugToImageIDMap { - assert.Equal(t, v, wh.SlugToImageID.Get(k), "Slug '%s' to image ID map doesn’t match", k) - } + actualSlugToImageIDMap := make(map[string]string) + wh.SlugToImageID.Range(func(key, value any) bool { + actualSlugToImageIDMap[key.(string)] = value.(string) + return true + }) + assert.Equal(t, tc.expectedSlugToImageIDMap, actualSlugToImageIDMap, "Slug to image ID map doesn’t match") // test expectedWlidAndImageIDMap assert.Equal(t, len(tc.expectedWlidAndImageIDMap), wh.WlidAndImageID.Cardinality(), "Wlid and image ID map doesn’t match") diff --git a/watcher/sbomwatcher.go b/watcher/sbomwatcher.go index 7154c19..1a9ee4c 100644 --- a/watcher/sbomwatcher.go +++ b/watcher/sbomwatcher.go @@ -103,7 +103,7 @@ func (wh *WatchHandler) SBOMWatch(ctx context.Context, workerPool *ants.PoolWith containerStatuses := slices.Concat(pod.Status.ContainerStatuses, pod.Status.InitContainerStatuses, pod.Status.EphemeralContainerStatuses) for _, containerStatus := range containerStatuses { hash := hashFromImageID(containerStatus.ImageID) - wh.ImageToContainerData.Set(hash, utils.ContainerData{ + wh.ImageToContainerData.Store(hash, utils.ContainerData{ ContainerName: containerStatus.Name, Wlid: wlid, }) @@ -169,7 +169,10 @@ func (wh *WatchHandler) HandleSBOMEvents(eventQueue *CooldownQueue, producedComm } imageID := obj.ObjectMeta.Annotations[helpersv1.ImageIDMetadataKey] - imageContainerData := wh.ImageToContainerData.Get(hashFromImageID(imageID)) + var imageContainerData utils.ContainerData + if cached, ok := wh.ImageToContainerData.Load(hashFromImageID(imageID)); ok { + imageContainerData = cached.(utils.ContainerData) + } containerData := &utils.ContainerData{ ContainerName: imageContainerData.ContainerName, ImageID: imageID, @@ -185,7 +188,10 @@ func (wh *WatchHandler) HandleSBOMEvents(eventQueue *CooldownQueue, producedComm // command with an empty Wlid — kubevuln silently drops those // from the platform submission path. key := obj.ObjectMeta.Namespace + "/" + obj.ObjectMeta.Name - attempt := wh.sbomRetryAttempts.Get(key) + var attempt int + if cached, ok := wh.sbomRetryAttempts.Load(key); ok { + attempt = cached.(int) + } if attempt >= sbomRetryMaxAttempts { wh.sbomRetryAttempts.Delete(key) logger.L().Warning("dropping SBOM scan after exhausting retries waiting for Wlid", @@ -196,7 +202,7 @@ func (wh *WatchHandler) HandleSBOMEvents(eventQueue *CooldownQueue, producedComm errorCh <- err continue } - wh.sbomRetryAttempts.Set(key, attempt+1) + wh.sbomRetryAttempts.Store(key, attempt+1) delay := sbomRetryBackoff(attempt) logger.L().Debug("Wlid not yet known for SBOM, re-enqueueing", helpers.String("name", obj.ObjectMeta.Name), diff --git a/watcher/sbomwatcher_test.go b/watcher/sbomwatcher_test.go index 246b364..99cf07b 100644 --- a/watcher/sbomwatcher_test.go +++ b/watcher/sbomwatcher_test.go @@ -156,7 +156,7 @@ func TestHandleSBOMEvents(t *testing.T) { ctx := context.Background() wh := newTestHandler(t, startingObjects...) if tc.seedContainerData { - wh.ImageToContainerData.Set(testImageHashOnly, utils.ContainerData{ + wh.ImageToContainerData.Store(testImageHashOnly, utils.ContainerData{ ContainerName: testContainerName, Wlid: testWlid, }) @@ -299,14 +299,22 @@ func TestHandleSBOMEvents_WlidArrivesLate(t *testing.T) { cmdCh := make(chan *apis.Command, 4) errorCh := make(chan error, 4) - go wh.HandleSBOMEvents(eventQueue, cmdCh, errorCh) + handlerDone := make(chan struct{}) + go func() { + defer close(handlerDone) + wh.HandleSBOMEvents(eventQueue, cmdCh, errorCh) + }() + t.Cleanup(func() { + eventQueue.Stop() + <-handlerDone + }) // Enqueue the SBOM while ImageToContainerData is empty. eventQueue.Enqueue(watch.Event{Type: watch.Added, Object: sbom}) // After a couple of retry cycles, simulate the pod informer populating the map. time.AfterFunc(350*time.Millisecond, func() { - wh.ImageToContainerData.Set(testImageHashOnly, utils.ContainerData{ + wh.ImageToContainerData.Store(testImageHashOnly, utils.ContainerData{ ContainerName: testContainerName, Wlid: testWlid, }) @@ -332,9 +340,8 @@ func TestHandleSBOMEvents_WlidArrivesLate(t *testing.T) { // Bookkeeping should be cleared after success. key := sbom.Namespace + "/" + sbom.Name - assert.Equal(t, 0, wh.sbomRetryAttempts.Get(key), "retry counter must be cleared on success") - - eventQueue.Stop() + _, exists := wh.sbomRetryAttempts.Load(key) + assert.False(t, exists, "retry counter must be cleared on success") } // TestHandleSBOMEvents_WlidNeverArrives_ExhaustsRetries verifies the @@ -381,7 +388,17 @@ func TestHandleSBOMEvents_WlidNeverArrives_ExhaustsRetries(t *testing.T) { } }() - go wh.HandleSBOMEvents(eventQueue, cmdCh, errorCh) + handlerDone := make(chan struct{}) + go func() { + defer close(handlerDone) + wh.HandleSBOMEvents(eventQueue, cmdCh, errorCh) + }() + t.Cleanup(func() { + eventQueue.Stop() + <-handlerDone + close(cmdCh) + <-cmdDone + }) eventQueue.Enqueue(watch.Event{Type: watch.Added, Object: sbom}) @@ -400,11 +417,6 @@ func TestHandleSBOMEvents_WlidNeverArrives_ExhaustsRetries(t *testing.T) { // Bookkeeping must be cleared on exhaustion to avoid leaking memory if the // SBOM is later re-observed. key := sbom.Namespace + "/" + sbom.Name - assert.Equal(t, 0, wh.sbomRetryAttempts.Get(key), "retry counter must be cleared on exhaustion") - - eventQueue.Stop() - // HandleSBOMEvents closes cmdCh implicitly? No - it only closes errorCh. - // Close cmdCh manually so the drain goroutine exits, then wait. - close(cmdCh) - <-cmdDone + _, exists := wh.sbomRetryAttempts.Load(key) + assert.False(t, exists, "retry counter must be cleared on exhaustion") } diff --git a/watcher/watchhandler.go b/watcher/watchhandler.go index 5ec9fcd..3cf76ae 100644 --- a/watcher/watchhandler.go +++ b/watcher/watchhandler.go @@ -3,13 +3,12 @@ package watcher import ( "errors" "fmt" + "sync" "time" mapset "github.com/deckarep/golang-set/v2" - "github.com/goradd/maps" "github.com/kubescape/k8s-interface/k8sinterface" "github.com/kubescape/operator/config" - "github.com/kubescape/operator/utils" kssc "github.com/kubescape/storage/pkg/generated/clientset/versioned" ) @@ -27,10 +26,10 @@ var ( ) type WatchHandler struct { - ImageToContainerData maps.SafeMap[string, utils.ContainerData] // map of : - SlugToImageID maps.SafeMap[string, string] // map of : string - WlidAndImageID mapset.Set[string] // set of - sbomRetryAttempts maps.SafeMap[string, int] // map of : retry attempts so far + ImageToContainerData sync.Map // string image hash -> utils.ContainerData + SlugToImageID sync.Map // string slug -> string image ID + WlidAndImageID mapset.Set[string] // set of + sbomRetryAttempts sync.Map // string SBOM key -> int retry attempts so far storageClient kssc.Interface cfg config.IConfig k8sAPI *k8sinterface.KubernetesApi From ba266b9b149bdbf3ebaba453225764c9df8501be Mon Sep 17 00:00:00 2001 From: Matthias Bertschy Date: Tue, 15 Sep 2026 13:23:20 +0200 Subject: [PATCH 2/2] fix: retain SafeMap and pin upstream concurrency fix Signed-off-by: Matthias Bertschy --- admission/rulebinding/cache/cache.go | 61 ++++++----------------- admission/rulebinding/cache/cache_test.go | 11 ++-- go.mod | 6 ++- go.sum | 6 +-- watcher/podwatcher.go | 4 +- watcher/podwatcher_test.go | 10 ++-- watcher/sbomwatcher.go | 14 ++---- watcher/sbomwatcher_test.go | 4 +- watcher/watchhandler.go | 11 ++-- 9 files changed, 46 insertions(+), 81 deletions(-) diff --git a/admission/rulebinding/cache/cache.go b/admission/rulebinding/cache/cache.go index 980d3aa..9f2cafe 100644 --- a/admission/rulebinding/cache/cache.go +++ b/admission/rulebinding/cache/cache.go @@ -2,8 +2,8 @@ package cache import ( "context" - "sync" + "github.com/goradd/maps" "github.com/kubescape/go-logger" "github.com/kubescape/go-logger/helpers" "github.com/kubescape/node-agent/pkg/k8sclient" @@ -27,9 +27,8 @@ var _ watcher.Adaptor = (*RBCache)(nil) type RBCache struct { k8sClient k8sclient.K8sClientInterface - mu sync.RWMutex // protects rbNameToRB and rbNameToRules - rbNameToRB map[string]typesv1.RuntimeAlertRuleBinding // rule binding name -> rule binding - rbNameToRules map[string][]rules.RuleEvaluator // rule binding name -> []created rules + rbNameToRB maps.SafeMap[string, typesv1.RuntimeAlertRuleBinding] // rule binding name -> rule binding + rbNameToRules maps.SafeMap[string, []rules.RuleEvaluator] // rule binding name -> []created rules ruleCreator rules.RuleCreator watchResources []watcher.WatchResource notifiers []*chan rulebindingmanager.RuleBindingNotify @@ -41,35 +40,17 @@ func NewCache(k8sClient k8sclient.K8sClientInterface, ruleCreator rules.RuleCrea return &RBCache{ k8sClient: k8sClient, ruleCreator: ruleCreator, - rbNameToRB: make(map[string]typesv1.RuntimeAlertRuleBinding), - rbNameToRules: make(map[string][]rules.RuleEvaluator), + rbNameToRB: maps.SafeMap[string, typesv1.RuntimeAlertRuleBinding]{}, watchResources: resourcesToWatch(), ignoreRuleBindings: ignoreRuleBindings, } } func (c *RBCache) RefreshRules() { - for _, rb := range c.ruleBindings() { + for _, rb := range c.rbNameToRB.Values() { rbName := uniqueName(&rb) - createdRules := c.createRules(rb.Spec.Rules) - c.mu.Lock() - if c.rbNameToRules == nil { - c.rbNameToRules = make(map[string][]rules.RuleEvaluator) - } - c.rbNameToRules[rbName] = createdRules - c.mu.Unlock() - } -} - -// ruleBindings returns a shallow snapshot so callers can process bindings without holding the lock. -func (c *RBCache) ruleBindings() []typesv1.RuntimeAlertRuleBinding { - c.mu.RLock() - defer c.mu.RUnlock() - bindings := make([]typesv1.RuntimeAlertRuleBinding, 0, len(c.rbNameToRB)) - for _, binding := range c.rbNameToRB { - bindings = append(bindings, binding) + c.rbNameToRules.Set(rbName, c.createRules(rb.Spec.Rules)) } - return bindings } // ----------------- watcher.WatchResources methods ----------------- @@ -96,7 +77,7 @@ func (c *RBCache) ListRulesForObject(ctx context.Context, object *unstructured.U var rulesSlice []rules.RuleEvaluator var rbNames []string - for _, rb := range c.ruleBindings() { + for _, rb := range c.rbNameToRB.Values() { rbName := uniqueName(&rb) // check if the object is cluster object if object.GetNamespace() == "" { @@ -147,11 +128,11 @@ func (c *RBCache) ListRulesForObject(ctx context.Context, object *unstructured.U rbNames = append(rbNames, rbName) } - c.mu.RLock() for _, ruleName := range rbNames { - rulesSlice = append(rulesSlice, c.rbNameToRules[ruleName]...) + if c.rbNameToRules.Has(ruleName) { + rulesSlice = append(rulesSlice, c.rbNameToRules.Get(ruleName)...) + } } - c.mu.RUnlock() return rulesSlice } @@ -220,20 +201,8 @@ func (c *RBCache) addRuleBinding(ruleBinding *typesv1.RuntimeAlertRuleBinding) [ logger.L().Info("RuleBinding added/modified", helpers.String("name", rbName)) // add the rule binding to the cache - c.mu.Lock() - if c.rbNameToRB == nil { - c.rbNameToRB = make(map[string]typesv1.RuntimeAlertRuleBinding) - } - c.rbNameToRB[rbName] = *ruleBinding - c.mu.Unlock() - - createdRules := c.createRules(ruleBinding.Spec.Rules) - c.mu.Lock() - if c.rbNameToRules == nil { - c.rbNameToRules = make(map[string][]rules.RuleEvaluator) - } - c.rbNameToRules[rbName] = createdRules - c.mu.Unlock() + c.rbNameToRB.Set(rbName, *ruleBinding) + c.rbNameToRules.Set(rbName, c.createRules(ruleBinding.Spec.Rules)) return rbs } @@ -242,10 +211,8 @@ func (c *RBCache) deleteRuleBinding(uniqueName string) []rulebindingmanager.Rule var rbs []rulebindingmanager.RuleBindingNotify // remove the rule binding from the cache - c.mu.Lock() - delete(c.rbNameToRB, uniqueName) - delete(c.rbNameToRules, uniqueName) - c.mu.Unlock() + c.rbNameToRB.Delete(uniqueName) + c.rbNameToRules.Delete(uniqueName) logger.L().Info("DeleteRuleBinding", helpers.String("name", uniqueName)) return rbs diff --git a/admission/rulebinding/cache/cache_test.go b/admission/rulebinding/cache/cache_test.go index 42b9346..944659b 100644 --- a/admission/rulebinding/cache/cache_test.go +++ b/admission/rulebinding/cache/cache_test.go @@ -5,6 +5,7 @@ import ( "sync" "testing" + "github.com/goradd/maps" "github.com/kubescape/k8s-interface/k8sinterface" typesv1 "github.com/kubescape/node-agent/pkg/rulebindingmanager/types/v1" "github.com/kubescape/operator/admission/rules" @@ -18,6 +19,7 @@ func NewCacheMock() *RBCache { return &RBCache{ k8sClient: k8sinterface.NewKubernetesApiMock(), ruleCreator: &rules.RuleCreatorMock{}, + rbNameToRules: maps.SafeMap[string, []rules.RuleEvaluator]{}, // rule binding name -> []created rules ignoreRuleBindings: false, } } @@ -291,21 +293,21 @@ func TestHandlersIgnoreNonRuleBindingKinds(t *testing.T) { t.Run("AddHandler ignores Rules CRD", func(t *testing.T) { c := NewCacheMock() c.AddHandler(context.Background(), rulesEvent) - assert.Len(t, c.rbNameToRB, 0, "no rule binding should be stored") + assert.Equal(t, 0, c.rbNameToRB.Len(), "no rule binding should be stored") }) t.Run("ModifyHandler ignores Rules CRD", func(t *testing.T) { c := NewCacheMock() c.ModifyHandler(context.Background(), rulesEvent) - assert.Len(t, c.rbNameToRB, 0) + assert.Equal(t, 0, c.rbNameToRB.Len()) }) t.Run("DeleteHandler ignores Rules CRD", func(t *testing.T) { c := NewCacheMock() // Seed a binding so we can detect spurious deletes. - c.rbNameToRB = map[string]typesv1.RuntimeAlertRuleBinding{"kubescape/admission-test-rules": {}} + c.rbNameToRB.Set("kubescape/admission-test-rules", typesv1.RuntimeAlertRuleBinding{}) c.DeleteHandler(context.Background(), rulesEvent) - assert.Len(t, c.rbNameToRB, 1, "the seeded binding must not be deleted by a Rules CRD event") + assert.Equal(t, 1, c.rbNameToRB.Len(), "the seeded binding must not be deleted by a Rules CRD event") }) } @@ -313,6 +315,7 @@ func TestListRulesForObjectIgnoreBindings(t *testing.T) { c := &RBCache{ k8sClient: k8sinterface.NewKubernetesApiMock(), ruleCreator: &rules.RuleCreatorMock{}, + rbNameToRules: maps.SafeMap[string, []rules.RuleEvaluator]{}, ignoreRuleBindings: true, } diff --git a/go.mod b/go.mod index f29fcc6..78edaa6 100644 --- a/go.mod +++ b/go.mod @@ -20,6 +20,7 @@ require ( github.com/go-openapi/runtime v0.28.0 github.com/google/cel-go v0.29.0 github.com/google/uuid v1.6.0 + github.com/goradd/maps v1.3.0 github.com/kubescape/backend v0.0.37 github.com/kubescape/go-logger v0.0.28 github.com/kubescape/k8s-interface v0.0.214 @@ -196,7 +197,6 @@ require ( github.com/google/s2a-go v0.1.9 // indirect github.com/googleapis/enterprise-certificate-proxy v0.3.11 // indirect github.com/googleapis/gax-go/v2 v2.17.0 // indirect - github.com/goradd/maps v1.3.0 // indirect github.com/gorilla/mux v1.8.1 // indirect github.com/gregjones/httpcache v0.0.0-20190611155906-901d90724c79 // indirect github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0 // indirect @@ -361,3 +361,7 @@ replace github.com/project-copacetic/copacetic => github.com/anubhav06/copacetic // runtime-spec v1.3.0 changed LinuxPids.Limit from int64 to *int64, which breaks // containerd v1.7.32 under Go 1.25.8+. replace github.com/opencontainers/runtime-spec => github.com/opencontainers/runtime-spec v1.2.1 + +// Temporary concurrency fix: https://github.com/goradd/maps/pull/22 +// Remove this replacement when an upstream release includes the fix. +replace github.com/goradd/maps => github.com/matthyx/maps v0.0.0-20260915111345-e9181ad40421 diff --git a/go.sum b/go.sum index 57bcec8..c37f972 100644 --- a/go.sum +++ b/go.sum @@ -145,8 +145,6 @@ github.com/armon/go-metrics v0.0.0-20180917152333-f0300d1749da/go.mod h1:Q73ZrmV github.com/armon/go-metrics v0.3.10/go.mod h1:4O98XIr/9W0sxpJ8UaYkvjk10Iff7SnFrb4QAOwNTFc= github.com/armon/go-radix v0.0.0-20180808171621-7fddfc383310/go.mod h1:ufUuZ+zHj4x4TnLV4JWEpy2hxWSpsRywHrMgIH9cCH8= github.com/armon/go-radix v1.0.0/go.mod h1:ufUuZ+zHj4x4TnLV4JWEpy2hxWSpsRywHrMgIH9cCH8= -github.com/armosec/armoapi-go v0.0.720 h1:mtxUw2wWPRSQWcUf89Eoc9J81SBIC0YaK66XqAXuhCQ= -github.com/armosec/armoapi-go v0.0.720/go.mod h1:9jAH0g8ZsryhiBDd/aNMX4+n10bGwTx/doWCyyjSxts= github.com/armosec/armoapi-go v0.0.761 h1:/idEh/lGFLGUIF64/ecuusCYHPzs7F6T/VQ9zFHEaxA= github.com/armosec/armoapi-go v0.0.761/go.mod h1:1l+70fBK09F7zI2jArrPUWVHaLkijg+sQutFTmE6HRs= github.com/armosec/gojay v1.2.17 h1:VSkLBQzD1c2V+FMtlGFKqWXNsdNvIKygTKJI9ysY8eM= @@ -587,8 +585,6 @@ github.com/gookit/color v1.2.5/go.mod h1:AhIE+pS6D4Ql0SQWbBeXPHw7gY0/sjHoA4s/n1K github.com/gookit/color v1.6.0 h1:JjJXBTk1ETNyqyilJhkTXJYYigHG24TM9Xa2M1xAhRA= github.com/gookit/color v1.6.0/go.mod h1:9ACFc7/1IpHGBW8RwuDm/0YEnhg3dwwXpoMsmtyHfjs= github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1/go.mod h1:wJfORRmW1u3UXTncJ5qlYoELFm8eSnnEO6hX4iZ3EWY= -github.com/goradd/maps v1.3.0 h1:toF7ALsgbjQBmmmRSACTAEO+9g2rApW8dU1WirFQyrE= -github.com/goradd/maps v1.3.0/go.mod h1:O3i5k17BAjHa9h5dzGWWfRJizF03umiBDZsNSqFdbVA= github.com/gorilla/mux v1.8.1 h1:TuBL49tXwgrFYWhqrNgrUNEY92u81SPhu7sTdzQEiWY= github.com/gorilla/mux v1.8.1/go.mod h1:AKf9I4AEqPTmMytcMc0KkNouC66V3BtZ4qD5fmWSiMQ= github.com/gorilla/securecookie v1.1.1/go.mod h1:ra0sb63/xPlUeL+yeDciTfxMRAA+MP+HVt/4epWDjd4= @@ -740,6 +736,8 @@ github.com/mailru/easyjson v0.9.0 h1:PrnmzHw7262yW8sTBwxi1PdJA3Iw/EKBa8psRf7d9a4 github.com/mailru/easyjson v0.9.0/go.mod h1:1+xMtQp2MRNVL/V1bOzuP3aP8VNwRW55fQUto+XFtTU= github.com/matthyx/inspektor-gadget v0.0.0-20260203101533-6ef87216d3dd h1:n8zR1L5t5UWzmQ/DgQ98DF/NrYJL7gUI57GkiDlyu9Y= github.com/matthyx/inspektor-gadget v0.0.0-20260203101533-6ef87216d3dd/go.mod h1:V4TgEmWo37K72pQvC7XuRQssysrxIIkrNX4TtEkgiE0= +github.com/matthyx/maps v0.0.0-20260915111345-e9181ad40421 h1:9XVp2iZdJ7AVWvlNb7e9AOgGzXUyZmX79GzNJDNOCLM= +github.com/matthyx/maps v0.0.0-20260915111345-e9181ad40421/go.mod h1:O3i5k17BAjHa9h5dzGWWfRJizF03umiBDZsNSqFdbVA= github.com/mattn/go-colorable v0.0.9/go.mod h1:9vuHe8Xs5qXnSaW/c/ABM9alt+Vo+STaOChaDxuIBZU= github.com/mattn/go-colorable v0.1.4/go.mod h1:U0ppj6V5qS13XJ6of8GYAs25YV2eR4EVcfRqFIhoBtE= github.com/mattn/go-colorable v0.1.6/go.mod h1:u6P/XSegPjTcexA+o6vUJrdnUu04hMope9wVRipJSqc= diff --git a/watcher/podwatcher.go b/watcher/podwatcher.go index 4eaac24..de15740 100644 --- a/watcher/podwatcher.go +++ b/watcher/podwatcher.go @@ -114,7 +114,7 @@ func (wh *WatchHandler) handlePodWatcher(ctx context.Context, pod *corev1.Pod, w wh.scanImage(ctx, pod, containerData, workerPool) } - wh.SlugToImageID.Store(containerData.Slug, containerData.ImageID) + wh.SlugToImageID.Set(containerData.Slug, containerData.ImageID) wh.WlidAndImageID.Add(getWlidAndImageID(containerData)) } } else { @@ -127,7 +127,7 @@ func (wh *WatchHandler) handlePodWatcher(ctx context.Context, pod *corev1.Pod, w } // cache the new slug - wh.SlugToImageID.Store(containerData.Slug, containerData.ImageID) + wh.SlugToImageID.Set(containerData.Slug, containerData.ImageID) if wh.WlidAndImageID.Contains(getWlidAndImageID(containerData)) { // wlid+imageID already exists, ignoring event diff --git a/watcher/podwatcher_test.go b/watcher/podwatcher_test.go index 8be0b6e..4b65214 100644 --- a/watcher/podwatcher_test.go +++ b/watcher/podwatcher_test.go @@ -390,12 +390,10 @@ func Test_handlePodWatcher(t *testing.T) { resourcesCreatedWg.Wait() // test slug to image ID map - actualSlugToImageIDMap := make(map[string]string) - wh.SlugToImageID.Range(func(key, value any) bool { - actualSlugToImageIDMap[key.(string)] = value.(string) - return true - }) - assert.Equal(t, tc.expectedSlugToImageIDMap, actualSlugToImageIDMap, "Slug to image ID map doesn’t match") + assert.Equal(t, len(tc.expectedSlugToImageIDMap), wh.SlugToImageID.Len(), "Slug to image ID map doesn’t match") + for k, v := range tc.expectedSlugToImageIDMap { + assert.Equal(t, v, wh.SlugToImageID.Get(k), "Slug '%s' to image ID map doesn’t match", k) + } // test expectedWlidAndImageIDMap assert.Equal(t, len(tc.expectedWlidAndImageIDMap), wh.WlidAndImageID.Cardinality(), "Wlid and image ID map doesn’t match") diff --git a/watcher/sbomwatcher.go b/watcher/sbomwatcher.go index 1a9ee4c..7154c19 100644 --- a/watcher/sbomwatcher.go +++ b/watcher/sbomwatcher.go @@ -103,7 +103,7 @@ func (wh *WatchHandler) SBOMWatch(ctx context.Context, workerPool *ants.PoolWith containerStatuses := slices.Concat(pod.Status.ContainerStatuses, pod.Status.InitContainerStatuses, pod.Status.EphemeralContainerStatuses) for _, containerStatus := range containerStatuses { hash := hashFromImageID(containerStatus.ImageID) - wh.ImageToContainerData.Store(hash, utils.ContainerData{ + wh.ImageToContainerData.Set(hash, utils.ContainerData{ ContainerName: containerStatus.Name, Wlid: wlid, }) @@ -169,10 +169,7 @@ func (wh *WatchHandler) HandleSBOMEvents(eventQueue *CooldownQueue, producedComm } imageID := obj.ObjectMeta.Annotations[helpersv1.ImageIDMetadataKey] - var imageContainerData utils.ContainerData - if cached, ok := wh.ImageToContainerData.Load(hashFromImageID(imageID)); ok { - imageContainerData = cached.(utils.ContainerData) - } + imageContainerData := wh.ImageToContainerData.Get(hashFromImageID(imageID)) containerData := &utils.ContainerData{ ContainerName: imageContainerData.ContainerName, ImageID: imageID, @@ -188,10 +185,7 @@ func (wh *WatchHandler) HandleSBOMEvents(eventQueue *CooldownQueue, producedComm // command with an empty Wlid — kubevuln silently drops those // from the platform submission path. key := obj.ObjectMeta.Namespace + "/" + obj.ObjectMeta.Name - var attempt int - if cached, ok := wh.sbomRetryAttempts.Load(key); ok { - attempt = cached.(int) - } + attempt := wh.sbomRetryAttempts.Get(key) if attempt >= sbomRetryMaxAttempts { wh.sbomRetryAttempts.Delete(key) logger.L().Warning("dropping SBOM scan after exhausting retries waiting for Wlid", @@ -202,7 +196,7 @@ func (wh *WatchHandler) HandleSBOMEvents(eventQueue *CooldownQueue, producedComm errorCh <- err continue } - wh.sbomRetryAttempts.Store(key, attempt+1) + wh.sbomRetryAttempts.Set(key, attempt+1) delay := sbomRetryBackoff(attempt) logger.L().Debug("Wlid not yet known for SBOM, re-enqueueing", helpers.String("name", obj.ObjectMeta.Name), diff --git a/watcher/sbomwatcher_test.go b/watcher/sbomwatcher_test.go index 99cf07b..848d9e2 100644 --- a/watcher/sbomwatcher_test.go +++ b/watcher/sbomwatcher_test.go @@ -156,7 +156,7 @@ func TestHandleSBOMEvents(t *testing.T) { ctx := context.Background() wh := newTestHandler(t, startingObjects...) if tc.seedContainerData { - wh.ImageToContainerData.Store(testImageHashOnly, utils.ContainerData{ + wh.ImageToContainerData.Set(testImageHashOnly, utils.ContainerData{ ContainerName: testContainerName, Wlid: testWlid, }) @@ -314,7 +314,7 @@ func TestHandleSBOMEvents_WlidArrivesLate(t *testing.T) { // After a couple of retry cycles, simulate the pod informer populating the map. time.AfterFunc(350*time.Millisecond, func() { - wh.ImageToContainerData.Store(testImageHashOnly, utils.ContainerData{ + wh.ImageToContainerData.Set(testImageHashOnly, utils.ContainerData{ ContainerName: testContainerName, Wlid: testWlid, }) diff --git a/watcher/watchhandler.go b/watcher/watchhandler.go index 3cf76ae..5ec9fcd 100644 --- a/watcher/watchhandler.go +++ b/watcher/watchhandler.go @@ -3,12 +3,13 @@ package watcher import ( "errors" "fmt" - "sync" "time" mapset "github.com/deckarep/golang-set/v2" + "github.com/goradd/maps" "github.com/kubescape/k8s-interface/k8sinterface" "github.com/kubescape/operator/config" + "github.com/kubescape/operator/utils" kssc "github.com/kubescape/storage/pkg/generated/clientset/versioned" ) @@ -26,10 +27,10 @@ var ( ) type WatchHandler struct { - ImageToContainerData sync.Map // string image hash -> utils.ContainerData - SlugToImageID sync.Map // string slug -> string image ID - WlidAndImageID mapset.Set[string] // set of - sbomRetryAttempts sync.Map // string SBOM key -> int retry attempts so far + ImageToContainerData maps.SafeMap[string, utils.ContainerData] // map of : + SlugToImageID maps.SafeMap[string, string] // map of : string + WlidAndImageID mapset.Set[string] // set of + sbomRetryAttempts maps.SafeMap[string, int] // map of : retry attempts so far storageClient kssc.Interface cfg config.IConfig k8sAPI *k8sinterface.KubernetesApi