Move everything common between cla and mesh cache in a common cache i…

…mplementation

Signed-off-by: Charly Molter <charly@koyeb.com>

pkg/xds/cache/cla/cache.go

@@ -9,13 +9,9 @@ import (
 
 	"github.com/kumahq/kuma/pkg/core/datasource"
 
-	"github.com/prometheus/client_golang/prometheus"
-
 	"github.com/kumahq/kuma/pkg/core/resources/model"
 	"github.com/kumahq/kuma/pkg/metrics"
 
-	"github.com/patrickmn/go-cache"
-
 	"github.com/kumahq/kuma/pkg/core"
 	"github.com/kumahq/kuma/pkg/core/dns/lookup"
 	core_mesh "github.com/kumahq/kuma/pkg/core/resources/apis/mesh"
@@ -35,13 +31,11 @@ var (
 // which reconcile Dataplane's state. In scope of one mesh ClusterLoadAssignment
 // will be the same for each service so no need to reconcile for each dataplane.
 type Cache struct {
-	cache   *cache.Cache
-	rm      manager.ReadOnlyResourceManager
-	dsl     datasource.Loader
-	ipFunc  lookup.LookupIPFunc
-	zone    string
-	onceMap *once.Map
-	metrics *prometheus.GaugeVec
+	cache  *once.Cache
+	rm     manager.ReadOnlyResourceManager
+	dsl    datasource.Loader
+	ipFunc lookup.LookupIPFunc
+	zone   string
 }
 
 func NewCache(
@@ -51,36 +45,22 @@ func NewCache(
 	ipFunc lookup.LookupIPFunc,
 	metrics metrics.Metrics,
 ) (*Cache, error) {
-	metric := prometheus.NewGaugeVec(prometheus.GaugeOpts{
-		Name: "cla_cache",
-		Help: "Summary of CLA Cache",
-	}, []string{"operation", "result"})
-	if err := metrics.Register(metric); err != nil {
+	c, err := once.New(expirationTime, "cla_cache", metrics)
+	if err != nil {
 		return nil, err
 	}
 	return &Cache{
-		cache:   cache.New(expirationTime, time.Duration(int64(float64(expirationTime)*0.9))),
-		rm:      rm,
-		dsl:     dsl,
-		zone:    zone,
-		ipFunc:  ipFunc,
-		onceMap: once.NewMap(),
-		metrics: metric,
+		cache:  c,
+		rm:     rm,
+		dsl:    dsl,
+		zone:   zone,
+		ipFunc: ipFunc,
 	}, nil
 }
 
 func (c *Cache) GetCLA(ctx context.Context, meshName, meshHash, service string, apiVersion envoy_common.APIVersion) (proto.Message, error) {
 	key := fmt.Sprintf("%s:%s:%s:%s", apiVersion, meshName, service, meshHash)
-	value, found := c.cache.Get(key)
-	if found {
-		c.metrics.WithLabelValues("get", "hit").Inc()
-		return value.(proto.Message), nil
-	}
-	o := c.onceMap.Get(key)
-	c.metrics.WithLabelValues("get", "hit-wait").Inc()
-	o.Do(func() (interface{}, error) {
-		c.metrics.WithLabelValues("get", "hit-wait").Dec()
-		c.metrics.WithLabelValues("get", "miss").Inc()
+	elt, err := c.cache.Get(ctx, key, func(ctx context.Context, key string) (interface{}, error) {
 		dataplanes, err := topology.GetDataplanes(claCacheLog, ctx, c.rm, c.ipFunc, meshName)
 		if err != nil {
 			return nil, err
@@ -94,16 +74,10 @@ func (c *Cache) GetCLA(ctx context.Context, meshName, meshHash, service string,
 			return nil, err
 		}
 		endpointMap := topology.BuildEndpointMap(mesh, c.zone, dataplanes.Items, externalServices.Items, c.dsl)
-		cla, err := envoy_endpoints.CreateClusterLoadAssignment(service, endpointMap[service], apiVersion)
-		if err != nil {
-			return nil, err
-		}
-		c.cache.SetDefault(key, cla)
-		c.onceMap.Delete(key)
-		return cla, nil
+		return envoy_endpoints.CreateClusterLoadAssignment(service, endpointMap[service], apiVersion)
 	})
-	if o.Err != nil {
-		return nil, o.Err
+	if err != nil {
+		return nil, err
 	}
-	return o.Value.(proto.Message), nil
+	return elt.(proto.Message), nil
 }

pkg/xds/cache/mesh/cache.go

@@ -4,14 +4,10 @@ import (
 	"context"
 	"time"
 
-	"github.com/prometheus/client_golang/prometheus"
-
 	"github.com/kumahq/kuma/pkg/metrics"
 
 	"github.com/kumahq/kuma/pkg/xds/cache/once"
 
-	"github.com/patrickmn/go-cache"
-
 	"github.com/kumahq/kuma/pkg/core/dns/lookup"
 	"github.com/kumahq/kuma/pkg/core/resources/manager"
 	core_model "github.com/kumahq/kuma/pkg/core/resources/model"
@@ -21,12 +17,10 @@ import (
 // reconcile Dataplane's state. Calculating hash is a heavy operation
 // that requires fetching all the resources belonging to the Mesh.
 type Cache struct {
-	cache   *cache.Cache
-	rm      manager.ReadOnlyResourceManager
-	types   []core_model.ResourceType
-	ipFunc  lookup.LookupIPFunc
-	onceMap *once.Map
-	metrics *prometheus.GaugeVec
+	cache  *once.Cache
+	rm     manager.ReadOnlyResourceManager
+	types  []core_model.ResourceType
+	ipFunc lookup.LookupIPFunc
 }
 
 func NewCache(
@@ -36,47 +30,28 @@ func NewCache(
 	ipFunc lookup.LookupIPFunc,
 	metrics metrics.Metrics,
 ) (*Cache, error) {
-	metric := prometheus.NewGaugeVec(prometheus.GaugeOpts{
-		Name: "mesh_cache",
-		Help: "Summary of Mesh Cache",
-	}, []string{"operation", "result"})
-	if err := metrics.Register(metric); err != nil {
+	c, err := once.New(expirationTime, "mesh_cache", metrics)
+	if err != nil {
 		return nil, err
 	}
 	return &Cache{
-		rm:      rm,
-		types:   types,
-		ipFunc:  ipFunc,
-		onceMap: once.NewMap(),
-		cache:   cache.New(expirationTime, time.Duration(int64(float64(expirationTime)*0.9))),
-		metrics: metric,
+		rm:     rm,
+		types:  types,
+		ipFunc: ipFunc,
+		cache:  c,
 	}, nil
 }
 
 func (c *Cache) GetHash(ctx context.Context, mesh string) (string, error) {
-	hash, found := c.cache.Get(mesh)
-	if found {
-		c.metrics.WithLabelValues("get", "hit").Inc()
-		return hash.(string), nil
-	}
-	o := c.onceMap.Get(mesh)
-	c.metrics.WithLabelValues("get", "hit-wait").Inc()
-	o.Do(func() (interface{}, error) {
-		c.metrics.WithLabelValues("get", "hit-wait").Dec()
-		c.metrics.WithLabelValues("get", "miss").Inc()
-		snapshot, err := GetMeshSnapshot(ctx, mesh, c.rm, c.types, c.ipFunc)
+	elt, err := c.cache.Get(ctx, mesh, func(ctx context.Context, key string) (interface{}, error) {
+		snapshot, err := GetMeshSnapshot(ctx, key, c.rm, c.types, c.ipFunc)
 		if err != nil {
-			// Don't cache errors
-			c.onceMap.Delete(mesh)
 			return nil, err
 		}
-		hash = snapshot.hash()
-		c.cache.SetDefault(mesh, hash)
-		c.onceMap.Delete(mesh)
-		return hash, nil
+		return snapshot.hash(), nil
 	})
-	if o.Err != nil {
-		return "", o.Err
+	if err != nil {
+		return "", err
 	}
-	return o.Value.(string), nil
+	return elt.(string), nil
 }

pkg/xds/cache/once/cache.go

@@ -0,0 +1,58 @@
+package once
+
+import (
+	"context"
+	"time"
+
+	"github.com/patrickmn/go-cache"
+	"github.com/prometheus/client_golang/prometheus"
+
+	"github.com/kumahq/kuma/pkg/metrics"
+)
+
+type Cache struct {
+	cache   *cache.Cache
+	onceMap *omap
+	metrics *prometheus.GaugeVec
+}
+
+func New(expirationTime time.Duration, name string, metrics metrics.Metrics) (*Cache, error) {
+	metric := prometheus.NewGaugeVec(prometheus.GaugeOpts{
+		Name: name,
+		Help: "Summary of " + name,
+	}, []string{"operation", "result"})
+	if err := metrics.Register(metric); err != nil {
+		return nil, err
+	}
+	return &Cache{
+		cache:   cache.New(expirationTime, time.Duration(int64(float64(expirationTime)*0.9))),
+		onceMap: newMap(),
+		metrics: metric,
+	}, nil
+}
+
+func (c *Cache) Get(ctx context.Context, key string, fn func(context.Context, string) (interface{}, error)) (interface{}, error) {
+	v, found := c.cache.Get(key)
+	if found {
+		c.metrics.WithLabelValues("get", "hit").Inc()
+		return v, nil
+	}
+	o := c.onceMap.Get(key)
+	c.metrics.WithLabelValues("get", "hit-wait").Inc()
+	o.Do(func() (interface{}, error) {
+		defer c.onceMap.Delete(key)
+		c.metrics.WithLabelValues("get", "hit-wait").Dec()
+		c.metrics.WithLabelValues("get", "miss").Inc()
+		res, err := fn(ctx, key)
+		if err != nil {
+			c.metrics.WithLabelValues("get", "error").Inc()
+			return nil, err
+		}
+		c.cache.SetDefault(key, res)
+		return res, nil
+	})
+	if o.Err != nil {
+		return "", o.Err
+	}
+	return o.Value, nil
+}