K8SPXC-1341: Resize PVCs on user request

pkg/controller/pxc/controller.go

@@ -294,6 +294,11 @@ func (r *ReconcilePerconaXtraDBCluster) Reconcile(ctx context.Context, request r
 			log.Info("failed to ensure version, running with default", "error", err)
 		}
 	}
+
+	err = r.reconcilePersistentVolumes(ctx, o)
+	if err != nil {
+		return reconcile.Result{}, errors.Wrap(err, "reconcile persistent volumes")
+	}
 
 	err = r.deploy(ctx, o)
 	if err != nil {

pkg/controller/pxc/volumes.go

@@ -0,0 +1,120 @@
+package pxc
+
+import (
+	"context"
+	"strings"
+
+	"github.com/pkg/errors"
+	corev1 "k8s.io/api/core/v1"
+	storagev1 "k8s.io/api/storage/v1"
+	k8serrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	logf "sigs.k8s.io/controller-runtime/pkg/log"
+
+	api "github.com/percona/percona-xtradb-cluster-operator/pkg/apis/pxc/v1"
+	"github.com/percona/percona-xtradb-cluster-operator/pkg/pxc/app/statefulset"
+)
+
+func (r *ReconcilePerconaXtraDBCluster) reconcilePersistentVolumes(ctx context.Context, cr *api.PerconaXtraDBCluster) error {
+	log := logf.FromContext(ctx)
+
+	pxcSet := statefulset.NewNode(cr)
+	sts := pxcSet.StatefulSet()
+
+	err := r.client.Get(ctx, client.ObjectKeyFromObject(sts), sts)
+	if err != nil {
+		if k8serrors.IsNotFound(err) {
+			return nil
+		}
+		return errors.Wrapf(err, "get statefulset/%s", sts.Name)
+	}
+
+	if cr.Spec.PXC.VolumeSpec.PersistentVolumeClaim == nil {
+		return nil
+	}
+
+	var volumeTemplate corev1.PersistentVolumeClaim
+	for _, vct := range sts.Spec.VolumeClaimTemplates {
+		if vct.Name == "datadir" {
+			volumeTemplate = vct
+		}
+	}
+
+	requested := cr.Spec.PXC.VolumeSpec.PersistentVolumeClaim.Resources.Requests[corev1.ResourceStorage]
+	actual := volumeTemplate.Spec.Resources.Requests[corev1.ResourceStorage]
+
+	if requested.Cmp(actual) < 0 {
+		return errors.Wrap(err, "requested storage is less than actual")
+	}
+
+	if requested.Cmp(actual) == 0 {
+		return nil
+	}
+
+	// Check storage class for allowVolumeExpansion field
+
+	labels := map[string]string{
+		"app.kubernetes.io/component":  "pxc",
+		"app.kubernetes.io/instance":   cr.Name,
+		"app.kubernetes.io/managed-by": "percona-xtradb-cluster-operator",
+		"app.kubernetes.io/name":       "percona-xtradb-cluster",
+		"app.kubernetes.io/part-of":    "percona-xtradb-cluster",
+	}
+
+	pvcList := corev1.PersistentVolumeClaimList{}
+	if err := r.client.List(ctx, &pvcList, client.InNamespace(cr.Namespace), client.MatchingLabels(labels)); err != nil {
+		return errors.Wrap(err, "list persistentvolumeclaims")
+	}
+
+	pvcNames := make([]string, 0, len(pvcList.Items))
+	for _, pvc := range pvcList.Items {
+		pvcNames = append(pvcNames, pvc.Name)
+	}
+
+	log.Info("Resizing PVCs", "requested", requested, "actual", actual, "pvcList", strings.Join(pvcNames, ","))
+
+	// check storage class for allowVolumeExpansion field
+	// we check only the first PVC, because all PVCs should
+	// have the same storage class since they're created from
+	// the same template
+	firstPVC := pvcList.Items[0]
+
+	if firstPVC.Spec.StorageClassName == nil {
+		return errors.Errorf("storageClassName is not set in pvc/%s", firstPVC.Name)
+	}
+
+	storageClass := &storagev1.StorageClass{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: *firstPVC.Spec.StorageClassName,
+		},
+	}
+	if err := r.client.Get(ctx, client.ObjectKeyFromObject(storageClass), storageClass); err != nil {
+		return errors.Wrapf(err, "get storageclass/%s", storageClass.Name)
+	}
+
+	if storageClass.AllowVolumeExpansion == nil || !*storageClass.AllowVolumeExpansion {
+		return errors.Errorf("storageclass/%s does not allow volume expansion", storageClass.Name)
+	}
+
+	log.Info("Deleting statefulset", "name", sts.Name)
+
+	if err := r.client.Delete(ctx, sts, client.PropagationPolicy("Orphan")); err != nil {
+		return errors.Wrapf(err, "delete statefulset/%s", sts.Name)
+	}
+
+	for _, pvc := range pvcList.Items {
+		if !strings.HasPrefix(pvc.Name, "datadir-"+sts.Name) {
+			continue
+		}
+
+		log.Info("Resizing PVC", "name", pvc.Name)
+		pvc.Spec.Resources.Requests[corev1.ResourceStorage] = requested
+
+		if err := r.client.Update(ctx, &pvc); err != nil {
+			return errors.Wrapf(err, "update persistentvolumeclaim/%s", pvc.Name)
+		}
+	}
+
+	return nil
+}