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)) +}