Add queue size prometheus metric

discovery/pkg/k8s/controller.go

@@ -30,6 +30,7 @@ import (
 const (
 	kubesystemNamespace = "kube-system"
 	kubesystemService   = "kubernetes"
+	clusterType         = "kubernetes"
 )
 
 // Controller receives notifications from the Kubernetes API and translates those
@@ -53,7 +54,7 @@ func NewController(log *logrus.Logger, gimbalKubeClient kubernetes.Interface, ku
 
 	c := &Controller{
 		Logger:          log,
-		syncqueue:       sync.NewQueue(log, clusterName, gimbalKubeClient, threadiness, metrics),
+		syncqueue:       sync.NewQueue(log, clusterName, clusterType, gimbalKubeClient, threadiness, metrics),
 		servicesSynced:  serviceInformer.Informer().HasSynced,
 		endpointsSynced: endpointsInformer.Informer().HasSynced,
 		clusterName:     clusterName,

discovery/pkg/metrics/metrics.go

@@ -110,7 +110,6 @@ func NewMetrics() DiscovererMetrics {
 func (d *DiscovererMetrics) RegisterPrometheus() {
 	// Register with Prometheus's default registry
 	for _, v := range d.metrics {
-
 		prometheus.MustRegister(v)
 	}
 }
@@ -154,3 +153,11 @@ func (d *DiscovererMetrics) EndpointsEventTimestampMetric(namespace, clusterName
 		m.WithLabelValues(namespace, clusterName, name).Set(float64(timestamp))
 	}
 }
+
+// QueueSizeGaugeMetric records the queue size prometheus metric
+func (d *DiscovererMetrics) QueueSizeGaugeMetric(namespace, clusterName, clusterType string, size int) {
+	m, ok := d.metrics[QueueSizeGauge].(*prometheus.GaugeVec)
+	if ok {
+		m.WithLabelValues(namespace, clusterName, clusterType).Set(float64(size))
+	}
+}

discovery/pkg/openstack/reconciler.go

@@ -29,6 +29,8 @@ import (
 	"k8s.io/client-go/kubernetes"
 )
 
+const clusterType = "openstack"
+
 type ProjectLister interface {
 	ListProjects() ([]projects.Project, error)
 }
@@ -69,7 +71,7 @@ func NewReconciler(clusterName string, gimbalKubeClient kubernetes.Interface, sy
 		ProjectLister:      projectLister,
 		Logger:             log,
 		Metrics:            metrics,
-		syncqueue:          sync.NewQueue(log, clusterName, gimbalKubeClient, queueWorkers, metrics),
+		syncqueue:          sync.NewQueue(log, clusterName, clusterType, gimbalKubeClient, queueWorkers, metrics),
 	}
 }
 

discovery/pkg/sync/endpoints.go

@@ -48,6 +48,11 @@ type endpointsAction struct {
 	endpoints *v1.Endpoints
 }
 
+// ObjectMeta returns the objectMeta piece of the Action interface object
+func (action endpointsAction) ObjectMeta() *metav1.ObjectMeta {
+	return &action.endpoints.ObjectMeta
+}
+
 // Sync performs the action on the given Endpoints resource
 func (action endpointsAction) Sync(kubeClient kubernetes.Interface, metrics localmetrics.DiscovererMetrics, clusterName string) error {
 

discovery/pkg/sync/queue.go

@@ -19,6 +19,7 @@ import (
 
 	localmetrics "github.com/heptio/gimbal/discovery/pkg/metrics"
 	"github.com/sirupsen/logrus"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/kubernetes"
@@ -40,10 +41,11 @@ type Queue struct {
 	Threadiness int
 	Metrics     localmetrics.DiscovererMetrics
 	ClusterName string
+	ClusterType string
 }
 
 // NewQueue returns an initialized sync.Queue for syncing resources with a Gimbal cluster.
-func NewQueue(logger *logrus.Logger, clusterName string, kubeClient kubernetes.Interface,
+func NewQueue(logger *logrus.Logger, clusterName, clusterType string, kubeClient kubernetes.Interface,
 	threadiness int, metrics localmetrics.DiscovererMetrics) Queue {
 	return Queue{
 		KubeClient:  kubeClient,
@@ -52,17 +54,20 @@ func NewQueue(logger *logrus.Logger, clusterName string, kubeClient kubernetes.I
 		Threadiness: threadiness,
 		Metrics:     metrics,
 		ClusterName: clusterName,
+		ClusterType: clusterType,
 	}
 }
 
 // Action that is added to the queue for processing
 type Action interface {
 	Sync(kube kubernetes.Interface, lm localmetrics.DiscovererMetrics, clusterName string) error
+	ObjectMeta() *metav1.ObjectMeta
 }
 
 // Enqueue adds a new resource action to the worker queue
 func (sq *Queue) Enqueue(action Action) {
 	sq.Workqueue.AddRateLimited(action)
+	sq.Metrics.QueueSizeGaugeMetric(action.ObjectMeta().GetNamespace(), sq.ClusterName, sq.ClusterType, sq.Workqueue.Len())
 }
 
 // Run starts the queue workers. It blocks until the stopCh is closed.

discovery/pkg/sync/service.go

@@ -48,6 +48,11 @@ type serviceAction struct {
 	service *v1.Service
 }
 
+// ObjectMeta returns the objectMeta piece of the Action interface object
+func (action serviceAction) ObjectMeta() *metav1.ObjectMeta {
+	return &action.service.ObjectMeta
+}
+
 // Sync performs the action on the given service
 func (action serviceAction) Sync(kubeClient kubernetes.Interface, metrics localmetrics.DiscovererMetrics, clusterName string) error {