diff --git a/pkg/lib/queueinformer/queueinformer_operator.go b/pkg/lib/queueinformer/queueinformer_operator.go index 0c1ca168c3..1e601bbec8 100644 --- a/pkg/lib/queueinformer/queueinformer_operator.go +++ b/pkg/lib/queueinformer/queueinformer_operator.go @@ -142,9 +142,21 @@ func (o *operator) RegisterInformer(informer cache.SharedIndexInformer) error { } func (o *operator) registerInformer(informer cache.SharedIndexInformer) { - // never double-register an informer + // Never double-register an informer. + // + // Comparing interface identity alone is not sufficient: client-go's + // typed informer factories return a new typedSharedIndexInformer wrapper + // for every Informer() call around the same cached underlying informer, + // and two such wrappers of one informer compare unequal. Registering + // both would make RunInformers run the underlying informer twice, which + // races sharedIndexInformer's start bookkeeping and eventually panics + // with "close of closed channel". Deduplicate on the underlying + // informer instead: every sharedIndexInformer owns its indexer from + // construction, so informers backed by the same underlying informer + // report the same indexer. + indexer := informer.GetIndexer() for i := range o.informers { - if o.informers[i] == informer { + if o.informers[i] == informer || o.informers[i].GetIndexer() == indexer { return } } diff --git a/pkg/lib/queueinformer/queueinformer_operator_test.go b/pkg/lib/queueinformer/queueinformer_operator_test.go index 3801f2f4f9..d02041bc85 100644 --- a/pkg/lib/queueinformer/queueinformer_operator_test.go +++ b/pkg/lib/queueinformer/queueinformer_operator_test.go @@ -6,7 +6,9 @@ import ( "time" "github.com/pkg/errors" + corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/version" + "k8s.io/client-go/tools/cache" ) type versionFunc func() (*version.Info, error) @@ -92,3 +94,31 @@ func TestOperatorRunChannelClosure(t *testing.T) { }) } } + +func TestRegisterInformerDeduplicatesTypedWrappers(t *testing.T) { + // client-go's typed informer factories return a new + // typedSharedIndexInformer wrapper for every Informer() call around the + // same cached underlying informer, and two such wrappers of one + // informer compare unequal. Registering both must not enqueue the + // underlying informer twice, or RunInformers runs it twice and races + // sharedIndexInformer's start bookkeeping. + o, err := newOperatorFromConfig(defaultOperatorConfig()) + if err != nil { + t.Fatalf("could not create operator from default config: %s", err) + } + + underlying := cache.NewSharedIndexInformer(&cache.ListWatch{}, &corev1.Namespace{}, 0, cache.Indexers{}) + first := cache.NewTypedSharedIndexInformer[*corev1.Namespace](underlying) + second := cache.NewTypedSharedIndexInformer[*corev1.Namespace](underlying) + + if first == second { + t.Fatal("expected distinct wrapper instances around the same underlying informer") + } + + o.registerInformer(first) + o.registerInformer(second) + + if len(o.informers) != 1 { + t.Fatalf("expected the second wrapper of the same underlying informer to be deduplicated, got %d registered informers", len(o.informers)) + } +}