From 171910707b1b55fc80f37abf6ecff3a1378d299b Mon Sep 17 00:00:00 2001 From: Travis Nielsen Date: Wed, 27 May 2020 16:15:50 -0600 Subject: [PATCH 1/4] ceph: refactor pod deletion to k8sutil package The pod deletion is needed by other daemons besides the osds, so we move the helper method into the k8sutil package. Signed-off-by: Travis Nielsen --- pkg/operator/ceph/cluster/osd/health.go | 30 ++------------------ pkg/operator/ceph/cluster/osd/health_test.go | 14 ++++----- pkg/operator/k8sutil/pod.go | 28 ++++++++++++++++++ 3 files changed, 36 insertions(+), 36 deletions(-) diff --git a/pkg/operator/ceph/cluster/osd/health.go b/pkg/operator/ceph/cluster/osd/health.go index ca3cad6aa..1992019ec 100644 --- a/pkg/operator/ceph/cluster/osd/health.go +++ b/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" ) @@ -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 } diff --git a/pkg/operator/ceph/cluster/osd/health_test.go b/pkg/operator/ceph/cluster/osd/health_test.go index fd159dd48..2534068ca 100644 --- a/pkg/operator/ceph/cluster/osd/health_test.go +++ b/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{}) diff --git a/pkg/operator/k8sutil/pod.go b/pkg/operator/k8sutil/pod.go index 50fbbac64..c78cc4715 100644 --- a/pkg/operator/k8sutil/pod.go +++ b/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 +} From 07f095c77275ea6d4c39c74b4ffbccf14526189e Mon Sep 17 00:00:00 2001 From: Travis Nielsen Date: Wed, 27 May 2020 16:17:33 -0600 Subject: [PATCH 2/4] ceph: set osd health check interval to 60s The osd health check every 5 min is not frequent enough to respond quickly enough to osds going down. Reducing the time interval to one minute for faster response time. Signed-off-by: Travis Nielsen --- pkg/operator/ceph/cluster/osd/health.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pkg/operator/ceph/cluster/osd/health.go b/pkg/operator/ceph/cluster/osd/health.go index 1992019ec..aefeb8b8c 100644 --- a/pkg/operator/ceph/cluster/osd/health.go +++ b/pkg/operator/ceph/cluster/osd/health.go @@ -35,7 +35,7 @@ const ( ) var ( - healthCheckInterval = 300 * time.Second + healthCheckInterval = 60 * time.Second ) // OSDHealthMonitor defines OSD process monitoring From ff6a3f80cd39e0b88da56e4fb074f4e57be95e0f Mon Sep 17 00:00:00 2001 From: Travis Nielsen Date: Wed, 27 May 2020 17:15:18 -0600 Subject: [PATCH 3/4] ceph: improve osd update logging The operator should only print helpful info messages when an OSD is going to be updated. If the OSD hasn't changed it is just a debug message. Signed-off-by: Travis Nielsen --- pkg/operator/ceph/cluster/osd/osd.go | 5 +++-- pkg/operator/k8sutil/deployment.go | 3 ++- 2 files changed, 5 insertions(+), 3 deletions(-) diff --git a/pkg/operator/ceph/cluster/osd/osd.go b/pkg/operator/ceph/cluster/osd/osd.go index 0563e0e73..cdd79f454 100644 --- a/pkg/operator/ceph/cluster/osd/osd.go +++ b/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) } } diff --git a/pkg/operator/k8sutil/deployment.go b/pkg/operator/k8sutil/deployment.go index a8b23fe8d..ec335a729 100644 --- a/pkg/operator/k8sutil/deployment.go +++ b/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) } From 3ab060965c4561daa59af3664a66d42d08fbef90 Mon Sep 17 00:00:00 2001 From: Travis Nielsen Date: Wed, 27 May 2020 17:17:08 -0600 Subject: [PATCH 4/4] ceph: force restart mon pods if on bad node If a mon is stuck in terminating state on a failed node, force delete the pod so it can attempt to restart. If the mon is backed by a PV the mon can start on another node as well. Signed-off-by: Travis Nielsen --- pkg/operator/ceph/cluster/mon/health.go | 78 ++++++++++++------- pkg/operator/ceph/cluster/mon/health_test.go | 79 ++++++++++++++++++++ 2 files changed, 130 insertions(+), 27 deletions(-) diff --git a/pkg/operator/ceph/cluster/mon/health.go b/pkg/operator/ceph/cluster/mon/health.go index c520672b2..f48f7d511 100644 --- a/pkg/operator/ceph/cluster/mon/health.go +++ b/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 +} diff --git a/pkg/operator/ceph/cluster/mon/health_test.go b/pkg/operator/ceph/cluster/mon/health_test.go index 8073d48a1..5d85b6e83 100644 --- a/pkg/operator/ceph/cluster/mon/health_test.go +++ b/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 +}