Ignore PVC overrides when storage == 0

This fixes the corner-case where storage is zero but overrides are used
to define it differently. To avoid unnecessarily confusing
configurations, if storage==0, all PVC overrides are ignored and a
warning is logged.

internal/resource/statefulset.go

@@ -82,18 +82,27 @@ func (builder *StatefulSetBuilder) Build() (client.Object, error) {
 		}
 
 		if len(overrideSts.Spec.VolumeClaimTemplates) != 0 {
-			override := overrideSts.Spec.VolumeClaimTemplates
-			pvcList := make([]corev1.PersistentVolumeClaim, len(override))
-			for i := range override {
-				copyObjectMeta(&pvcList[i].ObjectMeta, override[i].EmbeddedObjectMeta)
-				pvcList[i].Namespace = sts.Namespace // PVC should always be in the same namespace as the Stateful Set
-				pvcList[i].Spec = override[i].Spec
-				if err := controllerutil.SetControllerReference(builder.Instance, &pvcList[i], builder.Scheme); err != nil {
-					return nil, fmt.Errorf("failed setting controller reference: %v", err)
+			// If spec.persistence.storage == 0, ignore PVC overrides.
+			// Main reason for that is that there is no default PVC in such case (emptyDir is used instead)
+			// other PVCs could technically be still overridden/added but we would be entering a very confusing territory
+			// where storage is set to 0 and yet there are PVCs with data
+			if builder.Instance.Spec.Persistence.Storage.Cmp(k8sresource.MustParse("0Gi")) == 0 {
+				logger := ctrl.Log.WithName("statefulset").WithName("RabbitmqCluster")
+				logger.Info(fmt.Sprintf("Warning: persistentVolumeClaim overrides are ignored for cluster \"%s\", becasue spec.persistence.storage is set to zero.", sts.GetName()))
+			} else {
+				override := overrideSts.Spec.VolumeClaimTemplates
+				pvcList := make([]corev1.PersistentVolumeClaim, len(override))
+				for i := range override {
+					copyObjectMeta(&pvcList[i].ObjectMeta, override[i].EmbeddedObjectMeta)
+					pvcList[i].Namespace = sts.Namespace // PVC should always be in the same namespace as the Stateful Set
+					pvcList[i].Spec = override[i].Spec
+					if err := controllerutil.SetControllerReference(builder.Instance, &pvcList[i], builder.Scheme); err != nil {
+						return nil, fmt.Errorf("failed setting controller reference: %v", err)
+					}
+					disableBlockOwnerDeletion(pvcList[i])
 				}
-				disableBlockOwnerDeletion(pvcList[i])
+				sts.Spec.VolumeClaimTemplates = pvcList
 			}
-			sts.Spec.VolumeClaimTemplates = pvcList
 		}
 	}
 

internal/resource/statefulset_test.go

@@ -152,6 +152,27 @@ var _ = Describe("StatefulSet", func() {
 				zero := k8sresource.MustParse("0Gi")
 
 				builder.Instance.Spec.Persistence.Storage = &zero
+				// we shouldn't create the `persistence` PVC if storage==0, even if overrides are used
+				builder.Instance.Spec.Override.StatefulSet = &rabbitmqv1beta1.StatefulSet{
+					Spec: &rabbitmqv1beta1.StatefulSetSpec{
+						VolumeClaimTemplates: []rabbitmqv1beta1.PersistentVolumeClaim{
+							{
+								EmbeddedObjectMeta: rabbitmqv1beta1.EmbeddedObjectMeta{
+									Name:      "persistence",
+									Namespace: instance.Namespace,
+								},
+								Spec: corev1.PersistentVolumeClaimSpec{
+									Resources: corev1.ResourceRequirements{
+										Requests: corev1.ResourceList{
+											corev1.ResourceStorage: k8sresource.MustParse("10Gi"),
+										},
+									},
+								},
+							},
+						},
+					},
+				}
+
 				obj, err := stsBuilder.Build()
 				Expect(err).NotTo(HaveOccurred())
 				statefulSet := obj.(*appsv1.StatefulSet)