Add QueueLength Metric

Author: JoelSpeed

pkg/internal/controller/controller.go

@@ -30,6 +30,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/cache"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
+	ctrlmetrics "sigs.k8s.io/controller-runtime/pkg/internal/controller/metrics"
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 	"sigs.k8s.io/controller-runtime/pkg/runtime/inject"
@@ -171,6 +172,9 @@ func (c *Controller) Start(stop <-chan struct{}) error {
 func (c *Controller) processNextWorkItem() bool {
 	// This code copy-pasted from the sample-Controller.
 
+	// Update metrics after processing each item
+	defer c.updateMetrics()
+
 	obj, shutdown := c.Queue.Get()
 	if obj == nil {
 		// Sometimes the Queue gives us nil items when it starts up
@@ -233,3 +237,8 @@ func (c *Controller) InjectFunc(f inject.Func) error {
 	c.SetFields = f
 	return nil
 }
+
+// updateMetrics updates prometheus metrics within the controller
+func (c *Controller) updateMetrics() {
+	ctrlmetrics.QueueLength.WithLabelValues(c.Name).Set(float64(c.Queue.Len()))
+}

pkg/internal/controller/controller_test.go

@@ -23,6 +23,8 @@ import (
 
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
+	"github.com/prometheus/client_golang/prometheus"
+	dto "github.com/prometheus/client_model/go"
 	"k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/types"
@@ -31,6 +33,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/cache/informertest"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllertest"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
+	ctrlmetrics "sigs.k8s.io/controller-runtime/pkg/internal/controller/metrics"
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile/reconciletest"
@@ -404,6 +407,46 @@ var _ = Describe("controller", func() {
 		It("should create a new go routine for MaxConcurrentReconciles", func() {
 			// TODO(community): write this test
 		})
+
+		Context("should update prometheus metrics", func() {
+			It("should requeue a Request if there is an error and continue processing items", func(done Done) {
+				ctrlmetrics.QueueLength = prometheus.NewGaugeVec(prometheus.GaugeOpts{
+					Name: "controller_runtime_reconcile_queue_length",
+					Help: "Length of reconcile queue per controller",
+				}, []string{"controller"})
+
+				fakeReconcile.Err = fmt.Errorf("expected error: reconcile")
+				go func() {
+					defer GinkgoRecover()
+					Expect(ctrl.Start(stop)).NotTo(HaveOccurred())
+				}()
+				ctrl.Queue.Add(request)
+
+				// Reduce the jitterperiod so we don't have to wait a second before the reconcile function is rerun.
+				ctrl.JitterPeriod = time.Millisecond
+
+				By("Invoking Reconciler which will give an error")
+				Expect(<-reconciled).To(Equal(request))
+				var queueLength dto.Metric
+				Eventually(func() error {
+					ctrlmetrics.QueueLength.WithLabelValues(ctrl.Name).Write(&queueLength)
+					if queueLength.GetGauge().GetValue() != 1.0 {
+						return fmt.Errorf("metrics not updated")
+					}
+					return nil
+				}, 2.0).Should(Succeed())
+
+				By("Invoking Reconciler a second time without error")
+				fakeReconcile.Err = nil
+				Expect(<-reconciled).To(Equal(request))
+
+				By("Removing the item from the queue")
+				Eventually(ctrl.Queue.Len).Should(Equal(0))
+				Eventually(func() int { return ctrl.Queue.NumRequeues(request) }).Should(Equal(0))
+
+				close(done)
+			}, 2.0)
+		})
 	})
 })
 

pkg/internal/controller/metrics/metrics.go

@@ -0,0 +1,35 @@
+/*
+Copyright 2018 The Kubernetes Authors.
+
+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 metrics
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+	"sigs.k8s.io/controller-runtime/pkg/metrics"
+)
+
+var (
+	// QueueLength is a prometheus metric which counts the current reconcile
+	// queue length per controller
+	QueueLength = prometheus.NewGaugeVec(prometheus.GaugeOpts{
+		Name: "controller_runtime_reconcile_queue_length",
+		Help: "Length of reconcile queue per controller",
+	}, []string{"controller"})
+)
+
+func init() {
+	metrics.Registry.MustRegister(QueueLength)
+}