Create task queues with a name so that usage metrics can be tracked by Prometheus

pkg/controller/controller.go

@@ -124,7 +124,7 @@ func NewLoadBalancerController(
 	}
 	lbc.ingSyncer = ingsync.NewIngressSyncer(&lbc)
 
-	lbc.ingQueue = utils.NewPeriodicTaskQueue("ingresses", lbc.sync)
+	lbc.ingQueue = utils.NewPeriodicTaskQueue("ingress", "ingresses", lbc.sync)
 
 	// Ingress event handlers.
 	ctx.IngressInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{

pkg/controller/node.go

@@ -42,7 +42,7 @@ func NewNodeController(ctx *context.ControllerContext, instancePool instances.No
 		lister:       ctx.NodeInformer.GetIndexer(),
 		instancePool: instancePool,
 	}
-	c.queue = utils.NewPeriodicTaskQueue("nodes", c.sync)
+	c.queue = utils.NewPeriodicTaskQueue("", "nodes", c.sync)
 
 	ctx.NodeInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
 		AddFunc: func(obj interface{}) {

pkg/firewalls/controller.go

@@ -67,7 +67,7 @@ func NewFirewallController(
 		hasSynced:    ctx.HasSynced,
 	}
 
-	fwc.queue = utils.NewPeriodicTaskQueue("firewall", fwc.sync)
+	fwc.queue = utils.NewPeriodicTaskQueue("", "firewall", fwc.sync)
 
 	// Ingress event handlers.
 	ctx.IngressInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{

pkg/utils/taskqueue.go

@@ -90,18 +90,25 @@ func (t *PeriodicTaskQueue) Shutdown() {
 }
 
 // NewPeriodicTaskQueue creates a new task queue with the default rate limiter.
-func NewPeriodicTaskQueue(resource string, syncFn func(string) error) *PeriodicTaskQueue {
+func NewPeriodicTaskQueue(name, resource string, syncFn func(string) error) *PeriodicTaskQueue {
 	rl := workqueue.DefaultControllerRateLimiter()
-	return NewPeriodicTaskQueueWithLimiter(resource, syncFn, rl)
+	return NewPeriodicTaskQueueWithLimiter(name, resource, syncFn, rl)
 }
 
 // NewPeriodicTaskQueueWithLimiter creates a new task queue with the given sync function
 // and rate limiter. The sync function is called for every element inserted into the queue.
-func NewPeriodicTaskQueueWithLimiter(resource string, syncFn func(string) error, rl workqueue.RateLimiter) *PeriodicTaskQueue {
+func NewPeriodicTaskQueueWithLimiter(name, resource string, syncFn func(string) error, rl workqueue.RateLimiter) *PeriodicTaskQueue {
+	var queue workqueue.RateLimitingInterface
+	if name == "" {
+		queue = workqueue.NewRateLimitingQueue(rl)
+	} else {
+		queue = workqueue.NewNamedRateLimitingQueue(rl, name)
+	}
+
 	return &PeriodicTaskQueue{
 		resource:   resource,
 		keyFunc:    KeyFunc,
-		queue:      workqueue.NewRateLimitingQueue(rl),
+		queue:      queue,
 		sync:       syncFn,
 		workerDone: make(chan struct{}),
 	}

pkg/utils/taskqueue_test.go

@@ -42,7 +42,7 @@ func TestPeriodicTaskQueue(t *testing.T) {
 		}
 		return nil
 	}
-	tq = NewPeriodicTaskQueue("test", sync)
+	tq = NewPeriodicTaskQueue("", "test", sync)
 
 	go tq.Run()
 	tq.Enqueue(cache.ExplicitKey("a"))