Merge pull request #513 from mxinden/unit

main_test.go: Introduce overarching benchmark test

main.go

@@ -30,6 +30,7 @@ import (
 	"github.com/openshift/origin/pkg/util/proc"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
+	"k8s.io/client-go/informers"
 	clientset "k8s.io/client-go/kubernetes"
 	_ "k8s.io/client-go/plugin/pkg/client/auth"
 	"k8s.io/client-go/tools/clientcmd"
@@ -219,11 +220,20 @@ func metricsServer(registry prometheus.Gatherer, host string, port int) {
 // registerCollectors creates and starts informers and initializes and
 // registers metrics for collection.
 func registerCollectors(registry prometheus.Registerer, kubeClient clientset.Interface, enabledCollectors options.CollectorSet, namespaces options.NamespaceList, opts *options.Options) {
+	informerFactories := []informers.SharedInformerFactory{}
+	for _, ns := range namespaces {
+		informerFactories = append(
+			informerFactories,
+			informers.NewSharedInformerFactoryWithOptions(
+				kubeClient, 0, informers.WithNamespace(ns),
+			),
+		)
+	}
 	activeCollectors := []string{}
 	for c := range enabledCollectors {
 		f, ok := kcollectors.AvailableCollectors[c]
 		if ok {
-			f(registry, kubeClient, namespaces, opts)
+			f(registry, informerFactories, opts)
 			activeCollectors = append(activeCollectors, c)
 		}
 	}

main_test.go

@@ -0,0 +1,115 @@
+/*
+Copyright 2015 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 main
+
+import (
+	"net/http/httptest"
+	"strconv"
+	"testing"
+	"time"
+
+	"k8s.io/kube-state-metrics/pkg/options"
+
+	"k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes/fake"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+)
+
+func BenchmarkKubeStateMetrics(t *testing.B) {
+	kubeClient := fake.NewSimpleClientset()
+
+	if err := injectFixtures(kubeClient, 1000); err != nil {
+		t.Errorf("error injecting pod add: %v", err)
+	}
+
+	opts := options.NewOptions()
+	collectors := options.DefaultCollectors
+	namespaces := options.DefaultNamespaces
+
+	registry := prometheus.NewRegistry()
+	registerCollectors(registry, kubeClient, collectors, namespaces, opts)
+	handler := promhttp.HandlerFor(registry, promhttp.HandlerOpts{ErrorLog: promLogger{}})
+
+	req := httptest.NewRequest("GET", "http://localhost:8080/metrics", nil)
+
+	// Wait for informers to sync
+	time.Sleep(time.Second)
+
+	amountRequests := 10
+	for i := 0; i < amountRequests; i++ {
+		w := httptest.NewRecorder()
+		handler.ServeHTTP(w, req)
+	}
+}
+
+func injectFixtures(client *fake.Clientset, multiplier int) error {
+	creators := []func(*fake.Clientset, int) error{
+		configMap,
+		pod,
+	}
+
+	for _, c := range creators {
+		for i := 0; i < multiplier; i++ {
+			err := c(client, i)
+
+			if err != nil {
+				return err
+			}
+		}
+	}
+
+	return nil
+}
+
+func configMap(client *fake.Clientset, index int) error {
+	i := strconv.Itoa(index)
+
+	configMap := v1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:            "configmap" + i,
+			ResourceVersion: "123456",
+		},
+	}
+	_, err := client.CoreV1().ConfigMaps(metav1.NamespaceDefault).Create(&configMap)
+	return err
+}
+
+func pod(client *fake.Clientset, index int) error {
+	i := strconv.Itoa(index)
+
+	pod := v1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "pod" + i,
+		},
+		Status: v1.PodStatus{
+			ContainerStatuses: []v1.ContainerStatus{
+				v1.ContainerStatus{
+					Name:        "container1",
+					Image:       "k8s.gcr.io/hyperkube1",
+					ImageID:     "docker://sha256:aaa",
+					ContainerID: "docker://ab123",
+				},
+			},
+		},
+	}
+
+	_, err := client.CoreV1().Pods(metav1.NamespaceDefault).Create(&pod)
+	return err
+}

pkg/collectors/collectors.go

@@ -25,7 +25,7 @@ import (
 	"k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/runtime"
-	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/informers"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/kube-state-metrics/pkg/options"
@@ -53,7 +53,7 @@ var (
 	invalidLabelCharRE = regexp.MustCompile(`[^a-zA-Z0-9_]`)
 )
 
-var AvailableCollectors = map[string]func(registry prometheus.Registerer, kubeClient kubernetes.Interface, namespaces []string, opts *options.Options){
+var AvailableCollectors = map[string]func(registry prometheus.Registerer, informerFactories []informers.SharedInformerFactory, opts *options.Options){
 	"cronjobs":                 RegisterCronJobCollector,
 	"daemonsets":               RegisterDaemonSetCollector,
 	"deployments":              RegisterDeploymentCollector,

pkg/collectors/configmap.go

@@ -21,7 +21,8 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 	"golang.org/x/net/context"
 	"k8s.io/api/core/v1"
-	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/informers"
+	"k8s.io/client-go/tools/cache"
 	"k8s.io/kube-state-metrics/pkg/options"
 )
 
@@ -56,14 +57,14 @@ func (l ConfigMapLister) List() ([]v1.ConfigMap, error) {
 	return l()
 }
 
-func RegisterConfigMapCollector(registry prometheus.Registerer, kubeClient kubernetes.Interface, namespaces []string, opts *options.Options) {
-	client := kubeClient.CoreV1().RESTClient()
-	glog.Infof("collect configmap with %s", client.APIVersion())
-
-	cminfs := NewSharedInformerList(client, "configmaps", namespaces, &v1.ConfigMap{})
+func RegisterConfigMapCollector(registry prometheus.Registerer, informerFactories []informers.SharedInformerFactory, opts *options.Options) {
+	infs := SharedInformerList{}
+	for _, f := range informerFactories {
+		infs = append(infs, f.Core().V1().ConfigMaps().Informer().(cache.SharedInformer))
+	}
 
 	configMapLister := ConfigMapLister(func() (configMaps []v1.ConfigMap, err error) {
-		for _, cminf := range *cminfs {
+		for _, cminf := range infs {
 			for _, m := range cminf.GetStore().List() {
 				configMaps = append(configMaps, *m.(*v1.ConfigMap))
 			}
@@ -72,7 +73,7 @@ func RegisterConfigMapCollector(registry prometheus.Registerer, kubeClient kuber
 	})
 
 	registry.MustRegister(&configMapCollector{store: configMapLister, opts: opts})
-	cminfs.Run(context.Background().Done())
+	infs.Run(context.Background().Done())
 }
 
 type configMapStore interface {

pkg/collectors/cronjob.go

@@ -25,7 +25,8 @@ import (
 	"github.com/robfig/cron"
 	"golang.org/x/net/context"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/informers"
+	"k8s.io/client-go/tools/cache"
 
 	batchv1beta1 "k8s.io/api/batch/v1beta1"
 	"k8s.io/kube-state-metrics/pkg/options"
@@ -92,23 +93,24 @@ func (l CronJobLister) List() ([]batchv1beta1.CronJob, error) {
 	return l()
 }
 
-func RegisterCronJobCollector(registry prometheus.Registerer, kubeClient kubernetes.Interface, namespaces []string, opts *options.Options) {
-	client := kubeClient.BatchV1beta1().RESTClient()
-	glog.Infof("collect cronjob with %s", client.APIVersion())
+func RegisterCronJobCollector(registry prometheus.Registerer, informerFactories []informers.SharedInformerFactory, opts *options.Options) {
 
-	cjinfs := NewSharedInformerList(client, "cronjobs", namespaces, &batchv1beta1.CronJob{})
+	infs := SharedInformerList{}
+	for _, f := range informerFactories {
+		infs = append(infs, f.Batch().V1beta1().CronJobs().Informer().(cache.SharedInformer))
+	}
 
 	cronJobLister := CronJobLister(func() (cronjobs []batchv1beta1.CronJob, err error) {
-		for _, cjinf := range *cjinfs {
-			for _, c := range cjinf.GetStore().List() {
+		for _, inf := range infs {
+			for _, c := range inf.GetStore().List() {
 				cronjobs = append(cronjobs, *(c.(*batchv1beta1.CronJob)))
 			}
 		}
 		return cronjobs, nil
 	})
 
 	registry.MustRegister(&cronJobCollector{store: cronJobLister, opts: opts})
-	cjinfs.Run(context.Background().Done())
+	infs.Run(context.Background().Done())
 }
 
 type cronJobStore interface {

pkg/collectors/daemonset.go

@@ -21,7 +21,8 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 	"golang.org/x/net/context"
 	"k8s.io/api/extensions/v1beta1"
-	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/informers"
+	"k8s.io/client-go/tools/cache"
 	"k8s.io/kube-state-metrics/pkg/options"
 )
 
@@ -98,14 +99,15 @@ func (l DaemonSetLister) List() ([]v1beta1.DaemonSet, error) {
 	return l()
 }
 
-func RegisterDaemonSetCollector(registry prometheus.Registerer, kubeClient kubernetes.Interface, namespaces []string, opts *options.Options) {
-	client := kubeClient.ExtensionsV1beta1().RESTClient()
-	glog.Infof("collect daemonset with %s", client.APIVersion())
+func RegisterDaemonSetCollector(registry prometheus.Registerer, informerFactories []informers.SharedInformerFactory, opts *options.Options) {
 
-	dsinfs := NewSharedInformerList(client, "daemonsets", namespaces, &v1beta1.DaemonSet{})
+	infs := SharedInformerList{}
+	for _, f := range informerFactories {
+		infs = append(infs, f.Extensions().V1beta1().DaemonSets().Informer().(cache.SharedInformer))
+	}
 
 	dsLister := DaemonSetLister(func() (daemonsets []v1beta1.DaemonSet, err error) {
-		for _, dsinf := range *dsinfs {
+		for _, dsinf := range infs {
 			for _, c := range dsinf.GetStore().List() {
 				daemonsets = append(daemonsets, *(c.(*v1beta1.DaemonSet)))
 			}
@@ -114,7 +116,7 @@ func RegisterDaemonSetCollector(registry prometheus.Registerer, kubeClient kuber
 	})
 
 	registry.MustRegister(&daemonsetCollector{store: dsLister, opts: opts})
-	dsinfs.Run(context.Background().Done())
+	infs.Run(context.Background().Done())
 }
 
 type daemonsetStore interface {

pkg/collectors/deployment.go

@@ -22,7 +22,8 @@ import (
 	"golang.org/x/net/context"
 	"k8s.io/api/extensions/v1beta1"
 	"k8s.io/apimachinery/pkg/util/intstr"
-	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/informers"
+	"k8s.io/client-go/tools/cache"
 	"k8s.io/kube-state-metrics/pkg/options"
 )
 
@@ -118,14 +119,15 @@ func (l DeploymentLister) List() ([]v1beta1.Deployment, error) {
 	return l()
 }
 
-func RegisterDeploymentCollector(registry prometheus.Registerer, kubeClient kubernetes.Interface, namespaces []string, opts *options.Options) {
-	client := kubeClient.ExtensionsV1beta1().RESTClient()
-	glog.Infof("collect deployment with %s", client.APIVersion())
+func RegisterDeploymentCollector(registry prometheus.Registerer, informerFactories []informers.SharedInformerFactory, opts *options.Options) {
 
-	dinfs := NewSharedInformerList(client, "deployments", namespaces, &v1beta1.Deployment{})
+	infs := SharedInformerList{}
+	for _, f := range informerFactories {
+		infs = append(infs, f.Extensions().V1beta1().Deployments().Informer().(cache.SharedInformer))
+	}
 
 	dplLister := DeploymentLister(func() (deployments []v1beta1.Deployment, err error) {
-		for _, dinf := range *dinfs {
+		for _, dinf := range infs {
 			for _, c := range dinf.GetStore().List() {
 				deployments = append(deployments, *(c.(*v1beta1.Deployment)))
 			}
@@ -134,7 +136,7 @@ func RegisterDeploymentCollector(registry prometheus.Registerer, kubeClient kube
 	})
 
 	registry.MustRegister(&deploymentCollector{store: dplLister, opts: opts})
-	dinfs.Run(context.Background().Done())
+	infs.Run(context.Background().Done())
 }
 
 type deploymentStore interface {

pkg/collectors/endpoint.go

@@ -23,7 +23,8 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 
 	"k8s.io/api/core/v1"
-	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/informers"
+	"k8s.io/client-go/tools/cache"
 	"k8s.io/kube-state-metrics/pkg/options"
 )
 
@@ -74,14 +75,15 @@ func (l EndpointLister) List() ([]v1.Endpoints, error) {
 	return l()
 }
 
-func RegisterEndpointCollector(registry prometheus.Registerer, kubeClient kubernetes.Interface, namespaces []string, opts *options.Options) {
-	client := kubeClient.CoreV1().RESTClient()
-	glog.Infof("collect endpoint with %s", client.APIVersion())
+func RegisterEndpointCollector(registry prometheus.Registerer, informerFactories []informers.SharedInformerFactory, opts *options.Options) {
 
-	sinfs := NewSharedInformerList(client, "endpoints", namespaces, &v1.Endpoints{})
+	infs := SharedInformerList{}
+	for _, f := range informerFactories {
+		infs = append(infs, f.Core().V1().Endpoints().Informer().(cache.SharedInformer))
+	}
 
 	endpointLister := EndpointLister(func() (endpoints []v1.Endpoints, err error) {
-		for _, sinf := range *sinfs {
+		for _, sinf := range infs {
 			for _, m := range sinf.GetStore().List() {
 				endpoints = append(endpoints, *m.(*v1.Endpoints))
 			}
@@ -90,7 +92,7 @@ func RegisterEndpointCollector(registry prometheus.Registerer, kubeClient kubern
 	})
 
 	registry.MustRegister(&endpointCollector{store: endpointLister, opts: opts})
-	sinfs.Run(context.Background().Done())
+	infs.Run(context.Background().Done())
 }
 
 type endpointStore interface {