PWX-25414: Parallel/smart upgrades with minimum app downtime (#1629)

* [Cherry-pick into feature]: Cherry-pick of all tickets related to Smart and parallel upgrades merged into master (#1578) * adding k8snode name label to px pods * correcting go fmt error * making requested changes * Creating node PDB * go fmt error * Checking node version instead of cluster * making requested changes Conflicts: drivers/storage/portworx/component/disruption_budget.go * aggregating errors * adding logic to delete node PDB * correcting go fmt errors * made requested changes * correcting after merge conflict * [cherry-pick] PWX-36509 : StorageCluster schema changes to support parallel portworx upgrades (#1576) * StorageCluster schema changes to support parallel portworx upgrades Signed-off-by: hitesh-wani-px <hwani+github@purestorage.com> * change DisruptionSpec to Disruption and its description Signed-off-by: hitesh-wani-px <hwani+github@purestorage.com> --------- Signed-off-by: hitesh-wani-px <hwani+github@purestorage.com> * PWX-36477 : Updating Node PDB to allow k8s upgrades (#1580) * vendoring openstorage with nooverlappingnodes api * adding logic to update node PDB * Adding UTs and correcting logic * go fmt error * adding logic to delete cluster pdb after 3.1.2 * addressing comments requested * skip checking node version if version is empty * passing nodeIds instead of name to API * making requested changes and adding UTs * changing log to info * PWX-36510: No disruption of volumes for portworx upgrades (#1612) * non disruptive portworx upgrades * adding new method to driver interface * correcting UT * goimport fmt * gofmt file * update PX version in test * correcting new testcases --------- Signed-off-by: hitesh-wani-px <hwani+github@purestorage.com> Co-authored-by: Swarupa Vijaykumar <123443458+svijaykumar-px@users.noreply.github.com> Co-authored-by: hitesh-wani-px <hwani+github@purestorage.com> Co-authored-by: svijaykumar-px <svijaykumar@purestorage.com>
libopenstorage · Aug 12, 2024 · a8fb389 · a8fb389
1 parent 1a1964b
commit a8fb389
Show file tree

Hide file tree

Showing 48 changed files with 25,355 additions and 42,493 deletions.
diff --git a/deploy/crds/core_v1_storagecluster_crd.yaml b/deploy/crds/core_v1_storagecluster_crd.yaml
@@ -154,6 +154,16 @@ spec:
                           their place. Once the new pods are available, it then proceeds onto other
                           StorageCluster pods, thus ensuring that at least 70% of original number of
                           StorageCluster pods are available at all times during the update.
+                      disruption:
+                        type: object
+                        description: >-
+                          The default behavior is non-disruptive upgrades. This setting disables the default
+                          non-disruptive upgrades and reverts to the previous behavior of upgrading nodes in
+                          parallel without worrying about disruption.
+                        properties:
+                          allow:
+                            type: boolean
+                            description: Flag indicates whether updates are non-disruptive or disruptive.
               deleteStrategy:
                 type: object
                 description: Delete strategy to uninstall and wipe the storage cluster.

diff --git a/drivers/storage/portworx/component/disruption_budget.go b/drivers/storage/portworx/component/disruption_budget.go
@@ -1,16 +1,20 @@
 package component
 
 import (
+	"context"
 	"fmt"
 	"math"
 	"strconv"
+	"strings"
 
 	"github.com/hashicorp/go-version"
+	"github.com/libopenstorage/openstorage/api"
 	"github.com/sirupsen/logrus"
 	"google.golang.org/grpc"
 	policyv1 "k8s.io/api/policy/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	utilerrors "k8s.io/apimachinery/pkg/util/errors"
 	"k8s.io/apimachinery/pkg/util/intstr"
 	"k8s.io/client-go/tools/record"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -77,9 +81,59 @@ func (c *disruptionBudget) Reconcile(cluster *corev1.StorageCluster) error {
 	if err := c.createKVDBPodDisruptionBudget(cluster, ownerRef); err != nil {
 		return err
 	}
-	if err := c.createPortworxPodDisruptionBudget(cluster, ownerRef); err != nil {
+	// Create node PDB only if parallel upgrade is supported
+	var err error
+	c.sdkConn, err = pxutil.GetPortworxConn(c.sdkConn, c.k8sClient, cluster.Namespace)
+	if err != nil {
+		return err
+	}
+
+	// Get list of portworx storage nodes
+	nodeClient := api.NewOpenStorageNodeClient(c.sdkConn)
+	ctx, err := pxutil.SetupContextWithToken(context.Background(), cluster, c.k8sClient, false)
+	if err != nil {
 		return err
 	}
+	nodeEnumerateResponse, err := nodeClient.EnumerateWithFilters(
+		ctx,
+		&api.SdkNodeEnumerateWithFiltersRequest{},
+	)
+	if err != nil {
+		return fmt.Errorf("failed to enumerate nodes: %v", err)
+	}
+	if len(nodeEnumerateResponse.Nodes) == 0 {
+		logrus.Warnf("Cannot create/update storage PodDisruptionBudget as there are no storage nodes")
+		return nil
+	}
+
+	if pxutil.ClusterSupportsParallelUpgrade(nodeEnumerateResponse.Nodes) {
+		// Get the list of k8s nodes that are part of the current cluster
+		k8sNodeList := &v1.NodeList{}
+		err = c.k8sClient.List(context.TODO(), k8sNodeList)
+		if err != nil {
+			return err
+		}
+		if err := c.createPortworxNodePodDisruptionBudget(cluster, ownerRef, nodeEnumerateResponse, k8sNodeList); err != nil {
+			return err
+		}
+		if err := c.deleteClusterPodDisruptionBudget(cluster, ownerRef); err != nil {
+			logrus.Warnf("failed to delete cluster poddisruptionbudget if exists: %v", err)
+		}
+		if err := c.deletePortworxNodePodDisruptionBudget(cluster, ownerRef, nodeEnumerateResponse, k8sNodeList); err != nil {
+			return err
+		}
+		if err := c.updateMinAvailableForNodePDB(cluster, ownerRef, nodeEnumerateResponse, k8sNodeList); err != nil {
+			return err
+		}
+	} else {
+		if err := c.createPortworxPodDisruptionBudget(cluster, ownerRef, nodeEnumerateResponse); err != nil {
+			return err
+		}
+		if err := c.deleteAllNodePodDisruptionBudgets(cluster, ownerRef); err != nil {
+			logrus.Warnf("failed to delete node poddisruptionbudgets if exist: %v", err)
+		}
+	}
+
 	return nil
 }
 
@@ -106,6 +160,7 @@ func (c *disruptionBudget) MarkDeleted() {}
 func (c *disruptionBudget) createPortworxPodDisruptionBudget(
 	cluster *corev1.StorageCluster,
 	ownerRef *metav1.OwnerReference,
+	nodeEnumerateResponse *api.SdkNodeEnumerateWithFiltersResponse,
 ) error {
 	userProvidedMinValue, err := pxutil.MinAvailableForStoragePDB(cluster)
 	if err != nil {
@@ -114,12 +169,8 @@ func (c *disruptionBudget) createPortworxPodDisruptionBudget(
 	}
 
 	var minAvailable int
-	c.sdkConn, err = pxutil.GetPortworxConn(c.sdkConn, c.k8sClient, cluster.Namespace)
-	if err != nil {
-		return err
-	}
 
-	storageNodesCount, err := pxutil.CountStorageNodes(cluster, c.sdkConn, c.k8sClient)
+	storageNodesCount, err := pxutil.CountStorageNodes(cluster, c.sdkConn, c.k8sClient, nodeEnumerateResponse)
 	if err != nil {
 		c.closeSdkConn()
 		return err
@@ -178,6 +229,73 @@ func (c *disruptionBudget) createPortworxPodDisruptionBudget(
 	return err
 }
 
+func (c *disruptionBudget) createPortworxNodePodDisruptionBudget(
+	cluster *corev1.StorageCluster,
+	ownerRef *metav1.OwnerReference,
+	nodeEnumerateResponse *api.SdkNodeEnumerateWithFiltersResponse,
+	k8sNodeList *v1.NodeList,
+) error {
+	nodesNeedingPDB, err := pxutil.NodesNeedingPDB(c.k8sClient, nodeEnumerateResponse, k8sNodeList)
+	if err != nil {
+		return err
+	}
+	errors := []error{}
+	for _, node := range nodesNeedingPDB {
+		minAvailable := intstr.FromInt(1)
+		pdbName := "px-" + node
+		pdb := &policyv1.PodDisruptionBudget{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:            pdbName,
+				Namespace:       cluster.Namespace,
+				OwnerReferences: []metav1.OwnerReference{*ownerRef},
+			},
+			Spec: policyv1.PodDisruptionBudgetSpec{
+				MinAvailable: &minAvailable,
+				Selector: &metav1.LabelSelector{
+					MatchLabels: map[string]string{
+						constants.LabelKeyClusterName:      cluster.Name,
+						constants.OperatorLabelNodeNameKey: node,
+					},
+				},
+			},
+		}
+		err = k8sutil.CreateOrUpdatePodDisruptionBudget(c.k8sClient, pdb, ownerRef)
+		if err != nil {
+			logrus.Warnf("Failed to create PDB for node %s: %v", node, err)
+			errors = append(errors, err)
+		}
+	}
+	return utilerrors.NewAggregate(errors)
+
+}
+
+// Delete node pod disruption budget when kubertetes is not part of cluster or portworx does not run on it
+func (c *disruptionBudget) deletePortworxNodePodDisruptionBudget(
+	cluster *corev1.StorageCluster,
+	ownerRef *metav1.OwnerReference,
+	nodeEnumerateResponse *api.SdkNodeEnumerateWithFiltersResponse,
+	k8sNodeList *v1.NodeList,
+) error {
+	nodesToDeletePDB, err := pxutil.NodesToDeletePDB(c.k8sClient, nodeEnumerateResponse, k8sNodeList)
+	if err != nil {
+		return err
+	}
+	errors := []error{}
+
+	for _, node := range nodesToDeletePDB {
+		pdbName := "px-" + node
+		err = k8sutil.DeletePodDisruptionBudget(
+			c.k8sClient, pdbName,
+			cluster.Namespace, *ownerRef,
+		)
+		if err != nil {
+			logrus.Warnf("Failed to delete PDB for node %s: %v", node, err)
+			errors = append(errors, err)
+		}
+	}
+	return utilerrors.NewAggregate(errors)
+}
+
 func (c *disruptionBudget) createKVDBPodDisruptionBudget(
 	cluster *corev1.StorageCluster,
 	ownerRef *metav1.OwnerReference,
@@ -186,7 +304,6 @@ func (c *disruptionBudget) createKVDBPodDisruptionBudget(
 	if cluster.Spec.Kvdb != nil && !cluster.Spec.Kvdb.Internal {
 		return nil
 	}
-
 	clusterSize := kvdbClusterSize(cluster)
 	minAvailable := intstr.FromInt(clusterSize - 1)
 	pdb := &policyv1.PodDisruptionBudget{
@@ -262,3 +379,129 @@ func RegisterDisruptionBudgetComponent() {
 func init() {
 	RegisterDisruptionBudgetComponent()
 }
+
+func (c *disruptionBudget) updateMinAvailableForNodePDB(cluster *corev1.StorageCluster,
+	ownerRef *metav1.OwnerReference,
+	nodeEnumerateResponse *api.SdkNodeEnumerateWithFiltersResponse,
+	k8sNodeList *v1.NodeList,
+) error {
+
+	// GetNodesToUpgrade returns list of nodes that can be upgraded in parallel and the map of px node to k8s node name, number of nodes with PDB 0 + down nodes
+	// In the case when non disruptive upgrade is disabled, it returns all cordoned nodes which aren't already selected for the upgrade
+	nodesToUpgrade, cordonedPxNodesMap, numPxNodesDown, err := pxutil.GetNodesToUpgrade(cluster, nodeEnumerateResponse, k8sNodeList, c.k8sClient, c.sdkConn)
+	if err != nil {
+		return err
+	}
+	// Return if there are no nodes to upgrade
+	if len(nodesToUpgrade) == 0 {
+		return nil
+	}
+
+	// If user has mentioned minimum nodes that need to be running, then honor that value
+	storageNodesCount, err := pxutil.CountStorageNodes(cluster, c.sdkConn, c.k8sClient, nodeEnumerateResponse)
+	if err != nil {
+		c.closeSdkConn()
+		return err
+	}
+	userProvidedMinValue, err := pxutil.MinAvailableForStoragePDB(cluster)
+	if err != nil {
+		logrus.Warnf("Invalid value for annotation %s: %v", pxutil.AnnotationStoragePodDisruptionBudget, err)
+		userProvidedMinValue = -2
+	}
+
+	// Calculate the minimum number of nodes that should be available
+	quorumValue := int(math.Floor(float64(storageNodesCount)/2) + 1)
+	calculatedMinAvailable := quorumValue
+	// When non disruptive upgrades is disabled
+	if cluster.Annotations != nil && cluster.Annotations[pxutil.AnnotationsDisableNonDisruptiveUpgrade] == "true" {
+		if cluster.Annotations[pxutil.AnnotationStoragePodDisruptionBudget] == "" {
+			calculatedMinAvailable = storageNodesCount - 1
+		} else if userProvidedMinValue < quorumValue || userProvidedMinValue >= storageNodesCount {
+			calculatedMinAvailable = storageNodesCount - 1
+			// Log an error only if this is the first time we encounter an invalid value
+			if userProvidedMinValue != c.annotatedMinAvailable {
+				errmsg := fmt.Sprintf("Invalid minAvailable annotation value for storage pod disruption budget. Using default value: %d", calculatedMinAvailable)
+				c.recorder.Event(cluster, v1.EventTypeWarning, util.InvalidMinAvailable, errmsg)
+			}
+		} else if userProvidedMinValue >= quorumValue && userProvidedMinValue < storageNodesCount {
+			calculatedMinAvailable = userProvidedMinValue
+		}
+	} else {
+		// When non disruptive upgrades is enabled
+		if userProvidedMinValue >= storageNodesCount && userProvidedMinValue != c.annotatedMinAvailable {
+			errmsg := fmt.Sprintf("Invalid minAvailable annotation value for storage pod disruption budget. Using default value: %d", calculatedMinAvailable)
+			c.recorder.Event(cluster, v1.EventTypeWarning, util.InvalidMinAvailable, errmsg)
+		} else if userProvidedMinValue >= quorumValue && userProvidedMinValue < storageNodesCount {
+			calculatedMinAvailable = userProvidedMinValue
+		}
+	}
+	c.annotatedMinAvailable = userProvidedMinValue
+	downNodesCount := numPxNodesDown
+	// Update minAvailable to 0 for nodes in nodesToUpgrade
+	for _, node := range nodesToUpgrade {
+		// Ensure that portworx quorum or valid minAvailable provided by user is always maintained
+		maxUnavailable := storageNodesCount - calculatedMinAvailable
+		if downNodesCount >= maxUnavailable {
+			logrus.Infof("Number of down PX nodes: %d, is equal to or exceeds allowed maximum: %d. Total storage nodes: %d", downNodesCount, maxUnavailable, storageNodesCount)
+			break
+		}
+		k8sNode := cordonedPxNodesMap[node]
+		pdbName := "px-" + k8sNode
+		minAvailable := intstr.FromInt(0)
+		pdb := &policyv1.PodDisruptionBudget{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:            pdbName,
+				Namespace:       cluster.Namespace,
+				OwnerReferences: []metav1.OwnerReference{*ownerRef},
+			},
+			Spec: policyv1.PodDisruptionBudgetSpec{
+				MinAvailable: &minAvailable,
+				Selector: &metav1.LabelSelector{
+					MatchLabels: map[string]string{
+						constants.LabelKeyClusterName:      cluster.Name,
+						constants.OperatorLabelNodeNameKey: k8sNode,
+					},
+				},
+			},
+		}
+		err = k8sutil.CreateOrUpdatePodDisruptionBudget(c.k8sClient, pdb, ownerRef)
+		if err != nil {
+			logrus.Warnf("Failed to update PDB for node %s: %v", node, err)
+		} else {
+			downNodesCount++
+		}
+	}
+	return nil
+
+}
+
+func (c *disruptionBudget) deleteClusterPodDisruptionBudget(cluster *corev1.StorageCluster, ownerRef *metav1.OwnerReference) error {
+
+	err := k8sutil.DeletePodDisruptionBudget(c.k8sClient, StoragePodDisruptionBudgetName, cluster.Namespace, *ownerRef)
+	if err != nil {
+		logrus.Warnf("Failed to delete cluster PDB %s: %v", StoragePodDisruptionBudgetName, err)
+		return err
+	}
+
+	return nil
+}
+
+func (c *disruptionBudget) deleteAllNodePodDisruptionBudgets(cluster *corev1.StorageCluster, ownerRef *metav1.OwnerReference) error {
+	errors := []error{}
+	// Get the list of poddisruptionbudgets
+	pdbList := &policyv1.PodDisruptionBudgetList{}
+	err := c.k8sClient.List(context.TODO(), pdbList, client.InNamespace(cluster.Namespace))
+	if err != nil {
+		logrus.Warnf("failed to list poddisruptionbudgets: %v", err)
+	}
+	for _, pdb := range pdbList.Items {
+		if strings.HasPrefix(pdb.Name, "px") && pdb.Name != StoragePodDisruptionBudgetName && pdb.Name != "px-kvdb" {
+			err := k8sutil.DeletePodDisruptionBudget(c.k8sClient, pdb.Name, cluster.Namespace, *ownerRef)
+			if err != nil {
+				logrus.Warnf("Failed to delete node PDB %s: %v", pdb.Name, err)
+				errors = append(errors, err)
+			}
+		}
+	}
+	return utilerrors.NewAggregate(errors)
+}