Merge pull request #55 from dgrisonnet/instrument-informers

Instrument informers

Author: dgrisonnet

cmd/kube-events-exporter/main.go

@@ -60,18 +60,18 @@ func main() {
 		klog.Fatalf("failed to create cluster config: %v", err)
 	}
 
+	exporterRegistry := prometheus.NewRegistry()
+	exporter.RegisterExporterCollectors(exporterRegistry)
+
 	eventRegistry := prometheus.NewRegistry()
-	eventCollector := collector.NewEventCollector(kubeClient, opts)
+	eventCollector := collector.NewEventCollector(kubeClient, exporterRegistry, opts)
 
 	stopCh := make(chan struct{})
 	defer close(stopCh)
 
 	eventCollector.Run(stopCh)
 	eventRegistry.MustRegister(eventCollector)
 
-	exporterRegistry := prometheus.NewRegistry()
-	exporter.RegisterExporterCollectors(exporterRegistry)
-
 	eventMux := http.NewServeMux()
 	exporterhttp.RegisterEventsMuxHandlers(eventMux, eventRegistry, exporterRegistry)
 	exporterMux := http.NewServeMux()

go.sum

@@ -85,7 +85,6 @@ github.com/hashicorp/golang-lru v0.5.1 h1:0hERBMJE1eitiLkihrMvRVBYAkpHzc/J3QdDN+
 github.com/hashicorp/golang-lru v0.5.1/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
 github.com/hpcloud/tail v1.0.0 h1:nfCOvKYfkgYP8hkirhJocXT2+zOD8yUNjXaWfTlyFKI=
 github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU=
-github.com/imdario/mergo v0.3.5 h1:JboBksRwiiAJWvIYJVo46AfV+IAIKZpfrSzVKj42R4Q=
 github.com/imdario/mergo v0.3.5/go.mod h1:2EnlNZ0deacrJVfApfmtdGgDfMuh/nq6Ok1EcJh5FfA=
 github.com/imdario/mergo v0.3.9 h1:UauaLniWCFHWd+Jp9oCEkTBj8VO/9DKg3PV3VCNMDIg=
 github.com/imdario/mergo v0.3.9/go.mod h1:2EnlNZ0deacrJVfApfmtdGgDfMuh/nq6Ok1EcJh5FfA=
@@ -189,7 +188,6 @@ golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190227155943-e225da77a7e6/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e h1:vcxGaoTs7kV8m5Np9uUNQin4BrLOthgV7252N8V+FwY=
 golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20170830134202-bb24a47a89ea/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=

internal/collector/event.go

@@ -22,119 +22,103 @@ import (
 
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/rhobs/kube-events-exporter/internal/options"
+	"github.com/rhobs/kube-events-exporter/pkg/informer"
 
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/client-go/informers"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/tools/cache"
 )
 
 // EventCollector is a prometeus.Collector that bundles all the metrics related
 // to Kubernetes Events.
 type EventCollector struct {
-	eventsTotal *prometheus.CounterVec
-
-	lock              sync.Mutex
-	informerFactories []informers.SharedInformerFactory
-	filter            eventFilter
+	kclient   kubernetes.Interface
+	metrics   *exporterMetrics
+	lock      sync.Mutex
+	filter    eventFilter
+	informers []cache.SharedIndexInformer
 }
 
 // NewEventCollector returns a prometheus.Collector collecting metrics about
 // Kubernetes Events.
-func NewEventCollector(kubeClient kubernetes.Interface, opts *options.Options) *EventCollector {
-	var factories []informers.SharedInformerFactory
-	for _, ns := range opts.InvolvedObjectNamespaces {
-		for _, eventType := range opts.EventTypes {
-			factories = append(factories, newFilteredInformerFactory(kubeClient, ns, eventType))
-		}
-	}
-
+func NewEventCollector(kclient kubernetes.Interface, exporterRegistry *prometheus.Registry, opts *options.Options) *EventCollector {
 	collector := &EventCollector{
-		eventsTotal: prometheus.NewCounterVec(prometheus.CounterOpts{
-			Name: "kube_events_total",
-			Help: "Count of all Kubernetes Events",
-		}, []string{"type", "involved_object_namespace", "involved_object_kind", "reason"}),
-
-		lock:              sync.Mutex{},
-		informerFactories: factories,
+		kclient: kclient,
+		lock:    sync.Mutex{},
 		filter: eventFilter{
 			creationTimestamp: time.Now(),
 			apiGroups:         opts.InvolvedObjectAPIGroups,
 			controllers:       opts.ReportingControllers,
 		},
+		metrics: newExporterMetrics(exporterRegistry),
 	}
 
-	collector.initInformers()
-
+	for _, ns := range opts.InvolvedObjectNamespaces {
+		for _, evType := range opts.EventTypes {
+			inf := collector.newEventInformer(ns, evType)
+			inf.AddEventHandler(collector.eventHandler())
+			collector.informers = append(collector.informers, inf)
+		}
+	}
 	return collector
 }
 
-func newFilteredInformerFactory(kubeClient kubernetes.Interface, ns, eventType string) informers.SharedInformerFactory {
-	return informers.NewFilteredSharedInformerFactory(
-		kubeClient,
-		0,
-		metav1.NamespaceAll,
-		func(list *metav1.ListOptions) {
-			filterInvolvedObjectNs(list, ns)
-			filterEventType(list, eventType)
-		},
-	)
-}
-
 // Describe implements the prometheus.Collector interface.
 func (collector *EventCollector) Describe(ch chan<- *prometheus.Desc) {
-	collector.eventsTotal.Describe(ch)
+	collector.metrics.eventsTotal.Describe(ch)
 }
 
 // Collect implements the prometheus.Collector interface.
 func (collector *EventCollector) Collect(ch chan<- prometheus.Metric) {
-	collector.eventsTotal.Collect(ch)
+	collector.metrics.eventsTotal.Collect(ch)
 }
 
 // Run starts updating EventCollector metrics.
 func (collector *EventCollector) Run(stopCh <-chan struct{}) {
-	for _, factory := range collector.informerFactories {
-		go factory.Start(stopCh)
+	for _, informer := range collector.informers {
+		go informer.Run(stopCh)
 	}
 }
 
-func (collector *EventCollector) initInformers() {
-	for _, factory := range collector.informerFactories {
-		eventsTotalInformer := factory.Core().V1().Events().Informer()
-		eventsTotalInformer.AddEventHandler(collector.eventsTotalHandler())
-	}
+func (collector *EventCollector) newEventInformer(ns, evType string) cache.SharedIndexInformer {
+	return informer.NewInstrumentedEventInformer(
+		collector.kclient,
+		metav1.NamespaceAll,
+		collector.metrics.listWatchMetrics,
+		0,
+		cache.Indexers{},
+		func(list *metav1.ListOptions) {
+			filterInvolvedObjectNs(list, ns)
+			filterEventType(list, evType)
+		},
+	)
 }
 
-func (collector *EventCollector) eventsTotalHandler() cache.ResourceEventHandler {
+func (collector *EventCollector) eventHandler() cache.ResourceEventHandler {
 	return cache.FilteringResourceEventHandler{
 		FilterFunc: collector.filter.filter,
 		Handler: &cache.ResourceEventHandlerFuncs{
 			AddFunc: func(obj interface{}) {
+				collector.lock.Lock()
+				defer collector.lock.Unlock()
+
 				ev := obj.(*v1.Event)
-				collector.increaseEventsTotal(ev, 1)
+				collector.metrics.increaseEventsTotal(ev, 1)
 			},
 			UpdateFunc: func(oldObj, newObj interface{}) {
+				collector.lock.Lock()
+				defer collector.lock.Unlock()
+
 				oldEv := oldObj.(*v1.Event)
 				newEv := newObj.(*v1.Event)
 				nbNew := updatedEventNb(oldEv, newEv)
-				collector.increaseEventsTotal(newEv, float64(nbNew))
+				collector.metrics.increaseEventsTotal(newEv, float64(nbNew))
 			},
 		},
 	}
 }
 
-func (collector *EventCollector) increaseEventsTotal(event *v1.Event, nbNew float64) {
-	collector.lock.Lock()
-	collector.eventsTotal.With(prometheus.Labels{
-		"type":                      event.Type,
-		"involved_object_namespace": event.InvolvedObject.Namespace,
-		"involved_object_kind":      event.InvolvedObject.Kind,
-		"reason":                    event.Reason,
-	}).Add(nbNew)
-	collector.lock.Unlock()
-}
-
 func updatedEventNb(oldEv, newEv *v1.Event) int32 {
 	if newEv.Series != nil {
 		if oldEv.Series != nil {

internal/collector/metrics.go

@@ -0,0 +1,48 @@
+/*
+Copyright 2020 Red Hat, Inc. 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 collector
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/rhobs/kube-events-exporter/pkg/informer"
+
+	v1 "k8s.io/api/core/v1"
+)
+
+type exporterMetrics struct {
+	eventsTotal      *prometheus.CounterVec
+	listWatchMetrics *informer.ListWatchMetrics
+}
+
+func newExporterMetrics(exporterRegistry *prometheus.Registry) *exporterMetrics {
+	return &exporterMetrics{
+		eventsTotal: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Name: "kube_events_total",
+			Help: "Count of all Kubernetes Events",
+		}, []string{"type", "involved_object_namespace", "involved_object_kind", "reason"}),
+		listWatchMetrics: informer.NewListWatchMetrics(exporterRegistry),
+	}
+}
+
+func (m *exporterMetrics) increaseEventsTotal(event *v1.Event, nbNew float64) {
+	m.eventsTotal.With(prometheus.Labels{
+		"type":                      event.Type,
+		"involved_object_namespace": event.InvolvedObject.Namespace,
+		"involved_object_kind":      event.InvolvedObject.Kind,
+		"reason":                    event.Reason,
+	}).Add(nbNew)
+}

pkg/informer/event.go

@@ -0,0 +1,62 @@
+/*
+Copyright 2020 Red Hat, Inc. 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 informer
+
+import (
+	"context"
+	"time"
+
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/watch"
+	"k8s.io/client-go/informers/internalinterfaces"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/tools/cache"
+)
+
+// NewInstrumentedEventInformer constructs a new informer for Event type with
+// instrumented list watch.
+func NewInstrumentedEventInformer(client kubernetes.Interface, namespace string, metrics *ListWatchMetrics, resyncPeriod time.Duration, indexers cache.Indexers, tweakListOptions internalinterfaces.TweakListOptionsFunc) cache.SharedIndexInformer {
+	return cache.NewSharedIndexInformer(
+		NewInstrumentedListerWatcher(
+			NewEventListerWatcher(client, namespace, tweakListOptions),
+			metrics,
+		),
+		&v1.Event{},
+		resyncPeriod,
+		indexers,
+	)
+}
+
+// NewEventListerWatcher constructs a new lister watcher for Event type.
+func NewEventListerWatcher(client kubernetes.Interface, namespace string, tweakListOptions internalinterfaces.TweakListOptionsFunc) cache.ListerWatcher {
+	return &cache.ListWatch{
+		ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
+			if tweakListOptions != nil {
+				tweakListOptions(&options)
+			}
+			return client.CoreV1().Events(namespace).List(context.TODO(), options)
+		},
+		WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
+			if tweakListOptions != nil {
+				tweakListOptions(&options)
+			}
+			return client.CoreV1().Events(namespace).Watch(context.TODO(), options)
+		},
+	}
+}