refactor key function and ensure items always carry strings

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

Author: Per Goncalves da Silva

pkg/lib/queueinformer/config.go

@@ -82,24 +82,13 @@ func (c *queueInformerConfig) validateQueue() (err error) {
 
 func defaultKeyFunc(obj interface{}) (string, bool) {
 	// Get keys nested in resource events up to depth 2
-	keyable := false
-	for d := 0; d < 2 && !keyable; d++ {
-		switch v := obj.(type) {
-		case string:
-			return v, true
-		case kubestate.ResourceEvent:
-			obj = v.Resource()
-		default:
-			keyable = true
-		}
+	switch v := obj.(type) {
+	case string:
+		return v, true
+	default:
+		k, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj)
+		return k, err == nil
 	}
-
-	k, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj)
-	if err != nil {
-		return k, false
-	}
-
-	return k, true
 }
 
 func defaultConfig() *queueInformerConfig {

pkg/lib/queueinformer/queueinformer.go

@@ -35,29 +35,23 @@ func (q *QueueInformer) Sync(ctx context.Context, event kubestate.ResourceEvent)
 	return q.syncer.Sync(ctx, event)
 }
 
-// Enqueue adds a key to the queue. If obj is a key already it gets added directly.
+// enqueue adds a key to the queue. If obj is a key already it gets added directly.
 // Otherwise, the key is extracted via keyFunc.
-func (q *QueueInformer) Enqueue(event kubestate.ResourceEvent) {
-	if event == nil {
+func (q *QueueInformer) enqueue(eventType kubestate.ResourceEventType, obj interface{}) {
+	if obj == nil {
 		// Don't enqueue nil events
 		return
 	}
 
-	resource := event.Resource()
-	if event.Type() == kubestate.ResourceDeleted {
-		// Get object from tombstone if possible
-		if tombstone, ok := resource.(cache.DeletedFinalStateUnknown); ok {
-			resource = tombstone
-		}
-	} else {
-		// Extract key for add and update events
-		if key, ok := q.key(resource); ok {
-			resource = key
-		}
+	// Extract key for add and update events
+	key, ok := q.key(obj)
+	if !ok {
+		q.logger.Warnf("object %v not keyable: %v", obj, key)
+		return
 	}
 
 	// Create new resource event and add to queue
-	e := kubestate.NewResourceEvent(event.Type(), resource)
+	e := kubestate.NewResourceEvent(eventType, key)
 	q.logger.WithField("event", e).Trace("enqueuing resource event")
 	q.queue.Add(e)
 }
@@ -72,13 +66,16 @@ func (q *QueueInformer) key(obj interface{}) (string, bool) {
 func (q *QueueInformer) resourceHandlers(ctx context.Context) *cache.ResourceEventHandlerFuncs {
 	return &cache.ResourceEventHandlerFuncs{
 		AddFunc: func(obj interface{}) {
-			q.Enqueue(kubestate.NewResourceEvent(kubestate.ResourceUpdated, obj))
+			q.enqueue(kubestate.ResourceUpdated, obj)
 		},
 		UpdateFunc: func(oldObj, newObj interface{}) {
-			q.Enqueue(kubestate.NewResourceEvent(kubestate.ResourceUpdated, newObj))
+			q.enqueue(kubestate.ResourceUpdated, newObj)
 		},
 		DeleteFunc: func(obj interface{}) {
-			q.Enqueue(kubestate.NewResourceEvent(kubestate.ResourceDeleted, obj))
+			if tombstone, ok := obj.(cache.DeletedFinalStateUnknown); ok {
+				obj = tombstone
+			}
+			q.enqueue(kubestate.ResourceDeleted, obj)
 		},
 	}
 }

pkg/lib/queueinformer/queueinformer_operator.go

@@ -273,39 +273,39 @@ func (o *operator) processNextWorkItem(ctx context.Context, loop *QueueInformer)
 	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")
+	if _, ok := item.Resource().(string); !ok {
+		panic(fmt.Sprintf("unexpected item resource type: %T", item.Resource()))
+	}
+
+	key, keyable := loop.key(item.Resource())
+	if !keyable {
+		logger.WithField("item", item).Warn("could not form key")
+		queue.Forget(item)
+		return true
+	}
+
+	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
 		}
-
-		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
-			}
+		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)
 	}
+	event := kubestate.NewResourceEvent(item.Type(), resource)
 
 	// Sync and requeue on error (throw out failed deletion syncs)
 	err := loop.Sync(ctx, event)