Create ServiceMonitor automatically.

Author: neunhoef

manifests/templates/deployment/rbac.yaml

@@ -24,7 +24,7 @@ rules:
   verbs: ["*"]
 - apiGroups: ["apiextensions.k8s.io"]
   resources: ["customresourcedefinitions"]
-  verbs: ["get"]
+  verbs: ["get", "list", "watch"]
 - apiGroups: [""]
   resources: ["pods", "services", "endpoints", "persistentvolumeclaims", "events", "secrets"]
   verbs: ["*"]

pkg/deployment/deployment.go

@@ -31,6 +31,7 @@ import (
 	"github.com/arangodb/arangosync/client"
 	"github.com/rs/zerolog"
 	"github.com/rs/zerolog/log"
+	apiextensionsclient "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/tools/record"
@@ -59,6 +60,7 @@ type Config struct {
 type Dependencies struct {
 	Log           zerolog.Logger
 	KubeCli       kubernetes.Interface
+	KubeExtCli    apiextensionsclient.Interface
 	DatabaseCRCli versioned.Interface
 	EventRecorder record.EventRecorder
 }
@@ -98,6 +100,7 @@ type Deployment struct {
 	stopped int32
 
 	inspectTrigger            trigger.Trigger
+	inspectCRDTrigger         trigger.Trigger
 	updateDeploymentTrigger   trigger.Trigger
 	clientCache               *clientCache
 	recentInspectionErrors    int
@@ -107,6 +110,7 @@ type Deployment struct {
 	resources                 *resources.Resources
 	chaosMonkey               *chaos.Monkey
 	syncClientCache           client.ClientCache
+	haveServiceMonitorCRD     bool
 }
 
 // New creates a new Deployment from the given API object.
@@ -136,6 +140,7 @@ func New(config Config, deps Dependencies, apiObject *api.ArangoDeployment) (*De
 	go d.listenForPVCEvents(d.stopCh)
 	go d.listenForSecretEvents(d.stopCh)
 	go d.listenForServiceEvents(d.stopCh)
+	go d.listenForCRDEvents(d.stopCh)
 	if apiObject.Spec.GetMode() == api.DeploymentModeCluster {
 		ci := newClusterScalingIntegration(d)
 		d.clusterScalingIntegration = ci
@@ -202,8 +207,10 @@ func (d *Deployment) run() {
 		}
 
 		// Create service monitor
-		if err := d.resources.EnsureServiceMonitor(); err != nil {
-			d.CreateEvent(k8sutil.NewErrorEvent("Failed to create service monitor", err, d.GetAPIObject()))
+		if d.haveServiceMonitorCRD {
+			if err := d.resources.EnsureServiceMonitor(); err != nil {
+				d.CreateEvent(k8sutil.NewErrorEvent("Failed to create service monitor", err, d.GetAPIObject()))
+			}
 		}
 
 		// Create members
@@ -234,6 +241,8 @@ func (d *Deployment) run() {
 		log.Info().Msg("start running...")
 	}
 
+	d.lookForServiceMonitorCRD()
+
 	inspectionInterval := maxInspectionInterval
 	for {
 		select {
@@ -263,6 +272,8 @@ func (d *Deployment) run() {
 			inspectionInterval = d.inspectDeployment(inspectionInterval)
 			log.Debug().Str("interval", inspectionInterval.String()).Msg("...inspected deployment")
 
+		case <-d.inspectCRDTrigger.Done():
+			d.lookForServiceMonitorCRD()
 		case <-d.updateDeploymentTrigger.Done():
 			inspectionInterval = minInspectionInterval
 			if err := d.handleArangoDeploymentUpdatedEvent(); err != nil {
@@ -497,3 +508,29 @@ func (d *Deployment) isOwnerOf(obj metav1.Object) bool {
 	}
 	return ownerRefs[0].UID == d.apiObject.UID
 }
+
+// lookForServiceMonitorCRD checks if there is a CRD for the ServiceMonitor
+// CR and sets the flag haveServiceMonitorCRD accordingly. This is called
+// once at creation time of the deployment and then always if the CRD
+// informer is triggered.
+func (d *Deployment) lookForServiceMonitorCRD() {
+	_, err := d.deps.KubeExtCli.ApiextensionsV1beta1().CustomResourceDefinitions().Get("servicemonitors.monitoring.coreos.com", metav1.GetOptions{})
+	log := d.deps.Log
+	log.Debug().Msgf("looking for ServiceMonitor CRD...")
+	if err == nil {
+		if !d.haveServiceMonitorCRD {
+			log.Info().Msgf("have discovered ServiceMonitor CRD")
+		}
+		d.haveServiceMonitorCRD = true
+		d.triggerInspection()
+		return
+	} else if k8sutil.IsNotFound(err) {
+		if d.haveServiceMonitorCRD {
+			log.Info().Msgf("ServiceMonitor CRD no longer there")
+		}
+		d.haveServiceMonitorCRD = false
+		return
+	}
+	log.Warn().Err(err).Msgf("error when looking for ServiceMonitor CRD")
+	return
+}

pkg/deployment/deployment_inspector.go

@@ -152,6 +152,13 @@ func (d *Deployment) inspectDeployment(lastInterval util.Interval) util.Interval
 			hasError = true
 			d.CreateEvent(k8sutil.NewErrorEvent("Service creation failed", err, d.apiObject))
 		}
+		if d.haveServiceMonitorCRD {
+			if err := d.resources.EnsureServiceMonitor(); err != nil {
+				hasError = true
+				d.CreateEvent(k8sutil.NewErrorEvent("Service monitor creation failed", err, d.apiObject))
+			}
+		}
+
 		if err := d.resources.EnsurePVCs(); err != nil {
 			hasError = true
 			d.CreateEvent(k8sutil.NewErrorEvent("PVC creation failed", err, d.apiObject))
@@ -208,3 +215,8 @@ func (d *Deployment) inspectDeployment(lastInterval util.Interval) util.Interval
 func (d *Deployment) triggerInspection() {
 	d.inspectTrigger.Trigger()
 }
+
+// triggerCRDInspection ensures that an inspection is run soon.
+func (d *Deployment) triggerCRDInspection() {
+	d.inspectCRDTrigger.Trigger()
+}

pkg/deployment/informers.go

@@ -24,6 +24,7 @@ package deployment
 
 import (
 	"k8s.io/api/core/v1"
+	v1beta1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1"
 	"k8s.io/client-go/tools/cache"
 
 	"github.com/arangodb/kube-arangodb/pkg/util/k8sutil"
@@ -197,3 +198,36 @@ func (d *Deployment) listenForServiceEvents(stopCh <-chan struct{}) {
 
 	rw.Run(stopCh)
 }
+
+// listenForCRDEvents keep listening for changes in CRDs until the given channel is closed.
+func (d *Deployment) listenForCRDEvents(stopCh <-chan struct{}) {
+	//getCRD := func(obj interface{}) (*v1beta1.CustomResourceDefinition, bool) {
+	//		crd, ok := obj.(*v1beta1.CustomResourceDefinition)
+	//		if !ok {
+	//			tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
+	//			if !ok {
+	//				return nil, false
+	//			}
+	//			crd, ok = tombstone.Obj.(*v1beta1.CustomResourceDefinition)
+	//			return crd, ok
+	//		}
+	//		return crd, true
+	//	}
+
+	rw := k8sutil.NewResourceWatcher(
+		d.deps.Log,
+		d.deps.KubeExtCli.ApiextensionsV1beta1().RESTClient(),
+		"customresourcedefinitions",
+		"",
+		&v1beta1.CustomResourceDefinition{},
+		cache.ResourceEventHandlerFuncs{
+			AddFunc: func(obj interface{}) {
+				d.triggerCRDInspection()
+			},
+			DeleteFunc: func(obj interface{}) {
+				d.triggerCRDInspection()
+			},
+		})
+
+	rw.Run(stopCh)
+}

pkg/deployment/resources/servicemonitor.go

@@ -55,10 +55,9 @@ func (r *Resources) EnsureServiceMonitor() error {
 	ns := apiObject.GetNamespace()
 	owner := apiObject.AsOwner()
 	spec := r.context.GetSpec()
-	if !spec.Metrics.IsEnabled() {
-		return nil
-	}
+	wantMetrics := spec.Metrics.IsEnabled()
 	serviceMonitorName := deploymentName + "-exporter"
+	log.Debug().Msgf("EnsureServiceMonitor running %s", serviceMonitorName)
 
 	// First get a client:
 	var restConfig *rest.Config
@@ -77,6 +76,9 @@ func (r *Resources) EnsureServiceMonitor() error {
 	_, err = serviceMonitors.Get(serviceMonitorName, metav1.GetOptions{})
 	if err != nil {
 		if k8sutil.IsNotFound(err) {
+			if !wantMetrics {
+				return nil
+			}
 			// Need to create one:
 			smon := &coreosv1.ServiceMonitor{
 				ObjectMeta: metav1.ObjectMeta{
@@ -106,13 +108,24 @@ func (r *Resources) EnsureServiceMonitor() error {
 				log.Error().Err(err).Msgf("Failed to create ServiceMonitor %s", serviceMonitorName)
 				return maskAny(err)
 			}
+			log.Debug().Msgf("ServiceMonitor %s successfully created.", serviceMonitorName)
+			return nil
 		} else {
 			log.Error().Err(err).Msgf("Failed to get ServiceMonitor %s", serviceMonitorName)
 			return maskAny(err)
 		}
 	}
-
-	log.Debug().Msgf("ServiceMonitor %s already found, no need to create.",
-		serviceMonitorName)
-	return nil
+	if wantMetrics {
+		log.Debug().Msgf("ServiceMonitor %s already found, no need to create.",
+			serviceMonitorName)
+		return nil
+	}
+	// Need to get rid of the ServiceMonitor:
+	err = serviceMonitors.Delete(serviceMonitorName, &metav1.DeleteOptions{})
+	if err == nil {
+		log.Debug().Msgf("Deleted ServiceMonitor %s", serviceMonitorName)
+		return nil
+	}
+	log.Error().Err(err).Msgf("Could not delete ServiceMonitor %s.", serviceMonitorName)
+	return maskAny(err)
 }

pkg/operator/operator_deployment.go

@@ -212,6 +212,7 @@ func (o *Operator) makeDeploymentConfigAndDeps(apiObject *api.ArangoDeployment)
 			Str("deployment", apiObject.GetName()).
 			Logger(),
 		KubeCli:       o.Dependencies.KubeCli,
+		KubeExtCli:    o.Dependencies.KubeExtCli,
 		DatabaseCRCli: o.Dependencies.CRCli,
 		EventRecorder: o.Dependencies.EventRecorder,
 	}