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)) + }) })