Merge pull request #410 from vyzigold/scrape_rmq

Scrape RabbitMQ prometheus metrics

config/rbac/role.yaml

@@ -258,6 +258,14 @@ rules:
   - list
   - patch
   - watch
+- apiGroups:
+  - rabbitmq.com
+  resources:
+  - rabbitmqclusters
+  verbs:
+  - get
+  - list
+  - watch
 - apiGroups:
   - rabbitmq.openstack.org
   resources:

controllers/metricstorage_controller.go

@@ -59,6 +59,7 @@ import (
 	ceilometer "github.com/openstack-k8s-operators/telemetry-operator/pkg/ceilometer"
 	"github.com/openstack-k8s-operators/telemetry-operator/pkg/dashboards"
 	metricstorage "github.com/openstack-k8s-operators/telemetry-operator/pkg/metricstorage"
+	rabbitmqv1 "github.com/rabbitmq/cluster-operator/api/v1beta1"
 	monv1 "github.com/rhobs/obo-prometheus-operator/pkg/apis/monitoring/v1"
 	monv1alpha1 "github.com/rhobs/obo-prometheus-operator/pkg/apis/monitoring/v1alpha1"
 	obov1 "github.com/rhobs/observability-operator/pkg/apis/monitoring/v1alpha1"
@@ -103,6 +104,7 @@ func (r *MetricStorageReconciler) GetLogger(ctx context.Context) logr.Logger {
 //+kubebuilder:rbac:groups=monitoring.rhobs,resources=prometheuses,verbs=get;list;watch;update;patch;delete
 //+kubebuilder:rbac:groups=monitoring.rhobs,resources=alertmanagers,verbs=get;list;watch;update;patch;delete
 //+kubebuilder:rbac:groups=network.openstack.org,resources=ipsets,verbs=get;list;watch
+//+kubebuilder:rbac:groups=rabbitmq.com,resources=rabbitmqclusters,verbs=get;list;watch
 //+kubebuilder:rbac:groups=observability.openshift.io,resources=uiplugins,verbs=get;list;watch;create;patch
 //+kubebuilder:rbac:groups=core,resources=configmaps,verbs=get;list;watch;create;update;patch;delete
 
@@ -365,7 +367,7 @@ func (r *MetricStorageReconciler) reconcileNormal(
 		instance.Status.Conditions.MarkTrue(telemetryv1.MonitoringStackReadyCondition, condition.ReadyMessage)
 	}
 
-	// Deploy ServiceMonitor for ceilometer monitoring
+	// Deploy ServiceMonitors
 	err = r.ensureWatches(ctx, "servicemonitors.monitoring.rhobs", &monv1.ServiceMonitor{}, eventHandler)
 
 	if err != nil {
@@ -376,17 +378,20 @@ func (r *MetricStorageReconciler) reconcileNormal(
 		Log.Info("Can't own ServiceMonitor resource")
 		return ctrl.Result{RequeueAfter: telemetryv1.PauseBetweenWatchAttempts}, nil
 	}
+
+	// ServiceMonitor for ceilometer monitoring
+	ceilometerServerName := fmt.Sprintf("%s-internal.%s.svc", ceilometer.ServiceName, instance.Namespace)
 	ceilometerMonitor := &monv1.ServiceMonitor{
 		ObjectMeta: metav1.ObjectMeta{
-			Name:      instance.Name,
+			Name:      fmt.Sprintf("%s-%s", instance.Name, ceilometerServerName),
 			Namespace: instance.Namespace,
 		},
 	}
 	op, err = controllerutil.CreateOrPatch(ctx, r.Client, ceilometerMonitor, func() error {
 		ceilometerLabels := map[string]string{
 			common.AppSelector: ceilometer.ServiceName,
 		}
-		desiredCeilometerMonitor := metricstorage.ServiceMonitor(instance, serviceLabels, ceilometerLabels)
+		desiredCeilometerMonitor := metricstorage.ServiceMonitor(instance, serviceLabels, ceilometerLabels, ceilometerServerName, "")
 		desiredCeilometerMonitor.Spec.DeepCopyInto(&ceilometerMonitor.Spec)
 		ceilometerMonitor.ObjectMeta.Labels = desiredCeilometerMonitor.ObjectMeta.Labels
 		err = controllerutil.SetControllerReference(instance, ceilometerMonitor, r.Scheme)
@@ -398,6 +403,70 @@ func (r *MetricStorageReconciler) reconcileNormal(
 	if op != controllerutil.OperationResultNone {
 		Log.Info(fmt.Sprintf("Ceilometer ServiceMonitor %s successfully changed - operation: %s", ceilometerMonitor.Name, string(op)))
 	}
+	// ServiceMonitors for RabbitMQ monitoring
+	rabbitList := &rabbitmqv1.RabbitmqClusterList{}
+	listOpts := []client.ListOption{
+		client.InNamespace(instance.GetNamespace()),
+	}
+	err = r.Client.List(ctx, rabbitList, listOpts...)
+	if err != nil && !k8s_errors.IsNotFound(err) {
+		return ctrl.Result{}, err
+	}
+	for _, rabbit := range rabbitList.Items {
+		rabbitServerName := fmt.Sprintf("%s.%s.svc", rabbit.Name, rabbit.Namespace)
+		rabbitMonitor := &monv1.ServiceMonitor{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      fmt.Sprintf("%s-%s", instance.Name, rabbitServerName),
+				Namespace: instance.Namespace,
+			},
+		}
+		op, err = controllerutil.CreateOrPatch(ctx, r.Client, rabbitMonitor, func() error {
+			rabbitLabels := map[string]string{
+				"app.kubernetes.io/name": rabbit.Name,
+			}
+			desiredRabbitMonitor := metricstorage.ServiceMonitor(instance, serviceLabels, rabbitLabels, rabbitServerName, "prometheus-tls")
+			desiredRabbitMonitor.Spec.DeepCopyInto(&rabbitMonitor.Spec)
+			rabbitMonitor.ObjectMeta.Labels = desiredRabbitMonitor.ObjectMeta.Labels
+			err = controllerutil.SetControllerReference(instance, rabbitMonitor, r.Scheme)
+			return err
+		})
+		if err != nil {
+			return ctrl.Result{}, err
+		}
+		if op != controllerutil.OperationResultNone {
+			Log.Info(fmt.Sprintf("Rabbit ServiceMonitor %s successfully changed - operation: %s", rabbitMonitor.Name, string(op)))
+		}
+	}
+	// Check that RabbitMQ monitor's RabbitMQs still exist
+	// Delete the ServiceMonitors, which don't have a RabbitMQ anymore
+	svcMonitorList := &monv1.ServiceMonitorList{}
+	err = r.Client.List(ctx, svcMonitorList, listOpts...)
+	if err != nil && !k8s_errors.IsNotFound(err) {
+		return ctrl.Result{}, err
+	}
+	for _, svcMonitor := range svcMonitorList.Items {
+		if svcMonitor.OwnerReferences == nil ||
+			len(svcMonitor.OwnerReferences) < 1 ||
+			svcMonitor.OwnerReferences[0].Name != instance.Name {
+			continue
+		}
+		if svcMonitor.Name == fmt.Sprintf("%s-ceilometer-internal.%s.svc", instance.Name, instance.Namespace) {
+			continue
+		}
+		rabbitmqExists := false
+		for _, rabbit := range rabbitList.Items {
+			if svcMonitor.Name == fmt.Sprintf("%s-%s.%s.svc", instance.Name, rabbit.Name, instance.Namespace) {
+				rabbitmqExists = true
+			}
+		}
+		if !rabbitmqExists {
+			err = r.Client.Delete(ctx, svcMonitor)
+			if err != nil {
+				return ctrl.Result{}, err
+			}
+			Log.Info(fmt.Sprintf("Deleted ServiceMonitor: %s because its RabbitMQ doesn't exist", svcMonitor.Name))
+		}
+	}
 	instance.Status.Conditions.MarkTrue(telemetryv1.ServiceMonitorReadyCondition, condition.ReadyMessage)
 
 	endpointsNonTLS, endpointsTLS, err := getNodeExporterTargets(instance, helper)
@@ -861,6 +930,33 @@ func (r *MetricStorageReconciler) SetupWithManager(ctx context.Context, mgr ctrl
 		return nil
 	}
 
+	rabbitmqWatchFn := func(ctx context.Context, o client.Object) []reconcile.Request {
+		result := []reconcile.Request{}
+
+		// get all metricstorage CRs
+		metricStorages := &telemetryv1.MetricStorageList{}
+		listOpts := []client.ListOption{
+			client.InNamespace(o.GetNamespace()),
+		}
+		if err := r.Client.List(context.Background(), metricStorages, listOpts...); err != nil {
+			Log.Error(err, "Unable to retrieve MetricStorage CRs %w")
+			return nil
+		}
+
+		for _, cr := range metricStorages.Items {
+			// Reconcile all metricstorages
+			name := client.ObjectKey{
+				Namespace: o.GetNamespace(),
+				Name:      cr.Name,
+			}
+			result = append(result, reconcile.Request{NamespacedName: name})
+		}
+		if len(result) > 0 {
+			return result
+		}
+		return nil
+	}
+
 	// index prometheusCaBundleSecretNameField
 	if err := mgr.GetFieldIndexer().IndexField(context.Background(), &telemetryv1.MetricStorage{}, prometheusCaBundleSecretNameField, func(rawObj client.Object) []string {
 		// Extract the secret name from the spec, if one is provided
@@ -909,6 +1005,10 @@ func (r *MetricStorageReconciler) SetupWithManager(ctx context.Context, mgr ctrl
 			handler.EnqueueRequestsFromMapFunc(r.nodeSetWatchFn),
 			builder.WithPredicates(inventoryPredicator),
 		).
+		Watches(
+			&rabbitmqv1.RabbitmqCluster{},
+			handler.EnqueueRequestsFromMapFunc(rabbitmqWatchFn),
+		).
 		Build(r)
 	r.Controller = control
 	return err

go.mod

@@ -17,6 +17,7 @@ require (
 	github.com/openstack-k8s-operators/lib-common/modules/openstack v0.3.1-0.20240531085522-94fdcd5ff4fd
 	github.com/openstack-k8s-operators/mariadb-operator/api v0.3.1-0.20240514152407-b2bea62f05db
 	github.com/openstack-k8s-operators/telemetry-operator/api v0.3.1-0.20240529090522-c780bd90b147
+	github.com/rabbitmq/cluster-operator v1.14.0
 	github.com/rhobs/obo-prometheus-operator/pkg/apis/monitoring v0.69.0-rhobs1
 	github.com/rhobs/observability-operator v0.0.28
 	k8s.io/api v0.28.10

go.sum

@@ -99,6 +99,8 @@ github.com/prometheus/common v0.46.0 h1:doXzt5ybi1HBKpsZOL0sSkaNHJJqkyfEWZGGqqSc
 github.com/prometheus/common v0.46.0/go.mod h1:Tp0qkxpb9Jsg54QMe+EAmqXkSV7Evdy1BTn+g2pa/hQ=
 github.com/prometheus/procfs v0.12.0 h1:jluTpSng7V9hY0O2R9DzzJHYb2xULk9VTR1V1R/k6Bo=
 github.com/prometheus/procfs v0.12.0/go.mod h1:pcuDEFsWDnvcgNzo4EEweacyhjeA9Zk3cnaOZAZEfOo=
+github.com/rabbitmq/cluster-operator v1.14.0 h1:1/nMyd9v/8T5IHA1BVcWbV0nrzN31F+gLP+0Ges6Y5E=
+github.com/rabbitmq/cluster-operator v1.14.0/go.mod h1:7XVU6ngbVJSPDXld+uMVk6nu68GH7fM6yYYY2MdYKek=
 github.com/rhobs/obo-prometheus-operator/pkg/apis/monitoring v0.69.0-rhobs1 h1:d6UamuHl8GaEXcQqM7VO8jMnnhzHmTA0f6h4dQTQsGk=
 github.com/rhobs/obo-prometheus-operator/pkg/apis/monitoring v0.69.0-rhobs1/go.mod h1:SDL0R6499oleOOVL5ovIxHpyLOh2sR2g0rvBuuqHyQM=
 github.com/rhobs/observability-operator v0.0.28 h1:FFMBkmI1LWJVWyJKJqelYbV5iUc4u/wJClqowRJZRoU=

main.go

@@ -44,6 +44,7 @@ import (
 	rabbitmqv1 "github.com/openstack-k8s-operators/infra-operator/apis/rabbitmq/v1beta1"
 	keystonev1 "github.com/openstack-k8s-operators/keystone-operator/api/v1beta1"
 	mariadbv1beta1 "github.com/openstack-k8s-operators/mariadb-operator/api/v1beta1"
+	rabbitmqclusterv1 "github.com/rabbitmq/cluster-operator/api/v1beta1"
 	monv1 "github.com/rhobs/obo-prometheus-operator/pkg/apis/monitoring/v1"
 	monv1alpha1 "github.com/rhobs/obo-prometheus-operator/pkg/apis/monitoring/v1alpha1"
 	obov1 "github.com/rhobs/observability-operator/pkg/apis/monitoring/v1alpha1"
@@ -70,6 +71,7 @@ func init() {
 	utilruntime.Must(mariadbv1beta1.AddToScheme(scheme))
 	utilruntime.Must(heatv1.AddToScheme(scheme))
 	utilruntime.Must(infranetworkv1.AddToScheme(scheme))
+	utilruntime.Must(rabbitmqclusterv1.AddToScheme(scheme))
 	//+kubebuilder:scaffold:scheme
 }
 

pkg/metricstorage/service_monitor.go

@@ -21,7 +21,6 @@ import (
 
 	tls "github.com/openstack-k8s-operators/lib-common/modules/common/tls"
 	telemetryv1 "github.com/openstack-k8s-operators/telemetry-operator/api/v1beta1"
-	ceilometer "github.com/openstack-k8s-operators/telemetry-operator/pkg/ceilometer"
 	monv1 "github.com/rhobs/obo-prometheus-operator/pkg/apis/monitoring/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
@@ -31,6 +30,8 @@ func ServiceMonitor(
 	instance *telemetryv1.MetricStorage,
 	labels map[string]string,
 	selector map[string]string,
+	serverName string,
+	port string,
 ) *monv1.ServiceMonitor {
 	var scrapeInterval monv1.Duration
 	if instance.Spec.MonitoringStack != nil && instance.Spec.MonitoringStack.ScrapeInterval != "" {
@@ -43,7 +44,7 @@ func ServiceMonitor(
 
 	serviceMonitor := &monv1.ServiceMonitor{
 		ObjectMeta: metav1.ObjectMeta{
-			Name:      instance.Name,
+			Name:      fmt.Sprintf("%s-%s", instance.Name, serverName),
 			Namespace: instance.Namespace,
 			Labels:    labels,
 		},
@@ -85,12 +86,15 @@ func ServiceMonitor(
 			},
 		},
 	}
+	if port != "" {
+		serviceMonitor.Spec.Endpoints[0].Port = port
+	}
 	if instance.Spec.PrometheusTLS.Enabled() {
 		serviceMonitor.Spec.Endpoints[0].Scheme = "https"
 		serviceMonitor.Spec.Endpoints[0].TLSConfig = &monv1.TLSConfig{
 			CAFile: fmt.Sprintf("/etc/prometheus/secrets/%s/%s", instance.Spec.PrometheusTLS.CaBundleSecretName, tls.CABundleKey),
 			SafeTLSConfig: monv1.SafeTLSConfig{
-				ServerName: fmt.Sprintf("%s-internal.%s.svc", ceilometer.ServiceName, instance.Namespace),
+				ServerName: serverName,
 			},
 		}
 	}

tests/kuttl/suites/metricstorage/tests/01-assert.yaml

@@ -35,7 +35,7 @@ kind: ServiceMonitor
 metadata:
   labels:
     service: metricStorage
-  name: telemetry-kuttl
+  name: telemetry-kuttl-ceilometer-internal.telemetry-kuttl-tests.svc
   ownerReferences:
   - kind: MetricStorage
     name: telemetry-kuttl
@@ -58,6 +58,34 @@ spec:
     matchLabels:
       service: ceilometer
 ---
+apiVersion: monitoring.rhobs/v1
+kind: ServiceMonitor
+metadata:
+  labels:
+    service: metricStorage
+  name: telemetry-kuttl-rabbitmq.telemetry-kuttl-tests.svc
+  ownerReferences:
+  - kind: MetricStorage
+    name: telemetry-kuttl
+spec:
+  endpoints:
+  - interval: 30s
+    metricRelabelings:
+    - action: labeldrop
+      regex: pod
+    - action: labeldrop
+      regex: namespace
+    - action: labeldrop
+      regex: instance
+    - action: labeldrop
+      regex: job
+    - action: labeldrop
+      regex: publisher
+  namespaceSelector: {}
+  selector:
+    matchLabels:
+      app.kubernetes.io/name: rabbitmq
+---
 apiVersion: monitoring.rhobs/v1alpha1
 kind: ScrapeConfig
 metadata:

tests/kuttl/suites/tls/tests/02-assert.yaml

@@ -277,7 +277,7 @@ kind: ServiceMonitor
 metadata:
   labels:
     service: metricStorage
-  name: telemetry-kuttl-metricstorage
+  name: telemetry-kuttl-metricstorage-ceilometer-internal.telemetry-kuttl-tests.svc
   ownerReferences:
   - kind: MetricStorage
     name: telemetry-kuttl-metricstorage