Merge pull request #73 from integr8ly/reconcile-additiona-scrape-config

Reconcile additional scrape config

Makefile

@@ -3,7 +3,7 @@ NAMESPACE ?= application-monitoring
 PROJECT ?= application-monitoring-operator
 REG=quay.io
 SHELL=/bin/bash
-TAG ?= 0.0.20
+TAG ?= 0.0.21
 PKG=github.com/integr8ly/application-monitoring-operator
 TEST_DIRS?=$(shell sh -c "find $(TOP_SRC_DIRS) -name \\*_test.go -exec dirname {} \\; | sort | uniq")
 TEST_POD_NAME=application-monitoring-operator-test

deploy/operator.yaml

@@ -15,7 +15,7 @@ spec:
       serviceAccountName: application-monitoring-operator
       containers:
         - name: application-monitoring-operator
-          image: quay.io/integreatly/application-monitoring-operator:0.0.20
+          image: quay.io/integreatly/application-monitoring-operator:0.0.21
           ports:
             - containerPort: 60000
               name: metrics

pkg/controller/applicationmonitoring/applicationmonitoring_controller.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"crypto/md5"
 	"fmt"
+	"k8s.io/apimachinery/pkg/watch"
 	"time"
 
 	"k8s.io/api/apps/v1beta1"
@@ -32,7 +33,7 @@ import (
 const MonitoringFinalizerName = "monitoring.cleanup"
 
 // ReconcilePauseSeconds the number of seconds to wait before running the reconcile loop
-const ReconcilePauseSeconds = 10
+const ReconcilePauseSeconds = 30
 
 var log = logf.Log.WithName("controller_applicationmonitoring")
 
@@ -61,6 +62,7 @@ func newReconciler(mgr manager.Manager) reconcile.Reconciler {
 	return &ReconcileApplicationMonitoring{
 		client:      mgr.GetClient(),
 		scheme:      mgr.GetScheme(),
+		helper:      NewKubeHelper(),
 		extraParams: make(map[string]string),
 	}
 }
@@ -90,6 +92,8 @@ type ReconcileApplicationMonitoring struct {
 	// that reads objects from the cache and writes to the apiserver
 	client      client.Client
 	scheme      *runtime.Scheme
+	helper      *KubeHelperImpl
+	watch       watch.Interface
 	extraParams map[string]string
 }
 
@@ -121,6 +125,11 @@ func (r *ReconcileApplicationMonitoring) Reconcile(request reconcile.Request) (r
 	instanceCopy := instance.DeepCopy()
 
 	if instanceCopy.DeletionTimestamp != nil {
+		if r.watch != nil {
+			r.watch.Stop()
+			r.watch = nil
+		}
+
 		return r.cleanup(instanceCopy)
 	}
 
@@ -149,6 +158,7 @@ func (r *ReconcileApplicationMonitoring) Reconcile(request reconcile.Request) (r
 		log.Info("Finished installing application monitoring")
 		return r.updatePhase(instanceCopy, PhaseReconcileConfig)
 	case PhaseReconcileConfig:
+		r.tryWatchAdditionalScrapeConfigs(instanceCopy)
 		return r.reconcileConfig(instanceCopy)
 	}
 
@@ -180,6 +190,51 @@ func (r *ReconcileApplicationMonitoring) syncBlackboxTargets(cr *applicationmoni
 	return nil
 }
 
+func (r *ReconcileApplicationMonitoring) tryWatchAdditionalScrapeConfigs(cr *applicationmonitoringv1alpha1.ApplicationMonitoring) {
+	// Watch already set
+	if r.watch != nil {
+		return
+	}
+
+	watch, err := r.watchAdditionalScrapeConfigs(cr)
+	if err != nil {
+		log.Error(err, "error setting up watch for additional scrape config")
+		return
+	}
+
+	r.watch = watch
+	log.Info("watching additional scrape configs")
+}
+
+// Watches the additional scrape config secret for modifications and reconciles them into the prometheus
+// configuration
+func (r *ReconcileApplicationMonitoring) watchAdditionalScrapeConfigs(cr *applicationmonitoringv1alpha1.ApplicationMonitoring) (watch.Interface, error) {
+	if cr.Spec.AdditionalScrapeConfigSecretName == "" {
+		return nil, nil
+	}
+
+	events, err := r.helper.startSecretWatch(cr)
+	if err != nil {
+		return nil, err
+	}
+
+	go func() {
+		for update := range events.ResultChan() {
+			secret := update.Object.(*corev1.Secret)
+			if update.Type != watch.Error && secret.Name == cr.Spec.AdditionalScrapeConfigSecretName {
+				log.Info(fmt.Sprintf("watch event of type '%v' received for additional scrape config", update.Type))
+				err = r.createOrUpdateAdditionalScrapeConfig(cr)
+				if err != nil {
+					log.Error(err, "error updating additional scrape config")
+				}
+			}
+		}
+		log.Info("stop watching for additional scrape config")
+	}()
+
+	return events, nil
+}
+
 func (r *ReconcileApplicationMonitoring) installPrometheusOperator(cr *applicationmonitoringv1alpha1.ApplicationMonitoring) (reconcile.Result, error) {
 	log.Info("Phase: Install PrometheusOperator")
 
@@ -393,7 +448,7 @@ func (r *ReconcileApplicationMonitoring) readAdditionalScrapeConfigSecret(cr *ap
 	additionalScrapeConfig := corev1.Secret{}
 	err := r.client.Get(context.TODO(), selector, &additionalScrapeConfig)
 	if err != nil {
-		log.Error(err, fmt.Sprintf("error reading secret '%v'", cr.Spec.AdditionalScrapeConfigSecretName))
+		log.Info(fmt.Sprintf("can't find secret '%v'", cr.Spec.AdditionalScrapeConfigSecretName))
 		return nil, false
 	}
 

pkg/controller/applicationmonitoring/kubeHelper.go

@@ -0,0 +1,32 @@
+package applicationmonitoring
+
+import (
+	"fmt"
+	"github.com/integr8ly/application-monitoring-operator/pkg/apis/applicationmonitoring/v1alpha1"
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/watch"
+	"k8s.io/client-go/kubernetes"
+	"sigs.k8s.io/controller-runtime/pkg/client/config"
+)
+
+type KubeHelperImpl struct {
+	k8client *kubernetes.Clientset
+}
+
+func NewKubeHelper() *KubeHelperImpl {
+	config := config.GetConfigOrDie()
+
+	k8client := kubernetes.NewForConfigOrDie(config)
+
+	helper := new(KubeHelperImpl)
+	helper.k8client = k8client
+	return helper
+}
+
+// Watch secrets in the namespace that have a certain label
+func (h *KubeHelperImpl) startSecretWatch(cr *v1alpha1.ApplicationMonitoring) (watch.Interface, error) {
+	opts := v1.ListOptions{
+		LabelSelector: fmt.Sprintf("monitoring-key=%s", cr.Spec.LabelSelector),
+	}
+	return h.k8client.CoreV1().Secrets(cr.Namespace).Watch(opts)
+}

version/version.go

@@ -1,5 +1,5 @@
 package version
 
 var (
-	Version = "0.0.20"
+	Version = "0.0.21"
 )