From 8ae773a0c8bc5d6a4afba788002beb8706e1a005 Mon Sep 17 00:00:00 2001 From: abd-ulbasit Date: Wed, 21 Oct 2026 21:41:03 +0500 Subject: [PATCH] fix(metrics): bound the instance metrics Collect call The instance manager calls the plugin Collect RPC while it serves the /metrics endpoint of the instance, and it does not set a deadline. If Collect never returns, the whole endpoint hangs, not only the plugin metrics. Give Collect its own 5 second deadline, so it returns an error instead. The instance manager then logs the error, counts it in cnpg_collector_collection_errors_total, and still serves the other metrics. The deadline alone is not enough. The caching client held its lock for the whole request to the API server, and a mutex does not honour a context. One slow Secret or ObjectStore read blocked every other cached read, including the one in Collect. Release the lock while the request is in flight. Two lookups of the same object can now run at the same time. Stamp each cache entry with the time its request was sent, and do not replace an entry that was fetched later, so a slow request cannot overwrite fresher data or keep old data past the TTL. Refs #1045 Assisted-by: Claude Opus 5.5 Signed-off-by: abd-ulbasit --- .../cnpgi/instance/internal/client/client.go | 69 +++++++++++++------ .../instance/internal/client/client_test.go | 69 +++++++++++++++++++ internal/cnpgi/instance/metrics.go | 20 ++++++ internal/cnpgi/instance/metrics_test.go | 30 ++++++++ 4 files changed, 168 insertions(+), 20 deletions(-) diff --git a/internal/cnpgi/instance/internal/client/client.go b/internal/cnpgi/instance/internal/client/client.go index 536f1dddc..2167bdea1 100644 --- a/internal/cnpgi/instance/internal/client/client.go +++ b/internal/cnpgi/instance/internal/client/client.go @@ -100,12 +100,36 @@ 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 } @@ -113,13 +137,9 @@ func (e *ExtendedClient) getCachedObject( 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/ @@ -127,31 +147,40 @@ func (e *ExtendedClient) getCachedObject( 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 diff --git a/internal/cnpgi/instance/internal/client/client_test.go b/internal/cnpgi/instance/internal/client/client_test.go index 48947a0bb..19beab545 100644 --- a/internal/cnpgi/instance/internal/client/client_test.go +++ b/internal/cnpgi/instance/internal/client/client_test.go @@ -20,6 +20,7 @@ SPDX-License-Identifier: Apache-2.0 package client import ( + "context" "time" corev1 "k8s.io/api/core/v1" @@ -27,6 +28,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" 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")) + }) }) diff --git a/internal/cnpgi/instance/metrics.go b/internal/cnpgi/instance/metrics.go index d21a49c0d..87f6fb446 100644 --- a/internal/cnpgi/instance/metrics.go +++ b/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") diff --git a/internal/cnpgi/instance/metrics_test.go b/internal/cnpgi/instance/metrics_test.go index 6976d812f..62fa5a659 100644 --- a/internal/cnpgi/instance/metrics_test.go +++ b/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)) + }) })