primitive stampede support

Author: grokspawn

staging/operator-lifecycle-manager/pkg/controller/registry/resolver/cache/cache.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"slices"
 	"sort"
 	"sync"
 	"time"
@@ -147,7 +148,7 @@ func (c *NamespacedOperatorCache) Error() error {
 
 func (c *Cache) Namespaced(namespaces ...string) MultiCatalogOperatorFinder {
 	const (
-		CachePopulateTimeout = time.Minute
+		cachePopulateTimeout = time.Minute
 	)
 
 	sources := c.sp.Sources(namespaces...)
@@ -209,13 +210,27 @@ func (c *Cache) Namespaced(namespaces ...string) MultiCatalogOperatorFinder {
 	}
 	misses = misses[found:]
 
+	// remove any with a "live" outstanding request
+	misses = slices.DeleteFunc(misses, func(key SourceKey) bool {
+		hdr, _ := c.snapshots[key]
+
+		// if we already have a request timestamp, we have an outstanding request, so prevent stacking
+		// and just send new requests if the previous one has expired
+		if hdr.RequestSentinelActive() {
+			c.logger.Printf("Skipping new request for %s, already in progress", key)
+			return true
+		}
+		return false
+	})
+
 	for _, miss := range misses {
-		ctx, cancel := context.WithTimeout(context.Background(), CachePopulateTimeout)
+		ctx, cancel := context.WithTimeout(context.Background(), cachePopulateTimeout)
 
 		hdr := snapshotHeader{
-			key:      miss,
-			pop:      cancel,
-			priority: c.sourcePriorityProvider.Priority(miss),
+			key:             miss,
+			pop:             cancel,
+			priority:        c.sourcePriorityProvider.Priority(miss),
+			requestSentinel: time.Now().Add(cachePopulateTimeout), // set sentinel to prevent stacking requests
 		}
 
 		hdr.m.Lock()
@@ -228,6 +243,7 @@ func (c *Cache) Namespaced(namespaces ...string) MultiCatalogOperatorFinder {
 			defer func() { <-c.sem }()
 			if snapshot, err := source.Snapshot(ctx); err != nil {
 				hdr.err = err
+				// don't adjust request timestamp, so that we don't stampede sources
 			} else if snapshot != nil {
 				hdr.snapshot = snapshot
 			} else {
@@ -294,6 +310,8 @@ type snapshotHeader struct {
 	pop      context.CancelFunc
 	err      error
 	priority int
+
+	requestSentinel time.Time
 }
 
 func (hdr *snapshotHeader) Cancel() {
@@ -314,6 +332,12 @@ func (hdr *snapshotHeader) Valid() bool {
 	return true
 }
 
+func (hdr *snapshotHeader) RequestSentinelActive() bool {
+	hdr.m.RLock()
+	defer hdr.m.RUnlock()
+	return time.Now().Before(hdr.requestSentinel)
+}
+
 type sortableSnapshots struct {
 	snapshots          []*snapshotHeader
 	preferredNamespace string

staging/operator-lifecycle-manager/pkg/controller/registry/resolver/source_registry.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"runtime/debug"
 	"sync"
 	"time"
 
@@ -76,14 +75,16 @@ type RegistrySourceProvider struct {
 	invalidator  *sourceInvalidator
 }
 
+const defaultCacheLifetime time.Duration = 30 * time.Minute
+
 func SourceProviderFromRegistryClientProvider(rcp RegistryClientProvider, catsrcLister v1alpha1listers.CatalogSourceLister, logger logrus.StdLogger) *RegistrySourceProvider {
 	return &RegistrySourceProvider{
 		rcp:          rcp,
 		logger:       logger,
 		catsrcLister: catsrcLister,
 		invalidator: &sourceInvalidator{
 			validChans: make(map[cache.SourceKey]chan struct{}),
-			ttl:        30 * time.Minute,
+			ttl:        defaultCacheLifetime,
 		},
 	}
 }
@@ -153,9 +154,6 @@ func (s *registrySource) Snapshot(ctx context.Context) (*cache.Snapshot, error)
 	// or embed the information into Bundle.
 	packages := make(map[string]*api.Package)
 
-	s.logger.Printf("JEK >>>>> Listing packages for %#v", s.key)
-	debug.PrintStack()
-
 	it, err := s.client.ListBundles(ctx)
 	if err != nil {
 		return nil, fmt.Errorf("failed to list bundles: %w", err)