Implement EnqueueAfter on WorkerQueue

pkg/util/workerqueue/workerqueue.go

@@ -99,6 +99,20 @@ func (wq *WorkerQueue) EnqueueImmediately(obj interface{}) {
 	wq.queue.Add(key)
 }
 
+// EnqueueAfter delays an enqueuee operation by duration
+func (wq *WorkerQueue) EnqueueAfter(obj interface{}, duration time.Duration) {
+	var key string
+	var err error
+	if key, err = cache.MetaNamespaceKeyFunc(obj); err != nil {
+		err = errors.Wrap(err, "Error creating key for object")
+		runtime.HandleError(wq.logger.WithField("obj", obj), err)
+		return
+	}
+
+	wq.logger.WithField(wq.keyName, key).WithField("duration", duration).Info("Enqueueing after duration")
+	wq.queue.AddAfter(key, duration)
+}
+
 // runWorker is a long-running function that will continually call the
 // processNextWorkItem function in order to read and process a message on the
 // workqueue.

pkg/util/workerqueue/workerqueue_test.go

@@ -164,3 +164,32 @@ func TestWorkQueueHealthCheck(t *testing.T) {
 	assert.Error(t, wq.Healthy())
 	f(t, url, http.StatusServiceUnavailable)
 }
+
+func TestWorkerQueueEnqueueAfter(t *testing.T) {
+	t.Parallel()
+
+	updated := make(chan bool)
+	syncHandler := func(s string) error {
+		updated <- true
+		return nil
+	}
+	wq := NewWorkerQueue(syncHandler, logrus.WithField("source", "test"), "testKey", "test")
+	stop := make(chan struct{})
+	defer close(stop)
+
+	go wq.Run(1, stop)
+
+	wq.EnqueueAfter(cache.ExplicitKey("default/test"), 2*time.Second)
+
+	select {
+	case <-updated:
+		assert.FailNow(t, "should not be a result in queue yet")
+	case <-time.After(time.Second):
+	}
+
+	select {
+	case <-updated:
+	case <-time.After(2 * time.Second):
+		assert.Fail(t, "should have got a queue'd message by now")
+	}
+}