add safeguards around queue item resource types

Signed-off-by: Per Goncalves da Silva <[email protected]>

Author: Per Goncalves da Silva

pkg/controller/operators/catalog/operator.go

@@ -886,18 +886,16 @@ func (o *Operator) handleDeletion(obj interface{}) {
 func (o *Operator) handleCatSrcDeletion(obj interface{}) {
 	catsrc, ok := obj.(metav1.Object)
 	if !ok {
+		tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
 		if !ok {
-			tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
-			if !ok {
-				utilruntime.HandleError(fmt.Errorf("couldn't get object from tombstone %#v", obj))
-				return
-			}
+			utilruntime.HandleError(fmt.Errorf("couldn't get object from tombstone %#v", obj))
+			return
+		}
 
-			catsrc, ok = tombstone.Obj.(metav1.Object)
-			if !ok {
-				utilruntime.HandleError(fmt.Errorf("tombstone contained object that is not a Namespace %#v", obj))
-				return
-			}
+		catsrc, ok = tombstone.Obj.(metav1.Object)
+		if !ok {
+			utilruntime.HandleError(fmt.Errorf("tombstone contained object that is not a Namespace %#v", obj))
+			return
 		}
 	}
 	sourceKey := registry.CatalogKey{Name: catsrc.GetName(), Namespace: catsrc.GetNamespace()}

pkg/lib/queueinformer/queueinformer.go

@@ -44,12 +44,12 @@ func (q *QueueInformer) Enqueue(event kubestate.ResourceEvent) {
 	}
 
 	resource := event.Resource()
-	if event.Type() == kubestate.ResourceDeleted {
-		// Get object from tombstone if possible
-		if tombstone, ok := resource.(cache.DeletedFinalStateUnknown); ok {
-			resource = tombstone
-		}
-	} else {
+
+	// Delete operations should always carry either something assignable to
+	// metev1.Object or cache.DeletedFinalStateUnknown.
+	// Add/Update events coming from the informer should have their resource
+	// converted to a key (string) before being enqueued.
+	if event.Type() != kubestate.ResourceDeleted {
 		// Extract key for add and update events
 		if key, ok := q.key(resource); ok {
 			resource = key
@@ -69,7 +69,7 @@ func (q *QueueInformer) key(obj interface{}) (string, bool) {
 
 // resourceHandlers provides the default implementation for responding to events
 // these simply Log the event and add the object's key to the queue for later processing.
-func (q *QueueInformer) resourceHandlers(ctx context.Context) *cache.ResourceEventHandlerFuncs {
+func (q *QueueInformer) resourceHandlers(_ context.Context) *cache.ResourceEventHandlerFuncs {
 	return &cache.ResourceEventHandlerFuncs{
 		AddFunc: func(obj interface{}) {
 			q.Enqueue(kubestate.NewResourceEvent(kubestate.ResourceUpdated, obj))
@@ -104,25 +104,6 @@ func (q *QueueInformer) metricHandlers() *cache.ResourceEventHandlerFuncs {
 	}
 }
 
-func NewQueue(ctx context.Context, options ...Option) (*QueueInformer, error) {
-	config := defaultConfig()
-	config.apply(options)
-
-	if err := config.validateQueue(); err != nil {
-		return nil, err
-	}
-
-	queue := &QueueInformer{
-		MetricsProvider: config.provider,
-		logger:          config.logger,
-		queue:           config.queue,
-		keyFunc:         config.keyFunc,
-		syncer:          config.syncer,
-	}
-
-	return queue, nil
-}
-
 // NewQueueInformer returns a new QueueInformer configured with options.
 func NewQueueInformer(ctx context.Context, options ...Option) (*QueueInformer, error) {
 	// Get default config and apply given options

pkg/lib/queueinformer/queueinformer_operator.go

@@ -3,6 +3,7 @@ package queueinformer
 import (
 	"context"
 	"fmt"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sync"
 	"time"
 
@@ -270,41 +271,53 @@ func (o *operator) processNextWorkItem(ctx context.Context, loop *QueueInformer)
 	}
 	defer queue.Done(item)
 
+	// As per the 'validateQueueInformer' function in config.go in this package
+	// a valid queue informer must have either an indexer or an informer defined.
+	// As per the 'complete' function in the same file, if an indexer is not defined
+	// the informer indexer will be used.
+	// NewQueueInformer calls complete after applying the options. Therefore,
+	// an indexer should always be present.
+	if loop.indexer == nil {
+		panic("invalid queueinformer: no indexer defined")
+	}
+
 	logger := o.logger.WithField("item", item)
 	logger.WithField("queue-length", queue.Len()).Trace("popped queue")
 
 	var event = item
+
 	if item.Type() != kubestate.ResourceDeleted {
-		// Get the key
-		key, keyable := loop.key(item)
-		if !keyable {
-			logger.WithField("item", item).Warn("could not form key")
-			queue.Forget(item)
-			return true
+		// Get the key - add/update resource operations must always contain a string resource
+		key, ok := item.Resource().(string)
+		if !ok {
+			panic(fmt.Sprintf("unexpected item resource type: %T for add/update operation", item.Resource()))
 		}
 
 		logger = logger.WithField("cache-key", key)
 
 		var resource interface{}
-		if loop.indexer == nil {
-			resource = item.Resource()
-		} else {
-			// Get the current cached version of the resource
-			var exists bool
-			var err error
-			resource, exists, err = loop.indexer.GetByKey(key)
-			if err != nil {
-				logger.WithError(err).Error("cache get failed")
-				queue.Forget(item)
-				return true
-			}
-			if !exists {
-				logger.WithField("existing-cache-keys", loop.indexer.ListKeys()).Debug("cache get failed, key not in cache")
-				queue.Forget(item)
-				return true
-			}
+
+		// Get the current cached version of the resource
+		resource, exists, err := loop.indexer.GetByKey(key)
+		if err != nil {
+			logger.WithError(err).Error("cache get failed")
+			queue.Forget(item)
+			return true
+		}
+		if !exists {
+			logger.WithField("existing-cache-keys", loop.indexer.ListKeys()).Debug("cache get failed, key not in cache")
+			queue.Forget(item)
+			return true
 		}
+
 		event = kubestate.NewResourceEvent(item.Type(), resource)
+	} else {
+		switch item.Resource().(type) {
+		case client.Object:
+		case cache.DeletedFinalStateUnknown:
+		default:
+			panic(fmt.Sprintf("unexpected item resource type: %T for delete operation", item.Resource()))
+		}
 	}
 
 	// Sync and requeue on error (throw out failed deletion syncs)