Merge pull request #870 from fluxcd/token-cache

Add cache specialized for access tokens

cache/go.mod

@@ -3,6 +3,7 @@ module github.com/fluxcd/pkg/cache
 go 1.23.0
 
 require (
+	github.com/go-logr/logr v1.4.2
 	github.com/onsi/gomega v1.36.2
 	github.com/prometheus/client_golang v1.20.5
 )

cache/involved_object.go

@@ -0,0 +1,23 @@
+/*
+Copyright 2025 The Flux authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package cache
+
+type InvolvedObject struct {
+	Kind      string
+	Name      string
+	Namespace string
+}

cache/lru.go

@@ -17,8 +17,11 @@ limitations under the License.
 package cache
 
 import (
+	"context"
 	"fmt"
 	"sync"
+
+	"github.com/go-logr/logr"
 )
 
 // node is a node in a doubly linked list
@@ -124,6 +127,83 @@ func (c *LRU[T]) Set(key string, value T) error {
 	return nil
 }
 
+// GetIfOrSet returns an item in the cache for the given key if present and
+// if the condition is satisfied, or calls the fetch function to get a new
+// item and stores it in the cache. The operation is thread-safe and atomic.
+// The boolean return value indicates whether the item was retrieved from
+// the cache.
+func (c *LRU[T]) GetIfOrSet(ctx context.Context,
+	key string,
+	condition func(T) bool,
+	fetch func(context.Context) (T, error),
+	opts ...Options,
+) (value T, ok bool, err error) {
+
+	var evicted bool
+
+	c.mu.Lock()
+	defer func() {
+		c.mu.Unlock()
+
+		var o storeOptions
+		o.apply(opts...)
+
+		// Record metrics.
+		status := StatusSuccess
+		event := CacheEventTypeMiss
+		switch {
+		case ok:
+			event = CacheEventTypeHit
+		case evicted:
+			recordEviction(c.metrics)
+		case err == nil:
+			recordItemIncrement(c.metrics)
+		default:
+			status = StatusFailure
+		}
+		recordRequest(c.metrics, status)
+		if obj := o.involvedObject; obj != nil {
+			c.RecordCacheEvent(event, obj.Kind, obj.Name, obj.Namespace)
+		}
+
+		// Print debug logs. The involved object should already be set in the context logger.
+		switch l := logr.FromContextOrDiscard(ctx).V(1).WithValues("key", key); {
+		case err != nil:
+			l.Info("item refresh failed", "error", err)
+		case !ok:
+			l := l
+			if o.debugKey != "" {
+				l = l.WithValues(o.debugKey, o.debugValueFunc(value))
+			}
+			l.Info("item refreshed")
+		}
+	}()
+
+	var curNode *node[T]
+	curNode, ok = c.cache[key]
+
+	if ok {
+		c.delete(curNode)
+		if condition(curNode.value) {
+			_ = c.add(curNode)
+			value = curNode.value
+			return
+		}
+		ok = false
+	}
+
+	value, err = fetch(ctx)
+	if err != nil {
+		var zero T
+		value = zero
+		return
+	}
+
+	evicted = c.add(&node[T]{key: key, value: value})
+
+	return
+}
+
 func (c *LRU[T]) add(node *node[T]) (evicted bool) {
 	prev := c.tail.prev
 	prev.addNext(node)

cache/lru_test.go

@@ -17,6 +17,7 @@ limitations under the License.
 package cache
 
 import (
+	"context"
 	"fmt"
 	"math/rand/v2"
 	"sync"
@@ -326,3 +327,96 @@ func TestLRU_int(t *testing.T) {
 	g.Expect(err).To(Succeed())
 	g.Expect(got).To(Equal(4))
 }
+
+func TestLRU_GetIfOrSet(t *testing.T) {
+	for _, tt := range []struct {
+		name          string
+		cap           int
+		seed          map[string]int
+		key           string
+		condition     func(int) bool
+		fetch         func(context.Context) (int, error)
+		expectedValue int
+		expectedOk    bool
+		expectedErr   string
+	}{
+		{
+			name:          "cache hit",
+			cap:           1,
+			seed:          map[string]int{"key": 42},
+			key:           "key",
+			condition:     func(int) bool { return true },
+			expectedValue: 42,
+			expectedOk:    true,
+		},
+		{
+			name:        "cache hit but condition not satisfied, refresh fails",
+			cap:         1,
+			seed:        map[string]int{"key": 42},
+			key:         "key",
+			condition:   func(int) bool { return false },
+			fetch:       func(context.Context) (int, error) { return 0, fmt.Errorf("failed") },
+			expectedErr: "failed",
+		},
+		{
+			name:          "cache hit but condition not satisfied, refresh succeeds",
+			cap:           1,
+			seed:          map[string]int{"key": 42},
+			key:           "key",
+			condition:     func(int) bool { return false },
+			fetch:         func(context.Context) (int, error) { return 53, nil },
+			expectedValue: 53,
+			expectedOk:    false,
+		},
+		{
+			name:        "cache miss, refresh fails",
+			cap:         1,
+			seed:        map[string]int{},
+			key:         "key",
+			fetch:       func(context.Context) (int, error) { return 0, fmt.Errorf("failed") },
+			expectedErr: "failed",
+		},
+		{
+			name:          "cache miss, refresh succeeds",
+			cap:           1,
+			seed:          map[string]int{},
+			key:           "key",
+			fetch:         func(context.Context) (int, error) { return 42, nil },
+			expectedValue: 42,
+			expectedOk:    false,
+		},
+		{
+			name:          "cache miss, refresh succeeds, key evicted",
+			cap:           1,
+			seed:          map[string]int{"key": 42},
+			key:           "key2",
+			fetch:         func(context.Context) (int, error) { return 53, nil },
+			expectedValue: 53,
+			expectedOk:    false,
+		},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			g := NewWithT(t)
+
+			cache, err := NewLRU[int](tt.cap)
+			g.Expect(err).NotTo(HaveOccurred())
+
+			for k, v := range tt.seed {
+				g.Expect(cache.Set(k, v)).To(Succeed())
+			}
+
+			ctx := context.Background()
+			value, ok, err := cache.GetIfOrSet(ctx, tt.key, tt.condition, tt.fetch)
+
+			if tt.expectedErr == "" {
+				g.Expect(err).NotTo(HaveOccurred())
+				g.Expect(value).To(Equal(tt.expectedValue))
+				g.Expect(ok).To(Equal(tt.expectedOk))
+			} else {
+				g.Expect(err).To(MatchError(tt.expectedErr))
+				g.Expect(value).To(BeZero())
+				g.Expect(ok).To(BeFalse())
+			}
+		})
+	}
+}

cache/store.go

@@ -44,9 +44,21 @@ type Expirable[T any] interface {
 }
 
 type storeOptions struct {
-	interval      time.Duration
-	registerer    prometheus.Registerer
-	metricsPrefix string
+	interval       time.Duration
+	registerer     prometheus.Registerer
+	metricsPrefix  string
+	involvedObject *InvolvedObject
+	debugKey       string
+	debugValueFunc func(any) any
+}
+
+func (o *storeOptions) apply(opts ...Options) error {
+	for _, opt := range opts {
+		if err := opt(o); err != nil {
+			return err
+		}
+	}
+	return nil
 }
 
 // Options is a function that sets the store options.
@@ -75,3 +87,15 @@ func WithMetricsPrefix(prefix string) Options {
 		return nil
 	}
 }
+
+// WithInvolvedObject sets the involved object for the cache metrics.
+func WithInvolvedObject(kind, name, namespace string) Options {
+	return func(o *storeOptions) error {
+		o.involvedObject = &InvolvedObject{
+			Kind:      kind,
+			Name:      name,
+			Namespace: namespace,
+		}
+		return nil
+	}
+}