Create RabbitMQ TransportURL

This Patch is creating the required RabbitMQ infra via TransportURL CR.

- It adds RabbitMqClusterName param to the spec and
  WatcherRabbitMQTransportURLReadyCondition to the conditions list.
- It adds logic to the watcher controller to create a TransportURL
  object from rabbitmq in infra-operator.
- Adds functional envtest tests for the added feature.

api/bases/watcher.openstack.org_watchers.yaml

@@ -61,12 +61,19 @@ spec:
                       password from the Secret
                     type: string
                 type: object
+              rabbitMqClusterName:
+                default: rabbitmq
+                description: |-
+                  RabbitMQ instance name
+                  Needed to request a transportURL that is created and used in Barbican
+                type: string
               secret:
                 default: osp-secret
                 description: Secret containing all passwords / keys needed
                 type: string
             required:
             - databaseInstance
+            - rabbitMqClusterName
             type: object
           status:
             description: WatcherStatus defines the observed state of Watcher

api/v1beta1/common_types.go

@@ -44,6 +44,12 @@ type WatcherTemplate struct {
 	// INSERT ADDITIONAL SPEC FIELDS - desired state of cluster
 	// Important: Run "make" to regenerate code after modifying this file
 	WatcherCommon `json:",inline"`
+
+	// +kubebuilder:validation:Required
+	// +kubebuilder:default=rabbitmq
+	// RabbitMQ instance name
+	// Needed to request a transportURL that is created and used in Barbican
+	RabbitMqClusterName string `json:"rabbitMqClusterName"`
 }
 
 // PasswordSelector to identify the DB and AdminUser password from the Secret

api/v1beta1/conditions.go

@@ -0,0 +1,17 @@
+package v1beta1
+
+import "github.com/openstack-k8s-operators/lib-common/modules/common/condition"
+
+const (
+	// WatcherRabbitMQTransportURLReadyCondition -
+	WatcherRabbitMQTransportURLReadyCondition condition.Type = "WatcherRabbitMQTransportURLReady"
+)
+
+const (
+	// WatcherRabbitMQTransportURLReadyRunningMessage -
+	WatcherRabbitMQTransportURLReadyRunningMessage = "WatcherRabbitMQTransportURL creation in progress"
+	// WatcherRabbitMQTransportURLReadyMessage -
+	WatcherRabbitMQTransportURLReadyMessage = "WatcherRabbitMQTransportURL successfully created"
+	// WatcherRabbitMQTransportURLReadyErrorMessage -
+	WatcherRabbitMQTransportURLReadyErrorMessage = "WatcherRabbitMQTransportURL error occured %s"
+)

config/crd/bases/watcher.openstack.org_watchers.yaml

@@ -61,12 +61,19 @@ spec:
                       password from the Secret
                     type: string
                 type: object
+              rabbitMqClusterName:
+                default: rabbitmq
+                description: |-
+                  RabbitMQ instance name
+                  Needed to request a transportURL that is created and used in Barbican
+                type: string
               secret:
                 default: osp-secret
                 description: Secret containing all passwords / keys needed
                 type: string
             required:
             - databaseInstance
+            - rabbitMqClusterName
             type: object
           status:
             description: WatcherStatus defines the observed state of Watcher

config/rbac/role.yaml

@@ -58,6 +58,18 @@ rules:
   - patch
   - update
   - watch
+- apiGroups:
+  - rabbitmq.openstack.org
+  resources:
+  - transporturls
+  verbs:
+  - create
+  - delete
+  - get
+  - list
+  - patch
+  - update
+  - watch
 - apiGroups:
   - watcher.openstack.org
   resources:

controllers/watcher_common.go

@@ -29,6 +29,11 @@ var (
 	}
 )
 
+const (
+	// TransportURLSelector is the name of key in the secret created by TransportURL
+	TransportURLSelector = "transport_url"
+)
+
 // GetLogger returns a logger object with a prefix of "controller.name" and additional controller context fields
 func (r *ReconcilerBase) GetLogger(ctx context.Context) logr.Logger {
 	return log.FromContext(ctx).WithName("Controllers").WithName("ReconcilerBase")

controllers/watcher_controller.go

@@ -19,16 +19,22 @@ package controllers
 import (
 	"context"
 	"fmt"
+	"time"
 
+	"github.com/go-logr/logr"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/types"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 	"sigs.k8s.io/controller-runtime/pkg/log"
 
-	"github.com/go-logr/logr"
+	rabbitmqv1 "github.com/openstack-k8s-operators/infra-operator/apis/rabbitmq/v1beta1"
 	"github.com/openstack-k8s-operators/lib-common/modules/common"
 	"github.com/openstack-k8s-operators/lib-common/modules/common/condition"
 	"github.com/openstack-k8s-operators/lib-common/modules/common/helper"
+	"github.com/openstack-k8s-operators/lib-common/modules/common/util"
 	mariadbv1 "github.com/openstack-k8s-operators/mariadb-operator/api/v1beta1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	watcherv1beta1 "github.com/openstack-k8s-operators/watcher-operator/api/v1beta1"
 
@@ -63,6 +69,7 @@ func (r *WatcherReconciler) GetLogger(ctx context.Context) logr.Logger {
 //+kubebuilder:rbac:groups=mariadb.openstack.org,resources=mariadbdatabases,verbs=get;list;watch;create;update;patch;delete;
 //+kubebuilder:rbac:groups=core,resources=secrets,verbs=get;list;watch;create;update;patch;delete;
 //+kubebuilder:rbac:groups=core,resources=services,verbs=get;list;watch;create;update;patch;delete;
+//+kubebuilder:rbac:groups=rabbitmq.openstack.org,resources=transporturls,verbs=get;list;watch;create;update;patch;delete
 
 // Reconcile is part of the main kubernetes reconciliation loop which aims to
 // move the current state of the cluster closer to the desired state.
@@ -153,6 +160,35 @@ func (r *WatcherReconciler) Reconcile(ctx context.Context, req ctrl.Request) (re
 	_ = db
 	// create service DB - end
 
+	//
+	// create RabbitMQ transportURL CR and get the actual URL from the associated secret that is created
+	// not-ready condition is managed here instead of in ensureMQ to distinguish between Error (when receiving)
+	// an error, or Running when transportURL is empty.
+	//
+	transportURL, op, err := r.ensureMQ(ctx, instance, helper, serviceLabels)
+	if err != nil {
+		instance.Status.Conditions.Set(condition.FalseCondition(
+			watcherv1beta1.WatcherRabbitMQTransportURLReadyCondition,
+			condition.ErrorReason,
+			condition.SeverityWarning,
+			watcherv1beta1.WatcherRabbitMQTransportURLReadyErrorMessage,
+			err.Error()))
+		return ctrl.Result{}, err
+	}
+
+	if transportURL == nil {
+		Log.Info(fmt.Sprintf("Waiting for TransportURL for %s to be created", instance.Name))
+		instance.Status.Conditions.Set(condition.FalseCondition(
+			watcherv1beta1.WatcherRabbitMQTransportURLReadyCondition,
+			condition.RequestedReason,
+			condition.SeverityWarning,
+			watcherv1beta1.WatcherRabbitMQTransportURLReadyRunningMessage))
+		return ctrl.Result{RequeueAfter: time.Duration(10) * time.Second}, nil
+	}
+
+	_ = op
+	// end of TransportURL creation
+
 	// remove finalizers from unused MariaDBAccount records
 	// this assumes all database-depedendent deployments are up and
 	// running with current database account info
@@ -200,6 +236,10 @@ func (r *WatcherReconciler) initConditions(instance *watcherv1beta1.Watcher) err
 		// failure/in-progress operation
 		condition.UnknownCondition(condition.ReadyCondition, condition.InitReason, condition.ReadyInitMessage),
 		condition.UnknownCondition(condition.DBReadyCondition, condition.InitReason, condition.DBReadyInitMessage),
+		condition.UnknownCondition(
+			watcherv1beta1.WatcherRabbitMQTransportURLReadyCondition,
+			condition.InitReason,
+			condition.RabbitMqTransportURLReadyInitMessage),
 	)
 
 	instance.Status.Conditions.Init(&cl)
@@ -292,6 +332,66 @@ func (r *WatcherReconciler) ensureDB(
 	return db, ctrl.Result{}, err
 }
 
+// Create the required RabbitMQ
+func (r *WatcherReconciler) ensureMQ(
+	ctx context.Context,
+	instance *watcherv1beta1.Watcher,
+	h *helper.Helper,
+	serviceLabels map[string]string,
+) (*rabbitmqv1.TransportURL, controllerutil.OperationResult, error) {
+	Log := r.GetLogger(ctx)
+	Log.Info(fmt.Sprintf("Reconciling the RabbitMQ TransportURL for '%s'", instance.Name))
+
+	transportURL := &rabbitmqv1.TransportURL{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      fmt.Sprintf("%s-watcher-transport", instance.Name),
+			Namespace: instance.Namespace,
+			Labels:    serviceLabels,
+		},
+	}
+
+	op, err := controllerutil.CreateOrUpdate(ctx, r.Client, transportURL, func() error {
+		transportURL.Spec.RabbitmqClusterName = instance.Spec.RabbitMqClusterName
+
+		err := controllerutil.SetControllerReference(instance, transportURL, r.Scheme)
+		return err
+	})
+
+	if err != nil && !k8s_errors.IsNotFound(err) {
+		return nil, op, util.WrapErrorForObject(
+			fmt.Sprintf("Error create or update TransportURL object %s-watcher-transport", instance.Name),
+			transportURL,
+			err,
+		)
+	}
+
+	if op != controllerutil.OperationResultNone {
+		Log.Info(fmt.Sprintf("TransportURL %s successfully reconciled - operation: %s", transportURL.Name, string(op)))
+	}
+
+	// If transportURL is not ready, it returns nil
+	if !transportURL.IsReady() || transportURL.Status.SecretName == "" {
+		Log.Info(fmt.Sprintf("Waiting for TransportURL %s secret to be created", transportURL.Name))
+		return nil, op, nil
+	}
+
+	secretName := types.NamespacedName{Namespace: instance.Namespace, Name: transportURL.Status.SecretName}
+	secret := &corev1.Secret{}
+	err = h.GetClient().Get(ctx, secretName, secret)
+	if err != nil {
+		return nil, op, err
+	}
+
+	_, ok := secret.Data[TransportURLSelector]
+	if !ok {
+		return nil, op, fmt.Errorf(
+			"the TransportURL secret %s does not have 'transport_url' field", transportURL.Status.SecretName)
+	}
+
+	instance.Status.Conditions.MarkTrue(watcherv1beta1.WatcherRabbitMQTransportURLReadyCondition, watcherv1beta1.WatcherRabbitMQTransportURLReadyMessage)
+	return transportURL, op, nil
+}
+
 func (r *WatcherReconciler) reconcileDelete(ctx context.Context, instance *watcherv1beta1.Watcher, helper *helper.Helper) (ctrl.Result, error) {
 	Log := r.GetLogger(ctx)
 	Log.Info(fmt.Sprintf("Reconcile Service '%s' delete started", instance.Name))
@@ -322,5 +422,6 @@ func (r *WatcherReconciler) SetupWithManager(mgr ctrl.Manager) error {
 		Owns(&watcherv1beta1.WatcherApplier{}).
 		Owns(&mariadbv1.MariaDBDatabase{}).
 		Owns(&mariadbv1.MariaDBAccount{}).
+		Owns(&rabbitmqv1.TransportURL{}).
 		Complete(r)
 }

go.mod

@@ -7,6 +7,7 @@ require (
 	github.com/google/uuid v1.6.0
 	github.com/onsi/ginkgo/v2 v2.20.1
 	github.com/onsi/gomega v1.34.1
+	github.com/openstack-k8s-operators/infra-operator/apis v0.5.0
 	github.com/openstack-k8s-operators/lib-common/modules/common v0.5.0
 	github.com/openstack-k8s-operators/lib-common/modules/test v0.5.0
 	github.com/openstack-k8s-operators/mariadb-operator/api v0.5.0
@@ -22,13 +23,13 @@ require (
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cespare/xxhash/v2 v2.2.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
-	github.com/emicklei/go-restful/v3 v3.11.0 // indirect
-	github.com/evanphx/json-patch/v5 v5.8.0 // indirect
+	github.com/emicklei/go-restful/v3 v3.11.2 // indirect
+	github.com/evanphx/json-patch/v5 v5.9.0 // indirect
 	github.com/fsnotify/fsnotify v1.7.0 // indirect
 	github.com/go-logr/zapr v1.3.0 // indirect
-	github.com/go-openapi/jsonpointer v0.19.6 // indirect
-	github.com/go-openapi/jsonreference v0.20.2 // indirect
-	github.com/go-openapi/swag v0.22.3 // indirect
+	github.com/go-openapi/jsonpointer v0.20.2 // indirect
+	github.com/go-openapi/jsonreference v0.20.4 // indirect
+	github.com/go-openapi/swag v0.22.9 // indirect
 	github.com/go-task/slim-sprig/v3 v3.0.0 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
@@ -37,41 +38,40 @@ require (
 	github.com/google/go-cmp v0.6.0 // indirect
 	github.com/google/gofuzz v1.2.0 // indirect
 	github.com/google/pprof v0.0.0-20240727154555-813a5fbdbec8 // indirect
-	github.com/imdario/mergo v0.3.12 // indirect
+	github.com/imdario/mergo v0.3.16 // indirect
 	github.com/josharian/intern v1.0.0 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/k8snetworkplumbingwg/network-attachment-definition-client v1.7.5 // indirect
 	github.com/mailru/easyjson v0.7.7 // indirect
-	github.com/matttproud/golang_protobuf_extensions/v2 v2.0.0 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/openshift/api v3.9.0+incompatible // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/prometheus/client_golang v1.18.0 // indirect
 	github.com/prometheus/client_model v0.5.0 // indirect
-	github.com/prometheus/common v0.45.0 // indirect
+	github.com/prometheus/common v0.46.0 // indirect
 	github.com/prometheus/procfs v0.12.0 // indirect
 	github.com/spf13/pflag v1.0.5 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	golang.org/x/exp v0.0.0-20240719175910-8a7402abbf56 // indirect
 	golang.org/x/mod v0.20.0 // indirect
 	golang.org/x/net v0.28.0 // indirect
-	golang.org/x/oauth2 v0.12.0 // indirect
+	golang.org/x/oauth2 v0.16.0 // indirect
 	golang.org/x/sys v0.23.0 // indirect
 	golang.org/x/term v0.23.0 // indirect
 	golang.org/x/text v0.17.0 // indirect
-	golang.org/x/time v0.3.0 // indirect
+	golang.org/x/time v0.5.0 // indirect
 	golang.org/x/tools v0.24.0 // indirect
 	gomodules.xyz/jsonpatch/v2 v2.4.0 // indirect
-	google.golang.org/appengine v1.6.7 // indirect
+	google.golang.org/appengine v1.6.8 // indirect
 	google.golang.org/protobuf v1.34.1 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	k8s.io/apiextensions-apiserver v0.29.9 // indirect
 	k8s.io/component-base v0.29.9 // indirect
-	k8s.io/klog/v2 v2.110.1 // indirect
-	k8s.io/kube-openapi v0.0.0-20231010175941-2dd684a91f00 // indirect
+	k8s.io/klog/v2 v2.120.1 // indirect
+	k8s.io/kube-openapi v0.0.0-20240228011516-70dd3763d340 // indirect
 	k8s.io/utils v0.0.0-20240711033017-18e509b52bc8 // indirect
 	sigs.k8s.io/json v0.0.0-20221116044647-bc3834ca7abd // indirect
 	sigs.k8s.io/structured-merge-diff/v4 v4.4.1 // indirect