Add a logic for RolloutManager to create ServiceMonitor

Signed-off-by: Rizwana777 <[email protected]>

config/rbac/role.yaml

@@ -227,6 +227,14 @@ rules:
   - list
   - update
   - watch
+- apiGroups:
+  - monitoring.coreos.com
+  resources:
+  - prometheuses
+  - prometheusrules
+  - servicemonitors
+  verbs:
+  - '*'
 - apiGroups:
   - networking.istio.io
   resources:

controllers/argorollouts_controller.go

@@ -71,6 +71,7 @@ var log = logr.Log.WithName("rollouts-controller")
 //+kubebuilder:rbac:groups="traefik.containo.us",resources=traefikservices,verbs=watch;get;update
 //+kubebuilder:rbac:groups="x.getambassador.io",resources=ambassadormappings;mappings,verbs=create;watch;get;update;list;delete
 //+kubebuilder:rbac:groups="apisix.apache.org",resources=apisixroutes,verbs=watch;get;update
+//+kubebuilder:rbac:groups=monitoring.coreos.com,resources=prometheuses;prometheusrules;servicemonitors,verbs=*
 
 // Reconcile is part of the main kubernetes reconciliation loop which aims to
 // move the current state of the cluster closer to the desired state.

controllers/resources.go

@@ -6,11 +6,13 @@ import (
 	"reflect"
 
 	rolloutsmanagerv1alpha1 "github.com/argoproj-labs/argo-rollouts-manager/api/v1alpha1"
+	monitoringv1 "github.com/prometheus-operator/prometheus-operator/pkg/apis/monitoring/v1"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	rbacv1 "k8s.io/api/rbac/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/intstr"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 )
@@ -282,6 +284,18 @@ func (r *RolloutManagerReconciler) reconcileRolloutsMetricsService(ctx context.C
 		actualSvc.Spec.Ports = expectedSvc.Spec.Ports
 		return r.Client.Create(ctx, actualSvc)
 	}
+	// Checks if the cluster is an OpenShift cluster, if not logs an error and exits
+	clusterVersion, err := GetClusterVersion(r.Client)
+	if clusterVersion != "" && err == nil {
+		// Create ServiceMonitor for Rollouts metrics
+		err := r.createServiceMonitorIfAbsent(cr.Namespace, cr, actualSvc.Name, actualSvc.Name)
+		if err != nil {
+			return err
+		}
+	} else {
+		log.Info(fmt.Sprintf("Unable to get cluster version: %v", err))
+		return nil
+	}
 
 	return nil
 }
@@ -816,3 +830,59 @@ func getAggregateToViewPolicyRules() []rbacv1.PolicyRule {
 		},
 	}
 }
+
+func (r *RolloutManagerReconciler) createServiceMonitorIfAbsent(namespace string, rolloutManager *rolloutsmanagerv1alpha1.RolloutManager, name, serviceMonitorLabel string) error {
+	existingServiceMonitor := &monitoringv1.ServiceMonitor{}
+	err := r.Client.Get(context.TODO(), types.NamespacedName{Name: name, Namespace: namespace}, existingServiceMonitor)
+	if err == nil {
+		log.Info("A ServiceMonitor instance already exists",
+			"Namespace", existingServiceMonitor.Namespace, "Name", existingServiceMonitor.Name)
+		return nil
+	}
+	if apierrors.IsNotFound(err) {
+		serviceMonitor := newServiceMonitor(namespace, name, serviceMonitorLabel)
+		log.Info("Creating a new ServiceMonitor instance",
+			"Namespace", serviceMonitor.Namespace, "Name", serviceMonitor.Name)
+
+		// Set the RolloutManager instance as the owner and controller
+		if err := controllerutil.SetControllerReference(rolloutManager, serviceMonitor, r.Scheme); err != nil {
+			log.Error(err, "Error setting read role owner ref",
+				"Namespace", serviceMonitor.Namespace, "Name", serviceMonitor.Name, "RolloutManager Name", rolloutManager.Name)
+			return err
+		}
+
+		err = r.Client.Create(context.TODO(), serviceMonitor)
+		if err != nil {
+			log.Error(err, "Error creating a new ServiceMonitor instance",
+				"Namespace", serviceMonitor.Namespace, "Name", serviceMonitor.Name)
+			return err
+		}
+
+		return nil
+	}
+	log.Error(err, "Error querying for ServiceMonitor", "Namespace", namespace, "Name", name)
+	return err
+}
+
+func newServiceMonitor(namespace, name, matchLabel string) *monitoringv1.ServiceMonitor {
+	objectMeta := metav1.ObjectMeta{
+		Name:      name,
+		Namespace: namespace,
+	}
+	spec := monitoringv1.ServiceMonitorSpec{
+		Selector: metav1.LabelSelector{
+			MatchLabels: map[string]string{
+				"app.kubernetes.io/name": matchLabel,
+			},
+		},
+		Endpoints: []monitoringv1.Endpoint{
+			{
+				Port: "metrics",
+			},
+		},
+	}
+	return &monitoringv1.ServiceMonitor{
+		ObjectMeta: objectMeta,
+		Spec:       spec,
+	}
+}

controllers/resources_test.go

@@ -7,11 +7,13 @@ import (
 	"github.com/argoproj-labs/argo-rollouts-manager/api/v1alpha1"
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
+	monitoringv1 "github.com/prometheus-operator/prometheus-operator/pkg/apis/monitoring/v1"
 	corev1 "k8s.io/api/core/v1"
 	rbacv1 "k8s.io/api/rbac/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/intstr"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 )
@@ -183,4 +185,42 @@ var _ = Describe("ReconcileRolloutManager tests", func() {
 		}
 	})
 
+	It("Verify whether RolloutManager creating ServiceMonitor", func() {
+		existingSvc := &corev1.Service{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      DefaultArgoRolloutsMetricsServiceName,
+				Namespace: req.Namespace,
+				Labels: map[string]string{
+					"app.kubernetes.io/name":      DefaultArgoRolloutsMetricsServiceName,
+					"app.kubernetes.io/component": "server",
+				},
+			},
+			Spec: corev1.ServiceSpec{
+				Ports: []corev1.ServicePort{
+					{
+						Name:       "metrics",
+						Port:       8090,
+						Protocol:   corev1.ProtocolTCP,
+						TargetPort: intstr.FromInt(8090),
+					},
+				},
+				Selector: map[string]string{
+					DefaultRolloutsSelectorKey: DefaultArgoRolloutsResourceName,
+				},
+			},
+		}
+
+		Expect(r.Client.Create(ctx, existingSvc)).To(Succeed())
+
+		res, err := r.Reconcile(ctx, req)
+		Expect(err).ToNot(HaveOccurred())
+		Expect(res.Requeue).Should(BeFalse(), "reconcile should not requeue request")
+
+		sm := &monitoringv1.ServiceMonitor{}
+		Expect(r.Client.Get(ctx, types.NamespacedName{
+			Name:      DefaultArgoRolloutsMetricsServiceName,
+			Namespace: testNamespace,
+		}, sm)).To(Succeed())
+	})
+
 })

controllers/utils.go

@@ -8,12 +8,18 @@ import (
 	"sort"
 	"strings"
 
+	configv1 "github.com/openshift/api/config/v1"
 	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
+const (
+	clusterVersionName = "version"
+)
+
 func setRolloutsLabels(obj *metav1.ObjectMeta) {
 	obj.Labels = map[string]string{}
 	obj.Labels["app.kubernetes.io/name"] = DefaultArgoRolloutsResourceName
@@ -132,3 +138,16 @@ func isMergable(extraArgs []string, cmd []string) error {
 	}
 	return nil
 }
+
+// GetClusterVersion returns the OpenShift Cluster version in which the operator is installed
+func GetClusterVersion(client client.Client) (string, error) {
+	clusterVersion := &configv1.ClusterVersion{}
+	err := client.Get(context.TODO(), types.NamespacedName{Name: clusterVersionName}, clusterVersion)
+	if err != nil {
+		if apierrors.IsNotFound(err) {
+			return "", nil
+		}
+		return "", err
+	}
+	return clusterVersion.Status.Desired.Version, nil
+}

controllers/utils_test.go

@@ -5,6 +5,8 @@ import (
 
 	rolloutsmanagerv1alpha1 "github.com/argoproj-labs/argo-rollouts-manager/api/v1alpha1"
 	. "github.com/onsi/gomega"
+	configv1 "github.com/openshift/api/config/v1"
+	monitoringv1 "github.com/prometheus-operator/prometheus-operator/pkg/apis/monitoring/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -34,10 +36,17 @@ func makeTestRolloutManager(opts ...rolloutManagerOpt) *rolloutsmanagerv1alpha1.
 
 func makeTestReconciler(objs ...runtime.Object) *RolloutManagerReconciler {
 	s := scheme.Scheme
+	objs = append(objs, NewClusterVersion("4.14.1"))
 
 	err := rolloutsmanagerv1alpha1.AddToScheme(s)
 	Expect(err).ToNot(HaveOccurred())
 
+	err = monitoringv1.AddToScheme(s)
+	Expect(err).ToNot(HaveOccurred())
+
+	err = configv1.AddToScheme(s)
+	Expect(err).ToNot(HaveOccurred())
+
 	cl := fake.NewClientBuilder().WithScheme(s).WithRuntimeObjects(objs...).Build()
 	return &RolloutManagerReconciler{
 		Client:                       cl,
@@ -50,3 +59,19 @@ func createNamespace(r *RolloutManagerReconciler, n string) error {
 	ns := &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: n}}
 	return r.Client.Create(context.Background(), ns)
 }
+
+func NewClusterVersion(version string) *configv1.ClusterVersion {
+	return &configv1.ClusterVersion{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: clusterVersionName,
+		},
+		Spec: configv1.ClusterVersionSpec{
+			Channel: "stable",
+		},
+		Status: configv1.ClusterVersionStatus{
+			Desired: configv1.Update{
+				Version: version,
+			},
+		},
+	}
+}

go.mod

@@ -6,6 +6,8 @@ require (
 	github.com/argoproj/argo-rollouts v1.6.6
 	github.com/onsi/ginkgo/v2 v2.11.0
 	github.com/onsi/gomega v1.27.10
+	github.com/openshift/api v3.9.1-0.20190916204813-cdbe64fb0c91+incompatible
+	github.com/prometheus-operator/prometheus-operator/pkg/apis/monitoring v0.50.0
 	go.uber.org/zap v1.24.0
 	gopkg.in/yaml.v2 v2.4.0
 	k8s.io/api v0.26.0