diff --git a/cmd/rook/ceph/osd.go b/cmd/rook/ceph/osd.go index 5c2873ac3..1decf71d8 100644 --- a/cmd/rook/ceph/osd.go +++ b/cmd/rook/ceph/osd.go @@ -19,7 +19,6 @@ package ceph import ( "context" "encoding/json" - "fmt" "os" "path" "strconv" @@ -30,9 +29,6 @@ import ( "github.com/pkg/errors" "github.com/rook/rook/cmd/rook/rook" cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1" - "github.com/rook/rook/pkg/clusterd" - cleanup "github.com/rook/rook/pkg/daemon/ceph/cleanup" - "github.com/rook/rook/pkg/daemon/ceph/client" osddaemon "github.com/rook/rook/pkg/daemon/ceph/osd" "github.com/rook/rook/pkg/operator/ceph/cluster/mon" "github.com/rook/rook/pkg/operator/ceph/cluster/osd" @@ -263,11 +259,11 @@ func prepareOSD(cmd *cobra.Command, args []string) error { // destroy the OSD using the OSD ID var replaceOSD *osd.OSDReplaceInfo if replaceOSDID != -1 { - osdInfo, err := destroyOSD(context, &clusterInfo, replaceOSDID) + logger.Infof("destroying osd.%d and cleaning its backing device", replaceOSDID) + replaceOSD, err = osddaemon.DestroyOSD(context, &clusterInfo, replaceOSDID, cfg.pvcBacked, cfg.storeConfig.EncryptedDevice) if err != nil { - rook.TerminateFatal(errors.Wrapf(err, "failed to destroy OSD %d.", osdInfo.ID)) + rook.TerminateFatal(errors.Wrapf(err, "failed to destroy OSD %d.", replaceOSDID)) } - replaceOSD = &oposd.OSDReplaceInfo{ID: osdInfo.ID, Path: osdInfo.BlockPath} } agent := osddaemon.NewAgent(context, dataDevices, cfg.metadataDevice, forceFormat, @@ -417,36 +413,3 @@ func readCephSecret(path string) error { } return nil } - -func destroyOSD(context *clusterd.Context, clusterInfo *client.ClusterInfo, osdID int) (*oposd.OSDInfo, error) { - osdInfo, err := osddaemon.GetOSDInfoById(context, clusterInfo, osdID) - if err != nil { - return nil, errors.Wrapf(err, "failed to get OSD info for OSD.%d", osdID) - } - - // destroy the osd - logger.Infof("destroying OSD %d on path %q in %q mode", osdInfo.ID, osdInfo.BlockPath, osdInfo.CVMode) - destroyOSDArgs := []string{"osd", "destroy", fmt.Sprintf("osd.%d", osdInfo.ID), "--yes-i-really-mean-it"} - _, err = client.NewCephCommand(context, clusterInfo, destroyOSDArgs).Run() - if err != nil { - return nil, errors.Wrapf(err, "failed to destroy osd.%d.", osdInfo.ID) - } - - // Sanitize OSD disk - s := cleanup.NewDiskSanitizer(context, clusterInfo, - &cephv1.SanitizeDisksSpec{ - Method: cephv1.SanitizeMethodProperty(cephv1.SanitizeMethodComplete), - DataSource: cephv1.SanitizeDataSourceProperty(cephv1.SanitizeDataSourceZero), - Iteration: 1, - }, - ) - - // TODO: handle disk sanitization errors - if osdInfo.CVMode == "raw" { - s.SanitizeRawDisk([]oposd.OSDInfo{osdInfo}) - } else if osdInfo.CVMode == "lvm" { - s.SanitizeLVMDisk([]oposd.OSDInfo{osdInfo}) - } - - return &osdInfo, nil -} diff --git a/pkg/daemon/ceph/client/deviceclass.go b/pkg/daemon/ceph/client/deviceclass.go index fda23be07..53643f180 100644 --- a/pkg/daemon/ceph/client/deviceclass.go +++ b/pkg/daemon/ceph/client/deviceclass.go @@ -34,7 +34,7 @@ func GetDeviceClasses(context *clusterd.Context, clusterInfo *ClusterInfo) ([]st var deviceclass []string if err := json.Unmarshal(buf, &deviceclass); err != nil { - return nil, errors.Wrap(err, "failed to unmarshal osd crush class list response") + return nil, errors.Wrapf(err, "failed to unmarshal osd crush class list response %q", string(buf)) } return deviceclass, nil diff --git a/pkg/daemon/ceph/osd/daemon.go b/pkg/daemon/ceph/osd/daemon.go index 5514119f0..69d31ccf9 100644 --- a/pkg/daemon/ceph/osd/daemon.go +++ b/pkg/daemon/ceph/osd/daemon.go @@ -32,6 +32,7 @@ import ( "github.com/pkg/errors" "github.com/rook/rook/pkg/clusterd" "github.com/rook/rook/pkg/daemon/ceph/client" + "github.com/rook/rook/pkg/operator/ceph/cluster/osd" oposd "github.com/rook/rook/pkg/operator/ceph/cluster/osd" cephver "github.com/rook/rook/pkg/operator/ceph/version" "github.com/rook/rook/pkg/util/sys" @@ -371,6 +372,7 @@ func getAvailableDevices(context *clusterd.Context, agent *OsdAgent) (*DeviceOsd // This handles the case where the OSD deployment has been removed and the prepare // job kicks in again to re-deploy the OSD. continue + } else if device.Filesystem == "mpath_member" && agent.pvcBacked { logger.Infof("allowing multipath disk %q with filesystem %q", device.Name, device.Filesystem) } else { @@ -602,31 +604,31 @@ func getVolumeGroupName(lvPath string) string { return vgSlice[2] } -// GetOSDInfoByID returns the OSDInfo using the ceph volume list -func GetOSDInfoById(context *clusterd.Context, clusterInfo *client.ClusterInfo, osdID int) (oposd.OSDInfo, error) { +// GetOSDInfoById returns the osdInfo using the ceph volume list +func GetOSDInfoById(context *clusterd.Context, clusterInfo *client.ClusterInfo, osdID int) (*osd.OSDInfo, error) { // LVM mode OSDs osdLVMList, err := GetCephVolumeLVMOSDs(context, clusterInfo, clusterInfo.FSID, "", false, false) if err != nil { - return oposd.OSDInfo{}, errors.Wrap(err, "failed to list lvm osd(s)") + return nil, errors.Wrap(err, "failed to list lvm osd(s)") } for _, osdInfo := range osdLVMList { if osdInfo.ID == osdID { - return osdInfo, nil + return &osdInfo, nil } } // Raw mode OSDs osdRawList, err := GetCephVolumeRawOSDs(context, clusterInfo, clusterInfo.FSID, "", "", "", false, true) if err != nil { - return oposd.OSDInfo{}, errors.Wrap(err, "failed to list raw osd(s)") + return nil, errors.Wrap(err, "failed to list raw osd(s)") } for _, osdInfo := range osdRawList { if osdInfo.ID == osdID { - return osdInfo, nil + return &osdInfo, nil } } - return oposd.OSDInfo{}, fmt.Errorf("failed to get details for OSD %d using ceph-volume list", osdID) + return nil, fmt.Errorf("failed to get details for OSD %d using ceph-volume list", osdID) } diff --git a/pkg/daemon/ceph/osd/remove.go b/pkg/daemon/ceph/osd/remove.go index 1444ebb12..567d84296 100644 --- a/pkg/daemon/ceph/osd/remove.go +++ b/pkg/daemon/ceph/osd/remove.go @@ -18,6 +18,7 @@ package osd import ( "fmt" + "os" "strconv" "time" @@ -28,6 +29,7 @@ import ( "github.com/rook/rook/pkg/clusterd" "github.com/rook/rook/pkg/daemon/ceph/client" "github.com/rook/rook/pkg/operator/ceph/cluster/osd" + oposd "github.com/rook/rook/pkg/operator/ceph/cluster/osd" "github.com/rook/rook/pkg/operator/k8sutil" ) @@ -245,3 +247,51 @@ func archiveCrash(clusterdContext *clusterd.Context, clusterInfo *client.Cluster logger.Errorf("failed to archive the crash %q. %v", crashID, err) } } + +// DestroyOSD fetches the OSD to be replaced based on the ID and then destroys that OSD and zaps the backing device +func DestroyOSD(context *clusterd.Context, clusterInfo *client.ClusterInfo, id int, isPVC, isEncrypted bool) (*osd.OSDReplaceInfo, error) { + var block string + osdInfo, err := GetOSDInfoById(context, clusterInfo, id) + if err != nil { + return nil, errors.Wrapf(err, "failed to get OSD info for OSD.%d", id) + } + + block = osdInfo.BlockPath + + logger.Infof("destroying osd.%d", osdInfo.ID) + destroyOSDArgs := []string{"osd", "destroy", fmt.Sprintf("osd.%d", osdInfo.ID), "--yes-i-really-mean-it"} + _, err = client.NewCephCommand(context, clusterInfo, destroyOSDArgs).Run() + if err != nil { + return nil, errors.Wrapf(err, "failed to destroy osd.%d.", osdInfo.ID) + } + logger.Infof("successfully destroyed osd.%d", osdInfo.ID) + + if isPVC && isEncrypted { + // remove the dm device + pvcName := os.Getenv(oposd.PVCNameEnvVarName) + target := oposd.EncryptionDMName(pvcName, oposd.DmcryptBlockType) + err = removeEncryptedDevice(context, target) + if err != nil { + return nil, errors.Wrapf(err, "failed to remove dm device %q", target) + } + // ceph-volume uses `/dev/mapper/*` for encrypted disks. This is not a block device. So we need to fetch the corresponding + // block device for cleanup using `ceph-volume lvm zap` + blockPath := fmt.Sprintf("/mnt/%s", pvcName) + diskInfo, err := clusterd.PopulateDeviceInfo(blockPath, context.Executor) + if err != nil { + return nil, errors.Wrapf(err, "failed to get device info for %q", blockPath) + } + block = diskInfo.RealPath + } + + logger.Infof("zap OSD.%d path %q", osdInfo.ID, block) + output, err := context.Executor.ExecuteCommandWithCombinedOutput("stdbuf", "-oL", "ceph-volume", "lvm", "zap", block, "--destroy") + if err != nil { + return nil, errors.Wrapf(err, "failed to zap osd.%d path %q. %s.", osdInfo.ID, block, output) + } + + logger.Infof("%s\n", output) + logger.Infof("successfully zapped osd.%d path %q", osdInfo.ID, block) + + return &osd.OSDReplaceInfo{ID: osdInfo.ID, Path: block}, nil +} diff --git a/pkg/daemon/ceph/osd/volume.go b/pkg/daemon/ceph/osd/volume.go index f5afac586..d6b3c0890 100644 --- a/pkg/daemon/ceph/osd/volume.go +++ b/pkg/daemon/ceph/osd/volume.go @@ -1129,9 +1129,11 @@ func GetCephVolumeRawOSDs(context *clusterd.Context, clusterInfo *client.Cluster } // Close encrypted device - err = CloseEncryptedDevice(context, block) - if err != nil { - return nil, errors.Wrapf(err, "failed to close encrypted device %q for osd %d", block, osdID) + if block != "" { + err = CloseEncryptedDevice(context, block) + if err != nil { + return nil, errors.Wrapf(err, "failed to close encrypted device %q for osd %d", block, osdID) + } } // If there is a metadata block diff --git a/pkg/operator/ceph/cluster/osd/health.go b/pkg/operator/ceph/cluster/osd/health.go index 00febe89c..39d23c055 100644 --- a/pkg/operator/ceph/cluster/osd/health.go +++ b/pkg/operator/ceph/cluster/osd/health.go @@ -18,7 +18,6 @@ package osd import ( "fmt" - "reflect" "time" "github.com/pkg/errors" @@ -26,7 +25,6 @@ import ( "github.com/rook/rook/pkg/clusterd" "github.com/rook/rook/pkg/daemon/ceph/client" opcontroller "github.com/rook/rook/pkg/operator/ceph/controller" - "github.com/rook/rook/pkg/operator/ceph/reporting" "github.com/rook/rook/pkg/operator/k8sutil" kerrors "k8s.io/apimachinery/pkg/api/errors" ) @@ -82,7 +80,6 @@ func (m *OSDHealthMonitor) Start(monitoringRoutines map[string]*opcontroller.Clu case <-time.After(*m.interval): logger.Debug("checking osd processes status.") m.checkOSDHealth() - m.updateCephStorageStatus() case <-monitoringRoutines[daemon].InternalCtx.Done(): logger.Infof("stopping monitoring of OSDs in namespace %q", m.clusterInfo.Namespace) @@ -174,66 +171,3 @@ func (m *OSDHealthMonitor) removeOSDDeploymentIfSafeToDestroy(outOSDid int) erro } return nil } - -// updateCephStorageStatus updates the Storage details in the ceph cluster CR -func (m *OSDHealthMonitor) updateCephStorageStatus() { - cephCluster := cephv1.CephCluster{} - cephClusterStorage := cephv1.CephStorage{} - - deviceClasses, err := client.GetDeviceClasses(m.context, m.clusterInfo) - if err != nil { - logger.Errorf("failed to get osd device classes. %v", err) - return - } - - for _, deviceClass := range deviceClasses { - cephClusterStorage.DeviceClasses = append(cephClusterStorage.DeviceClasses, cephv1.DeviceClasses{Name: deviceClass}) - } - - osdStore, err := m.getOSDStoreStatus() - if err != nil { - logger.Errorf("failed to get osd store status. %v", err) - return - } - - cephClusterStorage.OSD = *osdStore - - err = m.context.Client.Get(m.clusterInfo.Context, m.clusterInfo.NamespacedName(), &cephCluster) - if err != nil { - if kerrors.IsNotFound(err) { - logger.Debug("CephCluster resource not found. Ignoring since object must be deleted.") - return - } - logger.Errorf("failed to retrieve ceph cluster %q to update ceph Storage. %v", m.clusterInfo.NamespacedName().Name, err) - return - } - if !reflect.DeepEqual(cephCluster.Status.CephStorage, cephClusterStorage) { - cephCluster.Status.CephStorage = &cephClusterStorage - if err := reporting.UpdateStatus(m.context.Client, &cephCluster); err != nil { - logger.Errorf("failed to update cluster %q Storage. %v", m.clusterInfo.NamespacedName().Name, err) - return - } - } -} - -func (m *OSDHealthMonitor) getOSDStoreStatus() (*cephv1.OSDStatus, error) { - label := fmt.Sprintf("%s=%s", k8sutil.AppAttr, AppName) - osdDeployments, err := k8sutil.GetDeployments(m.clusterInfo.Context, m.context.Clientset, m.clusterInfo.Namespace, label) - if err != nil { - if kerrors.IsNotFound(err) { - return nil, nil - } - return nil, errors.Wrap(err, "failed to get osd deployments") - } - - storeType := map[string]int{} - for i := range osdDeployments.Items { - if osdStore, ok := osdDeployments.Items[i].Labels[osdStore]; ok { - storeType[osdStore]++ - } - } - - return &cephv1.OSDStatus{ - StoreType: storeType, - }, nil -} diff --git a/pkg/operator/ceph/cluster/osd/health_test.go b/pkg/operator/ceph/cluster/osd/health_test.go index fe0189107..7fcad63f2 100644 --- a/pkg/operator/ceph/cluster/osd/health_test.go +++ b/pkg/operator/ceph/cluster/osd/health_test.go @@ -23,7 +23,6 @@ import ( "time" cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1" - "github.com/rook/rook/pkg/client/clientset/versioned/scheme" "github.com/rook/rook/pkg/clusterd" "github.com/rook/rook/pkg/daemon/ceph/client" opcontroller "github.com/rook/rook/pkg/operator/ceph/controller" @@ -33,8 +32,6 @@ import ( "github.com/stretchr/testify/assert" apps "k8s.io/api/apps/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - "k8s.io/apimachinery/pkg/runtime" - "sigs.k8s.io/controller-runtime/pkg/client/fake" ) func TestOSDHealthCheck(t *testing.T) { @@ -140,104 +137,3 @@ func TestNewOSDHealthMonitor(t *testing.T) { }) } } - -func TestUpdateCephStorageStatus(t *testing.T) { - ctx := context.TODO() - clusterInfo := client.AdminTestClusterInfo("fake") - executor := &exectest.MockExecutor{ - MockExecuteCommandWithOutput: func(command string, args ...string) (string, error) { - logger.Infof("ExecuteCommandWithOutputFile: %s %v", command, args) - if args[1] == "crush" && args[2] == "class" && args[3] == "ls" { - // Mock executor for OSD crush class list command, returning ssd as available device class - return `["ssd"]`, nil - } - return "", nil - }, - } - - cephCluster := &cephv1.CephCluster{ - ObjectMeta: metav1.ObjectMeta{ - Name: "testing", - Namespace: "fake", - }, - } - // Objects to track in the fake client. - object := []runtime.Object{ - cephCluster, - } - s := scheme.Scheme - // Create a fake client to mock API calls. - c := fake.NewClientBuilder().WithScheme(s).WithRuntimeObjects(object...).Build() - - context := &clusterd.Context{ - Executor: executor, - Client: c, - Clientset: testexec.New(t, 2), - } - - // Initializing an OSD monitoring - osdHealthMonitor := NewOSDHealthMonitor(context, clusterInfo, true, cephv1.CephClusterHealthCheckSpec{}) - - t.Run("verify ssd device class added to storage status", func(t *testing.T) { - osdHealthMonitor.updateCephStorageStatus() - err := context.Client.Get(clusterInfo.Context, clusterInfo.NamespacedName(), cephCluster) - assert.NoError(t, err) - assert.Equal(t, 1, len(cephCluster.Status.CephStorage.DeviceClasses)) - assert.Equal(t, "ssd", cephCluster.Status.CephStorage.DeviceClasses[0].Name) - }) - - t.Run("verify bluestore OSD count in storage status", func(t *testing.T) { - labels := map[string]string{ - k8sutil.AppAttr: AppName, - k8sutil.ClusterAttr: clusterInfo.Namespace, - OsdIdLabelKey: "0", - osdStore: "bluestore", - } - - deployment := &apps.Deployment{ - ObjectMeta: metav1.ObjectMeta{ - Name: "osd0", - Namespace: clusterInfo.Namespace, - Labels: labels, - }, - } - if _, err := context.Clientset.AppsV1().Deployments(clusterInfo.Namespace).Create(ctx, deployment, metav1.CreateOptions{}); err != nil { - logger.Errorf("Error creating fake deployment: %v", err) - } - osdHealthMonitor.updateCephStorageStatus() - err := context.Client.Get(clusterInfo.Context, clusterInfo.NamespacedName(), cephCluster) - assert.NoError(t, err) - assert.Equal(t, 1, len(cephCluster.Status.CephStorage.DeviceClasses)) - assert.Equal(t, "ssd", cephCluster.Status.CephStorage.DeviceClasses[0].Name) - assert.Equal(t, 1, cephCluster.Status.CephStorage.OSD.StoreType["bluestore"]) - assert.Equal(t, 0, cephCluster.Status.CephStorage.OSD.StoreType["bluestore-rdr"]) - - }) - - t.Run("verify bluestoreRDR OSD count in storage status", func(t *testing.T) { - labels := map[string]string{ - k8sutil.AppAttr: AppName, - k8sutil.ClusterAttr: clusterInfo.Namespace, - OsdIdLabelKey: "1", - osdStore: "bluestore-rdr", - } - - deployment := &apps.Deployment{ - ObjectMeta: metav1.ObjectMeta{ - Name: "osd1", - Namespace: clusterInfo.Namespace, - Labels: labels, - }, - } - if _, err := context.Clientset.AppsV1().Deployments(clusterInfo.Namespace).Create(ctx, deployment, metav1.CreateOptions{}); err != nil { - logger.Errorf("Error creating fake deployment: %v", err) - } - osdHealthMonitor.updateCephStorageStatus() - err := context.Client.Get(clusterInfo.Context, clusterInfo.NamespacedName(), cephCluster) - assert.NoError(t, err) - assert.Equal(t, 1, len(cephCluster.Status.CephStorage.DeviceClasses)) - assert.Equal(t, "ssd", cephCluster.Status.CephStorage.DeviceClasses[0].Name) - assert.Equal(t, 1, cephCluster.Status.CephStorage.OSD.StoreType["bluestore-rdr"]) - assert.Equal(t, 1, cephCluster.Status.CephStorage.OSD.StoreType["bluestore"]) - }) -} diff --git a/pkg/operator/ceph/cluster/osd/integration_test.go b/pkg/operator/ceph/cluster/osd/integration_test.go index 842815424..7918d239d 100644 --- a/pkg/operator/ceph/cluster/osd/integration_test.go +++ b/pkg/operator/ceph/cluster/osd/integration_test.go @@ -28,6 +28,7 @@ import ( "github.com/coreos/pkg/capnslog" "github.com/pkg/errors" cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1" + "github.com/rook/rook/pkg/client/clientset/versioned/scheme" "github.com/rook/rook/pkg/clusterd" cephclient "github.com/rook/rook/pkg/daemon/ceph/client" cephclientfake "github.com/rook/rook/pkg/daemon/ceph/client/fake" @@ -45,6 +46,7 @@ import ( "k8s.io/client-go/kubernetes" "k8s.io/client-go/kubernetes/fake" k8stesting "k8s.io/client-go/testing" + clientfake "sigs.k8s.io/controller-runtime/pkg/client/fake" ) var ( @@ -192,12 +194,28 @@ func testOSDIntegration(t *testing.T) { clusterInfo.Context = ctx executor := osdIntegrationTestExecutor(t, clientset, namespace) + cephCluster := &cephv1.CephCluster{ + ObjectMeta: metav1.ObjectMeta{ + Name: clusterName, + Namespace: namespace, + }, + Spec: cephv1.ClusterSpec{}, + } + // Objects to track in the fake client. + object := []runtime.Object{ + cephCluster, + } + s := scheme.Scheme + // Create a fake client to mock API calls. + client := clientfake.NewClientBuilder().WithScheme(s).WithRuntimeObjects(object...).Build() + rootCtx := &clusterd.Context{ Clientset: clientset, ConfigDir: "/var/lib/rook", Executor: executor, + Client: client, } - spec := cephv1.ClusterSpec{ + cephCluster.Spec = cephv1.ClusterSpec{ CephVersion: cephv1.CephVersionSpec{ Image: "quay.io/ceph/ceph:v16.2.0", }, @@ -219,13 +237,13 @@ func testOSDIntegration(t *testing.T) { } osdsPerNode := 2 // vda and vdb - c := New(rootCtx, clusterInfo, spec, "myversion") + c := New(rootCtx, clusterInfo, cephCluster.Spec, "myversion") var startErr error var done bool runReconcile := func(ctx context.Context) { // reset environment - c = New(rootCtx, clusterInfo, spec, "myversion") + c = New(rootCtx, clusterInfo, cephCluster.Spec, "myversion") clusterInfo.Context = ctx statusMapWatcher.Reset() @@ -280,10 +298,10 @@ func testOSDIntegration(t *testing.T) { }) t.Run("increase number of OSDs", func(t *testing.T) { - spec.Storage.Selection.DeviceFilter = "/dev/vd[abc]" // 3 more (1 more per node) - spec.Storage.StorageClassDeviceSets[0].Count = 8 // 2 more portable - spec.Storage.StorageClassDeviceSets[1].Count = 6 // 3 more (1 more per node) - osdsPerNode = 3 // vda, vdb, vdc + cephCluster.Spec.Storage.Selection.DeviceFilter = "/dev/vd[abc]" // 3 more (1 more per node) + cephCluster.Spec.Storage.StorageClassDeviceSets[0].Count = 8 // 2 more portable + cephCluster.Spec.Storage.StorageClassDeviceSets[1].Count = 6 // 3 more (1 more per node) + osdsPerNode = 3 // vda, vdb, vdc go runReconcile(contextCancel) @@ -301,9 +319,9 @@ func testOSDIntegration(t *testing.T) { }) t.Run("mixed create and update, cancel reconcile, and continue reconcile", func(t *testing.T) { - spec.Storage.Selection.DeviceFilter = "/dev/vd[abcd]" // 3 more (1 more per node) - spec.Storage.StorageClassDeviceSets[0].Count = 10 // 2 more portable - osdsPerNode = 4 // vd[a-d] + cephCluster.Spec.Storage.Selection.DeviceFilter = "/dev/vd[abcd]" // 3 more (1 more per node) + cephCluster.Spec.Storage.StorageClassDeviceSets[0].Count = 10 // 2 more portable + osdsPerNode = 4 // vd[a-d] go runReconcile(contextCancel) cms := waitForNumConfigMaps(clientset, namespace, 5) // 3 nodes + 2 new PVCs @@ -352,8 +370,8 @@ func testOSDIntegration(t *testing.T) { }) t.Run("failures reported in status configmaps", func(t *testing.T) { - spec.Storage.Selection.DeviceFilter = "/dev/vd[abcde]" // 3 more (1 more per node) - osdsPerNode = 5 // vd[a-e] + cephCluster.Spec.Storage.Selection.DeviceFilter = "/dev/vd[abcde]" // 3 more (1 more per node) + osdsPerNode = 5 // vd[a-e] go runReconcile(contextCancel) cms := waitForNumConfigMaps(clientset, namespace, 3) // 3 nodes @@ -416,8 +434,8 @@ func testOSDIntegration(t *testing.T) { }) t.Run("failures during deployment creation", func(t *testing.T) { - spec.Storage.Selection.DeviceFilter = "/dev/vd[abcdef]" // 3 more (1 more per node) - osdsPerNode = 6 // vd[a-f] + cephCluster.Spec.Storage.Selection.DeviceFilter = "/dev/vd[abcdef]" // 3 more (1 more per node) + osdsPerNode = 6 // vd[a-f] failCreatingDeployments = []string{"osd-31", "osd-33"} go runReconcile(contextCancel) @@ -454,7 +472,7 @@ func testOSDIntegration(t *testing.T) { Portable: true, VolumeClaimTemplates: []corev1.PersistentVolumeClaim{}, } - spec.Storage.StorageClassDeviceSets = append(spec.Storage.StorageClassDeviceSets, newSCDS) + cephCluster.Spec.Storage.StorageClassDeviceSets = append(cephCluster.Spec.Storage.StorageClassDeviceSets, newSCDS) go runReconcile(contextCancel) cms := waitForNumConfigMaps(clientset, namespace, 3) // 3 nodes @@ -469,7 +487,7 @@ func testOSDIntegration(t *testing.T) { assert.Len(t, deploymentsCreated, 0) assert.Len(t, deploymentsUpdated, 34) - spec.Storage.StorageClassDeviceSets[2].VolumeClaimTemplates = []corev1.PersistentVolumeClaim{ + cephCluster.Spec.Storage.StorageClassDeviceSets[2].VolumeClaimTemplates = []corev1.PersistentVolumeClaim{ newDummyPVC("data", namespace, "100Gi", "ec2"), newDummyPVC("metadata", namespace, "10Gi", "uncle-rogers-secret-stuff"), } @@ -562,6 +580,10 @@ func osdIntegrationTestExecutor(t *testing.T, clientset *fake.Clientset, namespa if args[2] == "get-device-class" { return cephclientfake.OSDDeviceClassOutput(args[3]), nil } + if args[2] == "class" && args[3] == "ls" { + // Mock executor for OSD crush class list command, returning ssd as available device class + return `["ssd"]`, nil + } } } if args[0] == "versions" { diff --git a/pkg/operator/ceph/cluster/osd/osd.go b/pkg/operator/ceph/cluster/osd/osd.go index 7df6b0169..5acd9ce67 100644 --- a/pkg/operator/ceph/cluster/osd/osd.go +++ b/pkg/operator/ceph/cluster/osd/osd.go @@ -21,6 +21,7 @@ import ( "bufio" "context" "fmt" + "reflect" "sort" "strconv" "strings" @@ -28,10 +29,12 @@ import ( 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" cephclient "github.com/rook/rook/pkg/daemon/ceph/client" osdconfig "github.com/rook/rook/pkg/operator/ceph/cluster/osd/config" "github.com/rook/rook/pkg/operator/ceph/cluster/osd/topology" "github.com/rook/rook/pkg/operator/ceph/controller" + "github.com/rook/rook/pkg/operator/ceph/reporting" cephver "github.com/rook/rook/pkg/operator/ceph/version" "github.com/rook/rook/pkg/operator/k8sutil" "github.com/rook/rook/pkg/util" @@ -117,6 +120,7 @@ type OSDInfo struct { Encrypted bool `json:"encrypted"` ExportService bool `json:"exportService"` NodeName string `json:"nodeName"` + PVCName string `json:"pvcName"` } // OrchestrationStatus represents the status of an OSD orchestration @@ -260,6 +264,11 @@ func (c *Cluster) Start() error { return errors.Wrapf(err, "failed to reconcile key rotation cron jobs") } + err = c.updateCephStorageStatus() + if err != nil { + return errors.Wrapf(err, "failed to update ceph storage status") + } + if c.replaceOSD != nil { delOpts := &k8sutil.DeleteOptions{MustDelete: true, WaitOptions: k8sutil.WaitOptions{Wait: true}} err := k8sutil.DeleteConfigMap(c.clusterInfo.Context, c.context.Clientset, OSDReplaceConfigName, namespace, delOpts) @@ -611,6 +620,10 @@ func (c *Cluster) getOSDInfo(d *appsv1.Deployment) (OSDInfo, error) { osd.Store = d.Labels[osdStore] + if isPVC { + osd.PVCName = d.Labels[OSDOverPVCLabelKey] + } + return osd, nil } @@ -871,3 +884,63 @@ func (c *Cluster) waitForHealthyPGs() (bool, error) { return true, nil } + +func (c *Cluster) updateCephStorageStatus() error { + cephCluster := cephv1.CephCluster{} + cephClusterStorage := cephv1.CephStorage{} + + deviceClasses, err := client.GetDeviceClasses(c.context, c.clusterInfo) + if err != nil { + return errors.Wrap(err, "failed to get osd device classes") + } + + for _, deviceClass := range deviceClasses { + cephClusterStorage.DeviceClasses = append(cephClusterStorage.DeviceClasses, cephv1.DeviceClasses{Name: deviceClass}) + } + + osdStore, err := c.getOSDStoreStatus() + if err != nil { + return errors.Wrapf(err, "failed to get osd store status") + } + + cephClusterStorage.OSD = *osdStore + + err = c.context.Client.Get(c.clusterInfo.Context, c.clusterInfo.NamespacedName(), &cephCluster) + if err != nil { + if kerrors.IsNotFound(err) { + logger.Debug("CephCluster resource not found. Ignoring since object must be deleted.") + return nil + } + return errors.Wrapf(err, "failed to retrieve ceph cluster %q to update ceph Storage", c.clusterInfo.NamespacedName().Name) + } + if !reflect.DeepEqual(cephCluster.Status.CephStorage, cephClusterStorage) { + cephCluster.Status.CephStorage = &cephClusterStorage + if err := reporting.UpdateStatus(c.context.Client, &cephCluster); err != nil { + return errors.Wrapf(err, "failed to update cluster %q Storage.", c.clusterInfo.NamespacedName().Name) + } + } + + return nil +} + +func (c *Cluster) getOSDStoreStatus() (*cephv1.OSDStatus, error) { + label := fmt.Sprintf("%s=%s", k8sutil.AppAttr, AppName) + osdDeployments, err := k8sutil.GetDeployments(c.clusterInfo.Context, c.context.Clientset, c.clusterInfo.Namespace, label) + if err != nil { + if kerrors.IsNotFound(err) { + return nil, nil + } + return nil, errors.Wrap(err, "failed to get osd deployments") + } + + storeType := map[string]int{} + for i := range osdDeployments.Items { + if osdStore, ok := osdDeployments.Items[i].Labels[osdStore]; ok { + storeType[osdStore]++ + } + } + + return &cephv1.OSDStatus{ + StoreType: storeType, + }, nil +} diff --git a/pkg/operator/ceph/cluster/osd/osd_test.go b/pkg/operator/ceph/cluster/osd/osd_test.go index d9d84a0ef..708359ed0 100644 --- a/pkg/operator/ceph/cluster/osd/osd_test.go +++ b/pkg/operator/ceph/cluster/osd/osd_test.go @@ -25,7 +25,9 @@ import ( "github.com/pkg/errors" cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1" fakeclient "github.com/rook/rook/pkg/client/clientset/versioned/fake" + "github.com/rook/rook/pkg/client/clientset/versioned/scheme" "github.com/rook/rook/pkg/clusterd" + "github.com/rook/rook/pkg/daemon/ceph/client" cephclient "github.com/rook/rook/pkg/daemon/ceph/client" cephclientfake "github.com/rook/rook/pkg/daemon/ceph/client/fake" discoverDaemon "github.com/rook/rook/pkg/daemon/discover" @@ -34,6 +36,7 @@ import ( cephver "github.com/rook/rook/pkg/operator/ceph/version" "github.com/rook/rook/pkg/operator/k8sutil" "github.com/rook/rook/pkg/operator/test" + testexec "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" @@ -45,6 +48,7 @@ import ( "k8s.io/apimachinery/pkg/watch" "k8s.io/client-go/kubernetes/fake" k8stesting "k8s.io/client-go/testing" + clientfake "sigs.k8s.io/controller-runtime/pkg/client/fake" ) const ( @@ -70,13 +74,40 @@ func TestOSDProperties(t *testing.T) { } func TestStart(t *testing.T) { + namespace := "ns" clientset := fake.NewSimpleClientset() + executor := &exectest.MockExecutor{ + MockExecuteCommandWithOutput: func(command string, args ...string) (string, error) { + logger.Infof("ExecuteCommandWithOutputFile: %s %v", command, args) + if args[1] == "crush" && args[2] == "class" && args[3] == "ls" { + // Mock executor for OSD crush class list command, returning ssd as available device class + return `["ssd"]`, nil + } + return "", nil + }, + } + + cephCluster := &cephv1.CephCluster{ + ObjectMeta: metav1.ObjectMeta{ + Name: "testing", + Namespace: namespace, + }, + Spec: cephv1.ClusterSpec{}, + } + // Objects to track in the fake client. + object := []runtime.Object{ + cephCluster, + } + s := scheme.Scheme + // Create a fake client to mock API calls. + client := clientfake.NewClientBuilder().WithScheme(s).WithRuntimeObjects(object...).Build() clusterInfo := &cephclient.ClusterInfo{ - Namespace: "ns", + Namespace: namespace, CephVersion: cephver.Quincy, Context: context.TODO(), } - context := &clusterd.Context{Clientset: clientset, ConfigDir: "/var/lib/rook", Executor: &exectest.MockExecutor{}} + clusterInfo.SetName("rook-ceph-test") + context := &clusterd.Context{Clientset: clientset, Client: client, ConfigDir: "/var/lib/rook", Executor: executor} spec := cephv1.ClusterSpec{} c := New(context, clusterInfo, spec, "myversion") @@ -163,6 +194,11 @@ func TestAddRemoveNode(t *testing.T) { generateKey := "expected key" executor := &exectest.MockExecutor{ MockExecuteCommandWithOutput: func(command string, args ...string) (string, error) { + logger.Infof("Command: %s %v", command, args) + if args[1] == "crush" && args[2] == "class" && args[3] == "ls" { + // Mock executor for OSD crush class list command, returning ssd as available device class + return `["ssd"]`, nil + } return "{\"key\": \"" + generateKey + "\"}", nil }, } @@ -173,20 +209,37 @@ func TestAddRemoveNode(t *testing.T) { Executor: executor, RookClientset: fakeclient.NewSimpleClientset(), } - spec := cephv1.ClusterSpec{ - DataDirHostPath: context.ConfigDir, - Storage: cephv1.StorageScopeSpec{ - Nodes: []cephv1.Node{ - { - Name: nodeName, - Selection: cephv1.Selection{ - Devices: []cephv1.Device{{Name: "sdx"}}, + + cephCluster := &cephv1.CephCluster{ + ObjectMeta: metav1.ObjectMeta{ + Name: "testing", + Namespace: namespace, + }, + Spec: cephv1.ClusterSpec{ + DataDirHostPath: context.ConfigDir, + Storage: cephv1.StorageScopeSpec{ + Nodes: []cephv1.Node{ + { + Name: nodeName, + Selection: cephv1.Selection{ + Devices: []cephv1.Device{{Name: "sdx"}}, + }, }, }, }, }, } - c := New(context, clusterInfo, spec, "myversion") + + // Objects to track in the fake client. + object := []runtime.Object{ + cephCluster, + } + s := scheme.Scheme + // Create a fake client to mock API calls. + client := clientfake.NewClientBuilder().WithScheme(s).WithRuntimeObjects(object...).Build() + context.Client = client + + c := New(context, clusterInfo, cephCluster.Spec, "myversion") // kick off the start of the orchestration in a goroutine var startErr error @@ -246,6 +299,9 @@ func TestAddRemoveNode(t *testing.T) { if args[2] == "get-device-class" { return cephclientfake.OSDDeviceClassOutput(args[3]), nil } + if args[2] == "class" { + return `["ssd"]`, nil + } } if args[1] == "out" { return "", nil @@ -276,8 +332,8 @@ func TestAddRemoveNode(t *testing.T) { } // modify the storage spec to remove the node from the cluster - spec.Storage.Nodes = []cephv1.Node{} - c = New(context, clusterInfo, spec, "myversion") + cephCluster.Spec.Storage.Nodes = []cephv1.Node{} + c = New(context, clusterInfo, cephCluster.Spec, "myversion") // reset the orchestration status watcher statusMapWatcher = watch.NewFake() @@ -767,7 +823,7 @@ func TestReplaceOSDForNewStore(t *testing.T) { fmt.Printf("%+v", c.replaceOSD) assert.NotNil(t, c.replaceOSD) assert.Equal(t, 2, c.replaceOSD.ID) - assert.Equal(t, "/some/path", c.replaceOSD.Path) + assert.Equal(t, "pvc1", c.replaceOSD.Path) // assert that OSD.2 deployment got deleted _, err = clientset.AppsV1().Deployments(clusterInfo.Namespace).Get(clusterInfo.Context, deploymentName(2), metav1.GetOptions{}) @@ -781,7 +837,7 @@ func TestReplaceOSDForNewStore(t *testing.T) { err = json.Unmarshal([]byte(actualCM.Data[OSDReplaceConfigKey]), &expectedOSDInfo) assert.NoError(t, err) assert.Equal(t, 2, expectedOSDInfo.ID) - assert.Equal(t, "/some/path", c.replaceOSD.Path) + assert.Equal(t, "pvc1", c.replaceOSD.Path) // delete configmap err = k8sutil.DeleteConfigMap(clusterInfo.Context, clientset, OSDReplaceConfigName, clusterInfo.Namespace, &k8sutil.DeleteOptions{}) @@ -813,3 +869,108 @@ func TestReplaceOSDForNewStore(t *testing.T) { assert.Nil(t, c.replaceOSD) }) } + +func TestUpdateCephStorageStatus(t *testing.T) { + ctx := context.TODO() + clusterInfo := client.AdminTestClusterInfo("fake") + executor := &exectest.MockExecutor{ + MockExecuteCommandWithOutput: func(command string, args ...string) (string, error) { + logger.Infof("ExecuteCommandWithOutputFile: %s %v", command, args) + if args[1] == "crush" && args[2] == "class" && args[3] == "ls" { + // Mock executor for OSD crush class list command, returning ssd as available device class + return `["ssd"]`, nil + } + return "", nil + }, + } + + cephCluster := &cephv1.CephCluster{ + ObjectMeta: metav1.ObjectMeta{ + Name: "testing", + Namespace: "fake", + }, + Spec: cephv1.ClusterSpec{}, + } + // Objects to track in the fake client. + object := []runtime.Object{ + cephCluster, + } + s := scheme.Scheme + // Create a fake client to mock API calls. + client := clientfake.NewClientBuilder().WithScheme(s).WithRuntimeObjects(object...).Build() + + context := &clusterd.Context{ + Executor: executor, + Client: client, + Clientset: testexec.New(t, 2), + } + + // Initializing an OSD monitoring + c := New(context, clusterInfo, cephCluster.Spec, "myversion") + + t.Run("verify ssd device class added to storage status", func(t *testing.T) { + err := c.updateCephStorageStatus() + assert.NoError(t, err) + err = context.Client.Get(clusterInfo.Context, clusterInfo.NamespacedName(), cephCluster) + assert.NoError(t, err) + assert.Equal(t, 1, len(cephCluster.Status.CephStorage.DeviceClasses)) + assert.Equal(t, "ssd", cephCluster.Status.CephStorage.DeviceClasses[0].Name) + }) + + t.Run("verify bluestore OSD count in storage status", func(t *testing.T) { + labels := map[string]string{ + k8sutil.AppAttr: AppName, + k8sutil.ClusterAttr: clusterInfo.Namespace, + OsdIdLabelKey: "0", + osdStore: "bluestore", + } + + deployment := &apps.Deployment{ + ObjectMeta: metav1.ObjectMeta{ + Name: "osd0", + Namespace: clusterInfo.Namespace, + Labels: labels, + }, + } + if _, err := context.Clientset.AppsV1().Deployments(clusterInfo.Namespace).Create(ctx, deployment, metav1.CreateOptions{}); err != nil { + logger.Errorf("Error creating fake deployment: %v", err) + } + err := c.updateCephStorageStatus() + assert.NoError(t, err) + err = context.Client.Get(clusterInfo.Context, clusterInfo.NamespacedName(), cephCluster) + assert.NoError(t, err) + assert.Equal(t, 1, len(cephCluster.Status.CephStorage.DeviceClasses)) + assert.Equal(t, "ssd", cephCluster.Status.CephStorage.DeviceClasses[0].Name) + assert.Equal(t, 1, cephCluster.Status.CephStorage.OSD.StoreType["bluestore"]) + assert.Equal(t, 0, cephCluster.Status.CephStorage.OSD.StoreType["bluestore-rdr"]) + + }) + + t.Run("verify bluestoreRDR OSD count in storage status", func(t *testing.T) { + labels := map[string]string{ + k8sutil.AppAttr: AppName, + k8sutil.ClusterAttr: clusterInfo.Namespace, + OsdIdLabelKey: "1", + osdStore: "bluestore-rdr", + } + + deployment := &apps.Deployment{ + ObjectMeta: metav1.ObjectMeta{ + Name: "osd1", + Namespace: clusterInfo.Namespace, + Labels: labels, + }, + } + if _, err := context.Clientset.AppsV1().Deployments(clusterInfo.Namespace).Create(ctx, deployment, metav1.CreateOptions{}); err != nil { + logger.Errorf("Error creating fake deployment: %v", err) + } + err := c.updateCephStorageStatus() + assert.NoError(t, err) + err = context.Client.Get(clusterInfo.Context, clusterInfo.NamespacedName(), cephCluster) + assert.NoError(t, err) + assert.Equal(t, 1, len(cephCluster.Status.CephStorage.DeviceClasses)) + assert.Equal(t, "ssd", cephCluster.Status.CephStorage.DeviceClasses[0].Name) + assert.Equal(t, 1, cephCluster.Status.CephStorage.OSD.StoreType["bluestore-rdr"]) + assert.Equal(t, 1, cephCluster.Status.CephStorage.OSD.StoreType["bluestore"]) + }) +} diff --git a/pkg/operator/ceph/cluster/osd/replace.go b/pkg/operator/ceph/cluster/osd/replace.go index 9fe6bc913..e7fe3738e 100644 --- a/pkg/operator/ceph/cluster/osd/replace.go +++ b/pkg/operator/ceph/cluster/osd/replace.go @@ -134,7 +134,13 @@ func (c *Cluster) getOSDWithNonMatchingStore() ([]OSDReplaceInfo, error) { if err != nil { return nil, errors.Wrapf(err, "failed to details about the OSD %q", deployments.Items[i].Name) } - osdReplaceList = append(osdReplaceList, OSDReplaceInfo{ID: osdInfo.ID, Path: osdInfo.BlockPath, Node: osdInfo.NodeName}) + var path string + if osdInfo.PVCName != "" { + path = osdInfo.PVCName + } else { + path = osdInfo.BlockPath + } + osdReplaceList = append(osdReplaceList, OSDReplaceInfo{ID: osdInfo.ID, Path: path, Node: osdInfo.NodeName}) } } } diff --git a/pkg/operator/ceph/cluster/osd/replace_test.go b/pkg/operator/ceph/cluster/osd/replace_test.go index e47e8e12a..b2daf5cef 100644 --- a/pkg/operator/ceph/cluster/osd/replace_test.go +++ b/pkg/operator/ceph/cluster/osd/replace_test.go @@ -160,7 +160,7 @@ func TestGetOSDWithNonMatchingStoreOnPVCs(t *testing.T) { assert.NoError(t, err) assert.Equal(t, 1, len(osdList)) assert.Equal(t, 0, osdList[0].ID) - assert.Equal(t, "/some/path", osdList[0].Path) + assert.Equal(t, "pvc0", osdList[0].Path) }) }