Deduplicate registered informers by their underlying informers

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

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