fix(metrics): bound the instance metrics Collect call

internal/cnpgi/instance/internal/client/client.go

@@ -100,58 +100,87 @@ func (e *ExtendedClient) getCachedObject(
 		WithValues("name", key.Name, "namespace", key.Namespace)
 
 	contextLogger.Trace("locking the cache")
+	found, err := e.loadFromCache(key, obj)
+	if found || err != nil {
+		if found {
+			contextLogger.Debug("object found, loaded it from cache")
+		}
+		return err
+	}
+
+	// The request to the API server is done without holding the cache
+	// lock. A slow or stuck request must only block its own caller,
+	// not every other lookup (e.g. the metrics Collect call).
+	// The TTL counts from when the request was sent.
+	fetchUnixTime := time.Now().Unix()
+	if err := e.Client.Get(ctx, key, obj, opts...); err != nil {
+		return err
+	}
+
+	contextLogger.Debug("setting object in the cache")
+	e.storeInCache(obj, fetchUnixTime)
+
+	return nil
+}
+
+// loadFromCache copies a non-expired cached entry matching key and the type
+// of obj into obj. It returns true if such an entry was found.
+func (e *ExtendedClient) loadFromCache(key client.ObjectKey, obj client.Object) (bool, error) {
 	e.mux.Lock()
 	defer e.mux.Unlock()
 
-	// check if in cache
-	expiredObjectIndex := -1
-	for idx, cacheEntry := range e.cachedObjects {
+	for _, cacheEntry := range e.cachedObjects {
 		if cacheEntry.entry.GetNamespace() != key.Namespace || cacheEntry.entry.GetName() != key.Name {
 			continue
 		}
 		if reflect.TypeOf(cacheEntry.entry) != reflect.TypeOf(obj) {
 			continue
 		}
 		if cacheEntry.isExpired() {
-			contextLogger.Trace("expired object found")
-			expiredObjectIndex = idx
-			break
+			return false, nil
 		}
 
-		contextLogger.Debug("object found, loading it from cache")
-
 		// Yes, this is a terrible hack, but that's exactly the way
 		// controller-runtime works.
 		// https://github.com/kubernetes-sigs/controller-runtime/blob/
 		// 717b32aede14c921d239cf1b974a11e718949865/pkg/cache/internal/cache_reader.go#L92
 		outVal := reflect.ValueOf(obj)
 		objVal := reflect.ValueOf(cacheEntry.entry)
 		if !objVal.Type().AssignableTo(outVal.Type()) {
-			return fmt.Errorf("cache had type %s, but %s was asked for", objVal.Type(), outVal.Type())
+			return false, fmt.Errorf("cache had type %s, but %s was asked for", objVal.Type(), outVal.Type())
 		}
 
 		reflect.Indirect(outVal).Set(reflect.Indirect(objVal))
-		return nil
+		return true, nil
 	}
 
-	if err := e.Client.Get(ctx, key, obj, opts...); err != nil {
-		return err
-	}
+	return false, nil
+}
 
+// storeInCache adds obj to the cache, replacing any entry with the same
+// namespace, name and type, unless that entry was fetched later
+func (e *ExtendedClient) storeInCache(obj client.Object, fetchUnixTime int64) {
 	cs := cachedEntry{
 		entry:         obj.DeepCopyObject().(client.Object),
-		fetchUnixTime: time.Now().Unix(),
+		fetchUnixTime: fetchUnixTime,
 		ttlSeconds:    DefaultTTLSeconds,
 	}
 
-	contextLogger.Debug("setting object in the cache")
-	if expiredObjectIndex != -1 {
-		e.cachedObjects[expiredObjectIndex] = cs
-	} else {
-		e.cachedObjects = append(e.cachedObjects, cs)
+	e.mux.Lock()
+	defer e.mux.Unlock()
+
+	for idx, cacheEntry := range e.cachedObjects {
+		if cacheEntry.entry.GetNamespace() == obj.GetNamespace() &&
+			cacheEntry.entry.GetName() == obj.GetName() &&
+			reflect.TypeOf(cacheEntry.entry) == reflect.TypeOf(obj) {
+			if cacheEntry.fetchUnixTime <= fetchUnixTime {
+				e.cachedObjects[idx] = cs
+			}
+			return
+		}
 	}
 
-	return nil
+	e.cachedObjects = append(e.cachedObjects, cs)
 }
 
 // removeObject ensures that a object is not present in the cache

internal/cnpgi/instance/internal/client/client_test.go

@@ -20,13 +20,15 @@ SPDX-License-Identifier: Apache-2.0
 package client
 
 import (
+	"context"
 	"time"
 
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+	"sigs.k8s.io/controller-runtime/pkg/client/interceptor"
 
 	barmancloudv1 "github.com/cloudnative-pg/plugin-barman-cloud/api/v1"
 
@@ -163,4 +165,71 @@ var _ = Describe("ExtendedClient Get", func() {
 			Expect(secretInClient.GetResourceVersion()).To(Equal("from cache"))
 			Expect(objectStore.GetResourceVersion()).To(Equal("from cache"))
 		})
+
+	It("does not block other lookups while a request to the API server is in flight", func(ctx SpecContext) {
+		requestStarted := make(chan struct{})
+		releaseRequest := make(chan struct{})
+		DeferCleanup(func() { close(releaseRequest) })
+
+		baseClient := fake.NewClientBuilder().
+			WithScheme(scheme).
+			WithObjects(secretInClient, objectStore).
+			WithInterceptorFuncs(interceptor.Funcs{
+				Get: func(
+					ctx context.Context,
+					c client.WithWatch,
+					key client.ObjectKey,
+					obj client.Object,
+					opts ...client.GetOption,
+				) error {
+					if _, isSecret := obj.(*corev1.Secret); isSecret {
+						// simulate a stuck request to the API server
+						close(requestStarted)
+						<-releaseRequest
+					}
+					return c.Get(ctx, key, obj, opts...)
+				},
+			}).Build()
+		extendedClient = NewExtendedClient(baseClient).(*ExtendedClient)
+
+		go func() {
+			defer GinkgoRecover()
+			_ = extendedClient.Get(context.Background(), client.ObjectKeyFromObject(secretInClient), &corev1.Secret{})
+		}()
+		Eventually(requestStarted).Should(BeClosed())
+
+		result := make(chan error, 1)
+		go func() {
+			result <- extendedClient.Get(ctx, client.ObjectKeyFromObject(objectStore), &barmancloudv1.ObjectStore{})
+		}()
+		Eventually(result).WithTimeout(2 * time.Second).Should(Receive(BeNil()))
+	})
+
+	It("keeps a newer cached entry when an older request returns", func(ctx SpecContext) {
+		baseClient := fake.NewClientBuilder().
+			WithScheme(scheme).
+			WithObjects(objectStore).
+			WithInterceptorFuncs(interceptor.Funcs{
+				Get: func(
+					ctx context.Context,
+					c client.WithWatch,
+					key client.ObjectKey,
+					obj client.Object,
+					opts ...client.GetOption,
+				) error {
+					// a request sent later caches its result while this one is in flight
+					addToCache(extendedClient, objectStore, time.Now().Unix()+1)
+					return c.Get(ctx, key, obj, opts...)
+				},
+			}).Build()
+		extendedClient = NewExtendedClient(baseClient).(*ExtendedClient)
+
+		err := extendedClient.Get(ctx, client.ObjectKeyFromObject(objectStore), &barmancloudv1.ObjectStore{})
+		Expect(err).NotTo(HaveOccurred())
+
+		var cached barmancloudv1.ObjectStore
+		err = extendedClient.Get(ctx, client.ObjectKeyFromObject(objectStore), &cached)
+		Expect(err).NotTo(HaveOccurred())
+		Expect(cached.GetResourceVersion()).To(Equal("from cache"))
+	})
 })

internal/cnpgi/instance/metrics.go

@@ -23,6 +23,7 @@ import (
 	"context"
 	"fmt"
 	"strings"
+	"time"
 
 	"github.com/cloudnative-pg/cnpg-i/pkg/metrics"
 	"github.com/cloudnative-pg/machinery/pkg/log"
@@ -36,10 +37,26 @@ import (
 // Sanitize the plugin name to be a valid Prometheus metric namespace
 var metricsDomain = strings.NewReplacer(".", "_", "-", "_").Replace(metadata.PluginName)
 
+// defaultMetricsCollectTimeout bounds a single Collect call. The instance
+// manager calls Collect while serving the /metrics endpoint of the
+// instance, so a Collect call that never returns would hang the whole
+// endpoint, not only the metrics of this plugin.
+const defaultMetricsCollectTimeout = 5 * time.Second
+
 type metricsImpl struct {
 	// important the client should be one with a underlying cache
 	Client client.Client
 	metrics.UnimplementedMetricsServer
+
+	// collectTimeout overrides defaultMetricsCollectTimeout when set
+	collectTimeout time.Duration
+}
+
+func (m metricsImpl) getCollectTimeout() time.Duration {
+	if m.collectTimeout > 0 {
+		return m.collectTimeout
+	}
+	return defaultMetricsCollectTimeout
 }
 
 func buildFqName(name string) string {
@@ -108,6 +125,9 @@ func (m metricsImpl) Collect(
 	contextLogger := log.FromContext(ctx)
 	contextLogger.Trace("metrics collect call received")
 
+	ctx, cancel := context.WithTimeout(ctx, m.getCollectTimeout())
+	defer cancel()
+
 	configuration, err := config.NewFromClusterJSON(req.ClusterDefinition)
 	if err != nil {
 		contextLogger.Error(err, "while creating configuration from cluster definition")

internal/cnpgi/instance/metrics_test.go

@@ -35,6 +35,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+	"sigs.k8s.io/controller-runtime/pkg/client/interceptor"
 )
 
 var _ = Describe("Metrics Collect method", func() {
@@ -139,4 +140,33 @@ var _ = Describe("Metrics Collect method", func() {
 		_, err := m.Collect(ctx, req)
 		Expect(err).To(HaveOccurred())
 	})
+
+	It("should return an error instead of hanging when the object store cannot be read", func() {
+		// Use a client whose requests never complete until the context is done
+		m.Client = fake.NewClientBuilder().
+			WithInterceptorFuncs(interceptor.Funcs{
+				Get: func(
+					ctx context.Context,
+					_ client.WithWatch,
+					_ client.ObjectKey,
+					_ client.Object,
+					_ ...client.GetOption,
+				) error {
+					<-ctx.Done()
+					return ctx.Err()
+				},
+			}).Build()
+		m.collectTimeout = 100 * time.Millisecond
+
+		result := make(chan error, 1)
+		go func() {
+			// the instance manager calls Collect without a deadline
+			_, err := m.Collect(context.Background(), req)
+			result <- err
+		}()
+
+		var err error
+		Eventually(result).WithTimeout(2 * time.Second).Should(Receive(&err))
+		Expect(err).To(MatchError(context.DeadlineExceeded))
+	})
 })