Merge pull request #606 from rabbitmq/persistence-resize

Support Persistence expansion

.github/workflows/pr.yml

@@ -62,7 +62,7 @@ jobs:
         make install-tools
         kind create cluster --image kindest/node:"$K8S_VERSION"
         DOCKER_REGISTRY_SERVER=local-server OPERATOR_IMAGE=local-operator make deploy-kind
-        make system-tests
+        SUPPORT_VOLUME_EXPANSION=false make system-tests
 
   kubectl_tests:
     name: kubectl rabbitmq tests

api/v1beta1/rabbitmqcluster_types.go

@@ -9,6 +9,7 @@
 package v1beta1
 
 import (
+	"strconv"
 	"strings"
 
 	appsv1 "k8s.io/api/apps/v1"
@@ -364,6 +365,10 @@ func (cluster RabbitmqCluster) ChildResourceName(name string) string {
 	return strings.TrimSuffix(strings.Join([]string{cluster.Name, name}, "-"), "-")
 }
 
+func (cluster RabbitmqCluster) PVCName(i int) string {
+	return strings.Join([]string{"persistence", cluster.Name, "server", strconv.Itoa(i)}, "-")
+}
+
 func init() {
 	SchemeBuilder.Register(&RabbitmqCluster{}, &RabbitmqClusterList{})
 }

api/v1beta1/rabbitmqcluster_types_test.go

@@ -456,6 +456,12 @@ var _ = Describe("RabbitmqCluster", func() {
 			Expect(updatedCondition.LastTransitionTime.Before(&notExpectedTime)).To(BeFalse())
 		})
 	})
+	Context("PVC Name helper function", func() {
+		It("returns the correct PVC name", func() {
+			r := generateRabbitmqClusterObject("testrabbit")
+			Expect(r.PVCName(0)).To(Equal("persistence-testrabbit-server-0"))
+		})
+	})
 })
 
 func getKey(cluster *RabbitmqCluster) types.NamespacedName {

config/rbac/role.yaml

@@ -40,6 +40,16 @@ rules:
   - create
   - get
   - patch
+- apiGroups:
+  - ""
+  resources:
+  - persistentvolumeclaims
+  verbs:
+  - create
+  - get
+  - list
+  - update
+  - watch
 - apiGroups:
   - ""
   resources:

controllers/rabbitmqcluster_controller.go

@@ -78,6 +78,7 @@ type RabbitmqClusterReconciler struct {
 // +kubebuilder:rbac:groups=rabbitmq.com,resources=rabbitmqclusters/finalizers,verbs=update
 // +kubebuilder:rbac:groups="",resources=events,verbs=get;create;patch
 // +kubebuilder:rbac:groups="",resources=serviceaccounts,verbs=get;list;watch;create;update
+// +kubebuilder:rbac:groups="",resources=persistentvolumeclaims,verbs=get;list;watch;create;update
 // +kubebuilder:rbac:groups="rbac.authorization.k8s.io",resources=roles,verbs=get;list;watch;create;update
 // +kubebuilder:rbac:groups="rbac.authorization.k8s.io",resources=rolebindings,verbs=get;list;watch;create;update
 
@@ -160,6 +161,17 @@ func (r *RabbitmqClusterReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 			return ctrl.Result{}, err
 		}
 
+		// only StatefulSetBuilder returns true
+		if  builder.UpdateMayRequireStsRecreate() {
+			if err = r.reconcilePVC(ctx, builder, rabbitmqCluster, resource); err != nil {
+				rabbitmqCluster.Status.SetCondition(status.ReconcileSuccess, corev1.ConditionFalse, "FailedReconcilePVC", err.Error())
+				if statusErr := r.Status().Update(ctx, rabbitmqCluster); statusErr != nil {
+					logger.Error(statusErr, "Failed to update ReconcileSuccess condition state")
+				}
+				return ctrl.Result{}, err
+			}
+		}
+
 		var operationResult controllerutil.OperationResult
 		err = clientretry.RetryOnConflict(clientretry.DefaultRetry, func() error {
 			var apiError error

controllers/reconcile_persistence.go

@@ -0,0 +1,173 @@
+package controllers
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"github.com/go-logr/logr"
+	rabbitmqv1beta1 "github.com/rabbitmq/cluster-operator/api/v1beta1"
+	"github.com/rabbitmq/cluster-operator/internal/resource"
+	appsv1 "k8s.io/api/apps/v1"
+	corev1 "k8s.io/api/core/v1"
+	k8serrors "k8s.io/apimachinery/pkg/api/errors"
+	k8sresource "k8s.io/apimachinery/pkg/api/resource"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"time"
+)
+
+func (r *RabbitmqClusterReconciler) reconcilePVC(ctx context.Context, builder resource.ResourceBuilder, cluster *rabbitmqv1beta1.RabbitmqCluster, resource client.Object) error {
+	logger := ctrl.LoggerFrom(ctx)
+
+	sts := resource.(*appsv1.StatefulSet)
+	current, err := r.statefulSet(ctx, cluster)
+
+	if client.IgnoreNotFound(err) != nil {
+		return err
+	} else if k8serrors.IsNotFound(err) {
+		logger.Info("statefulSet not created yet, skipping checks to expand PersistentVolumeClaims")
+		return nil
+	}
+
+	if err := builder.Update(sts); err != nil {
+		return err
+	}
+
+	resize, err := r.needsPVCResize(current, sts)
+	if err != nil {
+		return err
+	}
+
+	if resize {
+		if err := r.expandPVC(ctx, cluster, current, sts); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (r *RabbitmqClusterReconciler) expandPVC(ctx context.Context, rmq *rabbitmqv1beta1.RabbitmqCluster, current, desired *appsv1.StatefulSet) error {
+	logger := ctrl.LoggerFrom(ctx)
+
+	currentCapacity, err := persistenceStorageCapacity(current.Spec.VolumeClaimTemplates)
+	if err != nil {
+		return err
+	}
+
+	desiredCapacity, err := persistenceStorageCapacity(desired.Spec.VolumeClaimTemplates)
+	if err != nil {
+		return err
+	}
+
+	logger.Info(fmt.Sprintf("updating storage capacity from %s to %s", currentCapacity.String(), desiredCapacity.String()))
+
+	if err := r.deleteSts(ctx, rmq); err != nil {
+		return err
+	}
+
+	if err := r.updatePVC(ctx, rmq, *current.Spec.Replicas, desiredCapacity); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (r *RabbitmqClusterReconciler) updatePVC(ctx context.Context, rmq *rabbitmqv1beta1.RabbitmqCluster, replicas int32, desiredCapacity k8sresource.Quantity) error {
+	logger := ctrl.LoggerFrom(ctx)
+	logger.Info("expanding PersistentVolumeClaims")
+
+	for i := 0; i < int(replicas); i++ {
+		PVCName := rmq.PVCName(i)
+		PVC := corev1.PersistentVolumeClaim{}
+
+		if err := r.Client.Get(ctx, types.NamespacedName{Namespace: rmq.Namespace, Name: PVCName}, &PVC); err != nil {
+			msg := "failed to get PersistentVolumeClaim"
+			logger.Error(err, msg, "PersistentVolumeClaim", PVCName)
+			r.Recorder.Event(rmq, corev1.EventTypeWarning, "FailedReconcilePersistence", fmt.Sprintf("%s %s", msg, PVCName))
+			return fmt.Errorf("%s %s: %v", msg, PVCName, err)
+		}
+		PVC.Spec.Resources.Requests[corev1.ResourceStorage] = desiredCapacity
+		if err := r.Client.Update(ctx, &PVC, &client.UpdateOptions{}); err != nil {
+			msg := "failed to update PersistentVolumeClaim"
+			logger.Error(err, msg, "PersistentVolumeClaim", PVCName)
+			r.Recorder.Event(rmq, corev1.EventTypeWarning, "FailedReconcilePersistence", fmt.Sprintf("%s %s", msg, PVCName))
+			return fmt.Errorf("%s %s: %v", msg, PVCName, err)
+		}
+		logger.Info("successfully expanded", "PVC", PVCName)
+	}
+	return nil
+}
+
+func (r *RabbitmqClusterReconciler) needsPVCResize(current, desired *appsv1.StatefulSet) (bool, error) {
+	currentCapacity, err := persistenceStorageCapacity(current.Spec.VolumeClaimTemplates)
+	if err != nil {
+		return false, err
+	}
+
+	desiredCapacity, err := persistenceStorageCapacity(desired.Spec.VolumeClaimTemplates)
+	if err != nil {
+		return false, err
+	}
+
+	if currentCapacity.Cmp(desiredCapacity) != 0 {
+		return true, nil
+	}
+
+	return false, nil
+}
+
+func persistenceStorageCapacity(templates []corev1.PersistentVolumeClaim) (k8sresource.Quantity, error) {
+	for _, t := range templates {
+		if t.Name == "persistence" {
+			return t.Spec.Resources.Requests[corev1.ResourceStorage], nil
+		}
+	}
+	return k8sresource.Quantity{}, errors.New("cannot find PersistentVolumeClaim 'persistence'")
+}
+
+// deleteSts deletes a sts without deleting pods and PVCs
+// using DeletePropagationPolicy set to 'Orphan'
+func (r *RabbitmqClusterReconciler) deleteSts(ctx context.Context, rmq *rabbitmqv1beta1.RabbitmqCluster) error {
+	logger := ctrl.LoggerFrom(ctx)
+	logger.Info("deleting statefulSet (pods won't be deleted)", "statefulSet", rmq.ChildResourceName("server"))
+	deletePropagationPolicy := metav1.DeletePropagationOrphan
+	deleteOptions := &client.DeleteOptions{PropagationPolicy: &deletePropagationPolicy}
+	currentSts, err := r.statefulSet(ctx, rmq)
+	if err != nil {
+		return err
+	}
+	if err := r.Delete(ctx, currentSts, deleteOptions); err != nil {
+		msg := "failed to delete statefulSet"
+		logger.Error(err, msg, "statefulSet", currentSts.Name)
+		r.Recorder.Event(rmq, corev1.EventTypeWarning, "FailedReconcilePersistence", fmt.Sprintf("%s %s", msg, currentSts.Name))
+		return fmt.Errorf("%s %s: %v", msg, currentSts.Name, err)
+	}
+
+	if err := retryWithInterval(logger, "delete statefulSet", 10, 3*time.Second, func() bool {
+		_, getErr := r.statefulSet(ctx, rmq)
+		if k8serrors.IsNotFound(getErr) {
+			return true
+		}
+		return false
+	}); err != nil {
+		msg := "statefulSet not deleting after 30 seconds"
+		logger.Error(err, msg, "statefulSet", currentSts.Name)
+		r.Recorder.Event(rmq, corev1.EventTypeWarning, "FailedReconcilePersistence", fmt.Sprintf("%s %s", msg, currentSts.Name))
+		return fmt.Errorf("%s %s: %v", msg, currentSts.Name, err)
+	}
+	logger.Info("statefulSet deleted", "statefulSet", currentSts.Name)
+	return nil
+}
+
+func retryWithInterval(logger logr.Logger, msg string, retry int, interval time.Duration, f func() bool) (err error) {
+	for i := 0; i < retry; i++ {
+		if ok := f(); ok {
+			return
+		}
+		time.Sleep(interval)
+		logger.Info("retrying again", "action", msg, "interval", interval, "attempt", i+1)
+	}
+	return fmt.Errorf("failed to %s after %d retries", msg, retry)
+}

controllers/utils.go

@@ -2,15 +2,14 @@ package controllers
 
 import (
 	"context"
-	"sigs.k8s.io/controller-runtime/pkg/client"
-
 	rabbitmqv1beta1 "github.com/rabbitmq/cluster-operator/api/v1beta1"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/util/retry"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
 func (r *RabbitmqClusterReconciler) exec(namespace, podName, containerName string, command ...string) (string, string, error) {

internal/resource/configmap.go

@@ -77,6 +77,10 @@ func (builder *ServerConfigMapBuilder) Build() (client.Object, error) {
 	}, nil
 }
 
+func (builder *ServerConfigMapBuilder) UpdateMayRequireStsRecreate() bool {
+	return false
+}
+
 func (builder *ServerConfigMapBuilder) Update(object client.Object) error {
 	configMap := object.(*corev1.ConfigMap)
 

internal/resource/configmap_test.go

@@ -542,6 +542,12 @@ CONSOLE_LOG=new`
 			Expect(configMap.Annotations).To(BeEmpty())
 		})
 	})
+
+	Context("UpdateMayRequireStsRecreate", func() {
+		It("returns false", func() {
+			Expect(configMapBuilder.UpdateMayRequireStsRecreate()).To(BeFalse())
+		})
+	})
 })
 
 // iniString formats the input string using "gopkg.in/ini.v1"

internal/resource/default_user_secret.go

@@ -70,6 +70,10 @@ func (builder *DefaultUserSecretBuilder) Build() (client.Object, error) {
 	}, nil
 }
 
+func (builder *DefaultUserSecretBuilder) UpdateMayRequireStsRecreate() bool {
+	return false
+}
+
 func (builder *DefaultUserSecretBuilder) Update(object client.Object) error {
 	secret := object.(*corev1.Secret)
 	secret.Labels = metadata.GetLabels(builder.Instance.Name, builder.Instance.Labels)

internal/resource/default_user_secret_test.go

@@ -201,4 +201,10 @@ var _ = Describe("DefaultUserSecret", func() {
 		Expect(defaultUserSecretBuilder.Update(secret)).NotTo(HaveOccurred())
 		Expect(secret.OwnerReferences[0].Name).To(Equal(instance.Name))
 	})
+
+	Context("UpdateMayRequireStsRecreate", func() {
+		It("returns false", func() {
+			Expect(defaultUserSecretBuilder.UpdateMayRequireStsRecreate()).To(BeFalse())
+		})
+	})
 })

internal/resource/erlang_cookie.go

@@ -57,6 +57,10 @@ func (builder *ErlangCookieBuilder) Build() (client.Object, error) {
 	}, nil
 }
 
+func (builder *ErlangCookieBuilder) UpdateMayRequireStsRecreate() bool {
+	return false
+}
+
 func (builder *ErlangCookieBuilder) Update(object client.Object) error {
 	secret := object.(*corev1.Secret)
 	secret.Labels = metadata.GetLabels(builder.Instance.Name, builder.Instance.Labels)

internal/resource/erlang_cookie_test.go

@@ -174,4 +174,10 @@ var _ = Describe("ErlangCookie", func() {
 		Expect(erlangCookieBuilder.Update(secret)).NotTo(HaveOccurred())
 		Expect(secret.OwnerReferences[0].Name).To(Equal(instance.Name))
 	})
+
+	Context("UpdateMayRequireStsRecreate", func() {
+		It("returns false", func() {
+			Expect(erlangCookieBuilder.UpdateMayRequireStsRecreate()).To(BeFalse())
+		})
+	})
 })

internal/resource/headless_service.go

@@ -38,6 +38,10 @@ func (builder *RabbitmqResourceBuilder) HeadlessService() *HeadlessServiceBuilde
 	}
 }
 
+func (builder *HeadlessServiceBuilder) UpdateMayRequireStsRecreate() bool {
+	return false
+}
+
 func (builder *HeadlessServiceBuilder) Build() (client.Object, error) {
 	return &corev1.Service{
 		ObjectMeta: metav1.ObjectMeta{

internal/resource/headless_service_test.go

@@ -213,4 +213,10 @@ var _ = Describe("HeadlessService", func() {
 		Expect(serviceBuilder.Update(service)).NotTo(HaveOccurred())
 		Expect(service.OwnerReferences[0].Name).To(Equal(instance.Name))
 	})
+
+	Context("UpdateMayRequireStsRecreate", func() {
+		It("returns false", func() {
+			Expect(serviceBuilder.UpdateMayRequireStsRecreate()).To(BeFalse())
+		})
+	})
 })

internal/resource/rabbitmq_plugins.go

@@ -48,6 +48,10 @@ func (builder *RabbitmqPluginsConfigMapBuilder) Build() (client.Object, error) {
 	}, nil
 }
 
+func (builder *RabbitmqPluginsConfigMapBuilder) UpdateMayRequireStsRecreate() bool {
+	return false
+}
+
 func (builder *RabbitmqPluginsConfigMapBuilder) Update(object client.Object) error {
 	configMap := object.(*corev1.ConfigMap)
 

internal/resource/rabbitmq_plugins_test.go

@@ -248,5 +248,11 @@ var _ = Describe("RabbitMQPlugins", func() {
 				Expect(configMap.Annotations).To(BeEmpty())
 			})
 		})
+
+		Context("UpdateMayRequireStsRecreate", func() {
+			It("returns false", func() {
+				Expect(configMapBuilder.UpdateMayRequireStsRecreate()).To(BeFalse())
+			})
+		})
 	})
 })

internal/resource/rabbitmq_resource_builder.go

@@ -23,6 +23,7 @@ type RabbitmqResourceBuilder struct {
 type ResourceBuilder interface {
 	Build() (client.Object, error)
 	Update(client.Object) error
+	UpdateMayRequireStsRecreate() bool
 }
 
 func (builder *RabbitmqResourceBuilder) ResourceBuilders() ([]ResourceBuilder, error) {