From 0876a20e7c893a30006e6c9533d9f3e74727f11c Mon Sep 17 00:00:00 2001 From: rohan47 Date: Tue, 19 Jan 2021 00:23:01 +0530 Subject: [PATCH] ceph: clean pods stuck in terminating state when a node fails some pods remains in terminating while other manage to be rescheduled on another node successfully. this commit adds code to clean up all the Rook and CSI pods that are stuck in terminating state on a failed node. Signed-off-by: rohan47 --- pkg/operator/ceph/cluster/cephstatus.go | 59 ++++++++ pkg/operator/ceph/cluster/cephstatus_test.go | 138 +++++++++++++++++++ pkg/operator/k8sutil/node.go | 20 ++- pkg/operator/k8sutil/node_test.go | 48 +++++++ 4 files changed, 262 insertions(+), 3 deletions(-) diff --git a/pkg/operator/ceph/cluster/cephstatus.go b/pkg/operator/ceph/cluster/cephstatus.go index 1f09c8b3d..8773a9e89 100644 --- a/pkg/operator/ceph/cluster/cephstatus.go +++ b/pkg/operator/ceph/cluster/cephstatus.go @@ -29,6 +29,7 @@ import ( "github.com/rook/rook/pkg/operator/ceph/config" opcontroller "github.com/rook/rook/pkg/operator/ceph/controller" cephver "github.com/rook/rook/pkg/operator/ceph/version" + "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" @@ -123,6 +124,13 @@ func (c *cephStatusChecker) checkStatus() { if err := c.updateCephStatus(&status, condition, reason, message); err != nil { logger.Errorf("failed to query cluster status in namespace %q. %v", c.clusterInfo.Namespace, err) } + + if status.Health.Status != "HEALTH_OK" { + logger.Debug("checking for stuck pods on not ready nodes") + if err := c.forceDeleteStuckRookPodsOnNotReadyNodes(); err != nil { + logger.Errorf("failed to delete pod on not ready nodes. %v", err) + } + } } // updateStatus updates an object with a given status @@ -257,3 +265,54 @@ func (c *cephStatusChecker) conditionMessageReason(condition cephv1.ConditionTyp return condition, reason, message } + +// forceDeleteStuckPodsOnNotReadyNodes lists all the nodes that are in NotReady state and +// gets all the pods on the failed node and force delete the pods stuck in terminating state. +func (c *cephStatusChecker) forceDeleteStuckRookPodsOnNotReadyNodes() error { + nodes, err := k8sutil.GetNotReadyKubernetesNodes(c.context.Clientset) + if err != nil { + return errors.Wrap(err, "failed to get NotReady nodes") + } + for _, node := range nodes { + pods, err := c.getRookPodsOnNode(node.Name) + if err != nil { + logger.Errorf("failed to get pods on NotReady node %q. %v", node.Name, err) + } + for _, pod := range pods { + if err := k8sutil.ForceDeletePodIfStuck(c.context, pod); err != nil { + logger.Warningf("skipping forced delete of stuck pod %q. %v", pod.Name, err) + } + } + } + return nil +} + +func (c *cephStatusChecker) getRookPodsOnNode(node string) ([]v1.Pod, error) { + clusterName := c.clusterInfo.NamespacedName() + appLabels := []string{ + "csi-rbdplugin-provisioner", + "csi-rbdplugin", + "csi-cephfsplugin-provisioner", + "csi-cephfsplugin", + "rook-ceph-operator", + } + podsOnNode := []v1.Pod{} + pods, err := c.context.Clientset.CoreV1().Pods(clusterName.Namespace).List(context.TODO(), metav1.ListOptions{}) + if err != nil { + return podsOnNode, errors.Wrapf(err, "failed to get pods on node %q", node) + } + for _, pod := range pods.Items { + if pod.Labels["rook_cluster"] == clusterName.Name { + podsOnNode = append(podsOnNode, pod) + continue + } + for _, label := range appLabels { + if pod.Labels["app"] == label { + podsOnNode = append(podsOnNode, pod) + break + } + } + + } + return podsOnNode, nil +} diff --git a/pkg/operator/ceph/cluster/cephstatus_test.go b/pkg/operator/ceph/cluster/cephstatus_test.go index 57c9dd110..0809ac04d 100644 --- a/pkg/operator/ceph/cluster/cephstatus_test.go +++ b/pkg/operator/ceph/cluster/cephstatus_test.go @@ -18,7 +18,10 @@ limitations under the License. package cluster import ( + "context" + "fmt" "reflect" + "sort" "testing" "time" @@ -26,7 +29,11 @@ import ( "github.com/rook/rook/pkg/clusterd" "github.com/rook/rook/pkg/daemon/ceph/client" cephclient "github.com/rook/rook/pkg/daemon/ceph/client" + optest "github.com/rook/rook/pkg/operator/test" "github.com/stretchr/testify/assert" + v1 "k8s.io/api/core/v1" + kerrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" controllerclient "sigs.k8s.io/controller-runtime/pkg/client" ) @@ -197,3 +204,134 @@ func Test_cephStatusChecker_conditionMessageReason(t *testing.T) { }) } } + +func TestForceDeleteStuckRookPodsOnNotReadyNodes(t *testing.T) { + ctx := context.TODO() + clientset := optest.New(t, 1) + clusterInfo := client.NewClusterInfo("test", "test") + clusterName := clusterInfo.NamespacedName() + + context := &clusterd.Context{ + Clientset: clientset, + } + + c := newCephStatusChecker(context, clusterInfo, &cephv1.ClusterSpec{}) + + pod := v1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: "stuck-pod", + Namespace: clusterName.Namespace, + Labels: map[string]string{ + "rook_cluster": clusterName.Name, + }, + }, + } + pod.Spec.NodeName = "node0" + _, err := context.Clientset.CoreV1().Pods(clusterName.Namespace).Create(ctx, &pod, metav1.CreateOptions{}) + assert.NoError(t, err) + + // Create a non matching pod + notDeletePod := pod + notDeletePod.ObjectMeta.Labels = map[string]string{"app": "not-to-be-deleted"} + notDeletePod.ObjectMeta.Name = "not-to-be-deleted" + notDeletePod.DeletionTimestamp = &metav1.Time{Time: time.Now()} + _, err = context.Clientset.CoreV1().Pods(clusterName.Namespace).Create(ctx, ¬DeletePod, metav1.CreateOptions{}) + assert.NoError(t, err) + + // Set the node to NotReady state + nodes, err := context.Clientset.CoreV1().Nodes().List(ctx, metav1.ListOptions{}) + assert.NoError(t, err) + for _, node := range nodes.Items { + node.Status.Conditions[0].Status = v1.ConditionFalse + localnode := node + _, err := context.Clientset.CoreV1().Nodes().Update(ctx, &localnode, metav1.UpdateOptions{}) + assert.NoError(t, err) + } + + // There should be no error + err = c.forceDeleteStuckRookPodsOnNotReadyNodes() + assert.NoError(t, err) + + // The pod should still exist since its not deleted. + p, err := context.Clientset.CoreV1().Pods(clusterInfo.Namespace).Get(ctx, pod.Name, metav1.GetOptions{}) + assert.NoError(t, err) + assert.NotNil(t, p) + + // Add a deletion timestamp to the pod + pod.DeletionTimestamp = &metav1.Time{Time: time.Now()} + _, err = clientset.CoreV1().Pods(clusterName.Namespace).Update(ctx, &pod, metav1.UpdateOptions{}) + assert.NoError(t, err) + + // There should be no error as the pod is deleted + err = c.forceDeleteStuckRookPodsOnNotReadyNodes() + assert.NoError(t, err) + + // The pod should be deleted since the pod is marked as deleted and the node is in NotReady state + _, err = clientset.CoreV1().Pods(clusterName.Namespace).Get(ctx, pod.Name, metav1.GetOptions{}) + assert.Error(t, err) + assert.True(t, kerrors.IsNotFound(err)) + + // The pod should not be deleted as it does not have the matching labels + _, err = clientset.CoreV1().Pods(clusterName.Namespace).Get(ctx, notDeletePod.Name, metav1.GetOptions{}) + assert.NoError(t, err) +} + +func TestGetRookPodsOnNode(t *testing.T) { + ctx := context.TODO() + clientset := optest.New(t, 1) + clusterInfo := client.NewClusterInfo("test", "test") + clusterName := clusterInfo.NamespacedName() + context := &clusterd.Context{ + Clientset: clientset, + } + + c := newCephStatusChecker(context, clusterInfo, &cephv1.ClusterSpec{}) + labels := []map[string]string{ + {"rook_cluster": clusterName.Name}, + {"app": "csi-rbdplugin-provisioner"}, + {"app": "csi-rbdplugin"}, + {"app": "csi-cephfsplugin-provisioner"}, + {"app": "csi-cephfsplugin"}, + {"app": "rook-ceph-operator"}, + {"rook_cluster": "test", "app": "csi-cephfsplugin"}, + {"app": "user-app"}, + } + + pod := v1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: "pod-with-no-label", + Namespace: clusterName.Namespace, + }, + } + pod.Spec.NodeName = "node0" + _, err := context.Clientset.CoreV1().Pods(clusterName.Namespace).Create(ctx, &pod, metav1.CreateOptions{}) + assert.NoError(t, err) + + expectedPodNames := []string{} + for i, label := range labels { + pod.ObjectMeta.Name = fmt.Sprintf("pod-%d", i) + pod.ObjectMeta.Namespace = clusterName.Namespace + pod.ObjectMeta.Labels = label + if label["app"] != "user-app" { + expectedPodNames = append(expectedPodNames, pod.Name) + } + _, err := context.Clientset.CoreV1().Pods(clusterName.Namespace).Create(ctx, &pod, metav1.CreateOptions{}) + assert.NoError(t, err) + } + + pods, err := c.getRookPodsOnNode("node0") + assert.NoError(t, err) + // A pod is having two matching labels and its returned only once + assert.Equal(t, 7, len(pods)) + + podNames := []string{} + for _, pod := range pods { + // Check if the pods has labels + assert.NotEmpty(t, pod.Labels) + podNames = append(podNames, pod.Name) + } + + sort.Strings(expectedPodNames) + sort.Strings(podNames) + assert.Equal(t, expectedPodNames, podNames) +} diff --git a/pkg/operator/k8sutil/node.go b/pkg/operator/k8sutil/node.go index 234d203ab..b9eee998e 100644 --- a/pkg/operator/k8sutil/node.go +++ b/pkg/operator/k8sutil/node.go @@ -283,9 +283,7 @@ func GetNormalizedHostname(kubernetesNode v1.Node) string { func GetKubernetesNodesMatchingRookNodes(rookNodes []rookv1.Node, clientset kubernetes.Interface) ([]v1.Node, error) { ctx := context.TODO() nodes := []v1.Node{} - nodeOptions := metav1.ListOptions{} - nodeOptions.TypeMeta.Kind = "Node" - k8sNodes, err := clientset.CoreV1().Nodes().List(ctx, nodeOptions) + k8sNodes, err := clientset.CoreV1().Nodes().List(ctx, metav1.ListOptions{}) if err != nil { return nodes, fmt.Errorf("failed to list kubernetes nodes. %+v", err) } @@ -299,6 +297,22 @@ func GetKubernetesNodesMatchingRookNodes(rookNodes []rookv1.Node, clientset kube return nodes, nil } +// GetNotReadyKubernetesNodes lists all the nodes that are in NotReady state +func GetNotReadyKubernetesNodes(clientset kubernetes.Interface) ([]v1.Node, error) { + ctx := context.TODO() + nodes := []v1.Node{} + k8sNodes, err := clientset.CoreV1().Nodes().List(ctx, metav1.ListOptions{}) + if err != nil { + return nodes, fmt.Errorf("failed to list kubernetes nodes. %v", err) + } + for _, node := range k8sNodes.Items { + if !NodeIsReady(node) { + nodes = append(nodes, node) + } + } + return nodes, nil +} + // RookNodesMatchingKubernetesNodes returns only the given Rook nodes which have a corresponding // match in the list of Kubernetes nodes. func RookNodesMatchingKubernetesNodes(rookStorage rookv1.StorageScopeSpec, kubernetesNodes []v1.Node) []rookv1.Node { diff --git a/pkg/operator/k8sutil/node_test.go b/pkg/operator/k8sutil/node_test.go index 8ac5b4dc9..288cf92ef 100644 --- a/pkg/operator/k8sutil/node_test.go +++ b/pkg/operator/k8sutil/node_test.go @@ -446,3 +446,51 @@ func TestTopologyLabels(t *testing.T) { topology = ExtractTopologyFromLabels(nodeLabels, additionalTopologyLabels) assert.Equal(t, 0, len(topology)) } + +func TestGetNotReadyKubernetesNodes(t *testing.T) { + ctx := context.TODO() + clientset := optest.New(t, 0) + + //when there is no node + nodes, err := GetNotReadyKubernetesNodes(clientset) + assert.NoError(t, err) + assert.Equal(t, 0, len(nodes)) + + //when all the nodes are in ready state + clientset = optest.New(t, 2) + nodes, err = GetNotReadyKubernetesNodes(clientset) + assert.NoError(t, err) + assert.Equal(t, 0, len(nodes)) + + //when there is a not ready node + node := &v1.Node{ + ObjectMeta: metav1.ObjectMeta{ + Name: "failed", + }, + Status: v1.NodeStatus{ + Conditions: []v1.NodeCondition{ + { + Type: v1.NodeReady, Status: v1.ConditionFalse, + }, + }, + }, + } + _, err = clientset.CoreV1().Nodes().Create(ctx, node, metav1.CreateOptions{}) + assert.NoError(t, err) + nodes, err = GetNotReadyKubernetesNodes(clientset) + assert.NoError(t, err) + assert.Equal(t, 1, len(nodes)) + + // when all the nodes are not ready + allNodes, err := clientset.CoreV1().Nodes().List(ctx, metav1.ListOptions{}) + assert.NoError(t, err) + for _, n := range allNodes.Items { + n.Status.Conditions[0].Status = v1.ConditionFalse + updateNode := n + _, err := clientset.CoreV1().Nodes().Update(ctx, &updateNode, metav1.UpdateOptions{}) + assert.NoError(t, err) + } + nodes, err = GetNotReadyKubernetesNodes(clientset) + assert.NoError(t, err) + assert.Equal(t, 3, len(nodes)) +}