From c6810337e578cfcc2d121db04e856752de9e0fcf Mon Sep 17 00:00:00 2001 From: Matthias Bertschy Date: Fri, 2 Oct 2026 14:34:47 +0200 Subject: [PATCH 1/5] feat(cel): add dstNamespace and dstPodLabels with Service peer hook --- pkg/utils/cel.go | 55 ++++++++++++++++++++++++ pkg/utils/cel_dst_peer_test.go | 78 ++++++++++++++++++++++++++++++++++ 2 files changed, 133 insertions(+) create mode 100644 pkg/utils/cel_dst_peer_test.go diff --git a/pkg/utils/cel.go b/pkg/utils/cel.go index 39c3ad40f2..38ec187f4e 100644 --- a/pkg/utils/cel.go +++ b/pkg/utils/cel.go @@ -5,9 +5,11 @@ import ( "errors" "io" "net/http" + "sync" celtypes "github.com/google/cel-go/common/types" "github.com/google/cel-go/common/types/ref" + igtypes "github.com/inspektor-gadget/inspektor-gadget/pkg/types" "github.com/picatz/xcel" ) @@ -202,6 +204,39 @@ var CelFields = map[string]*celtypes.FieldType{ return celtypes.Int(x.Raw.GetDstPort()), nil }), }, + // dstNamespace / dstPodLabels carry the peer identity that IG's + // kubeipresolver resolves cluster-wide (independent of node-agent's + // node-local pod cache), so selector rules can match a peer on any node. + "dstNamespace": { + Type: celtypes.StringType, + IsSet: isSet, + GetFrom: ref.FieldGetter(func(target any) (any, error) { + x := target.(*xcel.Object[CelEvent]) + if x.Raw == nil { + return nil, errCelObjectNil + } + return celtypes.String(x.Raw.GetDstEndpoint().Namespace), nil + }), + }, + "dstPodLabels": { + Type: celtypes.MapType, + IsSet: isSet, + GetFrom: ref.FieldGetter(func(target any) (any, error) { + x := target.(*xcel.Object[CelEvent]) + if x.Raw == nil { + return nil, errCelObjectNil + } + ep := x.Raw.GetDstEndpoint() + pl := ep.PodLabels + if len(pl) == 0 && ep.Kind == igtypes.EndpointKindService { + pl = ServicePeerLabels(ep.Namespace, ep.Name) + } + if pl == nil { + pl = map[string]string{} + } + return pl, nil + }), + }, "exepath": { Type: celtypes.StringType, IsSet: isSet, @@ -613,3 +648,23 @@ var HttpRequestFields = map[string]*celtypes.FieldType{ }), }, } + +var ( + servicePeerLabelsMu sync.RWMutex + servicePeerLabels func(namespace, name string) map[string]string +) + +func SetServicePeerLabels(f func(namespace, name string) map[string]string) { + servicePeerLabelsMu.Lock() + defer servicePeerLabelsMu.Unlock() + servicePeerLabels = f +} + +func ServicePeerLabels(namespace, name string) map[string]string { + servicePeerLabelsMu.RLock() + defer servicePeerLabelsMu.RUnlock() + if servicePeerLabels == nil { + return nil + } + return servicePeerLabels(namespace, name) +} diff --git a/pkg/utils/cel_dst_peer_test.go b/pkg/utils/cel_dst_peer_test.go new file mode 100644 index 0000000000..a598e3e847 --- /dev/null +++ b/pkg/utils/cel_dst_peer_test.go @@ -0,0 +1,78 @@ +package utils + +import ( + "testing" + + celtypes "github.com/google/cel-go/common/types" + igtypes "github.com/inspektor-gadget/inspektor-gadget/pkg/types" + "github.com/picatz/xcel" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +type mockNetworkCelEvent struct { + CelEventImpl + endpoint igtypes.L4Endpoint +} + +func (m *mockNetworkCelEvent) GetDstEndpoint() igtypes.L4Endpoint { + return m.endpoint +} + +func TestCelFields_DstNamespaceAndDstPodLabels(t *testing.T) { + dstNsField, ok := CelFields["dstNamespace"] + require.True(t, ok) + dstLabelsField, ok := CelFields["dstPodLabels"] + require.True(t, ok) + + // 1. Pod peer with labels + event := &mockNetworkCelEvent{ + endpoint: igtypes.L4Endpoint{ + L3Endpoint: igtypes.L3Endpoint{ + Namespace: "prod", + Name: "api-pod", + Kind: igtypes.EndpointKindPod, + PodLabels: map[string]string{"app": "api", "tier": "backend"}, + }, + }, + } + wrapped, _ := xcel.NewObject[CelEvent](event) + + val, err := dstNsField.GetFrom(wrapped) + require.NoError(t, err) + assert.Equal(t, celtypes.String("prod"), val) + + val, err = dstLabelsField.GetFrom(wrapped) + require.NoError(t, err) + assert.Equal(t, map[string]string{"app": "api", "tier": "backend"}, val) + + // 2. Service peer without pod labels falls back to ServicePeerLabels hook + SetServicePeerLabels(func(ns, name string) map[string]string { + if ns == "prod" && name == "api-service" { + return map[string]string{"app": "api-resolved"} + } + return nil + }) + defer SetServicePeerLabels(nil) + + svcEvent := &mockNetworkCelEvent{ + endpoint: igtypes.L4Endpoint{ + L3Endpoint: igtypes.L3Endpoint{ + Namespace: "prod", + Name: "api-service", + Kind: igtypes.EndpointKindService, + }, + }, + } + svcWrapped, _ := xcel.NewObject[CelEvent](svcEvent) + + val, err = dstLabelsField.GetFrom(svcWrapped) + require.NoError(t, err) + assert.Equal(t, map[string]string{"app": "api-resolved"}, val) + + // 3. Fallback with no resolver yields empty map, not nil + SetServicePeerLabels(nil) + val, err = dstLabelsField.GetFrom(svcWrapped) + require.NoError(t, err) + assert.Equal(t, map[string]string{}, val) +} From c3da4da12bc0fab505de78a5b5503ac6c8639fdb Mon Sep 17 00:00:00 2001 From: Matthias Bertschy Date: Sat, 3 Oct 2026 13:21:34 +0200 Subject: [PATCH 2/5] fix(cel): resolve Service endpoint labels through hook regardless of metadata labels (5400495243) --- pkg/utils/cel.go | 6 ++++-- pkg/utils/cel_dst_peer_test.go | 3 ++- 2 files changed, 6 insertions(+), 3 deletions(-) diff --git a/pkg/utils/cel.go b/pkg/utils/cel.go index 38ec187f4e..df63a3e067 100644 --- a/pkg/utils/cel.go +++ b/pkg/utils/cel.go @@ -227,9 +227,11 @@ var CelFields = map[string]*celtypes.FieldType{ return nil, errCelObjectNil } ep := x.Raw.GetDstEndpoint() - pl := ep.PodLabels - if len(pl) == 0 && ep.Kind == igtypes.EndpointKindService { + var pl map[string]string + if ep.Kind == igtypes.EndpointKindService { pl = ServicePeerLabels(ep.Namespace, ep.Name) + } else { + pl = ep.PodLabels } if pl == nil { pl = map[string]string{} diff --git a/pkg/utils/cel_dst_peer_test.go b/pkg/utils/cel_dst_peer_test.go index a598e3e847..09242a57c8 100644 --- a/pkg/utils/cel_dst_peer_test.go +++ b/pkg/utils/cel_dst_peer_test.go @@ -46,7 +46,7 @@ func TestCelFields_DstNamespaceAndDstPodLabels(t *testing.T) { require.NoError(t, err) assert.Equal(t, map[string]string{"app": "api", "tier": "backend"}, val) - // 2. Service peer without pod labels falls back to ServicePeerLabels hook + // 2. Service peer with metadata labels resolves through ServicePeerLabels hook SetServicePeerLabels(func(ns, name string) map[string]string { if ns == "prod" && name == "api-service" { return map[string]string{"app": "api-resolved"} @@ -61,6 +61,7 @@ func TestCelFields_DstNamespaceAndDstPodLabels(t *testing.T) { Namespace: "prod", Name: "api-service", Kind: igtypes.EndpointKindService, + PodLabels: map[string]string{"app.kubernetes.io/name": "service-metadata-only"}, }, }, } From 289aab5937ea136768dd6df74e8cc3304bce3afb Mon Sep 17 00:00:00 2001 From: Matthias Bertschy Date: Sat, 3 Oct 2026 13:29:30 +0200 Subject: [PATCH 3/5] feat(rulemanager): register Service peer label resolver backed by k8sClient (5400517430) --- cmd/main.go | 1 + pkg/rulemanager/rule_manager.go | 1 + pkg/rulemanager/service_resolver.go | 25 +++++ pkg/rulemanager/service_resolver_test.go | 127 +++++++++++++++++++++++ 4 files changed, 154 insertions(+) create mode 100644 pkg/rulemanager/service_resolver.go create mode 100644 pkg/rulemanager/service_resolver_test.go diff --git a/cmd/main.go b/cmd/main.go index d936a36611..4b5f372c22 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -208,6 +208,7 @@ func main() { // Create clients logger.L().Info("Kubernetes mode is true") k8sClient := k8sinterface.NewKubernetesApi() + rulemanager.InitServicePeerLabelResolver(k8sClient) // Fetch cluster UID from kube-system namespace clusterUID := utils.GetClusterUID(k8sClient.GetKubernetesClient()) diff --git a/pkg/rulemanager/rule_manager.go b/pkg/rulemanager/rule_manager.go index cd6a1511a5..1e00d7fe02 100644 --- a/pkg/rulemanager/rule_manager.go +++ b/pkg/rulemanager/rule_manager.go @@ -92,6 +92,7 @@ func CreateRuleManager( ) (*RuleManager, error) { rulePolicyValidator := NewRulePolicyValidator(objectCache) detectorManager := detectors.NewDetectorManager(mntnsRegistry) + InitServicePeerLabelResolver(k8sClient) r := &RuleManager{ cfg: cfg, diff --git a/pkg/rulemanager/service_resolver.go b/pkg/rulemanager/service_resolver.go new file mode 100644 index 0000000000..ef12ed340d --- /dev/null +++ b/pkg/rulemanager/service_resolver.go @@ -0,0 +1,25 @@ +package rulemanager + +import ( + "github.com/kubescape/node-agent/pkg/k8sclient" + "github.com/kubescape/node-agent/pkg/utils" +) + +// InitServicePeerLabelResolver registers a ServicePeerLabels hook backed by the Kubernetes client +// so that Service destination endpoints in CEL rules resolve to their backend selector labels. +func InitServicePeerLabelResolver(k8sClient k8sclient.K8sClientInterface) { + if k8sClient == nil { + utils.SetServicePeerLabels(nil) + return + } + utils.SetServicePeerLabels(func(namespace, name string) map[string]string { + svc, err := k8sClient.GetWorkload(namespace, "Service", name) + if err != nil || svc == nil { + return nil + } + if svc.GetName() == "kubernetes" && svc.GetNamespace() == "default" { + return svc.GetLabels() + } + return svc.GetServiceSelector() + }) +} diff --git a/pkg/rulemanager/service_resolver_test.go b/pkg/rulemanager/service_resolver_test.go new file mode 100644 index 0000000000..dc052657db --- /dev/null +++ b/pkg/rulemanager/service_resolver_test.go @@ -0,0 +1,127 @@ +package rulemanager + +import ( + "errors" + "testing" + + igtypes "github.com/inspektor-gadget/inspektor-gadget/pkg/types" + "github.com/kubescape/k8s-interface/k8sinterface" + "github.com/kubescape/k8s-interface/workloadinterface" + "github.com/kubescape/node-agent/pkg/k8sclient" + "github.com/kubescape/node-agent/pkg/utils" + "github.com/picatz/xcel" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +type mockServiceK8sClient struct { + k8sclient.K8sClientMock + services map[string]k8sinterface.IWorkload +} + +func (m *mockServiceK8sClient) GetWorkload(namespace, kind, name string) (k8sinterface.IWorkload, error) { + if kind != "Service" { + return m.K8sClientMock.GetWorkload(namespace, kind, name) + } + key := namespace + "/" + name + if svc, ok := m.services[key]; ok { + return svc, nil + } + return nil, errors.New("service not found") +} + +func TestInitServicePeerLabelResolver(t *testing.T) { + defer utils.SetServicePeerLabels(nil) + + svcWorkload := workloadinterface.NewWorkloadObj(map[string]any{ + "apiVersion": "v1", + "kind": "Service", + "metadata": map[string]any{ + "name": "my-svc", + "namespace": "prod", + "labels": map[string]any{ + "helm.sh/chart": "my-chart-1.0.0", + }, + }, + "spec": map[string]any{ + "selector": map[string]any{ + "app": "backend", + "tier": "api", + }, + }, + }) + + k8sDefaultSvc := workloadinterface.NewWorkloadObj(map[string]any{ + "apiVersion": "v1", + "kind": "Service", + "metadata": map[string]any{ + "name": "kubernetes", + "namespace": "default", + "labels": map[string]any{ + "component": "apiserver", + "provider": "kubernetes", + }, + }, + "spec": map[string]any{}, + }) + + mockClient := &mockServiceK8sClient{ + services: map[string]k8sinterface.IWorkload{ + "prod/my-svc": svcWorkload, + "default/kubernetes": k8sDefaultSvc, + }, + } + + // 1. Initialize resolver with mock client + InitServicePeerLabelResolver(mockClient) + + // Verify standard service returns selector labels, not metadata labels + labels := utils.ServicePeerLabels("prod", "my-svc") + assert.Equal(t, map[string]string{"app": "backend", "tier": "api"}, labels) + + // Verify default/kubernetes service returns metadata labels + k8sLabels := utils.ServicePeerLabels("default", "kubernetes") + assert.Equal(t, map[string]string{"component": "apiserver", "provider": "kubernetes"}, k8sLabels) + + // Verify non-existent service returns nil + notFoundLabels := utils.ServicePeerLabels("prod", "unknown-svc") + assert.Nil(t, notFoundLabels) + + // 2. Verify integration with CEL dstPodLabels getter + dstLabelsField, ok := utils.CelFields["dstPodLabels"] + require.True(t, ok) + + event := &mockNetworkEvent{ + endpoint: igtypes.L4Endpoint{ + L3Endpoint: igtypes.L3Endpoint{ + Namespace: "prod", + Name: "my-svc", + Kind: igtypes.EndpointKindService, + PodLabels: map[string]string{"helm.sh/chart": "my-chart-1.0.0"}, // metadata label from kubeipresolver + }, + }, + } + wrapped, _ := xcel.NewObject[utils.CelEvent](event) + + val, err := dstLabelsField.GetFrom(wrapped) + require.NoError(t, err) + assert.Equal(t, map[string]string{"app": "backend", "tier": "api"}, val) + + // 3. Clear resolver on nil client + InitServicePeerLabelResolver(nil) + assert.Nil(t, utils.ServicePeerLabels("prod", "my-svc")) + + // Verify fallback to empty map when resolver is nil + val, err = dstLabelsField.GetFrom(wrapped) + require.NoError(t, err) + assert.Equal(t, map[string]string{}, val) +} + +type mockNetworkEvent struct { + utils.CelEventImpl + endpoint igtypes.L4Endpoint +} + +func (m *mockNetworkEvent) GetDstEndpoint() igtypes.L4Endpoint { + return m.endpoint +} From 439f85afb9d077f8d34ed6760fed8a43ceb9de4d Mon Sep 17 00:00:00 2001 From: Matthias Bertschy Date: Sat, 3 Oct 2026 13:36:04 +0200 Subject: [PATCH 4/5] perf(rulemanager): add bounded LRU cache with TTL refresh to service peer label resolver --- pkg/rulemanager/service_resolver.go | 61 +++++++++-- pkg/rulemanager/service_resolver_test.go | 128 ++++++++++++++++++++++- 2 files changed, 180 insertions(+), 9 deletions(-) diff --git a/pkg/rulemanager/service_resolver.go b/pkg/rulemanager/service_resolver.go index ef12ed340d..6710d2a529 100644 --- a/pkg/rulemanager/service_resolver.go +++ b/pkg/rulemanager/service_resolver.go @@ -1,25 +1,72 @@ package rulemanager import ( + "time" + + "github.com/hashicorp/golang-lru/v2/expirable" "github.com/kubescape/node-agent/pkg/k8sclient" "github.com/kubescape/node-agent/pkg/utils" ) +const ( + defaultServiceCacheSize = 1024 + defaultServiceCacheTTL = 1 * time.Minute +) + +type serviceCacheEntry struct { + selector map[string]string +} + // InitServicePeerLabelResolver registers a ServicePeerLabels hook backed by the Kubernetes client -// so that Service destination endpoints in CEL rules resolve to their backend selector labels. +// with a bounded LRU cache and TTL refresh policy, so that Service destination endpoints in CEL rules +// resolve to their backend selector labels without repeated synchronous Kubernetes API reads. func InitServicePeerLabelResolver(k8sClient k8sclient.K8sClientInterface) { + InitServicePeerLabelResolverWithCache(k8sClient, defaultServiceCacheSize, defaultServiceCacheTTL) +} + +// InitServicePeerLabelResolverWithCache initializes the ServicePeerLabels hook with explicit cache size and TTL. +func InitServicePeerLabelResolverWithCache(k8sClient k8sclient.K8sClientInterface, size int, ttl time.Duration) { if k8sClient == nil { utils.SetServicePeerLabels(nil) return } + if size <= 0 { + size = defaultServiceCacheSize + } + if ttl <= 0 { + ttl = defaultServiceCacheTTL + } + + cache := expirable.NewLRU[string, serviceCacheEntry](size, nil, ttl) + utils.SetServicePeerLabels(func(namespace, name string) map[string]string { - svc, err := k8sClient.GetWorkload(namespace, "Service", name) - if err != nil || svc == nil { - return nil + key := namespace + "/" + name + if entry, ok := cache.Get(key); ok { + return copyLabels(entry.selector) } - if svc.GetName() == "kubernetes" && svc.GetNamespace() == "default" { - return svc.GetLabels() + + var selector map[string]string + svc, err := k8sClient.GetWorkload(namespace, "Service", name) + if err == nil && svc != nil { + if svc.GetName() == "kubernetes" && svc.GetNamespace() == "default" { + selector = svc.GetLabels() + } else { + selector = svc.GetServiceSelector() + } } - return svc.GetServiceSelector() + + cache.Add(key, serviceCacheEntry{selector: copyLabels(selector)}) + return copyLabels(selector) }) } + +func copyLabels(labels map[string]string) map[string]string { + if labels == nil { + return nil + } + cp := make(map[string]string, len(labels)) + for k, v := range labels { + cp[k] = v + } + return cp +} diff --git a/pkg/rulemanager/service_resolver_test.go b/pkg/rulemanager/service_resolver_test.go index dc052657db..32cb0efcdc 100644 --- a/pkg/rulemanager/service_resolver_test.go +++ b/pkg/rulemanager/service_resolver_test.go @@ -2,7 +2,9 @@ package rulemanager import ( "errors" + "sync" "testing" + "time" igtypes "github.com/inspektor-gadget/inspektor-gadget/pkg/types" "github.com/kubescape/k8s-interface/k8sinterface" @@ -16,13 +18,18 @@ import ( type mockServiceK8sClient struct { k8sclient.K8sClientMock + mu sync.Mutex services map[string]k8sinterface.IWorkload + getCalls int } func (m *mockServiceK8sClient) GetWorkload(namespace, kind, name string) (k8sinterface.IWorkload, error) { if kind != "Service" { return m.K8sClientMock.GetWorkload(namespace, kind, name) } + m.mu.Lock() + defer m.mu.Unlock() + m.getCalls++ key := namespace + "/" + name if svc, ok := m.services[key]; ok { return svc, nil @@ -30,6 +37,24 @@ func (m *mockServiceK8sClient) GetWorkload(namespace, kind, name string) (k8sint return nil, errors.New("service not found") } +func (m *mockServiceK8sClient) CallCount() int { + m.mu.Lock() + defer m.mu.Unlock() + return m.getCalls +} + +func (m *mockServiceK8sClient) SetService(key string, svc k8sinterface.IWorkload) { + m.mu.Lock() + defer m.mu.Unlock() + m.services[key] = svc +} + +func (m *mockServiceK8sClient) DeleteService(key string) { + m.mu.Lock() + defer m.mu.Unlock() + delete(m.services, key) +} + func TestInitServicePeerLabelResolver(t *testing.T) { defer utils.SetServicePeerLabels(nil) @@ -67,8 +92,8 @@ func TestInitServicePeerLabelResolver(t *testing.T) { mockClient := &mockServiceK8sClient{ services: map[string]k8sinterface.IWorkload{ - "prod/my-svc": svcWorkload, - "default/kubernetes": k8sDefaultSvc, + "prod/my-svc": svcWorkload, + "default/kubernetes": k8sDefaultSvc, }, } @@ -117,6 +142,105 @@ func TestInitServicePeerLabelResolver(t *testing.T) { assert.Equal(t, map[string]string{}, val) } +func TestServicePeerLabelResolver_CachingAndRefresh(t *testing.T) { + defer utils.SetServicePeerLabels(nil) + + svcWorkload := workloadinterface.NewWorkloadObj(map[string]any{ + "apiVersion": "v1", + "kind": "Service", + "metadata": map[string]any{ + "name": "cached-svc", + "namespace": "test-ns", + }, + "spec": map[string]any{ + "selector": map[string]any{ + "app": "cached-v1", + }, + }, + }) + + mockClient := &mockServiceK8sClient{ + services: map[string]k8sinterface.IWorkload{ + "test-ns/cached-svc": svcWorkload, + }, + } + + // Initialize resolver with short TTL (50ms) for testing expiration and bounded size (2) + ttl := 50 * time.Millisecond + InitServicePeerLabelResolverWithCache(mockClient, 2, ttl) + + // 1. Initial lookup populates cache + labels1 := utils.ServicePeerLabels("test-ns", "cached-svc") + assert.Equal(t, map[string]string{"app": "cached-v1"}, labels1) + assert.Equal(t, 1, mockClient.CallCount(), "First call should query GetWorkload") + + // 2. Subsequent lookups within TTL hit cache without calling GetWorkload + for i := 0; i < 5; i++ { + labelsCached := utils.ServicePeerLabels("test-ns", "cached-svc") + assert.Equal(t, map[string]string{"app": "cached-v1"}, labelsCached) + } + assert.Equal(t, 1, mockClient.CallCount(), "Subsequent calls within TTL must hit cache") + + // 3. Negative caching for non-existent service + notFound1 := utils.ServicePeerLabels("test-ns", "non-existent") + assert.Nil(t, notFound1) + assert.Equal(t, 2, mockClient.CallCount()) + + notFound2 := utils.ServicePeerLabels("test-ns", "non-existent") + assert.Nil(t, notFound2) + assert.Equal(t, 2, mockClient.CallCount(), "Negative result should be cached within TTL") + + // 4. Update service selector and wait for TTL expiry + updatedSvc := workloadinterface.NewWorkloadObj(map[string]any{ + "apiVersion": "v1", + "kind": "Service", + "metadata": map[string]any{ + "name": "cached-svc", + "namespace": "test-ns", + }, + "spec": map[string]any{ + "selector": map[string]any{ + "app": "cached-v2", + }, + }, + }) + mockClient.SetService("test-ns/cached-svc", updatedSvc) + + time.Sleep(ttl + 20*time.Millisecond) + + // Lookup after TTL expiry refreshes from API + labelsUpdated := utils.ServicePeerLabels("test-ns", "cached-svc") + assert.Equal(t, map[string]string{"app": "cached-v2"}, labelsUpdated) + assert.Equal(t, 3, mockClient.CallCount(), "Call after TTL expiry must refresh from GetWorkload") + + // 5. Delete service and wait for TTL expiry + mockClient.DeleteService("test-ns/cached-svc") + time.Sleep(ttl + 20*time.Millisecond) + + labelsDeleted := utils.ServicePeerLabels("test-ns", "cached-svc") + assert.Nil(t, labelsDeleted) + assert.Equal(t, 4, mockClient.CallCount(), "Call after deletion and TTL expiry must reflect deletion") + + // 6. LRU eviction with bounded capacity + mockClient.SetService("test-ns/svc-a", svcWorkload) + mockClient.SetService("test-ns/svc-b", svcWorkload) + mockClient.SetService("test-ns/svc-c", svcWorkload) + + callsBefore := mockClient.CallCount() + utils.ServicePeerLabels("test-ns", "svc-a") // in cache: [svc-a] + utils.ServicePeerLabels("test-ns", "svc-b") // in cache: [svc-b, svc-a] + utils.ServicePeerLabels("test-ns", "svc-c") // capacity 2 -> evicts svc-a; in cache: [svc-c, svc-b] + assert.Equal(t, callsBefore+3, mockClient.CallCount()) + + // svc-b should still be cached + utils.ServicePeerLabels("test-ns", "svc-b") + assert.Equal(t, callsBefore+3, mockClient.CallCount(), "svc-b should still be cached") + + // svc-a was evicted, so accessing it calls GetWorkload again + utils.ServicePeerLabels("test-ns", "svc-a") + assert.Equal(t, callsBefore+4, mockClient.CallCount(), "svc-a was evicted and should re-query") +} + type mockNetworkEvent struct { utils.CelEventImpl endpoint igtypes.L4Endpoint From 54806f996c6ff9903be756c6bbd9ad39d180ca8b Mon Sep 17 00:00:00 2001 From: Matthias Bertschy Date: Sat, 3 Oct 2026 13:42:59 +0200 Subject: [PATCH 5/5] fix(rulemanager): coalesce concurrent service lookups and avoid caching transient failures --- pkg/rulemanager/rule_manager.go | 1 - pkg/rulemanager/service_resolver.go | 44 +++++-- pkg/rulemanager/service_resolver_test.go | 144 ++++++++++++++++++++--- 3 files changed, 164 insertions(+), 25 deletions(-) diff --git a/pkg/rulemanager/rule_manager.go b/pkg/rulemanager/rule_manager.go index 1e00d7fe02..cd6a1511a5 100644 --- a/pkg/rulemanager/rule_manager.go +++ b/pkg/rulemanager/rule_manager.go @@ -92,7 +92,6 @@ func CreateRuleManager( ) (*RuleManager, error) { rulePolicyValidator := NewRulePolicyValidator(objectCache) detectorManager := detectors.NewDetectorManager(mntnsRegistry) - InitServicePeerLabelResolver(k8sClient) r := &RuleManager{ cfg: cfg, diff --git a/pkg/rulemanager/service_resolver.go b/pkg/rulemanager/service_resolver.go index 6710d2a529..20872e694b 100644 --- a/pkg/rulemanager/service_resolver.go +++ b/pkg/rulemanager/service_resolver.go @@ -6,6 +6,8 @@ import ( "github.com/hashicorp/golang-lru/v2/expirable" "github.com/kubescape/node-agent/pkg/k8sclient" "github.com/kubescape/node-agent/pkg/utils" + "golang.org/x/sync/singleflight" + apierrors "k8s.io/apimachinery/pkg/api/errors" ) const ( @@ -18,8 +20,9 @@ type serviceCacheEntry struct { } // InitServicePeerLabelResolver registers a ServicePeerLabels hook backed by the Kubernetes client -// with a bounded LRU cache and TTL refresh policy, so that Service destination endpoints in CEL rules -// resolve to their backend selector labels without repeated synchronous Kubernetes API reads. +// with a bounded LRU cache, singleflight request coalescing, and TTL refresh policy, so that Service +// destination endpoints in CEL rules resolve to their backend selector labels without repeated +// synchronous Kubernetes API reads or duplicate concurrent in-flight requests. func InitServicePeerLabelResolver(k8sClient k8sclient.K8sClientInterface) { InitServicePeerLabelResolverWithCache(k8sClient, defaultServiceCacheSize, defaultServiceCacheTTL) } @@ -38,6 +41,7 @@ func InitServicePeerLabelResolverWithCache(k8sClient k8sclient.K8sClientInterfac } cache := expirable.NewLRU[string, serviceCacheEntry](size, nil, ttl) + var sf singleflight.Group utils.SetServicePeerLabels(func(namespace, name string) map[string]string { key := namespace + "/" + name @@ -45,18 +49,42 @@ func InitServicePeerLabelResolverWithCache(k8sClient k8sclient.K8sClientInterfac return copyLabels(entry.selector) } - var selector map[string]string - svc, err := k8sClient.GetWorkload(namespace, "Service", name) - if err == nil && svc != nil { + res, _, _ := sf.Do(key, func() (any, error) { + // Recheck cache inside singleflight in case another worker already populated it + if entry, ok := cache.Get(key); ok { + return copyLabels(entry.selector), nil + } + + svc, err := k8sClient.GetWorkload(namespace, "Service", name) + if err != nil { + // Only cache negative results for confirmed NotFound responses; + // transient errors (timeouts, network issues, RBAC) must not be cached. + if apierrors.IsNotFound(err) { + cache.Add(key, serviceCacheEntry{selector: nil}) + } + return nil, nil + } + if svc == nil { + cache.Add(key, serviceCacheEntry{selector: nil}) + return nil, nil + } + + var selector map[string]string if svc.GetName() == "kubernetes" && svc.GetNamespace() == "default" { selector = svc.GetLabels() } else { selector = svc.GetServiceSelector() } - } - cache.Add(key, serviceCacheEntry{selector: copyLabels(selector)}) - return copyLabels(selector) + copied := copyLabels(selector) + cache.Add(key, serviceCacheEntry{selector: copied}) + return copyLabels(copied), nil + }) + + if res == nil { + return nil + } + return res.(map[string]string) }) } diff --git a/pkg/rulemanager/service_resolver_test.go b/pkg/rulemanager/service_resolver_test.go index 32cb0efcdc..f1ed6cb634 100644 --- a/pkg/rulemanager/service_resolver_test.go +++ b/pkg/rulemanager/service_resolver_test.go @@ -14,13 +14,17 @@ import ( "github.com/picatz/xcel" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/runtime/schema" ) type mockServiceK8sClient struct { k8sclient.K8sClientMock mu sync.Mutex services map[string]k8sinterface.IWorkload + errs map[string]error getCalls int + delay time.Duration } func (m *mockServiceK8sClient) GetWorkload(namespace, kind, name string) (k8sinterface.IWorkload, error) { @@ -28,13 +32,27 @@ func (m *mockServiceK8sClient) GetWorkload(namespace, kind, name string) (k8sint return m.K8sClientMock.GetWorkload(namespace, kind, name) } m.mu.Lock() - defer m.mu.Unlock() m.getCalls++ + delay := m.delay key := namespace + "/" + name - if svc, ok := m.services[key]; ok { + if customErr, hasErr := m.errs[key]; hasErr { + m.mu.Unlock() + if delay > 0 { + time.Sleep(delay) + } + return nil, customErr + } + svc, ok := m.services[key] + m.mu.Unlock() + + if delay > 0 { + time.Sleep(delay) + } + + if ok { return svc, nil } - return nil, errors.New("service not found") + return nil, apierrors.NewNotFound(schema.GroupResource{Resource: "services"}, name) } func (m *mockServiceK8sClient) CallCount() int { @@ -46,13 +64,27 @@ func (m *mockServiceK8sClient) CallCount() int { func (m *mockServiceK8sClient) SetService(key string, svc k8sinterface.IWorkload) { m.mu.Lock() defer m.mu.Unlock() + if m.services == nil { + m.services = make(map[string]k8sinterface.IWorkload) + } + delete(m.errs, key) m.services[key] = svc } +func (m *mockServiceK8sClient) SetError(key string, err error) { + m.mu.Lock() + defer m.mu.Unlock() + if m.errs == nil { + m.errs = make(map[string]error) + } + m.errs[key] = err +} + func (m *mockServiceK8sClient) DeleteService(key string) { m.mu.Lock() defer m.mu.Unlock() delete(m.services, key) + delete(m.errs, key) } func TestInitServicePeerLabelResolver(t *testing.T) { @@ -181,16 +213,16 @@ func TestServicePeerLabelResolver_CachingAndRefresh(t *testing.T) { } assert.Equal(t, 1, mockClient.CallCount(), "Subsequent calls within TTL must hit cache") - // 3. Negative caching for non-existent service + // 3. Negative caching for confirmed NotFound notFound1 := utils.ServicePeerLabels("test-ns", "non-existent") assert.Nil(t, notFound1) assert.Equal(t, 2, mockClient.CallCount()) notFound2 := utils.ServicePeerLabels("test-ns", "non-existent") assert.Nil(t, notFound2) - assert.Equal(t, 2, mockClient.CallCount(), "Negative result should be cached within TTL") + assert.Equal(t, 2, mockClient.CallCount(), "NotFound should be negatively cached within TTL") - // 4. Update service selector and wait for TTL expiry + // 4. Update service selector and wait for TTL expiry using assert.Eventually updatedSvc := workloadinterface.NewWorkloadObj(map[string]any{ "apiVersion": "v1", "kind": "Service", @@ -206,20 +238,17 @@ func TestServicePeerLabelResolver_CachingAndRefresh(t *testing.T) { }) mockClient.SetService("test-ns/cached-svc", updatedSvc) - time.Sleep(ttl + 20*time.Millisecond) - - // Lookup after TTL expiry refreshes from API - labelsUpdated := utils.ServicePeerLabels("test-ns", "cached-svc") - assert.Equal(t, map[string]string{"app": "cached-v2"}, labelsUpdated) - assert.Equal(t, 3, mockClient.CallCount(), "Call after TTL expiry must refresh from GetWorkload") + assert.Eventually(t, func() bool { + labels := utils.ServicePeerLabels("test-ns", "cached-svc") + return labels != nil && labels["app"] == "cached-v2" + }, 1*time.Second, 10*time.Millisecond, "Service selector should refresh after TTL expires") // 5. Delete service and wait for TTL expiry mockClient.DeleteService("test-ns/cached-svc") - time.Sleep(ttl + 20*time.Millisecond) - labelsDeleted := utils.ServicePeerLabels("test-ns", "cached-svc") - assert.Nil(t, labelsDeleted) - assert.Equal(t, 4, mockClient.CallCount(), "Call after deletion and TTL expiry must reflect deletion") + assert.Eventually(t, func() bool { + return utils.ServicePeerLabels("test-ns", "cached-svc") == nil + }, 1*time.Second, 10*time.Millisecond, "Deleted service should resolve to nil after TTL expires") // 6. LRU eviction with bounded capacity mockClient.SetService("test-ns/svc-a", svcWorkload) @@ -241,6 +270,89 @@ func TestServicePeerLabelResolver_CachingAndRefresh(t *testing.T) { assert.Equal(t, callsBefore+4, mockClient.CallCount(), "svc-a was evicted and should re-query") } +func TestServicePeerLabelResolver_ConcurrentMisses(t *testing.T) { + defer utils.SetServicePeerLabels(nil) + + svcWorkload := workloadinterface.NewWorkloadObj(map[string]any{ + "apiVersion": "v1", + "kind": "Service", + "metadata": map[string]any{ + "name": "concurrent-svc", + "namespace": "concur-ns", + }, + "spec": map[string]any{ + "selector": map[string]any{ + "app": "concurrent", + }, + }, + }) + + mockClient := &mockServiceK8sClient{ + services: map[string]k8sinterface.IWorkload{ + "concur-ns/concurrent-svc": svcWorkload, + }, + delay: 20 * time.Millisecond, // simulate API latency to ensure concurrence + } + + InitServicePeerLabelResolverWithCache(mockClient, 100, 1*time.Minute) + + const workers = 10 + var wg sync.WaitGroup + wg.Add(workers) + + results := make([]map[string]string, workers) + for i := 0; i < workers; i++ { + go func(idx int) { + defer wg.Done() + results[idx] = utils.ServicePeerLabels("concur-ns", "concurrent-svc") + }(i) + } + + wg.Wait() + + for i := 0; i < workers; i++ { + assert.Equal(t, map[string]string{"app": "concurrent"}, results[i]) + } + assert.Equal(t, 1, mockClient.CallCount(), "Singleflight must coalesce concurrent requests into exactly 1 API call") +} + +func TestServicePeerLabelResolver_TransientFailureThenSuccess(t *testing.T) { + defer utils.SetServicePeerLabels(nil) + + svcWorkload := workloadinterface.NewWorkloadObj(map[string]any{ + "apiVersion": "v1", + "kind": "Service", + "metadata": map[string]any{ + "name": "flaky-svc", + "namespace": "flaky-ns", + }, + "spec": map[string]any{ + "selector": map[string]any{ + "app": "recovered", + }, + }, + }) + + mockClient := &mockServiceK8sClient{} + // Transient error: timeout / network failure (NOT NotFound) + mockClient.SetError("flaky-ns/flaky-svc", errors.New("i/o timeout: transient api server failure")) + + InitServicePeerLabelResolverWithCache(mockClient, 100, 1*time.Minute) + + // 1. Transient failure should return nil + res1 := utils.ServicePeerLabels("flaky-ns", "flaky-svc") + assert.Nil(t, res1) + assert.Equal(t, 1, mockClient.CallCount()) + + // 2. Immediately recover: set the valid workload (simulating transient issue resolved) + mockClient.SetService("flaky-ns/flaky-svc", svcWorkload) + + // 3. The very next call must NOT be blocked by a negative cache; it must re-fetch and succeed + res2 := utils.ServicePeerLabels("flaky-ns", "flaky-svc") + assert.Equal(t, map[string]string{"app": "recovered"}, res2) + assert.Equal(t, 2, mockClient.CallCount(), "Transient failure must not be negatively cached") +} + type mockNetworkEvent struct { utils.CelEventImpl endpoint igtypes.L4Endpoint