Merge pull request #9255 from rook/mergify/bp/release-1.7/pr-9230

osd: check if osd is safe-to-destroy before removal (backport #9230)

.github/workflows/canary-integration-test.yml

@@ -31,6 +31,9 @@ jobs:
       with:
         go-version: 1.16
 
+    - name: install deps
+      run: tests/scripts/github-action-helper.sh install_deps
+
     - name: setup minikube
       uses: manusa/actions-setup-minikube@v2.4.2
       with:
@@ -84,6 +87,21 @@ jobs:
     - name: check-ownerreferences
       run: tests/scripts/github-action-helper.sh check_ownerreferences
 
+    - name: test osd removal jobs
+      run: |
+        kubectl -n rook-ceph delete deploy/rook-ceph-operator
+        kubectl -n rook-ceph delete deploy/rook-ceph-osd-1 --grace-period=0 --force
+        sed -i 's/<OSD-IDs>/1/' cluster/examples/kubernetes/ceph/osd-purge.yaml
+        # the CI must force the deletion since we use replica 1 on 2 OSDs
+        sed -i 's/false/true/' cluster/examples/kubernetes/ceph/osd-purge.yaml
+        yq write -i cluster/examples/kubernetes/ceph/osd-purge.yaml "spec.template.spec.containers[0].image" rook/ceph:local-build
+        kubectl -n rook-ceph create -f cluster/examples/kubernetes/ceph/osd-purge.yaml
+        toolbox=$(kubectl get pod -l app=rook-ceph-tools -n rook-ceph -o jsonpath='{.items[*].metadata.name}')
+        kubectl -n rook-ceph exec $toolbox -- ceph status
+        timeout 120 sh -c "until kubectl -n rook-ceph exec $toolbox -- ceph osd tree|grep -qE 'osd.1.*.destroyed'; do echo 'waiting for ceph osd 1 to be destroyed'; sleep 1; done"
+        kubectl -n rook-ceph exec $toolbox -- ceph status
+        kubectl -n rook-ceph exec $toolbox -- ceph osd tree
+
     - name: collect common logs
       if: always()
       run: |

cluster/examples/kubernetes/ceph/osd-purge.yaml

@@ -29,7 +29,19 @@ spec:
           # TODO: Insert the OSD ID in the last parameter that is to be removed
           # The OSD IDs are a comma-separated list. For example: "0" or "0,2".
           # If you want to preserve the OSD PVCs, set `--preserve-pvc true`.
-          args: ["ceph", "osd", "remove", "--preserve-pvc", "false", "--osd-ids", "<OSD-IDs>"]
+          #
+          # A --force-osd-removal option is available if the OSD should be destroyed even though the
+          # removal could lead to data loss.
+          args:
+            - "ceph"
+            - "osd"
+            - "remove"
+            - "--preserve-pvc"
+            - "false"
+            - "--force-osd-removal"
+            - "false"
+            - "--osd-ids"
+            - "<OSD-IDs>"
           env:
             - name: POD_NAMESPACE
               valueFrom:

cmd/rook/ceph/osd.go

@@ -19,6 +19,7 @@ package ceph
 import (
 	"encoding/json"
 	"os"
+	"strconv"
 	"strings"
 
 	"k8s.io/client-go/kubernetes"
@@ -70,7 +71,8 @@ var (
 	blockPath               string
 	lvBackedPV              bool
 	osdIDsToRemove          string
-	preservePVC             bool
+	preservePVC             string
+	forceOSDRemoval         string
 )
 
 func addOSDFlags(command *cobra.Command) {
@@ -99,7 +101,8 @@ func addOSDFlags(command *cobra.Command) {
 
 	// flags for removing OSDs that are unhealthy or otherwise should be purged from the cluster
 	osdRemoveCmd.Flags().StringVar(&osdIDsToRemove, "osd-ids", "", "OSD IDs to remove from the cluster")
-	osdRemoveCmd.Flags().BoolVar(&preservePVC, "preserve-pvc", false, "Whether PVCs for OSDs will be deleted")
+	osdRemoveCmd.Flags().StringVar(&preservePVC, "preserve-pvc", "false", "Whether PVCs for OSDs will be deleted")
+	osdRemoveCmd.Flags().StringVar(&forceOSDRemoval, "force-osd-removal", "false", "Whether to force remove the OSD")
 
 	// add the subcommands to the parent osd command
 	osdCmd.AddCommand(osdConfigCmd,
@@ -261,11 +264,29 @@ func removeOSDs(cmd *cobra.Command, args []string) error {
 
 	context := createContext()
 
+	// We use strings instead of bool since the flag package has issues with parsing bools, or
+	// perhaps it's the translation between YAML and code... It's unclear but see:
+	// starting Rook v1.7.0-alpha.0.660.gb13faecc8 with arguments '/usr/local/bin/rook ceph osd remove --preserve-pvc false --force-osd-removal false --osd-ids 1'
+	// flag values: --force-osd-removal=true, --help=false, --log-level=DEBUG, --operator-image=,
+	// --osd-ids=1, --preserve-pvc=true, --service-account=
+	//
+	// Bools are false but they are interpreted true by the flag package.
+
+	forceOSDRemovalBool, err := strconv.ParseBool(forceOSDRemoval)
+	if err != nil {
+		return errors.Wrapf(err, "failed to parse --force-osd-removal flag")
+	}
+	preservePVCBool, err := strconv.ParseBool(preservePVC)
+	if err != nil {
+		return errors.Wrapf(err, "failed to parse --preserve-pvc flag")
+	}
+
 	// Run OSD remove sequence
-	err := osddaemon.RemoveOSDs(context, &clusterInfo, strings.Split(osdIDsToRemove, ","), preservePVC)
+	err = osddaemon.RemoveOSDs(context, &clusterInfo, strings.Split(osdIDsToRemove, ","), preservePVCBool, forceOSDRemovalBool)
 	if err != nil {
 		rook.TerminateFatal(err)
 	}
+
 	return nil
 }
 

pkg/daemon/ceph/client/crash.go

@@ -57,26 +57,31 @@ type CrashList struct {
 
 // GetCrashList gets the list of Crashes.
 func GetCrashList(context *clusterd.Context, clusterInfo *ClusterInfo) ([]CrashList, error) {
-	crashargs := []string{"crash", "ls"}
-	output, err := NewCephCommand(context, clusterInfo, crashargs).Run()
+	crashArgs := []string{"crash", "ls"}
+	output, err := NewCephCommand(context, clusterInfo, crashArgs).Run()
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to list ceph crash")
 	}
+
 	var crash []CrashList
 	err = json.Unmarshal(output, &crash)
 	if err != nil {
 		return nil, errors.Wrapf(err, "failed to unmarshal crash ls response. %s", string(output))
 	}
+
 	return crash, err
 }
 
 // ArchiveCrash archives the crash with respective crashID
 func ArchiveCrash(context *clusterd.Context, clusterInfo *ClusterInfo, crashID string) error {
-	crashsilenceargs := []string{"crash", "archive", crashID}
-	_, err := NewCephCommand(context, clusterInfo, crashsilenceargs).Run()
+	logger.Infof("silencing crash %q", crashID)
+	crashSilenceArgs := []string{"crash", "archive", crashID}
+	_, err := NewCephCommand(context, clusterInfo, crashSilenceArgs).Run()
 	if err != nil {
 		return errors.Wrapf(err, "failed to archive crash %q", crashID)
 	}
+
+	logger.Infof("successfully silenced crash %q", crashID)
 	return nil
 }
 
@@ -86,5 +91,6 @@ func GetCrash(context *clusterd.Context, clusterInfo *ClusterInfo) ([]CrashList,
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to list ceph crash")
 	}
+
 	return crash, nil
 }

pkg/daemon/ceph/client/osd.go

@@ -263,7 +263,7 @@ func OsdSafeToDestroy(context *clusterd.Context, clusterInfo *ClusterInfo, osdID
 
 	var output SafeToDestroyStatus
 	if err := json.Unmarshal(buf, &output); err != nil {
-		return false, errors.Wrap(err, "failed to unmarshal safe-to-destroy response")
+		return false, errors.Wrapf(err, "failed to unmarshal safe-to-destroy response. %s", string(buf))
 	}
 	if len(output.SafeToDestroy) != 0 && output.SafeToDestroy[0] == osdID {
 		return true, nil

pkg/daemon/ceph/osd/remove.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"fmt"
 	"strconv"
+	"time"
 
 	kerrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -32,8 +33,7 @@ import (
 )
 
 // RemoveOSDs purges a list of OSDs from the cluster
-func RemoveOSDs(context *clusterd.Context, clusterInfo *client.ClusterInfo, osdsToRemove []string, preservePVC bool) error {
-
+func RemoveOSDs(context *clusterd.Context, clusterInfo *client.ClusterInfo, osdsToRemove []string, preservePVC, forceOSDRemoval bool) error {
 	// Generate the ceph config for running ceph commands similar to the operator
 	if err := client.WriteCephConfig(context, clusterInfo); err != nil {
 		return errors.Wrap(err, "failed to write the ceph config")
@@ -59,8 +59,43 @@ func RemoveOSDs(context *clusterd.Context, clusterInfo *client.ClusterInfo, osds
 		if status == upStatus {
 			logger.Infof("osd.%d is healthy. It cannot be removed unless it is 'down'", osdID)
 			continue
+		} else {
+			logger.Infof("osd.%d is marked 'DOWN'", osdID)
+		}
+
+		// Check we can remove the OSD
+		// Loop forever until the osd is safe-to-destroy
+		for {
+			isSafeToDestroy, err := client.OsdSafeToDestroy(context, clusterInfo, osdID)
+			if err != nil {
+				// If we want to force remove the OSD and there was an error let's break outside of
+				// the loop and proceed with the OSD removal
+				if forceOSDRemoval {
+					logger.Errorf("failed to check if osd %d is safe to destroy, but force removal is enabled so proceeding with removal. %v", osdID, err)
+					break
+				} else {
+					logger.Errorf("failed to check if osd %d is safe to destroy, retrying in 1m. %v", osdID, err)
+					time.Sleep(1 * time.Minute)
+					continue
+				}
+			}
+
+			// If no error and the OSD is safe to destroy, we can proceed with the OSD removal
+			if isSafeToDestroy {
+				logger.Infof("osd.%d is safe to destroy, proceeding", osdID)
+				break
+			} else {
+				// If we arrive here and forceOSDRemoval is true, we should proceed with the OSD removal
+				if forceOSDRemoval {
+					logger.Infof("osd.%d is NOT be ok to destroy but force removal is enabled so proceeding with removal", osdID)
+					break
+				}
+				// Else we wait until the OSD can be removed
+				logger.Warningf("osd.%d is NOT be ok to destroy, retrying in 1m until success", osdID)
+				time.Sleep(1 * time.Minute)
+			}
 		}
-		logger.Infof("osd.%d is marked 'DOWN'. Removing it", osdID)
+
 		removeOSD(context, clusterInfo, osdID, preservePVC)
 	}
 
@@ -76,6 +111,7 @@ func removeOSD(clusterdContext *clusterd.Context, clusterInfo *client.ClusterInf
 	}
 
 	// Mark the OSD as out.
+	logger.Infof("marking osd.%d out", osdID)
 	args := []string{"osd", "out", fmt.Sprintf("osd.%d", osdID)}
 	_, err = client.NewCephCommand(clusterdContext, clusterInfo, args).Run()
 	if err != nil {
@@ -140,18 +176,21 @@ func removeOSD(clusterdContext *clusterd.Context, clusterInfo *client.ClusterInf
 	}
 
 	// purge the osd
-	purgeosdargs := []string{"osd", "purge", fmt.Sprintf("osd.%d", osdID), "--force", "--yes-i-really-mean-it"}
-	_, err = client.NewCephCommand(clusterdContext, clusterInfo, purgeosdargs).Run()
+	logger.Infof("destroying osd.%d", osdID)
+	purgeOSDArgs := []string{"osd", "destroy", fmt.Sprintf("osd.%d", osdID), "--yes-i-really-mean-it"}
+	_, err = client.NewCephCommand(clusterdContext, clusterInfo, purgeOSDArgs).Run()
 	if err != nil {
 		logger.Errorf("failed to purge osd.%d. %v", osdID, err)
 	}
 
 	// Attempting to remove the parent host. Errors can be ignored if there are other OSDs on the same host
-	hostargs := []string{"osd", "crush", "rm", hostName}
-	_, err = client.NewCephCommand(clusterdContext, clusterInfo, hostargs).Run()
+	logger.Infof("removing osd.%d from ceph", osdID)
+	hostArgs := []string{"osd", "crush", "rm", hostName}
+	_, err = client.NewCephCommand(clusterdContext, clusterInfo, hostArgs).Run()
 	if err != nil {
 		logger.Errorf("failed to remove CRUSH host %q. %v", hostName, err)
 	}
+
 	// call archiveCrash to silence crash warning in ceph health if any
 	archiveCrash(clusterdContext, clusterInfo, osdID)
 
@@ -169,13 +208,15 @@ func archiveCrash(clusterdContext *clusterd.Context, clusterInfo *client.Cluster
 		logger.Info("no ceph crash to silence")
 		return
 	}
+
 	var crashID string
 	for _, c := range crash {
 		if c.Entity == fmt.Sprintf("osd.%d", osdID) {
 			crashID = c.ID
 			break
 		}
 	}
+
 	err = client.ArchiveCrash(clusterdContext, clusterInfo, crashID)
 	if err != nil {
 		logger.Errorf("failed to archive the crash %q. %v", crashID, err)

tests/scripts/collect-logs.sh

@@ -19,11 +19,15 @@ $CEPH_CMD report > "${LOG_DIR}"/ceph-report.txt
 kubectl -n "${OPERATOR_NAMESPACE}" logs deploy/rook-ceph-operator > "${LOG_DIR}"/operator-logs.txt
 kubectl -n "${OPERATOR_NAMESPACE}" get pods -o wide > "${LOG_DIR}"/operator-pods-list.txt
 kubectl -n "${CLUSTER_NAMESPACE}" get pods -o wide > "${LOG_DIR}"/cluster-pods-list.txt
+kubectl -n "${CLUSTER_NAMESPACE}" get jobs -o wide > "${LOG_DIR}"/cluster-jobs-list.txt
 prepare_job="$(kubectl -n "${CLUSTER_NAMESPACE}" get job -l app=rook-ceph-osd-prepare --output name | awk 'FNR <= 1')" # outputs job/<name>
+removal_job="$(kubectl -n "${CLUSTER_NAMESPACE}" get job -l app=rook-ceph-purge-osd --output name | awk 'FNR <= 1')" # outputs job/<name>
 kubectl -n "${CLUSTER_NAMESPACE}" describe "${prepare_job}" > "${LOG_DIR}"/osd-prepare-describe.txt
 kubectl -n "${CLUSTER_NAMESPACE}" logs "${prepare_job}" > "${LOG_DIR}"/osd-prepare-logs.txt
 kubectl -n "${CLUSTER_NAMESPACE}" describe deploy/rook-ceph-osd-0 > "${LOG_DIR}"/rook-ceph-osd-0-describe.txt
 kubectl -n "${CLUSTER_NAMESPACE}" describe deploy/rook-ceph-osd-1 > "${LOG_DIR}"/rook-ceph-osd-1-describe.txt
+kubectl -n "${CLUSTER_NAMESPACE}" describe "${removal_job}" > "${LOG_DIR}"/osd-removal-describe.txt
+kubectl -n "${CLUSTER_NAMESPACE}" logs "${removal_job}" > "${LOG_DIR}"/osd-removal-logs.txt
 kubectl -n "${CLUSTER_NAMESPACE}" logs deploy/rook-ceph-osd-0 --all-containers > "${LOG_DIR}"/rook-ceph-osd-0-logs.txt
 kubectl -n "${CLUSTER_NAMESPACE}" logs deploy/rook-ceph-osd-1 --all-containers > "${LOG_DIR}"/rook-ceph-osd-1-logs.txt
 kubectl get all -n "${OPERATOR_NAMESPACE}" -o wide > "${LOG_DIR}"/operator-wide.txt