feat(kuma-cp): add observability to k8s auth cache

go.mod

@@ -19,6 +19,7 @@ require (
 	github.com/evanphx/json-patch/v5 v5.6.0
 	github.com/go-logr/logr v1.2.4
 	github.com/go-logr/zapr v1.2.4
+	github.com/goburrow/cache v0.1.4
 	github.com/golang-jwt/jwt/v4 v4.5.0
 	github.com/golang-migrate/migrate/v4 v4.16.2
 	github.com/golang/protobuf v1.5.3

go.sum

@@ -200,6 +200,8 @@ github.com/go-task/slim-sprig v0.0.0-20230315185526-52ccab3ef572 h1:tfuBGBXKqDEe
 github.com/go-task/slim-sprig v0.0.0-20230315185526-52ccab3ef572/go.mod h1:9Pwr4B2jHnOSGXyyzV8ROjYa2ojvAY6HCGYYfMoC3Ls=
 github.com/gobuffalo/flect v1.0.2 h1:eqjPGSo2WmjgY2XlpGwo2NXgL3RucAKo4k4qQMNA5sA=
 github.com/gobuffalo/flect v1.0.2/go.mod h1:A5msMlrHtLqh9umBSnvabjsMrCcCpAyzglnDvkbYKHs=
+github.com/goburrow/cache v0.1.4 h1:As4KzO3hgmzPlnaMniZU9+VmoNYseUhuELbxy9mRBfw=
+github.com/goburrow/cache v0.1.4/go.mod h1:cDFesZDnIlrHoNlMYqqMpCRawuXulgx+y7mXU8HZ+/c=
 github.com/gobwas/glob v0.2.3 h1:A4xDbljILXROh+kObIiy5kIaPYD8e96x1tgBhUI5J+Y=
 github.com/gobwas/glob v0.2.3/go.mod h1:d3Ez4x06l9bZtSvzIay5+Yzi0fmZzPgnTbPcKjJAkT8=
 github.com/godbus/dbus/v5 v5.0.4/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA=

pkg/util/cache/prometheus_status_counter.go

@@ -0,0 +1,43 @@
+package cache
+
+import (
+	"time"
+
+	"github.com/goburrow/cache"
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+const ResultLabel = "result"
+
+func NewMetric(name, help string) *prometheus.CounterVec {
+	return prometheus.NewCounterVec(prometheus.CounterOpts{
+		Name: name,
+		Help: help,
+	}, []string{ResultLabel})
+}
+
+type PrometheusStatsCounter struct {
+	Metric *prometheus.CounterVec
+}
+
+var _ cache.StatsCounter = &PrometheusStatsCounter{}
+
+func (p *PrometheusStatsCounter) RecordHits(count uint64) {
+	p.Metric.WithLabelValues("hit").Add(float64(count))
+}
+
+func (p *PrometheusStatsCounter) RecordMisses(count uint64) {
+	p.Metric.WithLabelValues("miss").Add(float64(count))
+}
+
+func (p *PrometheusStatsCounter) RecordLoadSuccess(loadTime time.Duration) {
+}
+
+func (p *PrometheusStatsCounter) RecordLoadError(loadTime time.Duration) {
+}
+
+func (p *PrometheusStatsCounter) RecordEviction() {
+}
+
+func (p *PrometheusStatsCounter) Snapshot(stats *cache.Stats) {
+}

pkg/xds/auth/components/components.go

@@ -14,31 +14,33 @@ import (
 	"github.com/kumahq/kuma/pkg/xds/auth"
 	k8s_auth "github.com/kumahq/kuma/pkg/xds/auth/k8s"
 	universal_auth "github.com/kumahq/kuma/pkg/xds/auth/universal"
+	xds_metrics "github.com/kumahq/kuma/pkg/xds/metrics"
 )
 
-type Context interface {
-	Config() kuma_cp.Config
-	Extensions() context.Context
-	ReadOnlyResourceManager() core_manager.ReadOnlyResourceManager
+type Deps struct {
+	Config                  kuma_cp.Config
+	Extensions              context.Context
+	ReadOnlyResourceManager core_manager.ReadOnlyResourceManager
+	XdsMetrics              *xds_metrics.Metrics
 }
 
-func NewKubeAuthenticator(rt Context) (auth.Authenticator, error) {
-	mgr, ok := k8s_extensions.FromManagerContext(rt.Extensions())
+func NewKubeAuthenticator(deps Deps) (auth.Authenticator, error) {
+	mgr, ok := k8s_extensions.FromManagerContext(deps.Extensions)
 	if !ok {
 		return nil, errors.Errorf("k8s controller runtime Manager hasn't been configured")
 	}
-	return k8s_auth.New(mgr.GetClient()), nil
+	return k8s_auth.New(mgr.GetClient(), deps.XdsMetrics), nil
 }
 
-func NewUniversalAuthenticator(rt Context) (auth.Authenticator, error) {
-	config := rt.Config()
+func NewUniversalAuthenticator(deps Deps) (auth.Authenticator, error) {
+	config := deps.Config
 
-	dataplaneValidator, err := builtin.NewDataplaneTokenValidator(rt.ReadOnlyResourceManager(), config.Store.Type, config.DpServer.Authn.DpProxy.DpToken.Validator)
+	dataplaneValidator, err := builtin.NewDataplaneTokenValidator(deps.ReadOnlyResourceManager, config.Store.Type, config.DpServer.Authn.DpProxy.DpToken.Validator)
 	if err != nil {
 		return nil, err
 	}
-	zoneIngressValidator := builtin.NewZoneIngressTokenValidator(rt.ReadOnlyResourceManager(), config.Store.Type)
-	zoneTokenValidator, err := builtin.NewZoneTokenValidator(rt.ReadOnlyResourceManager(), config.Mode, config.Store.Type, config.DpServer.Authn.ZoneProxy.ZoneToken.Validator)
+	zoneIngressValidator := builtin.NewZoneIngressTokenValidator(deps.ReadOnlyResourceManager, config.Store.Type)
+	zoneTokenValidator, err := builtin.NewZoneTokenValidator(deps.ReadOnlyResourceManager, config.Mode, config.Store.Type, config.DpServer.Authn.ZoneProxy.ZoneToken.Validator)
 	if err != nil {
 		return nil, err
 	}
@@ -47,12 +49,12 @@ func NewUniversalAuthenticator(rt Context) (auth.Authenticator, error) {
 	return universal_auth.NewAuthenticator(dataplaneValidator, adaptedValidator, config.Multizone.Zone.Name), nil
 }
 
-func DefaultAuthenticator(rt Context, typ string) (auth.Authenticator, error) {
+func DefaultAuthenticator(deps Deps, typ string) (auth.Authenticator, error) {
 	switch typ {
 	case dp_server.DpServerAuthServiceAccountToken:
-		return NewKubeAuthenticator(rt)
+		return NewKubeAuthenticator(deps)
 	case dp_server.DpServerAuthDpToken, dp_server.DpServerAuthZoneToken:
-		return NewUniversalAuthenticator(rt)
+		return NewUniversalAuthenticator(deps)
 	case dp_server.DpServerAuthNone:
 		return universal_auth.NewNoopAuthenticator(), nil
 	default:

pkg/xds/auth/k8s/authenticator.go

@@ -5,7 +5,7 @@ import (
 	"strings"
 	"time"
 
-	"github.com/patrickmn/go-cache"
+	"github.com/goburrow/cache"
 	"github.com/pkg/errors"
 	kube_auth "k8s.io/api/authentication/v1"
 	kube_core "k8s.io/api/core/v1"
@@ -14,29 +14,35 @@ import (
 
 	core_mesh "github.com/kumahq/kuma/pkg/core/resources/apis/mesh"
 	"github.com/kumahq/kuma/pkg/core/resources/model"
+	util_cache "github.com/kumahq/kuma/pkg/util/cache"
 	util_k8s "github.com/kumahq/kuma/pkg/util/k8s"
 	"github.com/kumahq/kuma/pkg/xds/auth"
+	xds_metrics "github.com/kumahq/kuma/pkg/xds/metrics"
 )
 
-func New(client kube_client.Client) auth.Authenticator {
+func New(client kube_client.Client, metrics *xds_metrics.Metrics) auth.Authenticator {
+	authCache := cache.New(
+		cache.WithExpireAfterAccess(1*time.Hour),
+		cache.WithMaximumSize(100000),
+		cache.WithStatsCounter(&util_cache.PrometheusStatsCounter{Metric: metrics.KubeAuthCache}),
+	)
+
 	return &kubeAuthenticator{
 		client:        client,
-		authenticated: cache.New(1*time.Hour, 1*time.Minute),
+		authenticated: authCache,
 	}
 }
 
 type kubeAuthenticator struct {
 	client kube_client.Client
 
-	authenticated *cache.Cache
+	authenticated cache.Cache
 }
 
 var _ auth.Authenticator = &kubeAuthenticator{}
 
 func (k *kubeAuthenticator) Authenticate(ctx context.Context, resource model.Resource, credential auth.Credential) error {
-	_, authenticated := k.authenticated.Get(credential)
-	if authenticated {
-		k.authenticated.Set(credential, struct{}{}, 1*time.Hour) // prolong the cache
+	if _, authenticated := k.authenticated.GetIfPresent(credential); authenticated {
 		return nil
 	}
 
@@ -52,7 +58,7 @@ func (k *kubeAuthenticator) Authenticate(ctx context.Context, resource model.Res
 		return err
 	}
 
-	k.authenticated.Set(credential, struct{}{}, 1*time.Hour)
+	k.authenticated.Put(credential, struct{}{})
 	return nil
 }
 

pkg/xds/metrics/metrics.go

@@ -4,11 +4,13 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 
 	core_metrics "github.com/kumahq/kuma/pkg/metrics"
+	util_cache "github.com/kumahq/kuma/pkg/util/cache"
 )
 
 type Metrics struct {
 	XdsGenerations       prometheus.Summary
 	XdsGenerationsErrors prometheus.Counter
+	KubeAuthCache        *prometheus.CounterVec
 }
 
 func NewMetrics(metrics core_metrics.Metrics) (*Metrics, error) {
@@ -17,19 +19,21 @@ func NewMetrics(metrics core_metrics.Metrics) (*Metrics, error) {
 		Help:       "Summary of XDS Snapshot generation",
 		Objectives: core_metrics.DefaultObjectives,
 	})
-	if err := metrics.Register(xdsGenerations); err != nil {
-		return nil, err
-	}
 	xdsGenerationsErrors := prometheus.NewCounter(prometheus.CounterOpts{
 		Name: "xds_generation_errors",
 		Help: "Counter of errors during XDS generation",
 	})
-	if err := metrics.Register(xdsGenerationsErrors); err != nil {
+	kubeAuthCache := util_cache.NewMetric(
+		"kube_auth_cache",
+		"Number of cache operations for Kubernetes authentication on XDS connection",
+	)
+	if err := metrics.BulkRegister(xdsGenerations, xdsGenerationsErrors, kubeAuthCache); err != nil {
 		return nil, err
 	}
 
 	return &Metrics{
 		XdsGenerations:       xdsGenerations,
 		XdsGenerationsErrors: xdsGenerationsErrors,
+		KubeAuthCache:        kubeAuthCache,
 	}, nil
 }

pkg/xds/runtime/context.go

@@ -1,37 +1,60 @@
 package runtime
 
 import (
+	"context"
+
+	kuma_cp "github.com/kumahq/kuma/pkg/config/app/kuma-cp"
+	core_manager "github.com/kumahq/kuma/pkg/core/resources/manager"
+	core_metrics "github.com/kumahq/kuma/pkg/metrics"
 	util_xds "github.com/kumahq/kuma/pkg/util/xds"
 	xds_auth "github.com/kumahq/kuma/pkg/xds/auth"
 	"github.com/kumahq/kuma/pkg/xds/auth/components"
 	xds_hooks "github.com/kumahq/kuma/pkg/xds/hooks"
+	xds_metrics "github.com/kumahq/kuma/pkg/xds/metrics"
 )
 
 type XDSRuntimeContext struct {
 	DpProxyAuthenticator   xds_auth.Authenticator
 	ZoneProxyAuthenticator xds_auth.Authenticator
 	Hooks                  *xds_hooks.Hooks
 	ServerCallbacks        util_xds.Callbacks
+	Metrics                *xds_metrics.Metrics
 }
 
 type ContextWithXDS interface {
-	components.Context
+	Config() kuma_cp.Config
+	Extensions() context.Context
+	ReadOnlyResourceManager() core_manager.ReadOnlyResourceManager
+	Metrics() core_metrics.Metrics
 	XDS() XDSRuntimeContext
 }
 
 func WithDefaults(ctx ContextWithXDS) (XDSRuntimeContext, error) {
 	currentXDS := ctx.XDS()
+	if currentXDS.Metrics == nil {
+		xdsMetrics, err := xds_metrics.NewMetrics(ctx.Metrics())
+		if err != nil {
+			return XDSRuntimeContext{}, err
+		}
+		currentXDS.Metrics = xdsMetrics
+	}
+	authDeps := components.Deps{
+		Config:                  ctx.Config(),
+		Extensions:              ctx.Extensions(),
+		ReadOnlyResourceManager: ctx.ReadOnlyResourceManager(),
+		XdsMetrics:              currentXDS.Metrics,
+	}
 
 	if currentXDS.DpProxyAuthenticator == nil {
-		dpProxyAuth, err := components.DefaultAuthenticator(ctx, ctx.Config().DpServer.Authn.DpProxy.Type)
+		dpProxyAuth, err := components.DefaultAuthenticator(authDeps, ctx.Config().DpServer.Authn.DpProxy.Type)
 		if err != nil {
 			return XDSRuntimeContext{}, err
 		}
 		currentXDS.DpProxyAuthenticator = dpProxyAuth
 	}
 
 	if currentXDS.ZoneProxyAuthenticator == nil {
-		zoneProxyAuth, err := components.DefaultAuthenticator(ctx, ctx.Config().DpServer.Authn.ZoneProxy.Type)
+		zoneProxyAuth, err := components.DefaultAuthenticator(authDeps, ctx.Config().DpServer.Authn.ZoneProxy.Type)
 		if err != nil {
 			return XDSRuntimeContext{}, err
 		}

pkg/xds/server/components.go

@@ -11,7 +11,6 @@ import (
 	util_xds "github.com/kumahq/kuma/pkg/util/xds"
 	"github.com/kumahq/kuma/pkg/xds/cache/cla"
 	xds_context "github.com/kumahq/kuma/pkg/xds/context"
-	xds_metrics "github.com/kumahq/kuma/pkg/xds/metrics"
 	"github.com/kumahq/kuma/pkg/xds/secrets"
 	v3 "github.com/kumahq/kuma/pkg/xds/server/v3"
 )
@@ -51,10 +50,6 @@ func RegisterXDS(rt core_runtime.Runtime) error {
 	if err != nil {
 		return err
 	}
-	xdsMetrics, err := xds_metrics.NewMetrics(rt.Metrics())
-	if err != nil {
-		return err
-	}
 	claCache, err := cla.NewCache(rt.Config().Store.Cache.ExpirationTime.Duration, rt.Metrics())
 	if err != nil {
 		return err
@@ -80,7 +75,7 @@ func RegisterXDS(rt core_runtime.Runtime) error {
 		Zone:     rt.Config().Multizone.Zone.Name,
 	}
 
-	if err := v3.RegisterXDS(statsCallbacks, xdsMetrics, envoyCpCtx, rt); err != nil {
+	if err := v3.RegisterXDS(statsCallbacks, rt.XDS().Metrics, envoyCpCtx, rt); err != nil {
 		return errors.Wrap(err, "could not register V3 XDS")
 	}
 	return nil