detect if ceph-volume is supported and allow legacy osds to be started

Signed-off-by: travisn <tnielsen@redhat.com>

pkg/daemon/ceph/osd/agent.go

@@ -143,15 +143,18 @@ func (a *OsdAgent) removeDirs(context *clusterd.Context, removedDirs map[string]
 }
 
 func (a *OsdAgent) configureAllDevices(context *clusterd.Context, devices *DeviceOsdMapping) ([]oposd.OSDInfo, error) {
-	var osds []oposd.OSDInfo
-	if devices == nil || len(devices.Entries) == 0 {
+
+	// prepare the OSDs with ceph-volume
+	osds, configured, err := a.configureDevices(context, devices)
+	if err != nil {
+		logger.Errorf("failed to configure devices with ceph-volume. %+v", err)
+	}
+
+	if devices == nil || len(devices.Entries) == 0 || configured {
 		return osds, nil
 	}
 
-	useCephVolume := true
-	if useCephVolume {
-		return a.configureDevices(context, devices)
-	}
+	// get info about legacy OSDs prepared without ceph-volume
 
 	// compute an OSD layout scheme that will optimize performance
 	scheme, err := a.getPartitionPerfScheme(context, devices)

pkg/daemon/ceph/osd/agent_test.go

@@ -229,7 +229,7 @@ func testOSDAgentWithDevicesHelper(t *testing.T, storeConfig config.StoreConfig)
 		"sdx": {Data: -1},
 		"sdy": {Data: -1},
 	}}
-	_, err = agent.configureDevices(context, devices)
+	_, err = agent.configureAllDevices(context, devices)
 	assert.Nil(t, err)
 
 	assert.Equal(t, int32(0), agent.configCounter)

pkg/daemon/ceph/osd/daemon.go

@@ -197,10 +197,10 @@ func getAvailableDevices(context *clusterd.Context, desiredDevices string, metad
 
 		if metadataDevice != "" && metadataDevice == device.Name {
 			// current device is desired as the metadata device
-			available.Entries[device.Name] = &DeviceOsdIDEntry{Data: unassignedOSDID, Metadata: []int{}}
+			available.Entries[device.Name] = &DeviceOsdIDEntry{Data: unassignedOSDID, Metadata: []int{}, LegacyPartitionsFound: ownPartitions}
 		} else if desiredDevices == "all" {
 			// user has specified all devices, use the current one for data
-			available.Entries[device.Name] = &DeviceOsdIDEntry{Data: unassignedOSDID}
+			available.Entries[device.Name] = &DeviceOsdIDEntry{Data: unassignedOSDID, LegacyPartitionsFound: ownPartitions}
 		} else if desiredDevices != "" {
 			var matched bool
 			var err error

pkg/daemon/ceph/osd/device.go

@@ -76,8 +76,9 @@ type DeviceOsdMapping struct {
 }
 
 type DeviceOsdIDEntry struct {
-	Data     int   // OSD ID that has data stored here
-	Metadata []int // OSD IDs (multiple) that have metadata stored here
+	Data                  int   // OSD ID that has data stored here
+	Metadata              []int // OSD IDs (multiple) that have metadata stored here
+	LegacyPartitionsFound bool  // Whether legacy rook partitions were found
 }
 
 type devicePartInfo struct {

pkg/daemon/ceph/osd/volume.go

@@ -21,47 +21,81 @@ import (
 	"fmt"
 	"path"
 	"strconv"
+	"syscall"
 
 	"github.com/rook/rook/pkg/clusterd"
 	oposd "github.com/rook/rook/pkg/operator/ceph/cluster/osd"
+	"github.com/rook/rook/pkg/util/exec"
 )
 
-func (a *OsdAgent) configureDevices(context *clusterd.Context, devices *DeviceOsdMapping) ([]oposd.OSDInfo, error) {
+func (a *OsdAgent) configureDevices(context *clusterd.Context, devices *DeviceOsdMapping) ([]oposd.OSDInfo, bool, error) {
+	var osds []oposd.OSDInfo
 
-	err := createOSDBootstrapKeyring(context, a.cluster.Name)
+	useCephVolume, err := getCephVolumeSupported(context)
 	if err != nil {
-		return nil, fmt.Errorf("failed to generate osd keyring. %+v", err)
+		return nil, false, fmt.Errorf("failed to detect if ceph-volume is available. %+v", err)
+	}
+	if !useCephVolume {
+		return osds, false, nil
 	}
 
-	var osds []oposd.OSDInfo
 	if devices == nil || len(devices.Entries) == 0 {
-		return osds, nil
+		logger.Infof("no new devices to configure. returning devices already configured with ceph-volume.")
+		osds, err = getCephVolumeOSDs(context, a.cluster.Name)
+		if err != nil {
+			logger.Infof("failed to get devices already provisioned by ceph-volume. %+v", err)
+		}
+		return osds, true, nil
+	}
+
+	err = createOSDBootstrapKeyring(context, a.cluster.Name)
+	if err != nil {
+		return nil, true, fmt.Errorf("failed to generate osd keyring. %+v", err)
 	}
 
 	volumeArgs := []string{"lvm", "batch", "--prepare", "--bluestore", "--yes"}
 	configured := 0
 	for name, device := range devices.Entries {
+		if device.LegacyPartitionsFound {
+			logger.Infof("skipping device %s configured with legacy rook osd", name)
+			continue
+		}
 		if device.Data == -1 {
 			logger.Infof("configuring new device %s", name)
 			volumeArgs = append(volumeArgs, path.Join("/dev", name))
 			configured++
 		} else {
-			logger.Infof("skipping existing device %s: %d", name, device.Data)
+			logger.Infof("skipping device %s with osd %d already configured", name, device.Data)
 		}
 	}
 
-	if configured == 0 {
-		logger.Infof("no osd devices attempted configuration on this node")
-		return osds, nil
+	if configured > 0 {
+		err = context.Executor.ExecuteCommand(false, "", "ceph-volume", volumeArgs...)
+		if err != nil {
+			return osds, true, fmt.Errorf("failed ceph-volume. %+v", err)
+		}
 	}
 
-	err = context.Executor.ExecuteCommand(false, "", "ceph-volume", volumeArgs...)
+	osds, err = getCephVolumeOSDs(context, a.cluster.Name)
+	return osds, true, err
+}
+
+func getCephVolumeSupported(context *clusterd.Context) (bool, error) {
+	_, err := context.Executor.ExecuteCommandWithOutput(false, "", "ceph-volume", "lvm", "batch", "--prepare")
 	if err != nil {
-		logger.Errorf("failed ceph-volume. %+v", err)
-		return osds, nil
+		if cmdErr, ok := err.(*exec.CommandError); ok {
+			exitStatus := cmdErr.ExitStatus()
+			if exitStatus == int(syscall.ENOENT) {
+				logger.Infof("supported version of ceph-volume not available")
+				return false, nil
+			}
+			logger.Warningf("unknown return code from ceph-volume when checking for compatibility: %d", exitStatus)
+		}
+		logger.Warningf("unknown ceph-volume failure. %+v", err)
+		return false, nil
 	}
 
-	return getCephVolumeOSDs(context, a.cluster.Name)
+	return true, nil
 }
 
 func getCephVolumeOSDs(context *clusterd.Context, clusterName string) ([]oposd.OSDInfo, error) {
@@ -89,12 +123,13 @@ func getCephVolumeOSDs(context *clusterd.Context, clusterName string) ([]oposd.O
 
 		configDir := "/var/lib/rook/osd" + name
 		osd := oposd.OSDInfo{
-			ID:          id,
-			DataPath:    configDir,
-			Config:      fmt.Sprintf("%s/%s.config", configDir, clusterName),
-			KeyringPath: path.Join(configDir, "keyring"),
-			Cluster:     "ceph",
-			UUID:        osdInfo[0].Tags.OSDFSID,
+			ID:                  id,
+			DataPath:            configDir,
+			Config:              fmt.Sprintf("%s/%s.config", configDir, clusterName),
+			KeyringPath:         path.Join(configDir, "keyring"),
+			Cluster:             "ceph",
+			UUID:                osdInfo[0].Tags.OSDFSID,
+			CephVolumeInitiated: true,
 		}
 		osds = append(osds, osd)
 	}

pkg/operator/ceph/cluster/osd/spec.go

@@ -99,7 +99,6 @@ func (c *Cluster) makeDeployment(nodeName string, devices []rookalpha.Device, se
 		}
 	} else {
 		dataDir = k8sutil.DataDir
-		osd.CephVolumeInitiated = true
 
 		// Create volume config for /dev so the pod can access devices on the host
 		devVolume := v1.Volume{Name: "devices", VolumeSource: v1.VolumeSource{HostPath: &v1.HostPathVolumeSource{Path: "/dev"}}}