Merge pull request #5556 from travisn/node-failure-cleanup

Handle node failure more gracefully

pkg/operator/ceph/cluster/mon/health.go

@@ -17,6 +17,7 @@ limitations under the License.
 package mon
 
 import (
+	"fmt"
 	"strings"
 	"time"
 
@@ -27,6 +28,7 @@ import (
 	cephconfig "github.com/rook/rook/pkg/daemon/ceph/config"
 	cephutil "github.com/rook/rook/pkg/daemon/ceph/util"
 	"github.com/rook/rook/pkg/operator/ceph/controller"
+	"github.com/rook/rook/pkg/operator/k8sutil"
 	kerrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
@@ -133,13 +135,13 @@ func (c *Cluster) checkHealth() error {
 			// when the mon isn't in the clusterInfo, but is in quorum and there are
 			// enough mons, remove it else remove it on the next run
 			if inQuorum && len(quorumStatus.MonMap.Mons) > desiredMonCount {
-				logger.Warningf("mon %s not in source of truth but in quorum, removing", mon.Name)
+				logger.Warningf("mon %q not in source of truth but in quorum, removing", mon.Name)
 				if err := c.removeMon(mon.Name); err != nil {
 					logger.Warningf("failed to remove mon %q. %v", mon.Name, err)
 				}
 			} else {
 				logger.Warningf(
-					"mon %s not in source of truth and not in quorum, not enough mons to remove now (wanted: %d, current: %d)",
+					"mon %q not in source of truth and not in quorum, not enough mons to remove now (wanted: %d, current: %d)",
 					mon.Name,
 					desiredMonCount,
 					len(quorumStatus.MonMap.Mons),
@@ -148,38 +150,42 @@ func (c *Cluster) checkHealth() error {
 		}
 
 		if inQuorum {
-			logger.Debugf("mon %s found in quorum", mon.Name)
+			logger.Debugf("mon %q found in quorum", mon.Name)
 			// delete the "timeout" for a mon if the pod is in quorum again
 			if _, ok := c.monTimeoutList[mon.Name]; ok {
 				delete(c.monTimeoutList, mon.Name)
-				logger.Infof("mon %s is back in quorum, removed from mon out timeout list", mon.Name)
+				logger.Infof("mon %q is back in quorum, removed from mon out timeout list", mon.Name)
 			}
-		} else {
-			logger.Debugf("mon %s NOT found in quorum. Mon quorum status: %+v", mon.Name, quorumStatus)
-			allMonsInQuorum = false
-
-			// If not yet set, add the current time, for the timeout
-			// calculation, to the list
-			if _, ok := c.monTimeoutList[mon.Name]; !ok {
-				c.monTimeoutList[mon.Name] = time.Now()
-			}
-
-			// when the timeout for the mon has been reached, continue to the
-			// normal failover/delete mon pod part of the code
-			if time.Since(c.monTimeoutList[mon.Name]) <= MonOutTimeout {
-				timeToFailover := int(MonOutTimeout.Seconds() - time.Since(c.monTimeoutList[mon.Name]).Seconds())
-				logger.Warningf("mon %s not found in quorum, waiting for timeout (%d seconds left) before failover", mon.Name, timeToFailover)
-				continue
-			}
-
-			logger.Warningf("mon %s NOT found in quorum and timeout exceeded, mon will be failed over", mon.Name)
-			c.failMon(len(quorumStatus.MonMap.Mons), desiredMonCount, mon.Name)
-			// only deal with one unhealthy mon per health check
-			return nil
+			continue
 		}
+
+		logger.Debugf("mon %q NOT found in quorum. Mon quorum status: %+v", mon.Name, quorumStatus)
+		allMonsInQuorum = false
+
+		// If not yet set, add the current time, for the timeout
+		// calculation, to the list
+		if _, ok := c.monTimeoutList[mon.Name]; !ok {
+			c.monTimeoutList[mon.Name] = time.Now()
+		}
+
+		// when the timeout for the mon has been reached, continue to the
+		// normal failover/delete mon pod part of the code
+		if time.Since(c.monTimeoutList[mon.Name]) <= MonOutTimeout {
+			timeToFailover := int(MonOutTimeout.Seconds() - time.Since(c.monTimeoutList[mon.Name]).Seconds())
+			logger.Warningf("mon %q not found in quorum, waiting for timeout (%d seconds left) before failover", mon.Name, timeToFailover)
+
+			// Restart the mon if it is stuck on a failed node
+			c.restartMonIfStuckTerminating(mon.Name)
+			continue
+		}
+
+		logger.Warningf("mon %q NOT found in quorum and timeout exceeded, mon will be failed over", mon.Name)
+		c.failMon(len(quorumStatus.MonMap.Mons), desiredMonCount, mon.Name)
+		// only deal with one unhealthy mon per health check
+		return nil
 	}
 
-	// after all unhealthy mons have been removed/failovered
+	// after all unhealthy mons have been removed or failed over
 	// handle all mons that haven't been in the Ceph mon map
 	for mon := range monsNotFound {
 		logger.Warningf("mon %s NOT found in ceph mon map, failover", mon)
@@ -422,3 +428,21 @@ func (c *Cluster) addOrRemoveExternalMonitor(status client.MonStatusResponse) (b
 	logger.Debugf("ClusterInfo.Monitors is %+v", c.ClusterInfo.Monitors)
 	return changed, nil
 }
+
+// restartMonIfStuckTerminating will check if a mon is on a node that is not ready.
+// If the pod is stuck in terminating state, go ahead and force delete the pod so K8s
+// will allow the mon to be restarted on another node.
+func (c *Cluster) restartMonIfStuckTerminating(monName string) error {
+	logger.Debugf("Checking for a stuck mon %q pod", monName)
+	labels := fmt.Sprintf("app=%s,mon=%s", AppName, monName)
+	pods, err := c.context.Clientset.CoreV1().Pods(c.Namespace).List(metav1.ListOptions{LabelSelector: labels})
+	if err != nil {
+		return errors.Wrapf(err, "failed to get pod for mon %q", monName)
+	}
+	for _, pod := range pods.Items {
+		if err := k8sutil.ForceDeletePodIfStuck(c.context, pod); err != nil {
+			logger.Warningf("skipping forced restart of mon %q. %v", monName, err)
+		}
+	}
+	return nil
+}

pkg/operator/ceph/cluster/mon/health_test.go

@@ -22,18 +22,23 @@ import (
 	"os"
 	"sync"
 	"testing"
+	"time"
 
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	"github.com/rook/rook/pkg/clusterd"
 	"github.com/rook/rook/pkg/daemon/ceph/client"
 	clienttest "github.com/rook/rook/pkg/daemon/ceph/client/test"
 	cephconfig "github.com/rook/rook/pkg/daemon/ceph/config"
+	"github.com/rook/rook/pkg/operator/k8sutil"
 	testopk8s "github.com/rook/rook/pkg/operator/k8sutil/test"
 	"github.com/rook/rook/pkg/operator/test"
 	exectest "github.com/rook/rook/pkg/util/exec/test"
 	"github.com/stretchr/testify/assert"
 	apps "k8s.io/api/apps/v1"
+	v1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
 )
 
 func TestCheckHealth(t *testing.T) {
@@ -284,3 +289,77 @@ func TestAddOrRemoveExternalMonitor(t *testing.T) {
 	// ClusterInfo should now have 2 monitors
 	assert.Equal(t, 2, len(c.ClusterInfo.Monitors))
 }
+
+func TestForceDeleteFailedMon(t *testing.T) {
+	clientset := test.New(t, 1)
+	context := &clusterd.Context{
+		Clientset: clientset,
+	}
+	c := &Cluster{
+		context:   context,
+		Namespace: "ns",
+	}
+
+	failedMonName := "a"
+	monToFail := createTestMonPod(t, clientset, c.Namespace, failedMonName)
+	createTestMonPod(t, clientset, c.Namespace, "b")
+	createTestMonPod(t, clientset, c.Namespace, "c")
+
+	assert.NoError(t, c.restartMonIfStuckTerminating(failedMonName))
+
+	// The mon should still exist since it wasn't in a deleted state
+	p, err := context.Clientset.CoreV1().Pods(c.Namespace).Get(monToFail.Name, metav1.GetOptions{})
+	assert.NoError(t, err)
+	assert.NotNil(t, p)
+
+	// Add a deletion timestamp to the pod
+	monToFail.DeletionTimestamp = &metav1.Time{Time: time.Now()}
+	_, err = context.Clientset.CoreV1().Pods(c.Namespace).Update(&monToFail)
+	assert.NoError(t, err)
+
+	assert.NoError(t, c.restartMonIfStuckTerminating(failedMonName))
+
+	// The pod should still exist since the node is ready
+	p, err = context.Clientset.CoreV1().Pods(c.Namespace).Get(monToFail.Name, metav1.GetOptions{})
+	assert.NoError(t, err)
+	assert.NotNil(t, p)
+
+	// Set the node to a not ready state
+	nodes, err := context.Clientset.CoreV1().Nodes().List(metav1.ListOptions{})
+	assert.NoError(t, err)
+	for _, node := range nodes.Items {
+		node.Status.Conditions[0].Status = v1.ConditionFalse
+		_, err := context.Clientset.CoreV1().Nodes().Update(&node)
+		assert.NoError(t, err)
+	}
+
+	assert.NoError(t, c.restartMonIfStuckTerminating(failedMonName))
+
+	// The pod should be deleted since the pod is marked as deleted and the node is not ready
+	_, err = context.Clientset.CoreV1().Pods(c.Namespace).Get(monToFail.Name, metav1.GetOptions{})
+	assert.Error(t, err)
+	assert.True(t, errors.IsNotFound(err))
+
+	// mons b and c should still exist
+	_, err = context.Clientset.CoreV1().Pods(c.Namespace).Get("b-test", metav1.GetOptions{})
+	assert.NoError(t, err)
+	_, err = context.Clientset.CoreV1().Pods(c.Namespace).Get("c-test", metav1.GetOptions{})
+	assert.NoError(t, err)
+}
+
+func createTestMonPod(t *testing.T, clientset kubernetes.Interface, namespace, name string) v1.Pod {
+	pod := v1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name + "-test",
+			Namespace: namespace,
+			Labels: map[string]string{
+				k8sutil.AppAttr: AppName,
+				"mon":           name,
+			},
+		},
+	}
+	pod.Spec.NodeName = "node0"
+	_, err := clientset.CoreV1().Pods(namespace).Create(&pod)
+	assert.NoError(t, err)
+	return pod
+}

pkg/operator/ceph/cluster/osd/health.go

@@ -24,7 +24,6 @@ import (
 	"github.com/rook/rook/pkg/clusterd"
 	"github.com/rook/rook/pkg/daemon/ceph/client"
 	"github.com/rook/rook/pkg/operator/k8sutil"
-	v1 "k8s.io/api/core/v1"
 	kerrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
@@ -36,7 +35,7 @@ const (
 )
 
 var (
-	healthCheckInterval = 300 * time.Second
+	healthCheckInterval = 60 * time.Second
 )
 
 // OSDHealthMonitor defines OSD process monitoring
@@ -160,35 +159,10 @@ func (m *OSDHealthMonitor) restartOSDIfStuck(osdID int) error {
 	if err != nil {
 		return errors.Wrapf(err, "failed to get OSD with ID %d", osdID)
 	}
-	return m.restartOSDPodsIfStuck(osdID, pods)
-}
-
-func (m *OSDHealthMonitor) restartOSDPodsIfStuck(osdID int, pods *v1.PodList) error {
 	for _, pod := range pods.Items {
-		logger.Debugf("checking if osd %d pod is stuck and should be force deleted", osdID)
-		if pod.DeletionTimestamp.IsZero() {
-			logger.Debugf("skipping restart of OSD %d since the pod is not deleted", osdID)
-			continue
+		if err := k8sutil.ForceDeletePodIfStuck(m.context, pod); err != nil {
+			logger.Warningf("skipping restart of OSD %d. %v", osdID, err)
 		}
-		node, err := m.context.Clientset.CoreV1().Nodes().Get(pod.Spec.NodeName, metav1.GetOptions{})
-		if err != nil {
-			logger.Warningf("skipping restart of OSD %d since the node status is not available. %v", osdID, err)
-			continue
-		}
-		if k8sutil.NodeIsReady(*node) {
-			logger.Debugf("skipping restart of OSD %d since the node status is ready", osdID)
-			continue
-		}
-
-		logger.Infof("force deleting pod %q that appears to be stuck terminating", pod.Name)
-		var gracePeriod int64
-		deleteOpts := &metav1.DeleteOptions{GracePeriodSeconds: &gracePeriod}
-		if err := m.context.Clientset.CoreV1().Pods(m.namespace).Delete(pod.Name, deleteOpts); err != nil {
-			logger.Warningf("pod %q deletion failed. %v", pod.Name, err)
-			continue
-		}
-		logger.Infof("pod %q deletion succeeded", pod.Name)
 	}
-
 	return nil
 }

pkg/operator/ceph/cluster/osd/health_test.go

@@ -117,7 +117,7 @@ func TestOSDRestartIfStuck(t *testing.T) {
 		Clientset: clientset,
 	}
 
-	pod := &v1.Pod{
+	pod := v1.Pod{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      "osd0",
 			Namespace: namespace,
@@ -128,13 +128,12 @@ func TestOSDRestartIfStuck(t *testing.T) {
 		},
 	}
 	pod.Spec.NodeName = "node0"
-	_, err := context.Clientset.CoreV1().Pods(namespace).Create(pod)
+	_, err := context.Clientset.CoreV1().Pods(namespace).Create(&pod)
 	assert.NoError(t, err)
 
 	m := NewOSDHealthMonitor(context, namespace, false)
 
-	podList := &v1.PodList{Items: []v1.Pod{*pod}}
-	assert.NoError(t, m.restartOSDPodsIfStuck(0, podList))
+	assert.NoError(t, k8sutil.ForceDeletePodIfStuck(m.context, pod))
 
 	// The pod should still exist since it wasn't in a deleted state
 	p, err := context.Clientset.CoreV1().Pods(namespace).Get(pod.Name, metav1.GetOptions{})
@@ -143,11 +142,10 @@ func TestOSDRestartIfStuck(t *testing.T) {
 
 	// Add a deletion timestamp to the pod
 	pod.DeletionTimestamp = &metav1.Time{Time: time.Now()}
-	_, err = context.Clientset.CoreV1().Pods(namespace).Update(pod)
+	_, err = context.Clientset.CoreV1().Pods(namespace).Update(&pod)
 	assert.NoError(t, err)
 
-	podList = &v1.PodList{Items: []v1.Pod{*pod}}
-	assert.NoError(t, m.restartOSDPodsIfStuck(0, podList))
+	assert.NoError(t, k8sutil.ForceDeletePodIfStuck(m.context, pod))
 
 	// The pod should still exist since the node is ready
 	p, err = context.Clientset.CoreV1().Pods(namespace).Get(pod.Name, metav1.GetOptions{})
@@ -163,7 +161,7 @@ func TestOSDRestartIfStuck(t *testing.T) {
 		assert.NoError(t, err)
 	}
 
-	assert.NoError(t, m.restartOSDPodsIfStuck(0, podList))
+	assert.NoError(t, k8sutil.ForceDeletePodIfStuck(m.context, pod))
 
 	// The pod should be deleted since the pod is marked as deleted and the node is not ready
 	_, err = context.Clientset.CoreV1().Pods(namespace).Get(pod.Name, metav1.GetOptions{})

pkg/operator/ceph/cluster/osd/osd.go

@@ -556,7 +556,7 @@ func (c *Cluster) startOSDDaemonsOnNode(nodeName string, config *provisionConfig
 		_, createErr := c.context.Clientset.AppsV1().Deployments(c.Namespace).Create(dp)
 		if createErr != nil {
 			if kerrors.IsAlreadyExists(createErr) {
-				logger.Infof("deployment for osd %d already exists. updating if needed", osd.ID)
+				logger.Debugf("deployment for osd %d already exists. updating if needed", osd.ID)
 				if err = updateDeploymentAndWait(c.context, dp, c.Namespace, opconfig.OsdType, strconv.Itoa(osd.ID), c.skipUpgradeChecks, c.continueUpgradeAfterChecksEvenIfNotHealthy); err != nil {
 					logger.Errorf("failed to update osd deployment %d. %v", osd.ID, err)
 				}
@@ -565,8 +565,9 @@ func (c *Cluster) startOSDDaemonsOnNode(nodeName string, config *provisionConfig
 				logger.Warningf("failed to create osd deployment for node %q, osd %+v. %v", n.Name, osd, createErr)
 				continue
 			}
+		} else {
+			logger.Infof("created deployment for osd %d", osd.ID)
 		}
-		logger.Infof("started deployment for osd %d", osd.ID)
 	}
 }
 

pkg/operator/k8sutil/deployment.go

@@ -74,6 +74,8 @@ func UpdateDeploymentAndWait(context *clusterd.Context, modifiedDeployment *apps
 
 	// If deployments are different, let's update!
 	if !patchResult.IsEmpty() {
+		logger.Infof("updating deployment %q after verifying it is safe to stop", modifiedDeployment.Name)
+
 		// Let's verify the deployment can be stopped
 		// retry for 5 times, every minute
 		err = util.Retry(5, 60*time.Second, func() error {
@@ -89,7 +91,6 @@ func UpdateDeploymentAndWait(context *clusterd.Context, modifiedDeployment *apps
 			return nil, fmt.Errorf("failed to set hash annotation on deployment %q. %v", modifiedDeployment.Name, err)
 		}
 
-		logger.Infof("updating deployment %q", modifiedDeployment.Name)
 		if _, err := context.Clientset.AppsV1().Deployments(namespace).Update(modifiedDeployment); err != nil {
 			return nil, fmt.Errorf("failed to update deployment %q. %v", modifiedDeployment.Name, err)
 		}

pkg/operator/k8sutil/pod.go

@@ -25,7 +25,9 @@ import (
 	"strconv"
 	"strings"
 
+	"github.com/pkg/errors"
 	rookv1 "github.com/rook/rook/pkg/apis/rook.io/v1"
+	"github.com/rook/rook/pkg/clusterd"
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -331,3 +333,29 @@ func SetNodeAntiAffinityForPod(pod *v1.PodSpec, p rookv1.Placement, requiredDuri
 			})
 	}
 }
+
+func ForceDeletePodIfStuck(context *clusterd.Context, pod v1.Pod) error {
+	logger.Debugf("checking if pod %q is stuck and should be force deleted", pod.Name)
+	if pod.DeletionTimestamp.IsZero() {
+		logger.Debugf("skipping pod %q restart since the pod is not deleted", pod.Name)
+		return nil
+	}
+	node, err := context.Clientset.CoreV1().Nodes().Get(pod.Spec.NodeName, metav1.GetOptions{})
+	if err != nil {
+		return errors.Wrap(err, "node status is not available")
+	}
+	if NodeIsReady(*node) {
+		logger.Debugf("skipping restart of pod %q since the node status is ready", pod.Name)
+		return nil
+	}
+
+	logger.Infof("force deleting pod %q that appears to be stuck terminating", pod.Name)
+	var gracePeriod int64
+	deleteOpts := &metav1.DeleteOptions{GracePeriodSeconds: &gracePeriod}
+	if err := context.Clientset.CoreV1().Pods(pod.Namespace).Delete(pod.Name, deleteOpts); err != nil {
+		logger.Warningf("pod %q deletion failed. %v", pod.Name, err)
+		return nil
+	}
+	logger.Infof("pod %q deletion succeeded", pod.Name)
+	return nil
+}