PWX-34271: pick the pods on unschedulable nodes for update first (#1544)

If we have marked the node unschedulable in the previous Reconcile(),
we have to pick that node first. Otherwise, we may end up making
too many nodes unschedulable.

Also, retry when add/remove of node label fails with conflict error
so we don't wait for another 30 seconds.

Signed-off-by: Neelesh Thakur <[email protected]>

Author: pureneelesh

pkg/constants/metadata.go

@@ -45,8 +45,8 @@ const (
 	// the custom registry, there is a list of hardcoded common registries, however the list
 	// may not be complete, users can use this annotation to add more.
 	AnnotationCommonImageRegistries = OperatorPrefix + "/common-image-registries"
-	// StorkAnnotationUnschedulable tells Stork to consider the node unschedulable
-	StorkAnnotationUnschedulable = "stork.libopenstorage.org/unschedulable"
+	// AnnotationUnschedulable tells Stork to consider the node unschedulable
+	AnnotationUnschedulable = OperatorPrefix + "/unschedulable"
 	// OperatorLabelManagedByKey is a label key that is added to any object that is
 	// managed the Portworx operator.
 	OperatorLabelManagedByKey = OperatorPrefix + "/managed-by"

pkg/controller/storagecluster/controller_test.go

@@ -2643,9 +2643,9 @@ func TestKubevirtVMsDuringUpgrade(t *testing.T) {
 	}
 
 	// add unschedulable label to the nodes and then verify that they get removed
-	err = storkUnschedulableAnnotationHelper(k8sClient, k8sNodes[0].Name, true)
+	err = nodeUnschedulableAnnotationHelper(k8sClient, k8sNodes[0].Name, true)
 	require.NoError(t, err)
-	err = storkUnschedulableAnnotationHelper(k8sClient, k8sNodes[2].Name, true)
+	err = nodeUnschedulableAnnotationHelper(k8sClient, k8sNodes[2].Name, true)
 	require.NoError(t, err)
 	verifyUnschedulableAnnotation(t, k8sClient, k8sNodes[0].Name, true)
 	verifyUnschedulableAnnotation(t, k8sClient, k8sNodes[1].Name, false)
@@ -2709,7 +2709,7 @@ func TestKubevirtVMsDuringUpgrade(t *testing.T) {
 
 	// reset the state before the next test
 	for _, k8sNode := range k8sNodes {
-		err = storkUnschedulableAnnotationHelper(k8sClient, k8sNode.Name, false)
+		err = nodeUnschedulableAnnotationHelper(k8sClient, k8sNode.Name, false)
 		require.NoError(t, err)
 		verifyUnschedulableAnnotation(t, k8sClient, k8sNode.Name, false)
 	}
@@ -7386,6 +7386,7 @@ func TestUpdateStorageClusterNodeSpec(t *testing.T) {
 		recorder:          recorder,
 		kubernetesVersion: k8sVersion,
 		nodeInfoMap:       maps.MakeSyncMap[string, *k8s.NodeInfo](),
+		kubevirt:          testutil.NoopKubevirtManager(mockCtrl),
 	}
 
 	driver.EXPECT().Validate(gomock.Any()).Return(nil).AnyTimes()
@@ -10648,7 +10649,7 @@ func createPxApiPod(
 
 func verifyUnschedulableAnnotation(t *testing.T, k8sClient client.Client, nodeName string, expectUnschedulable bool) {
 	node := getNode(t, k8sClient, nodeName)
-	val, ok := node.Annotations[constants.StorkAnnotationUnschedulable]
+	val, ok := node.Annotations[constants.AnnotationUnschedulable]
 	if expectUnschedulable {
 		require.True(t, ok && val == "true")
 	} else {

pkg/controller/storagecluster/storagecluster.go

@@ -836,18 +836,25 @@ func (c *Controller) syncStorageCluster(
 	}
 	hash := cur.Labels[util.DefaultStorageClusterUniqueLabelKey]
 
+	nodeList := &v1.NodeList{}
+	err = c.client.List(context.TODO(), nodeList, &client.ListOptions{})
+	if err != nil {
+		return fmt.Errorf("couldn't get list of nodes when syncing storage cluster %#v: %v",
+			cluster, err)
+	}
+
 	// TODO: Don't process a storage cluster until all its previous creations and
 	// deletions have been processed.
-	err = c.manage(cluster, hash)
+	err = c.manage(cluster, hash, nodeList)
 	if err != nil {
 		return err
 	}
 
 	switch cluster.Spec.UpdateStrategy.Type {
 	case corev1.OnDeleteStorageClusterStrategyType:
 	case corev1.RollingUpdateStorageClusterStrategyType:
-		if err := c.rollingUpdate(cluster, hash); err != nil {
-			return err
+		if err := c.rollingUpdate(cluster, hash, nodeList); err != nil {
+			return fmt.Errorf("rolling update failed: %s", err)
 		}
 	}
 
@@ -1020,6 +1027,7 @@ func (c *Controller) updateStorageClusterStatus(
 func (c *Controller) manage(
 	cluster *corev1.StorageCluster,
 	hash string,
+	nodeList *v1.NodeList,
 ) error {
 	// Run the pre install hook for the driver to ensure we are ready to create storage pods
 	if err := c.Driver.PreInstall(cluster); err != nil {
@@ -1034,13 +1042,6 @@ func (c *Controller) manage(
 	}
 	var nodesNeedingStoragePods, podsToDelete []string
 
-	nodeList := &v1.NodeList{}
-	err = c.client.List(context.TODO(), nodeList, &client.ListOptions{})
-	if err != nil {
-		return fmt.Errorf("couldn't get list of nodes when syncing storage cluster %#v: %v",
-			cluster, err)
-	}
-
 	// For each node, if the node is running the storage pod but isn't supposed to, kill the storage pod.
 	// If the node is supposed to run the storage pod, but isn't, create the storage pod on the node.
 	for _, node := range nodeList.Items {

pkg/controller/storagecluster/update.go

@@ -49,9 +49,13 @@ import (
 	"github.com/libopenstorage/operator/pkg/util/k8s"
 )
 
+const (
+	k8sNodeUpdateAttempts = 5
+)
+
 // rollingUpdate deletes old storage cluster pods making sure that no more than
 // cluster.Spec.UpdateStrategy.RollingUpdate.MaxUnavailable pods are unavailable
-func (c *Controller) rollingUpdate(cluster *corev1.StorageCluster, hash string) error {
+func (c *Controller) rollingUpdate(cluster *corev1.StorageCluster, hash string, nodeList *v1.NodeList) error {
 	logrus.Debug("Perform rolling update")
 	nodeToStoragePods, err := c.getNodeToStoragePods(cluster)
 	if err != nil {
@@ -74,7 +78,7 @@ func (c *Controller) rollingUpdate(cluster *corev1.StorageCluster, hash string)
 				continue
 			}
 			if _, ok := availablePods[pod.Name]; ok {
-				if err := c.removeStorkUnschedulableLabel(pod.Spec.NodeName); err != nil {
+				if err := c.removeNodeUnschedulableLabel(pod.Spec.NodeName); err != nil {
 					return err
 				}
 			}
@@ -94,7 +98,7 @@ func (c *Controller) rollingUpdate(cluster *corev1.StorageCluster, hash string)
 		oldPodsToDelete = append(oldPodsToDelete, pod.Name)
 	}
 
-	// TODO: Max unavaibable kvdb nodes has been set to 1 assuming default setup of 3 KVDB nodes.
+	// TODO: Max unavailable kvdb nodes has been set to 1 assuming default setup of 3 KVDB nodes.
 	// Need to generalise code for 5 KVDB nodes
 
 	// get the number of kvdb members which are unavailable in case of internal kvdb
@@ -107,7 +111,21 @@ func (c *Controller) rollingUpdate(cluster *corev1.StorageCluster, hash string)
 		}
 	}
 
+	nodesMarkedUnschedulable := map[string]bool{}
+	for _, node := range nodeList.Items {
+		if nodeMarkedUnschedulable(&node) {
+			nodesMarkedUnschedulable[node.Name] = true
+		}
+	}
 	logrus.Debugf("Marking old pods for deletion")
+	// sort the pods such that the pods on the nodes that we marked unschedulable are deleted first
+	// Otherwise, we will end up marking too many nodes as unschedulable (or ping-ponging VMs between the nodes).
+	sort.Slice(oldAvailablePods, func(i, j int) bool {
+		iUnschedulable := nodesMarkedUnschedulable[oldAvailablePods[i].Spec.NodeName]
+		jUnschedulable := nodesMarkedUnschedulable[oldAvailablePods[j].Spec.NodeName]
+
+		return iUnschedulable && !jUnschedulable
+	})
 	for _, pod := range oldAvailablePods {
 		if numUnavailable >= maxUnavailable {
 			logrus.Infof("Number of unavailable StorageCluster pods: %d, is equal "+
@@ -166,7 +184,7 @@ func (c *Controller) rollingUpdate(cluster *corev1.StorageCluster, hash string)
 					logrus.Warnf("Pod %s has no node name; cannot add unschedulable label to the node", podName)
 					continue
 				}
-				if err := c.addStorkUnschedulableAnnotation(pod.Spec.NodeName); err != nil {
+				if err := c.addNodeUnschedulableAnnotation(pod.Spec.NodeName); err != nil {
 					return err
 				}
 			}
@@ -220,15 +238,30 @@ func (c *Controller) rollingUpdate(cluster *corev1.StorageCluster, hash string)
 	return c.syncNodes(cluster, oldPodsToDeletePruned, []string{}, hash)
 }
 
-func (c *Controller) addStorkUnschedulableAnnotation(nodeName string) error {
-	return storkUnschedulableAnnotationHelper(c.client, nodeName, true)
+func (c *Controller) addNodeUnschedulableAnnotation(nodeName string) error {
+	return nodeUnschedulableAnnotationHelperWithRetries(c.client, nodeName, true)
 }
 
-func (c *Controller) removeStorkUnschedulableLabel(nodeName string) error {
-	return storkUnschedulableAnnotationHelper(c.client, nodeName, false)
+func (c *Controller) removeNodeUnschedulableLabel(nodeName string) error {
+	return nodeUnschedulableAnnotationHelperWithRetries(c.client, nodeName, false)
 }
 
-func storkUnschedulableAnnotationHelper(k8sClient client.Client, nodeName string, unschedulable bool) error {
+func nodeUnschedulableAnnotationHelperWithRetries(
+	k8sClient client.Client, nodeName string, unschedulable bool,
+) error {
+	var err error
+	for i := 0; i < k8sNodeUpdateAttempts; i++ {
+		err = nodeUnschedulableAnnotationHelper(k8sClient, nodeName, unschedulable)
+		if err == nil || !errors.IsConflict(err) {
+			return err
+		}
+		logrus.Warnf("Conflict while updating annotation %s on node %s in attempt %d: %v",
+			constants.AnnotationUnschedulable, nodeName, i, err)
+	}
+	return err
+}
+
+func nodeUnschedulableAnnotationHelper(k8sClient client.Client, nodeName string, unschedulable bool) error {
 	ctx := context.TODO()
 	node := &v1.Node{}
 	err := k8sClient.Get(ctx, types.NamespacedName{Name: nodeName}, node)
@@ -240,28 +273,25 @@ func storkUnschedulableAnnotationHelper(k8sClient client.Client, nodeName string
 		return fmt.Errorf("failed to get node %s: %w", nodeName, err)
 	}
 	needsUpdate := false
-	val, ok := node.Annotations[constants.StorkAnnotationUnschedulable]
-	if unschedulable && val != "true" {
+	val := nodeMarkedUnschedulable(node)
+	if unschedulable && !val {
+		logrus.Infof("Adding annotation %s to node %s", constants.AnnotationUnschedulable, nodeName)
 		if node.Annotations == nil {
 			node.Annotations = make(map[string]string)
 		}
-		node.Annotations[constants.StorkAnnotationUnschedulable] = "true"
+		node.Annotations[constants.AnnotationUnschedulable] = "true"
 		needsUpdate = true
-	} else if !unschedulable && ok {
-		delete(node.Annotations, constants.StorkAnnotationUnschedulable)
+	} else if !unschedulable && val {
+		logrus.Infof("Removing annotation %s from node %s", constants.AnnotationUnschedulable, nodeName)
+		delete(node.Annotations, constants.AnnotationUnschedulable)
 		needsUpdate = true
 	}
 	if !needsUpdate {
 		return nil
 	}
-	if unschedulable {
-		logrus.Infof("Adding annotation %s to node %s", constants.StorkAnnotationUnschedulable, nodeName)
-	} else {
-		logrus.Infof("Removing annotation %s from node %s", constants.StorkAnnotationUnschedulable, nodeName)
-	}
 	if err := k8sClient.Update(ctx, node); err != nil {
 		return fmt.Errorf("failed to update annotation %s on node %s: %w",
-			constants.StorkAnnotationUnschedulable, nodeName, err)
+			constants.AnnotationUnschedulable, nodeName, err)
 	}
 	return nil
 }

pkg/controller/storagecluster/util.go

@@ -140,16 +140,22 @@ func forceContinueUpgrade(
 }
 
 func evictVMsDuringUpdate(cluster *corev1.StorageCluster) bool {
-	value, exists := cluster.Annotations[constants.AnnotationEvictVMsDuringUpdate]
+	return getBoolVal(cluster.Annotations, constants.AnnotationEvictVMsDuringUpdate, true)
+}
+
+func nodeMarkedUnschedulable(node *v1.Node) bool {
+	return getBoolVal(node.Annotations, constants.AnnotationUnschedulable, false)
+}
+
+func getBoolVal(m map[string]string, key string, defaultVal bool) bool {
+	value, exists := m[key]
 	if !exists {
-		// default to true
-		return true
+		return defaultVal
 	}
 	boolval, err := strconv.ParseBool(value)
 	if err != nil {
-		logrus.Warnf("Invalid value %s for annotation %s; defaulting to true: %v",
-			value, constants.AnnotationEvictVMsDuringUpdate, err)
-		return true
+		logrus.Warnf("Invalid value %s for annotation %s; defaulting to %v: %v", value, key, defaultVal, err)
+		return defaultVal
 	}
 	return boolval
 }