[Bugfix] Fix Volume Resize on Single Mode (#802)

CHANGELOG.md

@@ -4,6 +4,7 @@
 - Replace `beta.kubernetes.io/arch` Pod label with `kubernetes.io/arch` using Silent Rotation
 - Add "Short Names" feature
 - Switch ArangoDB Image Discovery process from Headless Service to Pod IP
+- Fix PVC Resize for Single servers
 
 ## [1.2.3](https://github.com/arangodb/kube-arangodb/tree/1.2.3) (2021-09-24)
 - Update UBI Image to 8.4

pkg/deployment/reconcile/plan_builder_normal.go

@@ -104,6 +104,7 @@ func createNormalPlan(ctx context.Context, log zerolog.Logger, apiObject k8sutil
 		ApplySubPlanIfEmpty(createTLSStatusPropagatedFieldUpdate, createCAAppendPlan).
 		ApplyIfEmpty(createKeyfileRenewalPlan).
 		ApplyIfEmpty(createRotateServerStoragePlan).
+		ApplyIfEmpty(createRotateServerStorageResizePlan).
 		ApplySubPlanIfEmpty(createTLSStatusPropagatedFieldUpdate, createRotateTLSServerSNIPlan).
 		ApplyIfEmpty(createRestorePlan).
 		ApplySubPlanIfEmpty(createEncryptionKeyStatusPropagatedFieldUpdate, createEncryptionKeyCleanPlan).

pkg/deployment/reconcile/plan_builder_storage.go

@@ -45,10 +45,9 @@ func createRotateServerStoragePlan(ctx context.Context,
 		return nil
 	}
 	var plan api.Plan
-	var canContinue = true
 	status.Members.ForeachServerGroup(func(group api.ServerGroup, members api.MemberStatusList) error {
 		for _, m := range members {
-			if !plan.IsEmpty() && !canContinue {
+			if !plan.IsEmpty() {
 				// Only 1 change at a time
 				continue
 			}
@@ -74,9 +73,6 @@ func createRotateServerStoragePlan(ctx context.Context,
 			}
 
 			if util.StringOrDefault(pvc.Spec.StorageClassName) != storageClassName && storageClassName != "" {
-				// Do not append more than 1 operation if we replace storageClass
-				canContinue = false
-
 				// Storageclass has changed
 				log.Info().Str("pod-name", m.PodName).
 					Str("pvc-storage-class", util.StringOrDefault(pvc.Spec.StorageClassName)).
@@ -106,15 +102,8 @@ func createRotateServerStoragePlan(ctx context.Context,
 				if requestedSize, ok := res[core.ResourceStorage]; ok {
 					if volumeSize, ok := pvc.Spec.Resources.Requests[core.ResourceStorage]; ok {
 						cmp := volumeSize.Cmp(requestedSize)
-						if cmp < 0 {
-							if groupSpec.VolumeResizeMode.Get() == api.PVCResizeModeRotate {
-								// Do not append more than 1 operation if we hard restart member
-								canContinue = false
-							}
-							plan = append(plan, pvcResizePlan(log, group, groupSpec, m.ID)...)
-						} else if cmp > 0 {
-							// Do not append more than 1 operation if we schrink volume
-							canContinue = false
+						// Only schrink is possible
+						if cmp > 0 {
 
 							if groupSpec.GetVolumeAllowShrink() && group == api.ServerGroupDBServers && !m.Conditions.IsTrue(api.ConditionTypeMarkedToRemove) {
 								plan = append(plan, api.NewAction(api.ActionTypeMarkToRemoveMember, group, m.ID))
@@ -133,6 +122,61 @@ func createRotateServerStoragePlan(ctx context.Context,
 	return plan
 }
 
+// createRotateServerStorageResizePlan creates plan to resize storage
+func createRotateServerStorageResizePlan(ctx context.Context,
+	log zerolog.Logger, apiObject k8sutil.APIObject,
+	spec api.DeploymentSpec, status api.DeploymentStatus,
+	cachedStatus inspectorInterface.Inspector, context PlanBuilderContext) api.Plan {
+	var plan api.Plan
+
+	status.Members.ForeachServerGroup(func(group api.ServerGroup, members api.MemberStatusList) error {
+		for _, m := range members {
+			if m.Phase != api.MemberPhaseCreated {
+				// Only make changes when phase is created
+				continue
+			}
+			if m.PersistentVolumeClaimName == "" {
+				// Plan is irrelevant without PVC
+				continue
+			}
+			groupSpec := spec.GetServerGroupSpec(group)
+
+			if !plan.IsEmpty() && groupSpec.VolumeResizeMode.Get() == api.PVCResizeModeRotate {
+				// Only 1 change at a time
+				return nil
+			}
+
+			// Load PVC
+			pvc, exists := cachedStatus.PersistentVolumeClaim(m.PersistentVolumeClaimName)
+			if !exists {
+				log.Warn().
+					Str("role", group.AsRole()).
+					Str("id", m.ID).
+					Msg("Failed to get PVC")
+				continue
+			}
+
+			var res core.ResourceList
+			if groupSpec.HasVolumeClaimTemplate() {
+				res = groupSpec.GetVolumeClaimTemplate().Spec.Resources.Requests
+			} else {
+				res = groupSpec.Resources.Requests
+			}
+			if requestedSize, ok := res[core.ResourceStorage]; ok {
+				if volumeSize, ok := pvc.Spec.Resources.Requests[core.ResourceStorage]; ok {
+					cmp := volumeSize.Cmp(requestedSize)
+					if cmp < 0 {
+						plan = append(plan, pvcResizePlan(log, group, groupSpec, m.ID)...)
+					}
+				}
+			}
+		}
+		return nil
+	})
+
+	return plan
+}
+
 func pvcResizePlan(log zerolog.Logger, group api.ServerGroup, groupSpec api.ServerGroupSpec, memberID string) api.Plan {
 	mode := groupSpec.VolumeResizeMode.Get()
 	switch mode {