Collect and expose prometheus metrics #42

Collect counter metrics for total number of successful and failed
deletes of pods and jobs.

Signed-off-by: Sergey Nuzhdin <[email protected]>

Author: lwolf

cmd/main.go

@@ -5,6 +5,7 @@ import (
 	"flag"
 	"fmt"
 	"log"
+	"net/http"
 	"os"
 	"os/signal"
 	"path/filepath"
@@ -13,6 +14,7 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/VictoriaMetrics/metrics"
 	"k8s.io/client-go/kubernetes"
 	_ "k8s.io/client-go/plugin/pkg/client/auth/gcp" // TODO: Add all auth providers
 	_ "k8s.io/client-go/plugin/pkg/client/auth/oidc"
@@ -36,6 +38,7 @@ func setupLogging() {
 func main() {
 	runOutsideCluster := flag.Bool("run-outside-cluster", false, "Set this flag when running outside of the cluster.")
 	namespace := flag.String("namespace", "", "Limit scope to a single namespaces")
+	listenAddr := flag.String("listen-addr", "0.0.0.0:7000", "Address to expose metrics.")
 
 	deleteSuccessAfter := flag.Duration("delete-successful-after", 15*time.Minute, "Delete jobs and pods in successful state after X duration (golang duration format, e.g 5m), 0 - never delete")
 	deleteFailedAfter := flag.Duration("delete-failed-after", 0, "Delete jobs and pods in failed state after X duration (golang duration format, e.g 5m), 0 - never delete")
@@ -92,7 +95,7 @@ func main() {
 	if err != nil {
 		log.Fatal(err.Error())
 	}
-	ctx, cancel := context.WithCancel(context.Background())
+	ctx := context.Background()
 
 	wg.Add(1)
 	go func() {
@@ -105,7 +108,8 @@ func main() {
 				*legacyKeepSuccessHours,
 				*legacyKeepFailedHours,
 				*legacyKeepPendingHours,
-			).Run(stopCh)
+				stopCh,
+			).Run()
 		} else {
 			controller.NewKleaner(
 				ctx,
@@ -117,15 +121,46 @@ func main() {
 				*deletePendingAfter,
 				*deleteOrphanedAfter,
 				*deleteEvictedAfter,
-			).Run(stopCh)
+				stopCh,
+			).Run()
 		}
 		wg.Done()
 	}()
 	log.Printf("Controller started...")
 
+	server := http.Server{Addr: *listenAddr}
+	wg.Add(1)
+	go func() {
+		// Expose the registered metrics at `/metrics` path.
+		http.HandleFunc("/metrics", func(w http.ResponseWriter, req *http.Request) {
+			metrics.WritePrometheus(w, true)
+		})
+		err := server.ListenAndServe()
+		if err != nil {
+			log.Fatalf("failed to ListenAndServe metrics server: %v\n", err)
+		}
+		wg.Done()
+	}()
+	log.Printf("Listening at %s", *listenAddr)
+
+	wg.Add(1)
+	go func() {
+		for {
+			select {
+			case <-stopCh:
+				log.Println("shutting http server down")
+				err := server.Shutdown(ctx)
+				if err != nil {
+					log.Printf("failed to shutdown metrics server: %v\n", err)
+				}
+				wg.Done()
+				return
+			}
+		}
+	}()
+
 	<-sigsCh // Wait for signals (this hangs until a signal arrives)
-	log.Printf("Shutting down...")
-	cancel()
+	log.Printf("got termination signal...")
 	close(stopCh) // Tell goroutines to stopCh themselves
 	wg.Wait()     // Wait for all to be stopped
 }

go.mod

@@ -3,6 +3,7 @@ module github.com/lwolf/kube-cleanup-operator
 go 1.14
 
 require (
+	github.com/VictoriaMetrics/metrics v1.11.3
 	github.com/imdario/mergo v0.3.9 // indirect
 	golang.org/x/oauth2 v0.0.0-20200107190931-bf48bf16ab8d // indirect
 	golang.org/x/time v0.0.0-20200416051211-89c76fbcd5d1 // indirect

go.sum

@@ -14,6 +14,8 @@ github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03
 github.com/NYTimes/gziphandler v0.0.0-20170623195520-56545f4a5d46/go.mod h1:3wb06e3pkSAbeQ52E9H9iFoQsEEwGN64994WTCIhntQ=
 github.com/PuerkitoBio/purell v1.0.0/go.mod h1:c11w/QuzBsJSee3cPx9rAFu61PvFxuPbtSwDGJws/X0=
 github.com/PuerkitoBio/urlesc v0.0.0-20160726150825-5bd2802263f2/go.mod h1:uGdkoq3SwY9Y+13GIhn11/XLaGBb4BfwItxLd5jeuXE=
+github.com/VictoriaMetrics/metrics v1.11.3 h1:eSfXc0CrquKa1VTNUvhP+dhNjLUZHQGTFfp19mYCQWE=
+github.com/VictoriaMetrics/metrics v1.11.3/go.mod h1:LU2j9qq7xqZYXz8tF3/RQnB2z2MbZms5TDiIg9/NHiQ=
 github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
@@ -103,6 +105,10 @@ github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.4.0 h1:2E4SXV/wtOkTonXsotYi4li6zVWxYlZuYNCXe9XRJyk=
 github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
+github.com/valyala/fastrand v1.0.0 h1:LUKT9aKer2dVQNUi3waewTbKV+7H17kvWFNKs2ObdkI=
+github.com/valyala/fastrand v1.0.0/go.mod h1:HWqCzkrkg6QXT8V2EXWvXCoow7vLwOFN002oeRzjapQ=
+github.com/valyala/histogram v1.0.1 h1:FzA7n2Tz/wKRMejgu3PV1vw3htAklTjjuoI6z3d4KDg=
+github.com/valyala/histogram v1.0.1/go.mod h1:lQy0xA4wUz2+IUnf97SivorsJIp8FxsnRd6x25q7Mto=
 go.opencensus.io v0.21.0/go.mod h1:mSImk1erAIZhrmZN+AvHh14ztQfjbGwt4TtuofqLduU=
 golang.org/x/crypto v0.0.0-20190211182817-74369b46fc67/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2 h1:VklqNMn3ovrHsnt90PveolxSbWFaJdECFbxSq0Mqo2M=

pkg/controller/controller.go

@@ -2,10 +2,12 @@ package controller
 
 import (
 	"context"
+	"fmt"
 	"log"
 	"reflect"
 	"time"
 
+	"github.com/VictoriaMetrics/metrics"
 	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
 	apierrs "k8s.io/apimachinery/pkg/api/errors"
@@ -23,10 +25,20 @@ func ignoreNotFound(err error) error {
 	return err
 }
 
-const resyncPeriod = time.Second * 30
+func metricName(name string, namespace string) string {
+	return fmt.Sprintf(`%s{namespace=%q}`, name, namespace)
+}
+
+const (
+	resyncPeriod           = time.Second * 30
+	podDeletedMetric       = "pods_deleted_total"
+	podDeletedFailedMetric = "pods_deleted_failed_total"
+	jobDeletedFailedMetric = "jobs_deleted_failed_total"
+	jobDeletedMetric       = "jobs_deleted_total"
+)
 
-// Kleaner watches the kubernetes api for changes to Pods and
-// delete completed Pods without specific annotation
+// Kleaner watches the kubernetes api for changes to Pods and Jobs and
+// delete those according to configured timeouts
 type Kleaner struct {
 	podInformer cache.SharedIndexInformer
 	jobInformer cache.SharedIndexInformer
@@ -40,11 +52,12 @@ type Kleaner struct {
 
 	dryRun bool
 	ctx    context.Context
+	stopCh <-chan struct{}
 }
 
 // NewKleaner creates a new NewKleaner
 func NewKleaner(ctx context.Context, kclient *kubernetes.Clientset, namespace string, dryRun bool, deleteSuccessfulAfter,
-	deleteFailedAfter, deletePendingAfter, deleteOrphanedAfter, deleteEvictedAfter time.Duration) *Kleaner {
+	deleteFailedAfter, deletePendingAfter, deleteOrphanedAfter, deleteEvictedAfter time.Duration, stopCh <-chan struct{}) *Kleaner {
 	jobInformer := cache.NewSharedIndexInformer(
 		&cache.ListWatch{
 			ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
@@ -76,6 +89,7 @@ func NewKleaner(ctx context.Context, kclient *kubernetes.Clientset, namespace st
 		dryRun:                dryRun,
 		kclient:               kclient,
 		ctx:                   ctx,
+		stopCh:                stopCh,
 		deleteSuccessfulAfter: deleteSuccessfulAfter,
 		deleteFailedAfter:     deleteFailedAfter,
 		deletePendingAfter:    deletePendingAfter,
@@ -104,27 +118,33 @@ func NewKleaner(ctx context.Context, kclient *kubernetes.Clientset, namespace st
 }
 
 func (c *Kleaner) periodicCacheCheck() {
+	ticker := time.NewTicker(2 * resyncPeriod)
 	for {
-		for _, job := range c.jobInformer.GetStore().List() {
-			c.Process(job)
-		}
-		for _, obj := range c.podInformer.GetStore().List() {
-			c.Process(obj)
+		select {
+		case <-c.stopCh:
+			ticker.Stop()
+			return
+		case <-ticker.C:
+			for _, job := range c.jobInformer.GetStore().List() {
+				c.Process(job)
+			}
+			for _, obj := range c.podInformer.GetStore().List() {
+				c.Process(obj)
+			}
 		}
-		time.Sleep(2 * resyncPeriod)
 	}
 }
 
 // Run starts the process for listening for pod changes and acting upon those changes.
-func (c *Kleaner) Run(stopCh <-chan struct{}) {
+func (c *Kleaner) Run() {
 	log.Printf("Listening for changes...")
 
-	go c.podInformer.Run(stopCh)
-	go c.jobInformer.Run(stopCh)
+	go c.podInformer.Run(c.stopCh)
+	go c.jobInformer.Run(c.stopCh)
 
 	go c.periodicCacheCheck()
 
-	<-stopCh
+	<-c.stopCh
 }
 
 func (c *Kleaner) Process(obj interface{}) {
@@ -198,7 +218,10 @@ func (c *Kleaner) deleteJobs(job *batchv1.Job) {
 	jo := metav1.DeleteOptions{PropagationPolicy: &propagation}
 	if err := c.kclient.BatchV1().Jobs(job.Namespace).Delete(c.ctx, job.Name, jo); ignoreNotFound(err) != nil {
 		log.Printf("failed to delete job '%s:%s': %v", job.Namespace, job.Name, err)
+		metrics.GetOrCreateCounter(metricName(jobDeletedFailedMetric, job.Namespace)).Inc()
+		return
 	}
+	metrics.GetOrCreateCounter(metricName(jobDeletedMetric, job.Namespace)).Inc()
 }
 
 func (c *Kleaner) deletePods(pod *corev1.Pod) {
@@ -209,7 +232,10 @@ func (c *Kleaner) deletePods(pod *corev1.Pod) {
 	var po metav1.DeleteOptions
 	if err := c.kclient.CoreV1().Pods(pod.Namespace).Delete(c.ctx, pod.Name, po); ignoreNotFound(err) != nil {
 		log.Printf("failed to delete pod '%s:%s': %v", pod.Namespace, pod.Name, err)
+		metrics.GetOrCreateCounter(metricName(podDeletedFailedMetric, pod.Namespace)).Inc()
+		return
 	}
+	metrics.GetOrCreateCounter(metricName(podDeletedMetric, pod.Namespace)).Inc()
 }
 
 func (c *Kleaner) maybeDeletePod(podPhase corev1.PodPhase, timeSinceFinish time.Duration) bool {

pkg/controller/controller_legacy.go

@@ -9,6 +9,7 @@ import (
 	"strconv"
 	"time"
 
+	"github.com/VictoriaMetrics/metrics"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -30,6 +31,7 @@ type PodController struct {
 	dryRun           bool
 	isLegacySystem   bool
 	ctx              context.Context
+	stopCh           <-chan struct{}
 }
 
 // CreatedByAnnotation type used to match pods created by job
@@ -70,7 +72,7 @@ func isLegacySystem(v version.Info) bool {
 
 // NewPodController creates a new NewPodController
 func NewPodController(ctx context.Context, kclient *kubernetes.Clientset, namespace string, dryRun bool, keepSuccessHours,
-	keepFailedHours, keepPendingHours int64) *PodController {
+	keepFailedHours, keepPendingHours int64, stopCh <-chan struct{}) *PodController {
 
 	serverVersion, err := kclient.ServerVersion()
 	if err != nil {
@@ -84,6 +86,7 @@ func NewPodController(ctx context.Context, kclient *kubernetes.Clientset, namesp
 		dryRun:           dryRun,
 		isLegacySystem:   isLegacySystem(*serverVersion),
 		ctx:              ctx,
+		stopCh:           stopCh,
 	}
 	// Create informer for watching Namespaces
 	podInformer := cache.NewSharedIndexInformer(
@@ -126,17 +129,22 @@ func (c *PodController) periodicCacheCheck() {
 }
 
 // Run starts the process for listening for pod changes and acting upon those changes.
-func (c *PodController) Run(stopCh <-chan struct{}) {
+func (c *PodController) Run() {
 	log.Printf("Listening for changes...")
 
-	go c.podInformer.Run(stopCh)
+	go c.podInformer.Run(c.stopCh)
 	go c.periodicCacheCheck()
 
-	<-stopCh
+	<-c.stopCh
 }
 
 func (c *PodController) Process(obj interface{}) {
 	podObj := obj.(*corev1.Pod)
+	// skip pods that are already in the deleting process
+	if !podObj.DeletionTimestamp.IsZero() {
+		return
+	}
+
 	parentJobName := c.getParentJobName(podObj)
 	// if we couldn't find a prent job name, ignore this pod
 	if parentJobName == "" {
@@ -182,6 +190,9 @@ func (c *PodController) deleteObjects(podObj *corev1.Pod, parentJobName string)
 		var jo metav1.DeleteOptions
 		if err := c.kclient.BatchV1().Jobs(podObj.Namespace).Delete(c.ctx, parentJobName, jo); ignoreNotFound(err) != nil {
 			log.Printf("failed to delete job %s: %v", parentJobName, err)
+			metrics.GetOrCreateCounter(metricName(jobDeletedFailedMetric, podObj.Namespace)).Inc()
+		} else {
+			metrics.GetOrCreateCounter(metricName(jobDeletedMetric, podObj.Namespace)).Inc()
 		}
 	} else {
 		log.Printf("dry-run: Job '%s' would have been deleted", parentJobName)
@@ -192,6 +203,9 @@ func (c *PodController) deleteObjects(podObj *corev1.Pod, parentJobName string)
 		var po metav1.DeleteOptions
 		if err := c.kclient.CoreV1().Pods(podObj.Namespace).Delete(c.ctx, podObj.Name, po); ignoreNotFound(err) != nil {
 			log.Printf("failed to delete job's pod %s: %v", parentJobName, err)
+			metrics.GetOrCreateCounter(metricName(podDeletedFailedMetric, podObj.Namespace)).Inc()
+		} else {
+			metrics.GetOrCreateCounter(metricName(podDeletedMetric, podObj.Namespace)).Inc()
 		}
 	} else {
 		log.Printf("dry-run: Pod '%s' would have been deleted", podObj.Name)