Merge pull request #6999 from rohan47/clean_terminating_pods

ceph: clean pods stuck in terminating state on a failed node
2021-01-29 13:15:36 -07:00
parent deec85cd1d 0876a20e7c
commit 1761874da2
4 changed files with 262 additions and 3 deletions
@@ -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
+}
@@ -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, &notDeletePod, 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)
+}
@@ -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 {
@@ -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))
+}