Support Elasticsearch volumes expansion

pkg/apis/elasticsearch/v1/validations.go

@@ -9,13 +9,14 @@ import (
 	"net"
 	"strings"
 
-	apiequality "k8s.io/apimachinery/pkg/api/equality"
-	"k8s.io/apimachinery/pkg/util/validation/field"
-
 	commonv1 "github.com/elastic/cloud-on-k8s/pkg/apis/common/v1"
 	"github.com/elastic/cloud-on-k8s/pkg/controller/common/version"
 	esversion "github.com/elastic/cloud-on-k8s/pkg/controller/elasticsearch/version"
 	netutil "github.com/elastic/cloud-on-k8s/pkg/utils/net"
+	corev1 "k8s.io/api/core/v1"
+	apiequality "k8s.io/apimachinery/pkg/api/equality"
+	"k8s.io/apimachinery/pkg/api/resource"
+	"k8s.io/apimachinery/pkg/util/validation/field"
 )
 
 const (
@@ -29,7 +30,7 @@ const (
 	nodeRolesInOldVersionMsg = "node.roles setting is not available in this version of Elasticsearch"
 	parseStoredVersionErrMsg = "Cannot parse current Elasticsearch version. String format must be {major}.{minor}.{patch}[-{label}]"
 	parseVersionErrMsg       = "Cannot parse Elasticsearch version. String format must be {major}.{minor}.{patch}[-{label}]"
-	pvcImmutableMsg          = "Volume claim templates cannot be modified"
+	pvcImmutableMsg          = "Volume claim templates can only have their storage requests increased, if the storage class allows volume expansion. Any other change is forbidden."
 	unsupportedConfigErrMsg  = "Configuration setting is reserved for internal use. User-configured use is unsupported"
 	unsupportedUpgradeMsg    = "Unsupported version upgrade path. Check the Elasticsearch documentation for supported upgrade paths."
 	unsupportedVersionMsg    = "Unsupported version"
@@ -207,29 +208,44 @@ func checkNodeSetNameUniqueness(es *Elasticsearch) field.ErrorList {
 	return errs
 }
 
-// pvcModification ensures no PVCs are changed, as volume claim templates are immutable in stateful sets
+// pvcModification ensures the only part of volume claim templates that can be changed is storage requests.
+// VolumeClaimTemplates are immutable in StatefulSets, but we still manage to deal with volume expansion.
 func pvcModification(current, proposed *Elasticsearch) field.ErrorList {
 	var errs field.ErrorList
 	if current == nil || proposed == nil {
 		return errs
 	}
-	for i, node := range proposed.Spec.NodeSets {
-		currNode := getNode(node.Name, current)
-		if currNode == nil {
+	for i, proposedNodeSet := range proposed.Spec.NodeSets {
+		currNodeSet := getNode(proposedNodeSet.Name, current)
+		if currNodeSet == nil {
 			// this is a new sset, so there is nothing to check
 			continue
 		}
 
-		// ssets do not allow modifications to fields other than 'replicas', 'template', and 'updateStrategy'
-		// reflection isn't ideal, but okay here since the ES object does not have the status of the claims.
+		// Check that no modification was made to the volumeClaimTemplates, except on storage requests.
 		// Checking semantic equality here allows providing PVC storage size with different units (eg. 1Ti vs. 1024Gi).
-		if !apiequality.Semantic.DeepEqual(node.VolumeClaimTemplates, currNode.VolumeClaimTemplates) {
-			errs = append(errs, field.Invalid(field.NewPath("spec").Child("nodeSet").Index(i).Child("volumeClaimTemplates"), node.VolumeClaimTemplates, pvcImmutableMsg))
+		// We could conceptually also disallow storage decrease here (unsupported), but it would make it hard for users
+		// to revert to the previous size if tried increasing but their storage class doesn't support expansion.
+		currentClaims := currNodeSet.VolumeClaimTemplates
+		proposedClaims := proposedNodeSet.VolumeClaimTemplates
+		if !apiequality.Semantic.DeepEqual(claimsWithoutStorageReq(currentClaims), claimsWithoutStorageReq(proposedClaims)) {
+			errs = append(errs, field.Invalid(field.NewPath("spec").Child("nodeSet").Index(i).Child("volumeClaimTemplates"), proposedClaims, pvcImmutableMsg))
 		}
 	}
 	return errs
 }
 
+// claimsWithoutStorageReq returns a copy of the given claims, with all storage requests set to the empty quantity.
+func claimsWithoutStorageReq(claims []corev1.PersistentVolumeClaim) []corev1.PersistentVolumeClaim {
+	result := make([]corev1.PersistentVolumeClaim, 0, len(claims))
+	for _, claim := range claims {
+		patchedClaim := *claim.DeepCopy()
+		patchedClaim.Spec.Resources.Requests[corev1.ResourceStorage] = resource.Quantity{}
+		result = append(result, patchedClaim)
+	}
+	return result
+}
+
 func noDowngrades(current, proposed *Elasticsearch) field.ErrorList {
 	var errs field.ErrorList
 	if current == nil || proposed == nil {

pkg/apis/elasticsearch/v1/validations_test.go

@@ -326,7 +326,7 @@ func Test_pvcModified(t *testing.T) {
 		expectErrors bool
 	}{
 		{
-			name:    "resize fails",
+			name:    "resize succeeds",
 			current: current,
 			proposed: &Elasticsearch{
 				Spec: ElasticsearchSpec{
@@ -352,35 +352,6 @@ func Test_pvcModified(t *testing.T) {
 					},
 				},
 			},
-			expectErrors: true,
-		},
-		{
-			name:    "same size with different unit accepted",
-			current: current,
-			proposed: &Elasticsearch{
-				Spec: ElasticsearchSpec{
-					Version: "7.2.0",
-					NodeSets: []NodeSet{
-						{
-							Name: "master",
-							VolumeClaimTemplates: []corev1.PersistentVolumeClaim{
-								{
-									ObjectMeta: metav1.ObjectMeta{
-										Name: "elasticsearch-data",
-									},
-									Spec: corev1.PersistentVolumeClaimSpec{
-										Resources: corev1.ResourceRequirements{
-											Requests: corev1.ResourceList{
-												corev1.ResourceStorage: resource.MustParse("5120Mi"),
-											},
-										},
-									},
-								},
-							},
-						},
-					},
-				},
-			},
 			expectErrors: false,
 		},
 		{

pkg/controller/elasticsearch/driver/nodes.go

@@ -49,17 +49,28 @@ func (d *defaultDriver) reconcileNodeSpecs(
 		return results.WithResult(defaultRequeue)
 	}
 
-	actualStatefulSets, err := sset.RetrieveActualStatefulSets(d.Client, k8s.ExtractNamespacedName(&d.ES))
+	// recreate any StatefulSet that needs to account for PVC expansion
+	recreations, err := recreateStatefulSets(d.K8sClient(), d.ES)
 	if err != nil {
-		return results.WithError(err)
+		return results.WithError(fmt.Errorf("StatefulSet recreation: %w", err))
+	}
+	if recreations > 0 {
+		// Some StatefulSets are in the process of being recreated to handle PVC expansion:
+		// it is safer to requeue until the re-creation is done.
+		// Otherwise, some operation could be performed with wrong assumptions:
+		// the sset doesn't exist (was just deleted), but the Pods do actually exist.
+		log.V(1).Info("StatefulSets recreation in progress, re-queueing.",
+			"namespace", d.ES.Namespace, "es_name", d.ES.Name, "recreations", recreations)
+		return results.WithResult(defaultRequeue)
 	}
 
-	expectedResources, err := nodespec.BuildExpectedResources(d.ES, keystoreResources, actualStatefulSets, d.OperatorParameters.IPFamily, d.OperatorParameters.SetDefaultSecurityContext)
+	actualStatefulSets, err := sset.RetrieveActualStatefulSets(d.Client, k8s.ExtractNamespacedName(&d.ES))
 	if err != nil {
 		return results.WithError(err)
 	}
 
-	if err := GarbageCollectPVCs(d.K8sClient(), d.ES, actualStatefulSets, expectedResources.StatefulSets()); err != nil {
+	expectedResources, err := nodespec.BuildExpectedResources(d.ES, keystoreResources, actualStatefulSets, d.OperatorParameters.IPFamily, d.OperatorParameters.SetDefaultSecurityContext)
+	if err != nil {
 		return results.WithError(err)
 	}
 
@@ -74,7 +85,7 @@ func (d *defaultDriver) reconcileNodeSpecs(
 		esState:       esState,
 		expectations:  d.Expectations,
 	}
-	actualStatefulSets, err = HandleUpscaleAndSpecChanges(upscaleCtx, actualStatefulSets, expectedResources)
+	upscaleResults, err := HandleUpscaleAndSpecChanges(upscaleCtx, actualStatefulSets, expectedResources)
 	if err != nil {
 		reconcileState.AddEvent(corev1.EventTypeWarning, events.EventReconciliationError, fmt.Sprintf("Failed to apply spec change: %v", err))
 		var podTemplateErr *sset.PodTemplateError
@@ -84,12 +95,20 @@ func (d *defaultDriver) reconcileNodeSpecs(
 		}
 		return results.WithError(err)
 	}
+	if upscaleResults.Requeue {
+		return results.WithResult(defaultRequeue)
+	}
+	actualStatefulSets = upscaleResults.ActualStatefulSets
 
 	// Update PDB to account for new replicas.
 	if err := pdb.Reconcile(d.Client, d.ES, actualStatefulSets); err != nil {
 		return results.WithError(err)
 	}
 
+	if err := GarbageCollectPVCs(d.K8sClient(), d.ES, actualStatefulSets, expectedResources.StatefulSets()); err != nil {
+		return results.WithError(err)
+	}
+
 	// Phase 2: if there is any Pending or bootlooping Pod to upgrade, do it.
 	attempted, err := d.MaybeForceUpgrade(actualStatefulSets)
 	if err != nil || attempted {