From 87a32691907fc787f68bc433f866fb8479221fa1 Mon Sep 17 00:00:00 2001 From: "dependabot[bot]" <49699333+dependabot[bot]@users.noreply.github.com> Date: Fri, 13 Dec 2024 17:01:28 +0100 Subject: [PATCH 1/2] build(deps): bump golang.org/x/crypto from 0.30.0 to 0.31.0 (#3465) Bumps [golang.org/x/crypto](https://github.com/golang/crypto) from 0.30.0 to 0.31.0. - [Commits](https://github.com/golang/crypto/compare/v0.30.0...v0.31.0) --- updated-dependencies: - dependency-name: golang.org/x/crypto dependency-type: indirect ... Signed-off-by: dependabot[bot] Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> Signed-off-by: Per Goncalves da Silva --- .../registry/resolver/source_csvs.go | 26 +++++++++---------- 1 file changed, 13 insertions(+), 13 deletions(-) diff --git a/pkg/controller/registry/resolver/source_csvs.go b/pkg/controller/registry/resolver/source_csvs.go index bc7fe506cf..8a876272f9 100644 --- a/pkg/controller/registry/resolver/source_csvs.go +++ b/pkg/controller/registry/resolver/source_csvs.go @@ -93,19 +93,19 @@ func (s *csvSource) Snapshot(ctx context.Context) (*cache.Snapshot, error) { continue } - if cachedSubscription, ok := csvSubscriptions[csv]; !ok || cachedSubscription == nil { - // we might be in an incoherent state, so let's check with live clients to make sure - realSubscriptions, err := s.listSubscriptions(ctx) - if err != nil { - return nil, fmt.Errorf("failed to list subscriptions: %w", err) - } - for _, realSubscription := range realSubscriptions.Items { - if realSubscription.Status.InstalledCSV == csv.Name { - // oops, live cluster state is coherent - return nil, fmt.Errorf("lister caches incoherent for CSV %s/%s - found owning Subscription %s/%s", csv.Namespace, csv.Name, realSubscription.Namespace, realSubscription.Name) - } - } - } + //if cachedSubscription, ok := csvSubscriptions[csv]; !ok || cachedSubscription == nil { + // // we might be in an incoherent state, so let's check with live clients to make sure + // realSubscriptions, err := s.listSubscriptions(ctx) + // if err != nil { + // return nil, fmt.Errorf("failed to list subscriptions: %w", err) + // } + // for _, realSubscription := range realSubscriptions.Items { + // if realSubscription.Status.InstalledCSV == csv.Name { + // // oops, live cluster state is coherent + // return nil, fmt.Errorf("lister caches incoherent for CSV %s/%s - found owning Subscription %s/%s", csv.Namespace, csv.Name, realSubscription.Namespace, realSubscription.Name) + // } + // } + //} if failForwardEnabled { replacementChainEndsInFailure, err := isReplacementChainThatEndsInFailure(csv, ReplacementMapping(csvs)) From 4d188d95a8527269791bd28152375f1e50a09ed3 Mon Sep 17 00:00:00 2001 From: Per Goncalves da Silva Date: Thu, 19 Dec 2024 14:05:05 +0100 Subject: [PATCH 2/2] queue item fix Signed-off-by: Per Goncalves da Silva --- .../registry/resolver/source_csvs.go | 26 +++++++++---------- .../queueinformer/queueinformer_operator.go | 16 ++++++++---- 2 files changed, 24 insertions(+), 18 deletions(-) diff --git a/pkg/controller/registry/resolver/source_csvs.go b/pkg/controller/registry/resolver/source_csvs.go index 8a876272f9..bc7fe506cf 100644 --- a/pkg/controller/registry/resolver/source_csvs.go +++ b/pkg/controller/registry/resolver/source_csvs.go @@ -93,19 +93,19 @@ func (s *csvSource) Snapshot(ctx context.Context) (*cache.Snapshot, error) { continue } - //if cachedSubscription, ok := csvSubscriptions[csv]; !ok || cachedSubscription == nil { - // // we might be in an incoherent state, so let's check with live clients to make sure - // realSubscriptions, err := s.listSubscriptions(ctx) - // if err != nil { - // return nil, fmt.Errorf("failed to list subscriptions: %w", err) - // } - // for _, realSubscription := range realSubscriptions.Items { - // if realSubscription.Status.InstalledCSV == csv.Name { - // // oops, live cluster state is coherent - // return nil, fmt.Errorf("lister caches incoherent for CSV %s/%s - found owning Subscription %s/%s", csv.Namespace, csv.Name, realSubscription.Namespace, realSubscription.Name) - // } - // } - //} + if cachedSubscription, ok := csvSubscriptions[csv]; !ok || cachedSubscription == nil { + // we might be in an incoherent state, so let's check with live clients to make sure + realSubscriptions, err := s.listSubscriptions(ctx) + if err != nil { + return nil, fmt.Errorf("failed to list subscriptions: %w", err) + } + for _, realSubscription := range realSubscriptions.Items { + if realSubscription.Status.InstalledCSV == csv.Name { + // oops, live cluster state is coherent + return nil, fmt.Errorf("lister caches incoherent for CSV %s/%s - found owning Subscription %s/%s", csv.Namespace, csv.Name, realSubscription.Namespace, realSubscription.Name) + } + } + } if failForwardEnabled { replacementChainEndsInFailure, err := isReplacementChainThatEndsInFailure(csv, ReplacementMapping(csvs)) diff --git a/pkg/lib/queueinformer/queueinformer_operator.go b/pkg/lib/queueinformer/queueinformer_operator.go index ecdb4eb896..77795480fc 100644 --- a/pkg/lib/queueinformer/queueinformer_operator.go +++ b/pkg/lib/queueinformer/queueinformer_operator.go @@ -285,6 +285,16 @@ func (o *operator) processNextWorkItem(ctx context.Context, loop *QueueInformer) logger = logger.WithField("cache-key", key) + if !ok { + // we should really only (sync) resource event items + // otherwise we run the risk of processing the same object concurrently across the number of workers. + // As is, at least, the case for the namespace resolution informer, + // the namespace name gets added to the queue (e.g. nsResolveQueue.Add(obj.GetNamespace())), + // but the QueueInformer's Enqueue method always adds ResourceEvents. + queue.Add(kubestate.NewResourceEvent(kubestate.ResourceUpdated, key)) + return true + } + var resource interface{} if loop.indexer == nil { resource = event.Resource() @@ -305,11 +315,7 @@ func (o *operator) processNextWorkItem(ctx context.Context, loop *QueueInformer) } } - if !ok { - event = kubestate.NewResourceEvent(kubestate.ResourceUpdated, resource) - } else { - event = kubestate.NewResourceEvent(event.Type(), resource) - } + event = kubestate.NewResourceEvent(event.Type(), resource) } // Sync and requeue on error (throw out failed deletion syncs)