Use quorum queues if enabled

With openstack-k8s-operators/infra-operator#429
we added the ability to configure RabbitMQ to use quorum queues.

This commit adds a check to verify if "quorumqueues" has been added to
the TransportURL secret. If its value is true, the services will be
configured to use quorum queues.

Jira: https://issues.redhat.com/browse/OSPRH-19160

Assisted-by: claude-4-sonnet

Author: lmiccini

api/go.mod

@@ -4,7 +4,7 @@ go 1.21
 
 require (
 	github.com/google/go-cmp v0.7.0
-	github.com/openstack-k8s-operators/infra-operator/apis v0.6.1-0.20250813063935-fdc20530dcf1
+	github.com/openstack-k8s-operators/infra-operator/apis v0.6.1-0.20250909143828-e33d35ffd64f
 	github.com/openstack-k8s-operators/lib-common/modules/common v0.6.1-0.20250730071847-837b07f8d72f
 	github.com/robfig/cron/v3 v3.0.1
 	k8s.io/api v0.29.15

api/go.sum

@@ -73,8 +73,8 @@ github.com/onsi/ginkgo/v2 v2.20.1 h1:YlVIbqct+ZmnEph770q9Q7NVAz4wwIiVNahee6JyUzo
 github.com/onsi/ginkgo/v2 v2.20.1/go.mod h1:lG9ey2Z29hR41WMVthyJBGUBcBhGOtoPF2VFMvBXFCI=
 github.com/onsi/gomega v1.34.1 h1:EUMJIKUjM8sKjYbtxQI9A4z2o+rruxnzNvpknOXie6k=
 github.com/onsi/gomega v1.34.1/go.mod h1:kU1QgUvBDLXBJq618Xvm2LUX6rSAfRaFRTcdOeDLwwY=
-github.com/openstack-k8s-operators/infra-operator/apis v0.6.1-0.20250813063935-fdc20530dcf1 h1:77TRnwfSxNI5cn/RxMS9I+kqefMm7XRQsNknIhEE4tg=
-github.com/openstack-k8s-operators/infra-operator/apis v0.6.1-0.20250813063935-fdc20530dcf1/go.mod h1:Dv8qpmBIQy3Jv/EyQnOyc0w61X8vyfxpjcIQONP5CwY=
+github.com/openstack-k8s-operators/infra-operator/apis v0.6.1-0.20250909143828-e33d35ffd64f h1:chuu4iBT5sXHYw8aPeP/pWC+S3yGo6hdy39foP7c5vs=
+github.com/openstack-k8s-operators/infra-operator/apis v0.6.1-0.20250909143828-e33d35ffd64f/go.mod h1:Dv8qpmBIQy3Jv/EyQnOyc0w61X8vyfxpjcIQONP5CwY=
 github.com/openstack-k8s-operators/lib-common/modules/common v0.6.1-0.20250730071847-837b07f8d72f h1:DW8aNjEtDFrWiZ6vWuOXwdRB4eBD0n+bA9foQkOEx6U=
 github.com/openstack-k8s-operators/lib-common/modules/common v0.6.1-0.20250730071847-837b07f8d72f/go.mod h1:P+7F1wiwZUxOy4myYXFyc/uBtGATDFpk3yAllXe1Vzk=
 github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=

controllers/common.go

@@ -100,6 +100,14 @@ const (
 	// top level notification message bus transport URL
 	NotificationTransportURLSelector = "notification_transport_url"
 
+	// QuorumQueuesSelector is the name of key in the TransportURL Secret for
+	// the message bus quorum queues
+	QuorumQueuesSelector = "quorumqueues"
+
+	// QuorumQueuesTemplateKey is the name of key in template parameters for
+	// the message bus quorum queues configuration
+	QuorumQueuesTemplateKey = "quorum_queues"
+
 	// fields to index to reconcile when change
 	passwordSecretField        = ".spec.secret"
 	caBundleSecretNameField    = ".spec.tls.caBundleSecretName" // #nosec G101
@@ -747,3 +755,11 @@ func SortNovaCellListByName(cellList *novav1.NovaCellList) {
 		return cellList.Items[i].Name < cellList.Items[j].Name
 	})
 }
+
+// parseQuorumQueues parses the quorum queues value from secret data
+func parseQuorumQueues(data []byte) bool {
+	if data == nil {
+		return false
+	}
+	return string(data) == "true"
+}

controllers/nova_controller.go

@@ -366,7 +366,7 @@ func (r *NovaReconciler) Reconcile(ctx context.Context, req ctrl.Request) (resul
 	// Create TransportURLs to access the message buses of each cell. Cell0
 	// message bus is always the same as the top level API message bus so
 	// we create API MQ separately first
-	apiTransportURL, apiMQStatus, apiMQError := r.ensureMQ(
+	apiTransportURL, apiQuorumQueues, apiMQStatus, apiMQError := r.ensureMQ(
 		ctx, h, instance, instance.Name+"-api-transport", instance.Spec.APIMessageBusInstance)
 	switch apiMQStatus {
 	case nova.MQFailed:
@@ -403,7 +403,7 @@ func (r *NovaReconciler) Reconcile(ctx context.Context, req ctrl.Request) (resul
 
 	notificationTransportURLName := instance.Name + "-notification-transport"
 	if notificationBusName != "" {
-		notificationTransportURL, notificationMQStatus, notificationMQError = r.ensureMQ(
+		notificationTransportURL, _, notificationMQStatus, notificationMQError = r.ensureMQ(
 			ctx, h, instance, notificationTransportURLName, notificationBusName)
 
 		switch notificationMQStatus {
@@ -459,14 +459,16 @@ func (r *NovaReconciler) Reconcile(ctx context.Context, req ctrl.Request) (resul
 		var status nova.MessageBusStatus
 		var err error
 		cellTemplate := instance.Spec.CellTemplates[cellName]
+		var cellQuorumQueues bool
 		// cell0 does not need its own cell message bus it uses the
 		// API message bus instead
 		if cellName == novav1.Cell0Name {
 			cellTransportURL = apiTransportURL
+			cellQuorumQueues = apiQuorumQueues
 			status = apiMQStatus
 			err = apiMQError
 		} else {
-			cellTransportURL, status, err = r.ensureMQ(
+			cellTransportURL, cellQuorumQueues, status, err = r.ensureMQ(
 				ctx, h, instance, instance.Name+"-"+cellName+"-transport", cellTemplate.CellMessageBusInstance)
 		}
 		switch status {
@@ -478,7 +480,7 @@ func (r *NovaReconciler) Reconcile(ctx context.Context, req ctrl.Request) (resul
 		default:
 			return ctrl.Result{}, fmt.Errorf("%w from ensureMQ: %d for cell %s", util.ErrInvalidStatus, status, cellName)
 		}
-		cellMQs[cellName] = &nova.MessageBus{TransportURL: cellTransportURL, Status: status}
+		cellMQs[cellName] = &nova.MessageBus{TransportURL: cellTransportURL, QuorumQueues: cellQuorumQueues, Status: status}
 	}
 	if len(failedMQs) > 0 {
 		instance.Status.Conditions.Set(condition.FalseCondition(
@@ -541,7 +543,7 @@ func (r *NovaReconciler) Reconcile(ctx context.Context, req ctrl.Request) (resul
 		}
 		cell, status, err := r.ensureCell(
 			ctx, h, instance, cellName, cellTemplate,
-			cellDB.Database, apiDB, cellMQ.TransportURL, notificationTransportURL,
+			cellDB.Database, apiDB, cellMQ.TransportURL, cellMQ.QuorumQueues, notificationTransportURL,
 			keystoneInternalAuthURL, secret,
 		)
 		cells[cellName] = cell
@@ -608,7 +610,7 @@ func (r *NovaReconciler) Reconcile(ctx context.Context, req ctrl.Request) (resul
 
 	topLevelSecretName, err := r.ensureTopLevelSecret(
 		ctx, h, instance,
-		apiTransportURL,
+		apiTransportURL, apiQuorumQueues,
 		notificationTransportURL,
 		secret)
 	if err != nil {
@@ -1196,6 +1198,7 @@ func (r *NovaReconciler) ensureCell(
 	cellDB *mariadbv1.Database,
 	apiDB *mariadbv1.Database,
 	cellTransportURL string,
+	cellQuorumQueues bool,
 	notificationTransportURL string,
 	keystoneAuthURL string,
 	secret corev1.Secret,
@@ -1204,7 +1207,7 @@ func (r *NovaReconciler) ensureCell(
 
 	cellSecretName, err := r.ensureCellSecret(
 		ctx, h, instance, cellName, cellTemplate,
-		cellTransportURL, notificationTransportURL,
+		cellTransportURL, cellQuorumQueues, notificationTransportURL,
 		secret)
 	if err != nil {
 		return nil, nova.CellDeploying, err
@@ -1708,7 +1711,7 @@ func (r *NovaReconciler) ensureMQ(
 	instance *novav1.Nova,
 	transportName string,
 	messageBusInstanceName string,
-) (string, nova.MessageBusStatus, error) {
+) (string, bool, nova.MessageBusStatus, error) {
 	Log := r.GetLogger(ctx)
 	transportURL := &rabbitmqv1.TransportURL{
 		ObjectMeta: metav1.ObjectMeta{
@@ -1725,7 +1728,7 @@ func (r *NovaReconciler) ensureMQ(
 	})
 
 	if err != nil && !k8s_errors.IsNotFound(err) {
-		return "", nova.MQFailed, util.WrapErrorForObject(
+		return "", false, nova.MQFailed, util.WrapErrorForObject(
 			fmt.Sprintf("Error create or update TransportURL object %s", transportName),
 			transportURL,
 			err,
@@ -1734,20 +1737,20 @@ func (r *NovaReconciler) ensureMQ(
 
 	if op != controllerutil.OperationResultNone {
 		Log.Info(fmt.Sprintf("TransportURL object %s created or patched", transportName))
-		return "", nova.MQCreating, nil
+		return "", false, nova.MQCreating, nil
 	}
 
 	err = r.Client.Get(ctx, types.NamespacedName{Namespace: instance.Namespace, Name: transportName}, transportURL)
 	if err != nil && !k8s_errors.IsNotFound(err) {
-		return "", nova.MQFailed, util.WrapErrorForObject(
+		return "", false, nova.MQFailed, util.WrapErrorForObject(
 			fmt.Sprintf("Error reading TransportURL object %s", transportName),
 			transportURL,
 			err,
 		)
 	}
 
 	if k8s_errors.IsNotFound(err) || !transportURL.IsReady() || transportURL.Status.SecretName == "" {
-		return "", nova.MQCreating, nil
+		return "", false, nova.MQCreating, nil
 	}
 
 	secretName := types.NamespacedName{Namespace: instance.Namespace, Name: transportURL.Status.SecretName}
@@ -1756,17 +1759,24 @@ func (r *NovaReconciler) ensureMQ(
 	err = h.GetClient().Get(ctx, secretName, secret)
 	if err != nil {
 		if k8s_errors.IsNotFound(err) {
-			return "", nova.MQCreating, nil
+			return "", false, nova.MQCreating, nil
 		}
-		return "", nova.MQFailed, err
+		return "", false, nova.MQFailed, err
 	}
 
 	url, ok := secret.Data[TransportURLSelector]
 	if !ok {
-		return "", nova.MQFailed, fmt.Errorf(
+		return "", false, nova.MQFailed, fmt.Errorf(
 			"%w: the TransportURL secret %s does not have 'transport_url' field", util.ErrFieldNotFound, transportURL.Status.SecretName)
 	}
-	return string(url), nova.MQCompleted, nil
+
+	// Check if quorum queues are enabled
+	quorumQueues := false
+	if val, ok := secret.Data[QuorumQueuesSelector]; ok {
+		quorumQueues = string(val) == "true"
+	}
+
+	return string(url), quorumQueues, nova.MQCompleted, nil
 }
 
 func (r *NovaReconciler) ensureMQDeleted(
@@ -1993,15 +2003,22 @@ func (r *NovaReconciler) ensureCellSecret(
 	cellName string,
 	cellTemplate novav1.NovaCellTemplate,
 	cellTransportURL string,
+	cellQuorumQueues bool,
 	notificationTransportURL string,
 	externalSecret corev1.Secret,
 ) (string, error) {
 	// NOTE(gibi): We can move other sensitive data to the internal Secret from
 	// the NovaCellSpec fields, possibly hostnames or usernames.
+	quorumQueuesValue := "false"
+	if cellQuorumQueues {
+		quorumQueuesValue = "true"
+	}
+
 	data := map[string]string{
 		ServicePasswordSelector:          string(externalSecret.Data[instance.Spec.PasswordSelectors.Service]),
 		TransportURLSelector:             cellTransportURL,
 		NotificationTransportURLSelector: notificationTransportURL,
+		QuorumQueuesTemplateKey:          quorumQueuesValue,
 	}
 
 	// If metadata is enabled in the cell then the cell secret needs the
@@ -2045,16 +2062,23 @@ func (r *NovaReconciler) ensureTopLevelSecret(
 	h *helper.Helper,
 	instance *novav1.Nova,
 	apiTransportURL string,
+	apiQuorumQueues bool,
 	notificationTransportURL string,
 	externalSecret corev1.Secret,
 ) (string, error) {
 	// NOTE(gibi): We can move other sensitive data to the internal Secret from
 	// the subCR fields, possibly hostnames or usernames.
+	quorumQueuesValue := "false"
+	if apiQuorumQueues {
+		quorumQueuesValue = "true"
+	}
+
 	data := map[string]string{
 		ServicePasswordSelector:          string(externalSecret.Data[instance.Spec.PasswordSelectors.Service]),
 		MetadataSecretSelector:           string(externalSecret.Data[instance.Spec.PasswordSelectors.MetadataSecret]),
 		TransportURLSelector:             apiTransportURL,
 		NotificationTransportURLSelector: notificationTransportURL,
+		QuorumQueuesTemplateKey:          quorumQueuesValue,
 	}
 
 	// NOTE(gibi): When we switch to immutable secrets then we need to include

controllers/novaapi_controller.go

@@ -515,6 +515,7 @@ func (r *NovaAPIReconciler) generateConfigs(
 		"MemcachedServers":           memcachedInstance.GetMemcachedServerListString(),
 		"MemcachedServersWithInet":   memcachedInstance.GetMemcachedServerListWithInetString(),
 		"MemcachedTLS":               memcachedInstance.GetMemcachedTLSSupport(),
+		QuorumQueuesTemplateKey:      parseQuorumQueues(secret.Data[QuorumQueuesTemplateKey]),
 	}
 	// create httpd  vhost template parameters
 	httpdVhostConfig := map[string]interface{}{}

controllers/novacell_controller.go

@@ -780,6 +780,7 @@ func (r *NovaCellReconciler) generateComputeConfigs(
 		"compute_driver":             "libvirt.LibvirtDriver",
 		"transport_url":              string(secret.Data[TransportURLSelector]),
 		"notification_transport_url": string(secret.Data[NotificationTransportURLSelector]),
+		QuorumQueuesTemplateKey:      parseQuorumQueues(secret.Data[QuorumQueuesTemplateKey]),
 	}
 	// vnc is optional so we only need to configure it for the compute
 	// if the proxy service is deployed in the cell

controllers/novacompute_controller.go

@@ -369,7 +369,8 @@ func (r *NovaComputeReconciler) generateConfigs(
 		"notification_transport_url": string(secret.Data[NotificationTransportURLSelector]),
 		"compute_driver":             instance.Spec.ComputeDriver,
 		// Neither the ironic driver nor the fake driver support VNC
-		"vnc_enabled": false,
+		"vnc_enabled":           false,
+		QuorumQueuesTemplateKey: parseQuorumQueues(secret.Data[QuorumQueuesTemplateKey]),
 	}
 
 	extraData := map[string]string{}

controllers/novaconductor_controller.go

@@ -468,6 +468,7 @@ func (r *NovaConductorReconciler) generateConfigs(
 		"MemcachedServers":           memcachedInstance.GetMemcachedServerListString(),
 		"MemcachedServersWithInet":   memcachedInstance.GetMemcachedServerListWithInetString(),
 		"MemcachedTLS":               memcachedInstance.GetMemcachedTLSSupport(),
+		QuorumQueuesTemplateKey:      parseQuorumQueues(secret.Data[QuorumQueuesTemplateKey]),
 	}
 	if len(instance.Spec.APIDatabaseHostname) > 0 {
 		apiDatabaseAccount, apiDbSecret, err := mariadbv1.GetAccountAndSecret(ctx, h, instance.Spec.APIDatabaseAccount, instance.Namespace)

controllers/novametadata_controller.go

@@ -507,6 +507,7 @@ func (r *NovaMetadataReconciler) generateConfigs(
 		"MemcachedServers":         memcachedInstance.GetMemcachedServerListString(),
 		"MemcachedServersWithInet": memcachedInstance.GetMemcachedServerListWithInetString(),
 		"MemcachedTLS":             memcachedInstance.GetMemcachedTLSSupport(),
+		QuorumQueuesTemplateKey:    parseQuorumQueues(secret.Data[QuorumQueuesTemplateKey]),
 		"TimeOut":                  instance.Spec.APITimeout,
 	}
 

controllers/novanovncproxy_controller.go

@@ -483,6 +483,7 @@ func (r *NovaNoVNCProxyReconciler) generateConfigs(
 		"MemcachedServers":         memcachedInstance.GetMemcachedServerListString(),
 		"MemcachedServersWithInet": memcachedInstance.GetMemcachedServerListWithInetString(),
 		"MemcachedTLS":             memcachedInstance.GetMemcachedTLSSupport(),
+		QuorumQueuesTemplateKey:    parseQuorumQueues(secret.Data[QuorumQueuesTemplateKey]),
 	}
 	if instance.Spec.TLS.Service.Enabled() {
 		templateParameters["SSLCertificateFile"] = fmt.Sprintf("/etc/pki/tls/certs/%s.crt", novncproxy.ServiceName)