Auto-enable and remove federation plugins on RabbitmqCluster

When a RabbitMQFederation is reconciled, the controller now ensures
that rabbitmq_federation and rabbitmq_federation_management plugins
are enabled on the target RabbitmqCluster CR's additionalPlugins.
When the plugins are first added, reconciliation is requeued for 30s
to allow RabbitMQ to load them before configuring upstreams/policies.

On deletion, if no other RabbitMQFederation CRs reference the same
cluster, the federation plugins are removed from additionalPlugins.
This causes RabbitMQ to restart without those plugins, ensuring a
clean state.

Assisted-by: Claude 4.6 Opus
Co-authored-by: Cursor <cursoragent@cursor.com>

Author: xek

internal/controller/rabbitmq/rabbitmq_controller.go

@@ -380,6 +380,16 @@ func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (result ct
 		return ctrl.Result{}, fmt.Errorf("error configuring RabbitmqCluster: %w", err)
 	}
 
+	// Preserve additionalPlugins that may have been set by the federation controller
+	// on the existing RabbitmqCluster CR, since MarshalInto from the RabbitMq spec
+	// does not include them.
+	existingCluster := &rabbitmqv2.RabbitmqCluster{}
+	if err := r.Client.Get(ctx, types.NamespacedName{Name: instance.Name, Namespace: instance.Namespace}, existingCluster); err == nil {
+		if len(existingCluster.Spec.Rabbitmq.AdditionalPlugins) > 0 && len(rabbitmqCluster.Spec.Rabbitmq.AdditionalPlugins) == 0 {
+			rabbitmqCluster.Spec.Rabbitmq.AdditionalPlugins = existingCluster.Spec.Rabbitmq.AdditionalPlugins
+		}
+	}
+
 	rabbitmqImplCluster := impl.NewRabbitMqCluster(rabbitmqCluster, 5)
 	rmqres, rmqerr := rabbitmqImplCluster.CreateOrPatch(ctx, helper)
 	if rmqerr != nil {

internal/controller/rabbitmq/rabbitmqfederation_controller.go

@@ -32,6 +32,7 @@ import (
 	rabbitmqapi "github.com/openstack-k8s-operators/infra-operator/pkg/rabbitmq/api"
 	condition "github.com/openstack-k8s-operators/lib-common/modules/common/condition"
 	helper "github.com/openstack-k8s-operators/lib-common/modules/common/helper"
+	rabbitmqclusterv2 "github.com/rabbitmq/cluster-operator/v2/api/v1beta1"
 	k8s_errors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
@@ -57,7 +58,7 @@ type RabbitMQFederationReconciler struct {
 //+kubebuilder:rbac:groups=rabbitmq.openstack.org,resources=rabbitmqvhosts/finalizers,verbs=update
 //+kubebuilder:rbac:groups=rabbitmq.openstack.org,resources=rabbitmqs,verbs=get;list;watch;update;patch
 //+kubebuilder:rbac:groups=rabbitmq.openstack.org,resources=rabbitmqs/finalizers,verbs=update
-//+kubebuilder:rbac:groups=rabbitmq.com,resources=rabbitmqclusters,verbs=get;list;watch
+//+kubebuilder:rbac:groups=rabbitmq.com,resources=rabbitmqclusters,verbs=get;list;watch;update;patch
 //+kubebuilder:rbac:groups=core,resources=secrets,verbs=get;list;watch
 
 // Reconcile reconciles a RabbitMQFederation object
@@ -90,9 +91,8 @@ func (r *RabbitMQFederationReconciler) Reconcile(ctx context.Context, req ctrl.R
 		// Restore condition timestamps if they haven't changed
 		condition.RestoreLastTransitionTimes(&instance.Status.Conditions, savedConditions)
 
-		if instance.Status.Conditions.IsUnknown(condition.ReadyCondition) {
-			instance.Status.Conditions.Set(instance.Status.Conditions.Mirror(condition.ReadyCondition))
-		}
+		// Always mirror the ReadyCondition from the federation-specific condition
+		instance.Status.Conditions.Set(instance.Status.Conditions.Mirror(condition.ReadyCondition))
 		if err := h.PatchInstance(ctx, instance); err != nil {
 			Log.Error(err, "Failed to patch instance")
 		}
@@ -221,6 +221,23 @@ func (r *RabbitMQFederationReconciler) reconcileNormal(ctx context.Context, inst
 		return ctrl.Result{RequeueAfter: time.Duration(10) * time.Second}, err
 	}
 
+	// Ensure federation plugins are enabled on the RabbitmqCluster
+	pluginsUpdated, err := r.ensureFederationPlugins(ctx, instance.Spec.RabbitmqClusterName, instance.Namespace)
+	if err != nil {
+		instance.Status.Conditions.Set(condition.FalseCondition(
+			rabbitmqv1.RabbitMQFederationReadyCondition,
+			condition.ErrorReason,
+			condition.SeverityWarning,
+			rabbitmqv1.RabbitMQFederationReadyErrorMessage,
+			fmt.Sprintf("failed to enable federation plugins: %v", err)))
+		return ctrl.Result{RequeueAfter: time.Duration(10) * time.Second}, err
+	}
+	if pluginsUpdated {
+		// Plugins were just enabled - requeue to give RabbitMQ time to load them
+		Log.Info("Federation plugins enabled, requeueing to allow RabbitMQ to load them")
+		return ctrl.Result{RequeueAfter: time.Duration(30) * time.Second}, nil
+	}
+
 	// Get RabbitMQ cluster connection details
 	apiClient, err := r.getRabbitMQClient(ctx, instance.Spec.RabbitmqClusterName, instance.Namespace)
 	if err != nil {
@@ -287,10 +304,13 @@ func (r *RabbitMQFederationReconciler) reconcileNormal(ctx context.Context, inst
 
 	Log.Info("Created/updated federation policy", "policy", policyName, "vhost", vhostName)
 
-	// Set ready condition
+	// Set ready conditions
 	instance.Status.Conditions.Set(condition.TrueCondition(
 		rabbitmqv1.RabbitMQFederationReadyCondition,
 		rabbitmqv1.RabbitMQFederationReadyMessage))
+	instance.Status.Conditions.Set(condition.TrueCondition(
+		condition.ReadyCondition,
+		rabbitmqv1.RabbitMQFederationReadyMessage))
 
 	return ctrl.Result{}, nil
 }
@@ -357,6 +377,12 @@ func (r *RabbitMQFederationReconciler) reconcileDelete(ctx context.Context, inst
 		Log.Info("Deleted federation upstream", "upstream", instance.Spec.UpstreamName, "vhost", vhostName)
 	}
 
+	// Remove federation plugins if no other federations reference this cluster
+	if err := r.removeFederationPluginsIfUnused(ctx, instance.Spec.RabbitmqClusterName, instance.Namespace, instance.Name); err != nil {
+		Log.Info("Failed to remove federation plugins", "cluster", instance.Spec.RabbitmqClusterName, "error", err)
+		// Continue anyway - don't block deletion on cleanup failures
+	}
+
 	// Remove finalizer from vhost if VhostRef is set
 	if instance.Spec.VhostRef != "" {
 		vhostFinalizer := rabbitmqv1.FederationVhostFinalizerPrefix + instance.Name
@@ -468,6 +494,17 @@ func (r *RabbitMQFederationReconciler) buildUpstreamURI(ctx context.Context, ins
 
 // getRabbitMQClient creates a RabbitMQ API client for the specified cluster
 func (r *RabbitMQFederationReconciler) getRabbitMQClient(ctx context.Context, clusterName, namespace string) (*rabbitmqapi.Client, error) {
+	h, err := helper.NewHelper(&rabbitmqv1.RabbitMQFederation{}, r.Client, r.Kclient, r.Scheme, log.FromContext(ctx))
+	if err != nil {
+		return nil, fmt.Errorf("failed to create helper: %w", err)
+	}
+
+	// Get the RabbitmqCluster CR to determine TLS settings
+	rmqCluster := &rabbitmqclusterv2.RabbitmqCluster{}
+	if err := r.Get(ctx, types.NamespacedName{Name: clusterName, Namespace: namespace}, rmqCluster); err != nil {
+		return nil, fmt.Errorf("failed to get RabbitmqCluster %s: %w", clusterName, err)
+	}
+
 	// Get the RabbitMQ cluster credentials
 	secretName := fmt.Sprintf("%s-default-user", clusterName)
 	secret := &corev1.Secret{}
@@ -479,18 +516,128 @@ func (r *RabbitMQFederationReconciler) getRabbitMQClient(ctx context.Context, cl
 	if !ok {
 		return nil, fmt.Errorf("secret %s does not contain 'username' key", secretName)
 	}
-
 	password, ok := secret.Data["password"]
 	if !ok {
 		return nil, fmt.Errorf("secret %s does not contain 'password' key", secretName)
 	}
 
-	// Build the management API URL
-	serviceName := fmt.Sprintf("%s.%s.svc.cluster.local", clusterName, namespace)
-	baseURL := fmt.Sprintf("http://%s:15672", serviceName)
+	// Build the management API URL using the shared helper (handles TLS)
+	baseURL := getManagementURL(rmqCluster, secret)
+
+	// Get TLS CA cert if TLS is enabled
+	caCert, err := getTLSCACert(ctx, h, rmqCluster, namespace)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get TLS CA cert: %w", err)
+	}
+
+	tlsEnabled := rmqCluster.Spec.TLS.SecretName != ""
+	return rabbitmqapi.NewClient(baseURL, string(username), string(password), tlsEnabled, caCert), nil
+}
+
+// federationPlugins are the RabbitMQ plugins required for federation
+var federationPlugins = []rabbitmqclusterv2.Plugin{
+	"rabbitmq_federation",
+	"rabbitmq_federation_management",
+}
+
+// ensureFederationPlugins ensures that the federation plugins are enabled on the RabbitmqCluster CR.
+// The rabbitmq controller preserves additionalPlugins set here when reconciling.
+// Returns true if the cluster was updated (plugins were added).
+func (r *RabbitMQFederationReconciler) ensureFederationPlugins(ctx context.Context, clusterName, namespace string) (bool, error) {
+	Log := log.FromContext(ctx)
+
+	rmqCluster := &rabbitmqclusterv2.RabbitmqCluster{}
+	if err := r.Get(ctx, types.NamespacedName{Name: clusterName, Namespace: namespace}, rmqCluster); err != nil {
+		return false, fmt.Errorf("failed to get RabbitmqCluster %s: %w", clusterName, err)
+	}
+
+	existing := rmqCluster.Spec.Rabbitmq.AdditionalPlugins
+	updated := false
+	for _, plugin := range federationPlugins {
+		found := false
+		for _, p := range existing {
+			if p == plugin {
+				found = true
+				break
+			}
+		}
+		if !found {
+			existing = append(existing, plugin)
+			updated = true
+		}
+	}
+
+	if updated {
+		rmqCluster.Spec.Rabbitmq.AdditionalPlugins = existing
+		if err := r.Update(ctx, rmqCluster); err != nil {
+			return false, fmt.Errorf("failed to update RabbitmqCluster %s with federation plugins: %w", clusterName, err)
+		}
+		Log.Info("Enabled federation plugins on RabbitmqCluster", "cluster", clusterName,
+			"plugins", federationPlugins)
+	}
+
+	return updated, nil
+}
+
+// removeFederationPluginsIfUnused removes federation plugins from the RabbitmqCluster CR
+// if no other RabbitMQFederation CRs reference this cluster.
+func (r *RabbitMQFederationReconciler) removeFederationPluginsIfUnused(ctx context.Context, clusterName, namespace, excludeFederationName string) error {
+	Log := log.FromContext(ctx)
+
+	// Check if any other RabbitMQFederation CRs still reference this cluster
+	fedList := &rabbitmqv1.RabbitMQFederationList{}
+	if err := r.List(ctx, fedList, client.InNamespace(namespace)); err != nil {
+		return fmt.Errorf("failed to list RabbitMQFederation resources: %w", err)
+	}
+
+	for i := range fedList.Items {
+		fed := &fedList.Items[i]
+		// Skip the federation being deleted and any that are also being deleted
+		if fed.Name == excludeFederationName || !fed.DeletionTimestamp.IsZero() {
+			continue
+		}
+		if fed.Spec.RabbitmqClusterName == clusterName {
+			Log.Info("Other federation still references this cluster, keeping plugins",
+				"cluster", clusterName, "federation", fed.Name)
+			return nil
+		}
+	}
+
+	// No other federations reference this cluster - remove the plugins
+	rmqCluster := &rabbitmqclusterv2.RabbitmqCluster{}
+	if err := r.Get(ctx, types.NamespacedName{Name: clusterName, Namespace: namespace}, rmqCluster); err != nil {
+		if k8s_errors.IsNotFound(err) {
+			return nil
+		}
+		return fmt.Errorf("failed to get RabbitmqCluster %s: %w", clusterName, err)
+	}
+
+	filtered := make([]rabbitmqclusterv2.Plugin, 0, len(rmqCluster.Spec.Rabbitmq.AdditionalPlugins))
+	removed := false
+	for _, p := range rmqCluster.Spec.Rabbitmq.AdditionalPlugins {
+		isFedPlugin := false
+		for _, fp := range federationPlugins {
+			if p == fp {
+				isFedPlugin = true
+				break
+			}
+		}
+		if !isFedPlugin {
+			filtered = append(filtered, p)
+		} else {
+			removed = true
+		}
+	}
+
+	if removed {
+		rmqCluster.Spec.Rabbitmq.AdditionalPlugins = filtered
+		if err := r.Update(ctx, rmqCluster); err != nil {
+			return fmt.Errorf("failed to remove federation plugins from RabbitmqCluster %s: %w", clusterName, err)
+		}
+		Log.Info("Removed federation plugins from RabbitmqCluster", "cluster", clusterName)
+	}
 
-	// Create and return the client
-	return rabbitmqapi.NewClient(baseURL, string(username), string(password), false, nil), nil
+	return nil
 }
 
 // SetupWithManager sets up the controller with the Manager.