fix(kuma-cp): don't cache filtered data (backport #5574)

pkg/core/resources/manager/cache.go

@@ -85,6 +85,9 @@ func (c *cachedManager) Get(ctx context.Context, res model.Resource, fs ...store
 
 func (c *cachedManager) List(ctx context.Context, list model.ResourceList, fs ...store.ListOptionsFunc) error {
 	opts := store.NewListOptions(fs...)
+	if !opts.IsCacheable() {
+		return fmt.Errorf("filter functions are not allowed for cached store")
+	}
 	cacheKey := fmt.Sprintf("LIST:%s:%s", list.GetItemType(), opts.HashCode())
 	obj, found := c.cache.Get(cacheKey)
 	if !found {

pkg/core/resources/manager/cache_test.go

@@ -209,4 +209,53 @@ var _ = Describe("Cached Resource Manager", func() {
 		// then first request does not block request for other type
 		Expect(err).ToNot(HaveOccurred())
 	}))
+
+	It("should cache List() at different key when ordered", test.Within(5*time.Second, func() {
+		// when fetched resources multiple times
+		fetch := func(ordered bool) core_mesh.DataplaneResourceList {
+			fetched := core_mesh.DataplaneResourceList{}
+			var err error
+			if ordered {
+				err = cachedManager.List(context.Background(), &fetched, core_store.ListOrdered(), core_store.ListByMesh("default"))
+			} else {
+				err = cachedManager.List(context.Background(), &fetched, core_store.ListByMesh("default"))
+			}
+			Expect(err).ToNot(HaveOccurred())
+			return fetched
+		}
+
+		var wg sync.WaitGroup
+		for i := 0; i < 100; i++ {
+			wg.Add(1)
+			go func() {
+				fetch(false)
+				wg.Done()
+			}()
+		}
+		wg.Wait()
+
+		// then real manager should be called only once
+		list := fetch(false)
+		Expect(list.Items).To(HaveLen(1))
+		Expect(list.Items[0].GetSpec()).To(MatchProto(res.Spec))
+		Expect(countingManager.listQueries).To(Equal(1))
+
+		// when call for ordered data
+		list = fetch(true)
+
+		// then real manager should be called
+		Expect(list.Items).To(HaveLen(1))
+		Expect(list.Items[0].GetSpec()).To(MatchProto(res.Spec))
+		Expect(countingManager.listQueries).To(Equal(2))
+
+		// and metrics are published
+		Expect(test_metrics.FindMetric(metrics, "store_cache", "operation", "list", "result", "miss").Counter.GetValue()).To(Equal(2.0))
+		hits := test_metrics.FindMetric(metrics, "store_cache", "operation", "list", "result", "hit").Counter.GetValue()
+		hitWaits := 0.0
+		hitWaitMetric := test_metrics.FindMetric(metrics, "store_cache", "operation", "list", "result", "hit-wait")
+		if hitWaitMetric != nil {
+			hitWaits = hitWaitMetric.Counter.GetValue()
+		}
+		Expect(hits + hitWaits).To(Equal(100.0))
+	}))
 })

pkg/core/resources/store/options.go

@@ -204,6 +204,10 @@ func ListOrdered() ListOptionsFunc {
 	}
 }
 
+func (l *ListOptions) IsCacheable() bool {
+	return l.FilterFunc == nil
+}
+
 func (l *ListOptions) HashCode() string {
-	return l.Mesh
+	return fmt.Sprintf("%s:%t:%d:%s", l.Mesh, l.Ordered, l.PageSize, l.PageOffset)
 }

pkg/xds/sync/ingress_proxy_builder.go

@@ -9,6 +9,7 @@ import (
 	"github.com/kumahq/kuma/pkg/core/resources/manager"
 	core_model "github.com/kumahq/kuma/pkg/core/resources/model"
 	"github.com/kumahq/kuma/pkg/core/resources/registry"
+	"github.com/kumahq/kuma/pkg/core/resources/store"
 	core_store "github.com/kumahq/kuma/pkg/core/resources/store"
 	core_xds "github.com/kumahq/kuma/pkg/core/xds"
 	xds_cache "github.com/kumahq/kuma/pkg/xds/cache/mesh"
@@ -152,15 +153,16 @@ func (p *IngressProxyBuilder) updateIngress(ctx context.Context, zoneIngress *co
 
 func (p *IngressProxyBuilder) getIngressExternalServices(ctx context.Context) (*core_mesh.ExternalServiceResourceList, error) {
 	meshList := &core_mesh.MeshResourceList{}
-	if err := p.ReadOnlyResManager.List(ctx, meshList, core_store.ListOrdered(), core_store.ListByFilterFunc(func(rs core_model.Resource) bool {
-		return rs.(*core_mesh.MeshResource).ZoneEgressEnabled()
-	})); err != nil {
+	if err := p.ReadOnlyResManager.List(ctx, meshList, store.ListOrdered()); err != nil {
 		return nil, err
 	}
 
 	allMeshExternalServices := &core_mesh.ExternalServiceResourceList{}
 	var externalServices []*core_mesh.ExternalServiceResource
-	for _, mesh := range meshList.GetItems() {
+	for _, mesh := range meshList.Items {
+		if !mesh.ZoneEgressEnabled() {
+			continue
+		}
 		meshName := mesh.GetMeta().GetName()
 
 		meshCtx, err := p.meshCache.GetMeshContext(ctx, syncLog, meshName)