add built-in ratelimiter to workqueue

pkg/util/workqueue/rate_limitting_queue.go

@@ -0,0 +1,61 @@
+/*
+Copyright 2016 The Kubernetes Authors All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package workqueue
+
+// RateLimitingInterface is an Interface that can Add an item at a later time.  This makes it easier to
+// requeue items after failures without ending up in a hot-loop.
+type RateLimitingInterface interface {
+	DelayingInterface
+	// AddRateLimited adds an item to the workqueue after the rate limiter says its ok
+	AddRateLimited(item interface{})
+
+	// Forget indicates that an item is finished being retried.  Doesn't matter whether its for perm failing
+	// or for success, we'll stop the rate limiter from tracking it.  This only clears the `rateLimiter`, you
+	// still have to call `Done` on the queue.
+	Forget(item interface{})
+	// NumRequeues returns back how many times the item was requeued
+	NumRequeues(item interface{}) int
+}
+
+// NewRateLimitingQueue constructs a new workqueue with rateLimited queuing ability
+// Remember to call Forget!  If you don't, you may end up tracking failures forever.
+func NewRateLimitingQueue(rateLimiter RateLimiter) RateLimitingInterface {
+	return &rateLimitingType{
+		DelayingInterface: NewDelayingQueue(),
+		rateLimiter:       rateLimiter,
+	}
+}
+
+// rateLimitingType wraps an Interface and provides rateLimited re-enquing
+type rateLimitingType struct {
+	DelayingInterface
+
+	rateLimiter RateLimiter
+}
+
+// AddRateLimited AddAfter's the item based on the time when the rate limiter says its ok
+func (q *rateLimitingType) AddRateLimited(item interface{}) {
+	q.DelayingInterface.AddAfter(item, q.rateLimiter.When(item))
+}
+
+func (q *rateLimitingType) NumRequeues(item interface{}) int {
+	return q.rateLimiter.NumRequeues(item)
+}
+
+func (q *rateLimitingType) Forget(item interface{}) {
+	q.rateLimiter.Forget(item)
+}