Adds support for storageClassDeviceSet in rook-ceph operator

- Added code to support StorageClassDeviceSet spec provided in the cluster-on-pvc.yaml
- The code reads the StorageClassDeviceSet spec and creates pvc based on the ‘count’ field for each device set.
- OSD prepare job is started for each PVC which activates the ceph-volume on each PVC
- Finally OSD is started on each of the PVC device.

Co-authored-by: rohan47 <rohgupta@redhat.com>
Co-authored-by: Ashish Ranjan <aranjan@redhat.com>
Signed-off-by: Santosh Pillai <sapillai@redhat.com>

cluster/examples/kubernetes/ceph/cluster.yaml

@@ -99,8 +99,12 @@ spec:
 #    mon:
 #    osd:
   storage: # cluster level storage configuration and selection
-    useAllNodes: true
-    useAllDevices: true
+    persistentVolumeClaim:
+    - claimName: "example-pvc"
+    - claimName: "example-pvc1"
+    - claimName: "example-pvc4"
+    useAllNodes: false
+    useAllDevices: false
     deviceFilter:
     location:
     config:

cmd/rook/ceph/ceph.go

@@ -51,6 +51,7 @@ type config struct {
 	networkInfo        clusterd.NetworkInfo
 	monEndpoints       string
 	nodeName           string
+	pvcBacked          bool
 }
 
 func init() {

cmd/rook/ceph/osd.go

@@ -66,6 +66,7 @@ var (
 	osdStringID         string
 	osdUUID             string
 	osdIsDevice         bool
+	pvcBackedOSD        bool
 )
 
 func addOSDFlags(command *cobra.Command) {
@@ -79,6 +80,7 @@ func addOSDFlags(command *cobra.Command) {
 	provisionCmd.Flags().StringVar(&cfg.metadataDevice, "metadata-device", "", "device to use for metadata (e.g. a high performance SSD/NVMe device)")
 	provisionCmd.Flags().BoolVar(&cfg.forceFormat, "force-format", false,
 		"true to force the format of any specified devices, even if they already have a filesystem.  BE CAREFUL!")
+	provisionCmd.Flags().BoolVar(&cfg.pvcBacked, "pvc-backed-osd", false, "true to specify a block mode pvc is backing the OSD")
 
 	// flags for generating the osd config
 	osdConfigCmd.Flags().IntVar(&osdID, "osd-id", -1, "osd id for which to generate config")
@@ -92,6 +94,7 @@ func addOSDFlags(command *cobra.Command) {
 	osdStartCmd.Flags().StringVar(&osdStringID, "osd-id", "", "the osd ID")
 	osdStartCmd.Flags().StringVar(&osdUUID, "osd-uuid", "", "the osd UUID")
 	osdStartCmd.Flags().StringVar(&osdStoreType, "osd-store-type", "", "whether the osd is bluestore or filestore")
+	osdStartCmd.Flags().BoolVar(&pvcBackedOSD, "pvc-backed-osd", false, "Whether the OSD backing store in PVC or not")
 
 	// add the subcommands to the parent osd command
 	osdCmd.AddCommand(osdConfigCmd,
@@ -139,7 +142,7 @@ func startOSD(cmd *cobra.Command, args []string) error {
 	commonOSDInit(osdStartCmd)
 
 	context := createContext()
-	err := osddaemon.StartOSD(context, osdStoreType, osdStringID, osdUUID, args)
+	err := osddaemon.StartOSD(context, osdStoreType, osdStringID, osdUUID, pvcBackedOSD, args)
 	if err != nil {
 		rook.TerminateFatal(err)
 	}
@@ -244,14 +247,15 @@ func prepareOSD(cmd *cobra.Command, args []string) error {
 	ownerRef := cluster.ClusterOwnerRef(clusterInfo.Name, ownerRefID)
 	kv := k8sutil.NewConfigMapKVStore(clusterInfo.Name, context.Clientset, ownerRef)
 	agent := osddaemon.NewAgent(context, dataDevices, cfg.metadataDevice, cfg.directories, forceFormat,
-		crushLocation, cfg.storeConfig, &clusterInfo, cfg.nodeName, kv)
+		crushLocation, cfg.storeConfig, &clusterInfo, cfg.nodeName, kv, cfg.pvcBacked)
 
 	err = osddaemon.Provision(context, agent)
 	if err != nil {
 		// something failed in the OSD orchestration, update the status map with failure details
 		status := oposd.OrchestrationStatus{
-			Status:  oposd.OrchestrationStatusFailed,
-			Message: err.Error(),
+			Status:       oposd.OrchestrationStatusFailed,
+			Message:      err.Error(),
+			PvcBackedOSD: cfg.pvcBacked,
 		}
 		oposd.UpdateNodeStatus(kv, cfg.nodeName, status)
 

pkg/apis/rook.io/v1alpha2/types.go

@@ -16,7 +16,7 @@ limitations under the License.
 package v1alpha2
 
 import (
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
@@ -34,6 +34,8 @@ type StorageScopeSpec struct {
 	Location      string            `json:"location,omitempty"`
 	Config        map[string]string `json:"config"`
 	Selection
+	VolumeSources          []VolumeSource          `json:"volumeSources,omitempty"`
+	StorageClassDeviceSets []StorageClassDeviceSet `json:"storageClassDeviceSets"`
 }
 
 type Node struct {
@@ -127,3 +129,20 @@ type KeyType string
 type AnnotationsSpec map[KeyType]Annotations
 
 type Annotations map[string]string
+
+type StorageClassDeviceSet struct {
+	Name                 string                     `json:"name,omitempty"`                 // A unique identifier for the set
+	Count                int                        `json:"count,omitemity"`                // Number of devices in this set
+	Resources            v1.ResourceRequirements    `json:"resources,omitempty"`            // Requests/limits for the devices
+	Placement            Placement                  `json:"placement,omitempty"`            // Placement constraints for the devices
+	Config               map[string]string          `json:"config,omitempty"`               // Provider-specific device configuration
+	VolumeClaimTemplates []v1.PersistentVolumeClaim `json:"volumeClaimTemplates,omitempty"` // List of PVC templates for the underlying storage devices
+}
+
+type VolumeSource struct {
+	Name                        string                               `json:"name,omitempty"`
+	PersistentVolumeClaimSource v1.PersistentVolumeClaimVolumeSource `json:"persistentVolumeClaimSource,omitempty"`
+	Resources                   v1.ResourceRequirements              `json:"resources,omitempty"`
+	Placement                   Placement                            `json:"placement,omitempty"`
+	Config                      map[string]string                    `json:"config,omitempty"`
+}

pkg/apis/rook.io/v1alpha2/zz_generated.deepcopy.go

@@ -350,6 +350,38 @@ func (in *Selection) DeepCopy() *Selection {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *StorageClassDeviceSet) DeepCopyInto(out *StorageClassDeviceSet) {
+	*out = *in
+	in.Resources.DeepCopyInto(&out.Resources)
+	in.Placement.DeepCopyInto(&out.Placement)
+	if in.Config != nil {
+		in, out := &in.Config, &out.Config
+		*out = make(map[string]string, len(*in))
+		for key, val := range *in {
+			(*out)[key] = val
+		}
+	}
+	if in.VolumeClaimTemplates != nil {
+		in, out := &in.VolumeClaimTemplates, &out.VolumeClaimTemplates
+		*out = make([]v1.PersistentVolumeClaim, len(*in))
+		for i := range *in {
+			(*in)[i].DeepCopyInto(&(*out)[i])
+		}
+	}
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new StorageClassDeviceSet.
+func (in *StorageClassDeviceSet) DeepCopy() *StorageClassDeviceSet {
+	if in == nil {
+		return nil
+	}
+	out := new(StorageClassDeviceSet)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *StorageScopeSpec) DeepCopyInto(out *StorageScopeSpec) {
 	*out = *in
@@ -368,6 +400,20 @@ func (in *StorageScopeSpec) DeepCopyInto(out *StorageScopeSpec) {
 		}
 	}
 	in.Selection.DeepCopyInto(&out.Selection)
+	if in.VolumeSources != nil {
+		in, out := &in.VolumeSources, &out.VolumeSources
+		*out = make([]VolumeSource, len(*in))
+		for i := range *in {
+			(*in)[i].DeepCopyInto(&(*out)[i])
+		}
+	}
+	if in.StorageClassDeviceSets != nil {
+		in, out := &in.StorageClassDeviceSets, &out.StorageClassDeviceSets
+		*out = make([]StorageClassDeviceSet, len(*in))
+		for i := range *in {
+			(*in)[i].DeepCopyInto(&(*out)[i])
+		}
+	}
 	return
 }
 
@@ -444,3 +490,29 @@ func (in *VolumeList) DeepCopyObject() runtime.Object {
 	}
 	return nil
 }
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *VolumeSource) DeepCopyInto(out *VolumeSource) {
+	*out = *in
+	out.PersistentVolumeClaimSource = in.PersistentVolumeClaimSource
+	in.Resources.DeepCopyInto(&out.Resources)
+	in.Placement.DeepCopyInto(&out.Placement)
+	if in.Config != nil {
+		in, out := &in.Config, &out.Config
+		*out = make(map[string]string, len(*in))
+		for key, val := range *in {
+			(*out)[key] = val
+		}
+	}
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new VolumeSource.
+func (in *VolumeSource) DeepCopy() *VolumeSource {
+	if in == nil {
+		return nil
+	}
+	out := new(VolumeSource)
+	in.DeepCopyInto(out)
+	return out
+}

pkg/clusterd/disk.go

@@ -54,93 +54,104 @@ func DiscoverDevices(executor exec.Executor) ([]*sys.LocalDisk, error) {
 			// skip device
 			continue
 		}
-
-		diskProps, err := sys.GetDeviceProperties(d, executor)
-		if err != nil {
-			logger.Warningf("skipping device %s: %+v", d, err)
+		disk := PopulateDeviceInfo(d, executor)
+		if disk == nil {
 			continue
 		}
-
-		diskType, ok := diskProps["TYPE"]
-		if !ok || (diskType != sys.SSDType && diskType != sys.CryptType && diskType != sys.DiskType && diskType != sys.PartType && diskType != sys.LinearType) {
-			if !ok {
-				logger.Warningf("skipping device %s: diskType is empty", d)
-			} else {
-				logger.Warningf("skipping device %s: unsupported diskType %+s", d, diskType)
-			}
-			// unsupported disk type, just continue
-			continue
-		}
-
-		// get the UUID for disks
-		var diskUUID string
-		if diskType != sys.PartType {
-			diskUUID, err = sys.GetDiskUUID(d, executor)
-			if err != nil {
-				logger.Warningf("device %s has an unknown uuid. %+v", d, err)
-				continue
-			}
-		}
-
-		udevInfo, err := sys.GetUdevInfo(d, executor)
-		if err != nil {
-			logger.Warningf("failed to get udev info for device %s: %+v", d, err)
-			continue
-		}
-
-		disk := &sys.LocalDisk{Name: d, UUID: diskUUID}
-
-		if val, ok := diskProps["TYPE"]; ok {
-			disk.Type = val
-		}
-		if val, ok := diskProps["SIZE"]; ok {
-			if size, err := strconv.ParseUint(val, 10, 64); err == nil {
-				disk.Size = size
-			}
-		}
-		if val, ok := diskProps["ROTA"]; ok {
-			if rotates, err := strconv.ParseBool(val); err == nil {
-				disk.Rotational = rotates
-			}
-		}
-		if val, ok := diskProps["RO"]; ok {
-			if ro, err := strconv.ParseBool(val); err == nil {
-				disk.Readonly = ro
-			}
-		}
-		if val, ok := diskProps["PKNAME"]; ok {
-			disk.Parent = val
-		}
-
-		// parse udev info output
-		if val, ok := udevInfo["DEVLINKS"]; ok {
-			disk.DevLinks = val
-		}
-		if val, ok := udevInfo["ID_FS_TYPE"]; ok {
-			disk.Filesystem = val
-		}
-		if val, ok := udevInfo["ID_SERIAL"]; ok {
-			disk.Serial = val
-		}
-
-		if val, ok := udevInfo["ID_VENDOR"]; ok {
-			disk.Vendor = val
-		}
-
-		if val, ok := udevInfo["ID_MODEL"]; ok {
-			disk.Model = val
-		}
-
-		if val, ok := udevInfo["ID_WWN_WITH_EXTENSION"]; ok {
-			disk.WWNVendorExtension = val
-		}
-
-		if val, ok := udevInfo["ID_WWN"]; ok {
-			disk.WWN = val
-		}
-
+		disk = PopulateDeviceUdevInfo(d, executor, disk)
 		disks = append(disks, disk)
 	}
 
 	return disks, nil
 }
+
+func PopulateDeviceInfo(d string, executor exec.Executor) *sys.LocalDisk {
+	diskProps, err := sys.GetDeviceProperties(d, executor)
+	if err != nil {
+		logger.Warningf("skipping device %s: %+v", d, err)
+		return nil
+	}
+
+	diskType, ok := diskProps["TYPE"]
+	if !ok || (diskType != sys.SSDType && diskType != sys.CryptType && diskType != sys.DiskType && diskType != sys.PartType && diskType != sys.LinearType) {
+		if !ok {
+			logger.Warningf("skipping device %s: diskType is empty", d)
+		} else {
+			logger.Warningf("skipping device %s: unsupported diskType %+s", d, diskType)
+		}
+		// unsupported disk type, just continue
+		return nil
+	}
+
+	// get the UUID for disks
+	var diskUUID string
+	if diskType != sys.PartType {
+		diskUUID, err = sys.GetDiskUUID(d, executor)
+		if err != nil {
+			logger.Warningf("device %s has an unknown uuid. %+v", d, err)
+			return nil
+		}
+	}
+
+	disk := &sys.LocalDisk{Name: d, UUID: diskUUID}
+
+	if val, ok := diskProps["TYPE"]; ok {
+		disk.Type = val
+	}
+	if val, ok := diskProps["SIZE"]; ok {
+		if size, err := strconv.ParseUint(val, 10, 64); err == nil {
+			disk.Size = size
+		}
+	}
+	if val, ok := diskProps["ROTA"]; ok {
+		if rotates, err := strconv.ParseBool(val); err == nil {
+			disk.Rotational = rotates
+		}
+	}
+	if val, ok := diskProps["RO"]; ok {
+		if ro, err := strconv.ParseBool(val); err == nil {
+			disk.Readonly = ro
+		}
+	}
+	if val, ok := diskProps["PKNAME"]; ok {
+		disk.Parent = val
+	}
+
+	return disk
+}
+
+func PopulateDeviceUdevInfo(d string, executor exec.Executor, disk *sys.LocalDisk) *sys.LocalDisk {
+	udevInfo, err := sys.GetUdevInfo(d, executor)
+	if err != nil {
+		logger.Warningf("failed to get udev info for device %s: %+v", d, err)
+		return disk
+	}
+	// parse udev info output
+	if val, ok := udevInfo["DEVLINKS"]; ok {
+		disk.DevLinks = val
+	}
+	if val, ok := udevInfo["ID_FS_TYPE"]; ok {
+		disk.Filesystem = val
+	}
+	if val, ok := udevInfo["ID_SERIAL"]; ok {
+		disk.Serial = val
+	}
+
+	if val, ok := udevInfo["ID_VENDOR"]; ok {
+		disk.Vendor = val
+	}
+
+	if val, ok := udevInfo["ID_MODEL"]; ok {
+		disk.Model = val
+	}
+
+	if val, ok := udevInfo["ID_WWN_WITH_EXTENSION"]; ok {
+		disk.WWNVendorExtension = val
+	}
+
+	if val, ok := udevInfo["ID_WWN"]; ok {
+		disk.WWN = val
+	}
+
+	return disk
+}

pkg/daemon/ceph/osd/agent.go

@@ -56,6 +56,7 @@ type OsdAgent struct {
 	procMan        *proc.ProcManager
 	storeConfig    config.StoreConfig
 	kv             *k8sutil.ConfigMapKVStore
+	pvcBacked      bool
 	configCounter  int32
 	osdsCompleted  chan struct{}
 }
@@ -66,7 +67,7 @@ type device struct {
 }
 
 func NewAgent(context *clusterd.Context, devices []DesiredDevice, metadataDevice, directories string, forceFormat bool,
-	location string, storeConfig config.StoreConfig, cluster *cephconfig.ClusterInfo, nodeName string, kv *k8sutil.ConfigMapKVStore) *OsdAgent {
+	location string, storeConfig config.StoreConfig, cluster *cephconfig.ClusterInfo, nodeName string, kv *k8sutil.ConfigMapKVStore, pvcBacked bool) *OsdAgent {
 
 	return &OsdAgent{
 		devices:        devices,
@@ -78,6 +79,7 @@ func NewAgent(context *clusterd.Context, devices []DesiredDevice, metadataDevice
 		cluster:        cluster,
 		nodeName:       nodeName,
 		kv:             kv,
+		pvcBacked:      pvcBacked,
 		procMan:        proc.New(context.Executor),
 		osdProc:        make(map[int]*proc.MonitoredProc),
 	}
@@ -147,10 +149,17 @@ func (a *OsdAgent) removeDirs(context *clusterd.Context, removedDirs map[string]
 
 func (a *OsdAgent) configureDevices(context *clusterd.Context, devices *DeviceOsdMapping) ([]oposd.OSDInfo, error) {
 
-	cvSupported, err := getCephVolumeSupported(context)
-	if err != nil {
-		logger.Errorf("failed to detect if ceph-volume is available. %+v", err)
+	var cvSupported bool
+	var err error
+	if a.pvcBacked {
+		cvSupported = true //ceph-volume is always supported when OSD is backed by PVC
+	} else {
+		cvSupported, err = getCephVolumeSupported(context)
+		if err != nil {
+			logger.Errorf("failed to detect if ceph-volume is available. %+v", err)
+		}
 	}
+
 	if a.metadataDevice != "" {
 		// ceph-volume still is work in progress for accepting fast devices for the metadata
 		logger.Warningf("ceph-volume metadata support is experimental. osd provision might fail if vg on %s does not have enough space", a.metadataDevice)
@@ -160,7 +169,7 @@ func (a *OsdAgent) configureDevices(context *clusterd.Context, devices *DeviceOs
 	if devices == nil || len(devices.Entries) == 0 {
 		logger.Infof("no more devices to configure")
 		if cvSupported {
-			return getCephVolumeOSDs(context, a.cluster.Name, a.cluster.FSID)
+			return getCephVolumeOSDs(context, a.cluster.Name, a.cluster.FSID, "")
 		}
 		return osds, nil
 	}

pkg/daemon/ceph/osd/agent_test.go

@@ -247,6 +247,8 @@ func testOSDAgentWithDevicesHelper(t *testing.T, storeConfig config.StoreConfig,
 		"sdx": {Data: -1},
 		"sdy": {Data: -1},
 	}}
+
+	agent.pvcBacked = false
 	_, err = agent.configureDevices(context, devices)
 	assert.Nil(t, err)
 
@@ -398,7 +400,7 @@ func createTestAgent(t *testing.T, devices, configDir, nodeName string, storeCon
 	cluster := &cephconfig.ClusterInfo{Name: "myclust"}
 	context := &clusterd.Context{ConfigDir: configDir, Executor: executor, Clientset: testop.New(1)}
 	agent := NewAgent(context, desiredDevices, "", "", forceFormat, location, *storeConfig,
-		cluster, nodeName, mockKVStore())
+		cluster, nodeName, mockKVStore(), false)
 
 	return agent, executor, context
 }
@@ -465,7 +467,8 @@ func TestGetPartitionPerfScheme(t *testing.T) {
 	}
 	context.Executor = executor
 
-	devices, err := getAvailableDevices(context, []DesiredDevice{{Name: "sda"}, {Name: "sdb"}}, "sdc")
+	pvcBackedOSD := false
+	devices, err := getAvailableDevices(context, []DesiredDevice{{Name: "sda"}, {Name: "sdb"}}, "sdc", pvcBackedOSD)
 	assert.Nil(t, err)
 	scheme, _, err := a.getPartitionPerfScheme(context, devices, false)
 	assert.Nil(t, err)
@@ -542,7 +545,8 @@ func TestGetPartitionSchemeDiskInUse(t *testing.T) {
 
 	// get the partition scheme based on the available devices.  Since sda is already in use, the partition
 	// scheme returned should reflect that.
-	devices, err := getAvailableDevices(context, []DesiredDevice{{Name: "sda"}}, "")
+	pvcBackedOSD := false
+	devices, err := getAvailableDevices(context, []DesiredDevice{{Name: "sda"}}, "", pvcBackedOSD)
 	scheme, _, err := a.getPartitionPerfScheme(context, devices, false)
 	assert.Nil(t, err)
 
@@ -610,7 +614,8 @@ func TestGetPartitionSchemeDiskNameChanged(t *testing.T) {
 
 	// get the current partition scheme.  This should notice that the device names changed and update the
 	// partition scheme to have the latest device names
-	devices, err := getAvailableDevices(context, []DesiredDevice{{Name: "sda-changed"}}, "nvme01")
+	pvcBackedOSD := false
+	devices, err := getAvailableDevices(context, []DesiredDevice{{Name: "sda-changed"}}, "nvme01", pvcBackedOSD)
 	scheme, _, err := a.getPartitionPerfScheme(context, devices, false)
 	assert.Nil(t, err)
 	require.NotNil(t, scheme)

pkg/daemon/ceph/osd/daemon.go

@@ -39,7 +39,7 @@ var (
 )
 
 // StartOSD starts an OSD on a device that was provisioned by ceph-volume
-func StartOSD(context *clusterd.Context, osdType, osdID, osdUUID string, cephArgs []string) error {
+func StartOSD(context *clusterd.Context, osdType, osdID, osdUUID string, pvcBackedOSD bool, cephArgs []string) error {
 
 	// ensure the config mount point exists
 	configDir := fmt.Sprintf("/var/lib/ceph/osd/ceph-%s", osdID)
@@ -48,6 +48,16 @@ func StartOSD(context *clusterd.Context, osdType, osdID, osdUUID string, cephArg
 		logger.Errorf("failed to create config dir %s. %+v", configDir, err)
 	}
 
+	if pvcBackedOSD {
+		if err := updateLVMConfig(context); err != nil {
+			return fmt.Errorf("sed failure, %+v", err) // fail return here as validation provided by ceph-volume
+		}
+
+		context.Executor.ExecuteCommand(false, "", "/sbin/vgchange", "-an")
+
+		context.Executor.ExecuteCommand(false, "", "/sbin/vgchange", "-ay")
+	}
+
 	// activate the osd with ceph-volume
 	storeFlag := "--" + osdType
 	if err := context.Executor.ExecuteCommand(false, "", "stdbuf", "-oL", "ceph-volume", "lvm", "activate", "--no-systemd", storeFlag, osdID, osdUUID); err != nil {
@@ -100,16 +110,25 @@ func Provision(context *clusterd.Context, agent *OsdAgent) error {
 	}
 
 	logger.Infof("discovering hardware")
-	rawDevices, err := clusterd.DiscoverDevices(context.Executor)
-	if err != nil {
-		return fmt.Errorf("failed initial hardware discovery. %+v", err)
+	var rawDevices []*sys.LocalDisk
+	if agent.pvcBacked {
+		if len(agent.devices) > 1 {
+			return fmt.Errorf("more than one desired device found in case of PVC backed OSDs. we expect exactly one device")
+		}
+		rawDevices = append(rawDevices, clusterd.PopulateDeviceInfo(agent.devices[0].Name, context.Executor))
+	} else {
+		rawDevices, err = clusterd.DiscoverDevices(context.Executor)
+		if err != nil {
+			return fmt.Errorf("failed initial hardware discovery. %+v", err)
+		}
 	}
+
 	context.Devices = rawDevices
 
 	logger.Infof("creating and starting the osds")
 
 	// determine the set of devices that can/should be used for OSDs.
-	devices, err := getAvailableDevices(context, agent.devices, agent.metadataDevice)
+	devices, err := getAvailableDevices(context, agent.devices, agent.metadataDevice, agent.pvcBacked)
 	if err != nil {
 		return fmt.Errorf("failed to get available devices. %+v", err)
 	}
@@ -121,7 +140,7 @@ func Provision(context *clusterd.Context, agent *OsdAgent) error {
 	}
 
 	// orchestration is about to start, update the status
-	status = oposd.OrchestrationStatus{Status: oposd.OrchestrationStatusOrchestrating}
+	status = oposd.OrchestrationStatus{Status: oposd.OrchestrationStatusOrchestrating, PvcBackedOSD: agent.pvcBacked}
 	if err := oposd.UpdateNodeStatus(agent.kv, agent.nodeName, status); err != nil {
 		return err
 	}
@@ -168,7 +187,7 @@ func Provision(context *clusterd.Context, agent *OsdAgent) error {
 	osds := append(deviceOSDs, dirOSDs...)
 
 	// orchestration is completed, update the status
-	status = oposd.OrchestrationStatus{OSDs: osds, Status: oposd.OrchestrationStatusCompleted}
+	status = oposd.OrchestrationStatus{OSDs: osds, Status: oposd.OrchestrationStatusCompleted, PvcBackedOSD: agent.pvcBacked}
 	if err := oposd.UpdateNodeStatus(agent.kv, agent.nodeName, status); err != nil {
 		return err
 	}
@@ -176,7 +195,7 @@ func Provision(context *clusterd.Context, agent *OsdAgent) error {
 	return nil
 }
 
-func getAvailableDevices(context *clusterd.Context, desiredDevices []DesiredDevice, metadataDevice string) (*DeviceOsdMapping, error) {
+func getAvailableDevices(context *clusterd.Context, desiredDevices []DesiredDevice, metadataDevice string, pvcBacked bool) (*DeviceOsdMapping, error) {
 
 	available := &DeviceOsdMapping{Entries: map[string]*DeviceOsdIDEntry{}}
 
@@ -189,7 +208,7 @@ func getAvailableDevices(context *clusterd.Context, desiredDevices []DesiredDevi
 		if device.Type == sys.PartType {
 			continue
 		}
-		partCount, ownPartitions, fs, err := sys.CheckIfDeviceAvailable(context.Executor, device.Name)
+		partCount, ownPartitions, fs, err := sys.CheckIfDeviceAvailable(context.Executor, device.Name, pvcBacked)
 		if err != nil {
 			return nil, fmt.Errorf("failed to get device %s info. %+v", device.Name, err)
 		}

pkg/daemon/ceph/osd/daemon_test.go

@@ -70,6 +70,8 @@ func TestRunDaemon(t *testing.T) {
 	agent, _, context := createTestAgent(t, "none", configDir, "node5375", &config.StoreConfig{StoreType: config.Bluestore})
 	agent.devices[0].IsFilter = true
 
+	agent.pvcBacked = false
+	logger.Infof("Agent %+v", agent)
 	err := Provision(context, agent)
 	assert.Nil(t, err)
 }
@@ -209,7 +211,8 @@ NAME="sdb1" SIZE="30" TYPE="part" PKNAME="sdb"`, nil
 	}
 
 	// select all devices, including nvme01 for metadata
-	mapping, err := getAvailableDevices(context, []DesiredDevice{{Name: "all"}}, "nvme01")
+	pvcBackedOSD := false
+	mapping, err := getAvailableDevices(context, []DesiredDevice{{Name: "all"}}, "nvme01", pvcBackedOSD)
 	assert.Nil(t, err)
 	assert.Equal(t, 5, len(mapping.Entries))
 	assert.Equal(t, -1, mapping.Entries["sda"].Data)
@@ -221,29 +224,29 @@ NAME="sdb1" SIZE="30" TYPE="part" PKNAME="sdb"`, nil
 	assert.Equal(t, 0, len(mapping.Entries["nvme01"].Metadata))
 
 	// select no devices both using and not using a filter
-	mapping, err = getAvailableDevices(context, nil, "")
+	mapping, err = getAvailableDevices(context, nil, "", pvcBackedOSD)
 	assert.Nil(t, err)
 	assert.Equal(t, 0, len(mapping.Entries))
 
-	mapping, err = getAvailableDevices(context, nil, "")
+	mapping, err = getAvailableDevices(context, nil, "", pvcBackedOSD)
 	assert.Nil(t, err)
 	assert.Equal(t, 0, len(mapping.Entries))
 
 	// select the sd* devices
-	mapping, err = getAvailableDevices(context, []DesiredDevice{{Name: "^sd.$", IsFilter: true}}, "")
+	mapping, err = getAvailableDevices(context, []DesiredDevice{{Name: "^sd.$", IsFilter: true}}, "", pvcBackedOSD)
 	assert.Nil(t, err)
 	assert.Equal(t, 2, len(mapping.Entries))
 	assert.Equal(t, -1, mapping.Entries["sda"].Data)
 	assert.Equal(t, -1, mapping.Entries["sdd"].Data)
 
 	// select an exact device
-	mapping, err = getAvailableDevices(context, []DesiredDevice{{Name: "sdd"}}, "")
+	mapping, err = getAvailableDevices(context, []DesiredDevice{{Name: "sdd"}}, "", pvcBackedOSD)
 	assert.Nil(t, err)
 	assert.Equal(t, 1, len(mapping.Entries))
 	assert.Equal(t, -1, mapping.Entries["sdd"].Data)
 
 	// select all devices except those that have a prefix of "s"
-	mapping, err = getAvailableDevices(context, []DesiredDevice{{Name: "^[^s]", IsFilter: true}}, "")
+	mapping, err = getAvailableDevices(context, []DesiredDevice{{Name: "^[^s]", IsFilter: true}}, "", pvcBackedOSD)
 	assert.Nil(t, err)
 	assert.Equal(t, 3, len(mapping.Entries))
 	assert.Equal(t, -1, mapping.Entries["rda"].Data)

pkg/daemon/ceph/osd/device.go

@@ -111,7 +111,8 @@ func formatDevice(context *clusterd.Context, config *osdConfig, forceFormat bool
 	}
 
 	// check if partitions belong to rook
-	_, ownPartitions, devFS, err := sys.CheckIfDeviceAvailable(context.Executor, dataDetails.Device)
+	pvcBackedOSD := false
+	_, ownPartitions, devFS, err := sys.CheckIfDeviceAvailable(context.Executor, dataDetails.Device, pvcBackedOSD)
 	if err != nil {
 		return nil, fmt.Errorf("failed to format device. %+v", err)
 	}
@@ -171,7 +172,8 @@ func partitionMetadata(context *clusterd.Context, info *config.MetadataDeviceInf
 	}
 
 	// check one last time to make sure it's OK for us to format this metadata device
-	_, ownPartitions, fs, err := sys.CheckIfDeviceAvailable(context.Executor, info.Device)
+	pvcBackedOSD := false
+	_, ownPartitions, fs, err := sys.CheckIfDeviceAvailable(context.Executor, info.Device, pvcBackedOSD)
 	if err != nil {
 		return fmt.Errorf("failed to get metadata device %s info: %+v", info.Device, err)
 	} else if fs != "" || !ownPartitions {

pkg/daemon/ceph/osd/device_test.go

@@ -117,7 +117,7 @@ NAME="sda3" SIZE="20" TYPE="part" PKNAME="sda"`
 			assert.Equal(t, "lsblk /dev/sda", name)
 			output = ""
 		case 5:
-			assert.Equal(t, "get filesystem type for sda", name)
+			assert.Equal(t, "get filesystem type for /dev/sda", name)
 			output = ""
 		}
 		outputExecCount++

pkg/daemon/ceph/osd/volume.go

@@ -21,6 +21,7 @@ import (
 	"fmt"
 	"path"
 	"strconv"
+	"strings"
 	"syscall"
 
 	"github.com/rook/rook/pkg/clusterd"
@@ -28,6 +29,7 @@ import (
 	"github.com/rook/rook/pkg/operator/ceph/cluster/osd/config"
 	"github.com/rook/rook/pkg/util/display"
 	"github.com/rook/rook/pkg/util/exec"
+	"github.com/rook/rook/pkg/util/sys"
 )
 
 var cephConfigDir = "/var/lib/ceph"
@@ -42,11 +44,12 @@ const (
 
 func (a *OsdAgent) configureCVDevices(context *clusterd.Context, devices *DeviceOsdMapping) ([]oposd.OSDInfo, error) {
 	var osds []oposd.OSDInfo
+	var lv string
 
 	var err error
 	if len(devices.Entries) == 0 {
 		logger.Infof("no new devices to configure. returning devices already configured with ceph-volume.")
-		osds, err = getCephVolumeOSDs(context, a.cluster.Name, a.cluster.FSID)
+		osds, err = getCephVolumeOSDs(context, a.cluster.Name, a.cluster.FSID, lv)
 		if err != nil {
 			logger.Infof("failed to get devices already provisioned by ceph-volume. %+v", err)
 		}
@@ -57,15 +60,97 @@ func (a *OsdAgent) configureCVDevices(context *clusterd.Context, devices *Device
 	if err != nil {
 		return nil, fmt.Errorf("failed to generate osd keyring. %+v", err)
 	}
-
-	if err = a.initializeDevices(context, devices); err != nil {
-		return nil, fmt.Errorf("failed to initialize devices. %+v", err)
+	if a.pvcBacked {
+		if lv, err = a.initializeBlockPVC(context, devices); err != nil {
+			return nil, fmt.Errorf("failed to initialize devices. %+v", err)
+		}
+	} else {
+		if err = a.initializeDevices(context, devices); err != nil {
+			return nil, fmt.Errorf("failed to initialize devices. %+v", err)
+		}
 	}
 
-	osds, err = getCephVolumeOSDs(context, a.cluster.Name, a.cluster.FSID)
+	osds, err = getCephVolumeOSDs(context, a.cluster.Name, a.cluster.FSID, lv)
 	return osds, err
 }
 
+func (a *OsdAgent) initializeBlockPVC(context *clusterd.Context, devices *DeviceOsdMapping) (string, error) {
+	if err := updateLVMConfig(context); err != nil {
+		return "", fmt.Errorf("sed failure, %+v", err) // fail return here as validation provided by ceph-volume
+	}
+	baseCommand := "stdbuf"
+	baseArgs := []string{"-oL", cephVolumeCmd, "lvm", "prepare"}
+	var lvpath string
+	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)
+			deviceArg := device.Config.Name
+			immediateExecuteArgs := append(baseArgs, []string{
+				"--data",
+				deviceArg,
+			}...)
+			// execute ceph-volume with the device
+
+			if op, err := context.Executor.ExecuteCommandWithOutput(false, "", baseCommand, immediateExecuteArgs...); err != nil {
+				return "", fmt.Errorf("failed ceph-volume. %+v", err) // fail return here as validation provided by ceph-volume
+			} else {
+				logger.Infof("%v", op)
+				lvpath = getLVPath(op)
+				if lvpath == "" {
+					return "", fmt.Errorf("failed to get lvpath from ceph-volume lvm prepare output")
+				}
+			}
+		} else {
+			logger.Infof("skipping device %s with osd %d already configured", name, device.Data)
+		}
+	}
+
+	return lvpath, nil
+}
+
+func getLVPath(op string) string {
+	tmp := sys.Grep(op, "Volume group")
+	vgtmp := strings.Split(tmp, "\"")
+
+	tmp = sys.Grep(op, "Logical volume")
+	lvtmp := strings.Split(tmp, "\"")
+	if len(vgtmp) > 0 && len(lvtmp) > 0 {
+		if sys.Grep(vgtmp[1], "ceph") != "" && sys.Grep(lvtmp[1], "osd-block") != "" {
+			return fmt.Sprintf("/dev/%s/%s", vgtmp[1], lvtmp[1])
+		}
+	}
+	return ""
+}
+
+func updateLVMConfig(context *clusterd.Context) error {
+	testArgs := []string{"-oL", "sed", "-i", "-e", "s#udev_sync = 1#udev_sync = 0#", "/etc/lvm/lvm.conf"}
+	if err := context.Executor.ExecuteCommand(false, "", "stdbuf", testArgs...); err != nil {
+		return fmt.Errorf("failed ceph-volume. %+v", err) // fail return here as validation provided by ceph-volume
+	}
+	testArgs = []string{"-oL", "sed", "-i", "-e", "s#udev_rules = 1#udev_rules = 0#", "/etc/lvm/lvm.conf"}
+	if err := context.Executor.ExecuteCommand(false, "", "stdbuf", testArgs...); err != nil {
+		return fmt.Errorf("failed ceph-volume. %+v", err) // fail return here as validation provided by ceph-volume
+	}
+	testArgs = []string{"-oL", "sed", "-i", "-e", "s#use_lvmetad = 1#use_lvmetad = 0#", "/etc/lvm/lvm.conf"}
+	if err := context.Executor.ExecuteCommand(false, "", "stdbuf", testArgs...); err != nil {
+		return fmt.Errorf("failed ceph-volume. %+v", err) // fail return here as validation provided by ceph-volume
+	}
+	testArgs = []string{"-oL", "sed", "-i", "-e", "s#scan = \\[ \"/dev\" \\]#scan = [ \"/dev\", \"/mnt\" ]#", "/etc/lvm/lvm.conf"}
+	if err := context.Executor.ExecuteCommand(false, "", "stdbuf", testArgs...); err != nil {
+		return fmt.Errorf("failed ceph-volume. %+v", err) // fail return here as validation provided by ceph-volume
+	}
+	testArgs = []string{"-oL", "sed", "-i", "-e", "0,/# filter =.*/{s%# filter =.*% filter = [ \"a|^/mnt/.*| r|.*/|\" ]%}", "/etc/lvm/lvm.conf"}
+	if err := context.Executor.ExecuteCommand(false, "", "stdbuf", testArgs...); err != nil {
+		return fmt.Errorf("failed ceph-volume. %+v", err) // fail return here as validation provided by ceph-volume
+	}
+
+	return nil
+}
+
 func (a *OsdAgent) initializeDevices(context *clusterd.Context, devices *DeviceOsdMapping) error {
 	storeFlag := "--bluestore"
 	if a.storeConfig.StoreType == config.Filestore {
@@ -133,6 +218,9 @@ func (a *OsdAgent) initializeDevices(context *clusterd.Context, devices *DeviceO
 					"--report",
 				}...)
 
+				logger.Infof("Base command - %+v", baseCommand)
+				logger.Infof("immediateReportArgs - %+v", baseCommand)
+				logger.Infof("immediateExecuteArgs - %+v", immediateExecuteArgs)
 				if err := context.Executor.ExecuteCommand(false, "", baseCommand, immediateReportArgs...); err != nil {
 					return fmt.Errorf("failed ceph-volume report. %+v", err) // fail return here as validation provided by ceph-volume
 				}
@@ -200,7 +288,9 @@ func sanitizeOSDsPerDevice(count int) string {
 }
 
 func getCephVolumeSupported(context *clusterd.Context) (bool, error) {
+
 	_, err := context.Executor.ExecuteCommandWithOutput(false, "", cephVolumeCmd, "lvm", "batch", "--prepare")
+
 	if err != nil {
 		if cmdErr, ok := err.(*exec.CommandError); ok {
 			exitStatus := cmdErr.ExitStatus()
@@ -216,9 +306,9 @@ func getCephVolumeSupported(context *clusterd.Context) (bool, error) {
 	return true, nil
 }
 
-func getCephVolumeOSDs(context *clusterd.Context, clusterName string, cephfsid string) ([]oposd.OSDInfo, error) {
+func getCephVolumeOSDs(context *clusterd.Context, clusterName string, cephfsid string, lv string) ([]oposd.OSDInfo, error) {
 
-	result, err := context.Executor.ExecuteCommandWithOutput(false, "", cephVolumeCmd, "lvm", "list", "--format", "json")
+	result, err := context.Executor.ExecuteCommandWithOutput(false, "", cephVolumeCmd, "lvm", "list", lv, "--format", "json")
 	if err != nil {
 		return nil, fmt.Errorf("failed to retrieve ceph-volume results. %+v", err)
 	}

pkg/daemon/ceph/osd/volume_test.go

@@ -252,7 +252,7 @@ func TestParseCephVolumeResult(t *testing.T) {
 	}
 
 	context := &clusterd.Context{Executor: executor}
-	osds, err := getCephVolumeOSDs(context, "rook", "4bfe8b72-5e69-4330-b6c0-4d914db8ab89")
+	osds, err := getCephVolumeOSDs(context, "rook", "4bfe8b72-5e69-4330-b6c0-4d914db8ab89", "")
 	assert.Nil(t, err)
 	require.NotNil(t, osds)
 	assert.Equal(t, 2, len(osds))
@@ -272,7 +272,7 @@ func TestCephVolumeResultMultiClusterSingleOSD(t *testing.T) {
 	}
 
 	context := &clusterd.Context{Executor: executor}
-	osds, err := getCephVolumeOSDs(context, "rook", "451267e6-883f-4936-8dff-080d781c67d5")
+	osds, err := getCephVolumeOSDs(context, "rook", "451267e6-883f-4936-8dff-080d781c67d5", "")
 	assert.Nil(t, err)
 	require.NotNil(t, osds)
 	assert.Equal(t, 1, len(osds))
@@ -293,7 +293,7 @@ func TestCephVolumeResultMultiClusterMultiOSD(t *testing.T) {
 	}
 
 	context := &clusterd.Context{Executor: executor}
-	osds, err := getCephVolumeOSDs(context, "rook", "451267e6-883f-4936-8dff-080d781c67d5")
+	osds, err := getCephVolumeOSDs(context, "rook", "451267e6-883f-4936-8dff-080d781c67d5", "")
 	assert.Nil(t, err)
 	require.NotNil(t, osds)
 	assert.Equal(t, 1, len(osds))

pkg/daemon/discover/discover.go

@@ -34,7 +34,7 @@ import (
 	"github.com/rook/rook/pkg/operator/k8sutil"
 	"github.com/rook/rook/pkg/util/sys"
 
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
@@ -344,6 +344,7 @@ func updateDeviceCM(context *clusterd.Context) error {
 func probeDevices(context *clusterd.Context) ([]sys.LocalDisk, error) {
 	devices := make([]sys.LocalDisk, 0)
 	localDevices, err := clusterd.DiscoverDevices(context.Executor)
+	logger.Infof("localdevices - %+v", localDevices)
 	if err != nil {
 		return devices, fmt.Errorf("failed initial hardware discovery. %+v", err)
 	}

pkg/daemon/discover/discover_test.go

@@ -78,7 +78,7 @@ func TestProbeDevices(t *testing.T) {
 			output = "testa"
 		case "lsblk /dev/testa":
 			output = `SIZE="249510756352" ROTA="1" RO="0" TYPE="disk" PKNAME=""`
-		case "get filesystem type for testa":
+		case "get filesystem type for /dev/testa":
 			output = udevOutput
 		case "get parent for device testa":
 			output = `       testa
@@ -88,7 +88,7 @@ testa2   centos_host13-root
 testa2   centos_host13-swap
 testa2   centos_host13-home
 `
-		case "get disk testa uuid":
+		case "get disk /dev/testa uuid":
 			output = sgdiskOutput
 
 		case "get disk testa fs serial":

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

@@ -0,0 +1,129 @@
+/*
+Copyright 2016 The Rook Authors. All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+	http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package osd
+
+import (
+	"fmt"
+
+	rookalpha "github.com/rook/rook/pkg/apis/rook.io/v1alpha2"
+	opspec "github.com/rook/rook/pkg/operator/ceph/spec"
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+)
+
+func (c *Cluster) prepareStorageClassDeviceSets(config *provisionConfig) []rookalpha.VolumeSource {
+	volumeSources := []rookalpha.VolumeSource{}
+	for _, storageClassDeviceSet := range c.DesiredStorage.StorageClassDeviceSets {
+		if err := opspec.CheckPodMemory(storageClassDeviceSet.Resources, cephOsdPodMinimumMemory); err != nil {
+			config.addError("cannot use storageClassDeviceSet %s for creating osds %v", storageClassDeviceSet.Name, err)
+			continue
+		}
+		for i := 0; i < storageClassDeviceSet.Count; i++ {
+			pvc, err := c.createStorageClassDeviceSetPVC(storageClassDeviceSet, i)
+			if err != nil {
+				config.addError("%+v", err)
+				config.addError("OSD creation for storageClassDeviceSet %v failed for count %v", storageClassDeviceSet.Name, i)
+				continue
+			}
+			volumeSources = append(volumeSources, rookalpha.VolumeSource{
+				Name:      storageClassDeviceSet.Name,
+				Resources: storageClassDeviceSet.Resources,
+				Placement: storageClassDeviceSet.Placement,
+				Config:    storageClassDeviceSet.Config,
+				PersistentVolumeClaimSource: v1.PersistentVolumeClaimVolumeSource{
+					ClaimName: pvc.GetName(),
+					ReadOnly:  false,
+				},
+			})
+			logger.Infof("successfully provisioned osd for storageClassDeviceSet %s of set %v", storageClassDeviceSet.Name, i)
+		}
+	}
+	return volumeSources
+}
+
+func (c *Cluster) createStorageClassDeviceSetPVC(storageClassDeviceSet rookalpha.StorageClassDeviceSet, setIndex int) (*v1.PersistentVolumeClaim, error) {
+	if len(storageClassDeviceSet.VolumeClaimTemplates) == 0 {
+		return nil, fmt.Errorf("No PVC available for storageClassDeviceSet %s", storageClassDeviceSet.Name)
+	}
+	pvcStorageClassDeviceSetPVCId, pvcStorageClassDeviceSetPVCIdLabelSelector := makeStorageClassDeviceSetPVCID(storageClassDeviceSet.Name, setIndex, 0)
+
+	pvc := makeStorageClassDeviceSetPVC(storageClassDeviceSet.Name, pvcStorageClassDeviceSetPVCId, 0, setIndex, storageClassDeviceSet.VolumeClaimTemplates[0])
+	// Check if a PVC already exists with same StorageClassDeviceSet label
+	presentPVCs, err := c.context.Clientset.CoreV1().PersistentVolumeClaims(c.Namespace).List(metav1.ListOptions{LabelSelector: pvcStorageClassDeviceSetPVCIdLabelSelector})
+	if err != nil {
+		return nil, fmt.Errorf("failed to create pvc %v for storageClassDeviceSet %v, err %+v", pvc.GetGenerateName(), storageClassDeviceSet.Name, err)
+	}
+	if len(presentPVCs.Items) == 0 { // No PVC found, creating a new one
+		deployedPVC, err := c.context.Clientset.CoreV1().PersistentVolumeClaims(c.Namespace).Create(pvc)
+		if err != nil {
+			return nil, fmt.Errorf("failed to create pvc %v for storageClassDeviceSet %v, err %+v", pvc.GetGenerateName(), storageClassDeviceSet.Name, err)
+		}
+		return deployedPVC, nil
+	} else if len(presentPVCs.Items) == 1 { // The PVC is already present.
+		return &presentPVCs.Items[0], nil
+	}
+	// More than one PVC exists with same labelSelector
+	return nil, fmt.Errorf("more than one PVCs exists with label %v, pvcs %+v", pvcStorageClassDeviceSetPVCIdLabelSelector, presentPVCs)
+}
+
+// GetValidVolumeSources returns all volumes that has an OSDDeployment
+func GetValidVolumeSources(clientset kubernetes.Interface, namespace string, volumeSources []rookalpha.VolumeSource) (validVolumes []rookalpha.VolumeSource, err error) {
+	if clientset == nil {
+		return validVolumes, fmt.Errorf("nil clientset received")
+	}
+	for _, vol := range volumeSources {
+		osdDeployments, err := clientset.AppsV1().Deployments(namespace).List(metav1.ListOptions{LabelSelector: OSDOverPVCLabelKey + "=" + vol.PersistentVolumeClaimSource.ClaimName})
+		if err != nil {
+			return validVolumes, err
+		}
+		osdPrepareJob, err := clientset.BatchV1().Jobs(namespace).List(metav1.ListOptions{LabelSelector: OSDOverPVCLabelKey + "=" + vol.PersistentVolumeClaimSource.ClaimName})
+		if err != nil {
+			return validVolumes, err
+		}
+		if len(osdDeployments.Items) == 0 && len(osdPrepareJob.Items) == 0 {
+			validVolumes = append(validVolumes, vol)
+		}
+	}
+	return
+}
+
+func makeStorageClassDeviceSetPVC(storageClassDeviceSetName, pvcStorageClassDeviceSetPVCId string, pvcIndex, setIndex int, pvcTemplate v1.PersistentVolumeClaim) (pvcs *v1.PersistentVolumeClaim) {
+	pvcLabels := makeStorageClassDeviceSetPVCLabel(storageClassDeviceSetName, pvcStorageClassDeviceSetPVCId, pvcIndex, setIndex)
+
+	// pvc naming format <storageClassDeviceSetName>-<SetNumber>-<PVCIndex>
+	pvcGenerateName := pvcStorageClassDeviceSetPVCId + "-"
+
+	// Add pvcTemplate name to pvc name. i.e <storageClassDeviceSetName>-<SetNumber>-<PVCIndex>-<pvcTemplateName>
+	if len(pvcTemplate.GetName()) != 0 {
+		pvcGenerateName = pvcGenerateName + pvcTemplate.GetName() + "-"
+	}
+
+	// Add user provided labels to pvcTemplates
+	for k, v := range pvcTemplate.GetLabels() {
+		pvcLabels[k] = v
+	}
+
+	return &v1.PersistentVolumeClaim{
+		ObjectMeta: metav1.ObjectMeta{
+			GenerateName: pvcGenerateName,
+			Labels:       pvcLabels,
+			Annotations:  pvcTemplate.Annotations,
+		},
+		Spec: pvcTemplate.Spec,
+	}
+}

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

@@ -126,13 +126,29 @@ type OSDInfo struct {
 }
 
 type OrchestrationStatus struct {
-	OSDs    []OSDInfo `json:"osds"`
-	Status  string    `json:"status"`
-	Message string    `json:"message"`
+	OSDs         []OSDInfo `json:"osds"`
+	Status       string    `json:"status"`
+	PvcBackedOSD bool      `json:"pvc-backed-osd"`
+	Message      string    `json:"message"`
+}
+
+type osdProperties struct {
+	//crushHostname refers to the hostname or PVC name when the OSD is provisioned on Nodes or PVC block device, respectively.
+	crushHostname  string
+	devices        []rookalpha.Device
+	pvc            v1.PersistentVolumeClaimVolumeSource
+	selection      rookalpha.Selection
+	resources      v1.ResourceRequirements
+	storeConfig    osdconfig.StoreConfig
+	placement      rookalpha.Placement
+	metadataDevice string
+	location       string
 }
 
 // Start the osd management
 func (c *Cluster) Start() error {
+	config := newProvisionConfig()
+
 	// Validate pod's memory if specified
 	// This is valid for both Filestore and Bluestore
 	err := opspec.CheckPodMemory(c.resources, cephOsdPodMinimumMemory)
@@ -142,52 +158,17 @@ func (c *Cluster) Start() error {
 
 	logger.Infof("start running osds in namespace %s", c.Namespace)
 
-	if c.DesiredStorage.UseAllNodes == false && len(c.DesiredStorage.Nodes) == 0 {
-		logger.Warningf("useAllNodes is set to false and no nodes are specified, no OSD pods are going to be created")
+	if c.DesiredStorage.UseAllNodes == false && len(c.DesiredStorage.Nodes) == 0 && len(c.DesiredStorage.VolumeSources) == 0 && len(c.DesiredStorage.StorageClassDeviceSets) == 0 {
+		logger.Warningf("useAllNodes is set to false and no nodes, storageClassDevicesets or volumeSources are specified, no OSD pods are going to be created")
 	}
 
-	if c.DesiredStorage.UseAllNodes {
-		// Get the list of all nodes in the cluster. The placement settings will be applied below.
-		hostnameMap, err := k8sutil.GetNodeHostNames(c.context.Clientset)
-		if err != nil {
-			logger.Warningf("failed to get node hostnames: %v", err)
-			return err
-		}
-		c.DesiredStorage.Nodes = nil
-		for _, hostname := range hostnameMap {
-			storageNode := rookalpha.Node{
-				Name: hostname,
-			}
-			c.DesiredStorage.Nodes = append(c.DesiredStorage.Nodes, storageNode)
-		}
-		logger.Debugf("storage nodes: %+v", c.DesiredStorage.Nodes)
-	}
-	// generally speaking, this finds nodes which are capable of running new osds
-	validNodes := k8sutil.GetValidNodes(c.DesiredStorage, c.context.Clientset, c.placement)
-
-	// no valid node is ready to run an osd
-	if len(validNodes) == 0 {
-		logger.Warningf("no valid node available to run an osd in namespace %s. "+
-			"Rook will not create any new OSD nodes and will skip checking for removed nodes since "+
-			"removing all OSD nodes without destroying the Rook cluster is unlikely to be intentional", c.Namespace)
-		return nil
-	}
-	logger.Infof("%d of the %d storage nodes are valid", len(validNodes), len(c.DesiredStorage.Nodes))
-	c.ValidStorage = *c.DesiredStorage.DeepCopy()
-	c.ValidStorage.Nodes = validNodes
-
 	// start the jobs to provision the OSD devices and directories
-	config := newProvisionConfig()
+
+	logger.Infof("start provisioning the osds on pvcs, if needed")
+	c.startProvisioningOverPVCs(config)
+
 	logger.Infof("start provisioning the osds on nodes, if needed")
-	c.startProvisioning(config)
-
-	// start the OSD pods, waiting for the provisioning to be completed
-	logger.Infof("start osds after provisioning is completed, if needed")
-	c.completeProvision(config)
-
-	// handle the removed nodes and rebalance the PGs
-	logger.Infof("checking if any nodes were removed")
-	c.handleRemovedNodes(config)
+	c.startProvisioningOverNodes(config)
 
 	if len(config.errorMessages) > 0 {
 		return fmt.Errorf("%d failures encountered while running osds in namespace %s: %+v",
@@ -225,12 +206,105 @@ func (c *Cluster) Start() error {
 	return nil
 }
 
-func (c *Cluster) startProvisioning(config *provisionConfig) {
+func (c *Cluster) startProvisioningOverPVCs(config *provisionConfig) {
+	// Parsing storageClassDeviceSets and parsing it to volume sources
+	c.DesiredStorage.VolumeSources = append(c.DesiredStorage.VolumeSources, c.prepareStorageClassDeviceSets(config)...)
+
+	// Filtering out valid volume sources
+	validVolumeSources, err := GetValidVolumeSources(c.context.Clientset, c.Namespace, c.DesiredStorage.VolumeSources)
+	if err != nil {
+		config.addError(fmt.Sprintf("%v", err))
+		return
+	}
+	c.ValidStorage.VolumeSources = validVolumeSources
+
+	// no validVolumeSource is ready to run an osd
+	if len(c.DesiredStorage.VolumeSources) == 0 && len(c.DesiredStorage.StorageClassDeviceSets) == 0 {
+		logger.Warningf("no valid volumeSource available to run an osd in namespace %s. "+
+			"Rook will not create any new OSD nodes and will skip checking for removed pvcs "+
+			"removing all OSD nodes without destroying the Rook cluster is unlikely to be intentional", c.Namespace)
+		return
+	}
+
+	logger.Infof("%d of the %d volumeSources are valid", len(validVolumeSources), len(c.DesiredStorage.VolumeSources))
+	for _, volume := range c.ValidStorage.VolumeSources {
+		osdProps := osdProperties{
+			crushHostname: volume.PersistentVolumeClaimSource.ClaimName,
+			pvc:           volume.PersistentVolumeClaimSource,
+			resources:     volume.Resources,
+			placement:     volume.Placement,
+		}
+
+		// update the orchestration status of this pvc to the starting state
+		status := OrchestrationStatus{Status: OrchestrationStatusStarting, PvcBackedOSD: true}
+		if err := c.updateOSDStatus(osdProps.crushHostname, status); err != nil {
+			config.addError("failed to set orchestration starting status for pvc %s: %+v", osdProps.crushHostname, err)
+			continue
+		}
+
+		job, err := c.makeJob(osdProps)
+		if err != nil {
+			message := fmt.Sprintf("failed to create prepare job for pvc %s: %v", osdProps.crushHostname, err)
+			config.addError(message)
+			status := OrchestrationStatus{Status: OrchestrationStatusCompleted, Message: message, PvcBackedOSD: true}
+			if err := c.updateOSDStatus(osdProps.crushHostname, status); err != nil {
+				config.addError("failed to update pvc %s status. %+v", osdProps.crushHostname, err)
+				continue
+			}
+		}
+
+		if !c.runJob(job, osdProps.crushHostname, config, "provision") {
+			status := OrchestrationStatus{
+				Status:       OrchestrationStatusCompleted,
+				Message:      fmt.Sprintf("failed to start osd provisioning on pvc %s", osdProps.crushHostname),
+				PvcBackedOSD: true,
+			}
+			if err := c.updateOSDStatus(osdProps.crushHostname, status); err != nil {
+				config.addError("failed to update osd %s status. %+v", osdProps.crushHostname, err)
+			}
+		}
+	}
+	logger.Infof("start osds after provisioning is completed, if needed")
+	c.completeProvision(config)
+}
+
+func (c *Cluster) startProvisioningOverNodes(config *provisionConfig) {
 	if len(c.dataDirHostPath) == 0 {
 		logger.Warningf("skipping osd provisioning where no dataDirHostPath is set")
 		return
 	}
 
+	if c.DesiredStorage.UseAllNodes {
+		// Get the list of all nodes in the cluster. The placement settings will be applied below.
+		hostnameMap, err := k8sutil.GetNodeHostNames(c.context.Clientset)
+		if err != nil {
+			config.addError("failed to get node hostnames: %v", err)
+			return
+		}
+		c.DesiredStorage.Nodes = nil
+		for _, hostname := range hostnameMap {
+			storageNode := rookalpha.Node{
+				Name: hostname,
+			}
+			c.DesiredStorage.Nodes = append(c.DesiredStorage.Nodes, storageNode)
+		}
+		logger.Debugf("storage nodes: %+v", c.DesiredStorage.Nodes)
+	}
+	// generally speaking, this finds nodes which are capable of running new osds
+	validNodes := k8sutil.GetValidNodes(c.DesiredStorage, c.context.Clientset, c.placement)
+
+	logger.Infof("%d of the %d storage nodes are valid", len(validNodes), len(c.DesiredStorage.Nodes))
+
+	c.ValidStorage = *c.DesiredStorage.DeepCopy()
+	c.ValidStorage.Nodes = validNodes
+
+	// no valid node is ready to run an osd
+	if len(validNodes) == 0 {
+		logger.Warningf("no valid nodes available to run an osd in namespace %s. "+
+			"Rook will not create any new OSD nodes and will skip checking for removed nodes since "+
+			"removing all OSD nodes without destroying the Rook cluster is unlikely to be intentional", c.Namespace)
+		return
+	}
 	// start with nodes currently in the storage spec
 	for _, node := range c.ValidStorage.Nodes {
 		// fully resolve the storage config and resources for this node
@@ -247,7 +321,7 @@ func (c *Cluster) startProvisioning(config *provisionConfig) {
 
 		// update the orchestration status of this node to the starting state
 		status := OrchestrationStatus{Status: OrchestrationStatusStarting}
-		if err := c.updateNodeStatus(n.Name, status); err != nil {
+		if err := c.updateOSDStatus(n.Name, status); err != nil {
 			config.addError("failed to set orchestration starting status for node %s: %+v", n.Name, err)
 			continue
 		}
@@ -255,12 +329,21 @@ func (c *Cluster) startProvisioning(config *provisionConfig) {
 		// create the job that prepares osds on the node
 		storeConfig := osdconfig.ToStoreConfig(n.Config)
 		metadataDevice := osdconfig.MetadataDevice(n.Config)
-		job, err := c.makeJob(n.Name, n.Devices, n.Selection, n.Resources, storeConfig, metadataDevice, n.Location)
+		osdProps := osdProperties{
+			crushHostname:  n.Name,
+			devices:        n.Devices,
+			selection:      n.Selection,
+			resources:      n.Resources,
+			storeConfig:    storeConfig,
+			metadataDevice: metadataDevice,
+			location:       n.Location,
+		}
+		job, err := c.makeJob(osdProps)
 		if err != nil {
 			message := fmt.Sprintf("failed to create prepare job node %s: %v", n.Name, err)
 			config.addError(message)
 			status := OrchestrationStatus{Status: OrchestrationStatusCompleted, Message: message}
-			if err := c.updateNodeStatus(n.Name, status); err != nil {
+			if err := c.updateOSDStatus(n.Name, status); err != nil {
 				config.addError("failed to update node %s status. %+v", n.Name, err)
 				continue
 			}
@@ -268,11 +351,19 @@ func (c *Cluster) startProvisioning(config *provisionConfig) {
 
 		if !c.runJob(job, n.Name, config, "provision") {
 			status := OrchestrationStatus{Status: OrchestrationStatusCompleted, Message: fmt.Sprintf("failed to start osd provisioning on node %s", n.Name)}
-			if err := c.updateNodeStatus(n.Name, status); err != nil {
+			if err := c.updateOSDStatus(n.Name, status); err != nil {
 				config.addError("failed to update node %s status. %+v", n.Name, err)
 			}
 		}
 	}
+	logger.Infof("start osds after provisioning is completed, if needed")
+	c.completeProvision(config)
+
+	// start the OSD pods, waiting for the provisioning to be completed
+	// handle the removed nodes and rebalance the PGs
+	logger.Infof("checking if any nodes were removed")
+	c.handleRemovedNodes(config)
+
 }
 
 func (c *Cluster) runJob(job *batch.Job, nodeName string, config *provisionConfig, action string) bool {
@@ -291,6 +382,56 @@ func (c *Cluster) runJob(job *batch.Job, nodeName string, config *provisionConfi
 	return true
 }
 
+func (c *Cluster) startOSDDaemonsOnPVC(pvcName string, config *provisionConfig, configMap *v1.ConfigMap, status *OrchestrationStatus) {
+	osds := status.OSDs
+	logger.Infof("starting %d osd daemons on pvc %s", len(osds), pvcName)
+	conf := make(map[string]string)
+	storeConfig := osdconfig.ToStoreConfig(conf)
+	osdProps := osdProperties{
+		crushHostname: pvcName,
+		pvc: v1.PersistentVolumeClaimVolumeSource{
+			ClaimName: pvcName,
+		},
+	}
+
+	// start osds
+	for _, osd := range osds {
+		logger.Debugf("start osd %v", osd)
+		dp, err := c.makeDeployment(osdProps, osd)
+		if err != nil {
+			errMsg := fmt.Sprintf("failed to create deployment for pvc %s: %v", osdProps.crushHostname, err)
+			config.addError(errMsg)
+			continue
+		}
+
+		_, err = c.context.Clientset.AppsV1().Deployments(c.Namespace).Create(dp)
+		if err != nil {
+			if !errors.IsAlreadyExists(err) {
+				// we failed to create job, update the orchestration status for this pvc
+				logger.Warningf("failed to create osd deployment for pvc %s, osd %v: %+v", osdProps.crushHostname, osd, err)
+				continue
+			}
+			logger.Infof("deployment for osd %d already exists. updating if needed", osd.ID)
+			// Always invoke ceph version before an upgrade so we are sure to be up-to-date
+			daemon := string(opconfig.OsdType)
+			var cephVersionToUse cephver.CephVersion
+			currentCephVersion, err := client.LeastUptodateDaemonVersion(c.context, c.clusterInfo.Name, daemon)
+			if err != nil {
+				logger.Warningf("failed to retrieve current ceph %s version. %+v", daemon, err)
+				logger.Debug("could not detect ceph version during update, this is likely an initial bootstrap, proceeding with c.clusterInfo.CephVersion")
+				cephVersionToUse = c.clusterInfo.CephVersion
+			} else {
+				logger.Debugf("current cluster version for osds before upgrading is: %+v", currentCephVersion)
+				cephVersionToUse = currentCephVersion
+			}
+			if err = updateDeploymentAndWait(c.context, dp, c.Namespace, daemon, strconv.Itoa(osd.ID), cephVersionToUse); err != nil {
+				config.addError(fmt.Sprintf("failed to update osd deployment %d. %+v", osd.ID, err))
+			}
+		}
+		logger.Infof("started deployment for osd %d (dir=%t, type=%s)", osd.ID, osd.IsDirectory, storeConfig.StoreType)
+	}
+}
+
 func (c *Cluster) startOSDDaemonsOnNode(nodeName string, config *provisionConfig, configMap *v1.ConfigMap, status *OrchestrationStatus) {
 
 	osds := status.OSDs
@@ -302,14 +443,23 @@ func (c *Cluster) startOSDDaemonsOnNode(nodeName string, config *provisionConfig
 		config.addError("node %s did not resolve to start osds", nodeName)
 		return
 	}
-
 	storeConfig := osdconfig.ToStoreConfig(n.Config)
 	metadataDevice := osdconfig.MetadataDevice(n.Config)
 
+	osdProps := osdProperties{
+		crushHostname:  n.Name,
+		devices:        n.Devices,
+		selection:      n.Selection,
+		resources:      n.Resources,
+		storeConfig:    storeConfig,
+		metadataDevice: metadataDevice,
+		location:       n.Location,
+	}
+
 	// start osds
 	for _, osd := range osds {
 		logger.Debugf("start osd %v", osd)
-		dp, err := c.makeDeployment(n.Name, n.Selection, n.Resources, storeConfig, metadataDevice, n.Location, osd)
+		dp, err := c.makeDeployment(osdProps, osd)
 		if err != nil {
 			errMsg := fmt.Sprintf("failed to create deployment for node %s: %v", n.Name, err)
 			config.addError(errMsg)
@@ -349,6 +499,7 @@ func (c *Cluster) handleRemovedNodes(config *provisionConfig) {
 	removedNodes, err := c.findRemovedNodes()
 	if err != nil {
 		config.addError("failed to find removed nodes: %+v", err)
+		return
 	}
 	logger.Infof("processing %d removed nodes", len(removedNodes))
 
@@ -387,7 +538,7 @@ func (c *Cluster) handleRemovedNodes(config *provisionConfig) {
 			}
 		}
 
-		if err := c.updateNodeStatus(removedNode, OrchestrationStatus{Status: OrchestrationStatusCompleted}); err != nil {
+		if err := c.updateOSDStatus(removedNode, OrchestrationStatus{Status: OrchestrationStatusCompleted}); err != nil {
 			config.addError("failed to set orchestration starting status for removed node %s: %+v", removedNode, err)
 		}
 
@@ -403,20 +554,26 @@ func (c *Cluster) handleRemovedNodes(config *provisionConfig) {
 
 func (c *Cluster) cleanupRemovedNode(config *provisionConfig, nodeName, crushName string) {
 	// update the orchestration status of this removed node to the starting state
-	if err := c.updateNodeStatus(nodeName, OrchestrationStatus{Status: OrchestrationStatusStarting}); err != nil {
+	if err := c.updateOSDStatus(nodeName, OrchestrationStatus{Status: OrchestrationStatusStarting}); err != nil {
 		config.addError("failed to set orchestration starting status for removed node %s: %+v", nodeName, err)
 		return
 	}
 
 	// trigger orchestration on the removed node by telling it not to use any storage at all.  note that the directories are still passed in
 	// so that the pod will be able to mount them and migrate data from them.
-	job, err := c.makeJob(nodeName, []rookalpha.Device{}, rookalpha.Selection{DeviceFilter: "none"},
-		v1.ResourceRequirements{}, osdconfig.StoreConfig{}, "", "")
+	osdProps := osdProperties{
+		crushHostname: nodeName,
+		devices:       []rookalpha.Device{},
+		selection:     rookalpha.Selection{DeviceFilter: "none"},
+		resources:     v1.ResourceRequirements{},
+		storeConfig:   osdconfig.StoreConfig{},
+	}
+	job, err := c.makeJob(osdProps)
 	if err != nil {
 		message := fmt.Sprintf("failed to create prepare job node %s: %v", nodeName, err)
 		config.addError(message)
 		status := OrchestrationStatus{Status: OrchestrationStatusCompleted, Message: message}
-		if err := c.updateNodeStatus(nodeName, status); err != nil {
+		if err := c.updateOSDStatus(nodeName, status); err != nil {
 			config.addError("failed to update node %s status. %+v", nodeName, err)
 		}
 		return
@@ -424,7 +581,7 @@ func (c *Cluster) cleanupRemovedNode(config *provisionConfig, nodeName, crushNam
 
 	if !c.runJob(job, nodeName, config, "remove") {
 		status := OrchestrationStatus{Status: OrchestrationStatusCompleted, Message: fmt.Sprintf("failed to cleanup osd config on node %s", nodeName)}
-		if err := c.updateNodeStatus(nodeName, status); err != nil {
+		if err := c.updateOSDStatus(nodeName, status); err != nil {
 			config.addError("failed to update node %s status. %+v", nodeName, err)
 		}
 		return
@@ -458,7 +615,8 @@ func (c *Cluster) discoverStorageNodes() (map[string][]*apps.Deployment, error)
 		// get the node name from the node selector
 		nodeName, ok := osdPodSpec.NodeSelector[v1.LabelHostname]
 		if !ok || nodeName == "" {
-			return nil, fmt.Errorf("osd deployment %s doesn't have a node name on its node selector: %+v", osdDeployment.Name, osdPodSpec.NodeSelector)
+			logger.Debugf("skipping osd %s because osd deployment %s doesn't have a node name on its node selector: %+v", osdDeployment.Name, osdDeployment.Name, osdPodSpec.NodeSelector)
+			continue
 		}
 
 		if _, ok := discoveredNodes[nodeName]; !ok {

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

@@ -197,7 +197,7 @@ func TestAddRemoveNode(t *testing.T) {
 				return `{"stats":{"total_bytes":0,"total_used_bytes":0,"total_avail_bytes":3072}}`, nil
 			}
 			if args[0] == "pg" && args[1] == "dump" {
-				return `[]`, nil
+				return `{}`, nil
 			}
 			if args[0] == "auth" && args[1] == "del" {
 				assert.Equal(t, "osd.1", args[2])
@@ -257,17 +257,29 @@ func TestDiscoverOSDs(t *testing.T) {
 	node2 := "n2"
 
 	osd1 := OSDInfo{ID: 0, IsDirectory: true, IsFileStore: true, DataPath: "/rook/path"}
-	d1, err := c.makeDeployment(node1, rookalpha.Selection{}, v1.ResourceRequirements{}, config.StoreConfig{}, "", "", osd1)
+	osdProp := osdProperties{
+		crushHostname: node1,
+		selection:     rookalpha.Selection{},
+		resources:     v1.ResourceRequirements{},
+		storeConfig:   config.StoreConfig{},
+	}
+	d1, err := c.makeDeployment(osdProp, osd1)
 	assert.Nil(t, err)
 	assert.NotNil(t, d1)
 
 	osd2 := OSDInfo{ID: 101, IsDirectory: true, IsFileStore: true, DataPath: "/rook/path"}
-	d2, err := c.makeDeployment(node1, rookalpha.Selection{}, v1.ResourceRequirements{}, config.StoreConfig{}, "", "", osd2)
+	d2, err := c.makeDeployment(osdProp, osd2)
 	assert.Nil(t, err)
 	assert.NotNil(t, d2)
 
+	osdProp2 := osdProperties{
+		crushHostname: node2,
+		selection:     rookalpha.Selection{},
+		resources:     v1.ResourceRequirements{},
+		storeConfig:   config.StoreConfig{},
+	}
 	osd3 := OSDInfo{ID: 23, IsDirectory: true, IsFileStore: true, DataPath: "/rook/path"}
-	d3, err := c.makeDeployment(node2, rookalpha.Selection{}, v1.ResourceRequirements{}, config.StoreConfig{}, "", "", osd3)
+	d3, err := c.makeDeployment(osdProp2, osd3)
 	assert.Nil(t, err)
 	assert.NotNil(t, d3)
 

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

@@ -47,23 +47,35 @@ const (
 	osdsPerDeviceEnvVarName             = "ROOK_OSDS_PER_DEVICE"
 	encryptedDeviceEnvVarName           = "ROOK_ENCRYPTED_DEVICE"
 	osdMetadataDeviceEnvVarName         = "ROOK_METADATA_DEVICE"
+	pvcBackedOSDVarName                 = "ROOK_PVC_BACKED_OSD"
 	rookBinariesMountPath               = "/rook"
 	rookBinariesVolumeName              = "rook-binaries"
+	blockPVCMapperInitContainer         = "blkdevmapper"
 	osdMemoryTargetSafetyFactor float32 = 0.8
+	CephDeviceSetLabelKey               = "ceph.rook.io/DeviceSet"
+	CephSetIndexLabelKey                = "ceph.rook.io/setIndex"
+	CephDeviceSetPVCIDLabelKey          = "ceph.rook.io/DeviceSetPVCId"
+	OSDOverPVCLabelKey                  = "ceph.rook.io/pvc"
 )
 
-func (c *Cluster) makeJob(nodeName string, devices []rookalpha.Device,
-	selection rookalpha.Selection, resources v1.ResourceRequirements, storeConfig config.StoreConfig, metadataDevice, location string) (*batch.Job, error) {
+func (c *Cluster) makeJob(osdProps osdProperties) (*batch.Job, error) {
 
-	podSpec, err := c.provisionPodTemplateSpec(devices, selection, resources, storeConfig, metadataDevice, nodeName, location, v1.RestartPolicyOnFailure)
+	podSpec, err := c.provisionPodTemplateSpec(osdProps, v1.RestartPolicyOnFailure)
 	if err != nil {
 		return nil, err
 	}
-	podSpec.Spec.NodeSelector = map[string]string{v1.LabelHostname: nodeName}
+
+	if osdProps.pvc.ClaimName == "" {
+		podSpec.Spec.NodeSelector = map[string]string{v1.LabelHostname: osdProps.crushHostname}
+	} else {
+		podSpec.Spec.InitContainers = []v1.Container{
+			c.getPVCInitContainer(osdProps.pvc),
+		}
+	}
 
 	job := &batch.Job{
 		ObjectMeta: metav1.ObjectMeta{
-			Name:      k8sutil.TruncateNodeName(prepareAppNameFmt, nodeName),
+			Name:      k8sutil.TruncateNodeName(prepareAppNameFmt, osdProps.crushHostname),
 			Namespace: c.Namespace,
 			Labels: map[string]string{
 				k8sutil.AppAttr:     prepareAppName,
@@ -74,14 +86,18 @@ func (c *Cluster) makeJob(nodeName string, devices []rookalpha.Device,
 			Template: *podSpec,
 		},
 	}
+
+	if len(osdProps.pvc.ClaimName) > 0 {
+		k8sutil.AddLabelToJob(OSDOverPVCLabelKey, osdProps.pvc.ClaimName, job)
+	}
+
 	k8sutil.AddRookVersionLabelToJob(job)
 	opspec.AddCephVersionLabelToJob(c.clusterInfo.CephVersion, job)
 	k8sutil.SetOwnerRef(&job.ObjectMeta, &c.ownerRef)
 	return job, nil
 }
 
-func (c *Cluster) makeDeployment(nodeName string, selection rookalpha.Selection, resources v1.ResourceRequirements,
-	storeConfig config.StoreConfig, metadataDevice, location string, osd OSDInfo) (*apps.Deployment, error) {
+func (c *Cluster) makeDeployment(osdProps osdProperties, osd OSDInfo) (*apps.Deployment, error) {
 
 	replicaCount := int32(1)
 	volumeMounts := opspec.CephVolumeMounts()
@@ -113,6 +129,11 @@ func (c *Cluster) makeDeployment(nodeName string, selection rookalpha.Selection,
 		volumeMounts = append(volumeMounts, devMount)
 	}
 
+	if osdProps.pvc.ClaimName != "" {
+		// Create volume config for PVCs
+		volumes = append(volumes, getPVCOSDVolumes(&osdProps)...)
+	}
+
 	if len(volumes) == 0 {
 		return nil, fmt.Errorf("empty volumes")
 	}
@@ -125,7 +146,7 @@ func (c *Cluster) makeDeployment(nodeName string, selection rookalpha.Selection,
 	osdID := strconv.Itoa(osd.ID)
 	tiniEnvVar := v1.EnvVar{Name: "TINI_SUBREAPER", Value: ""}
 	envVars := []v1.EnvVar{
-		nodeNameEnvVar(nodeName),
+		nodeNameEnvVar(osdProps.crushHostname),
 		k8sutil.PodIPEnvVar(k8sutil.PrivateIPEnvVar),
 		k8sutil.PodIPEnvVar(k8sutil.PublicIPEnvVar),
 		tiniEnvVar,
@@ -136,7 +157,7 @@ func (c *Cluster) makeDeployment(nodeName string, selection rookalpha.Selection,
 		{Name: "ROOK_OSD_ID", Value: osdID},
 		{Name: "ROOK_OSD_STORE_TYPE", Value: storeType},
 	}...)
-	configEnvVars := append(c.getConfigEnvVars(storeConfig, dataDir, nodeName, location), []v1.EnvVar{
+	configEnvVars := append(c.getConfigEnvVars(osdProps.storeConfig, dataDir, osdProps.crushHostname, osdProps.location), []v1.EnvVar{
 		tiniEnvVar,
 		{Name: "ROOK_OSD_ID", Value: osdID},
 		{Name: "ROOK_CEPH_VERSION", Value: c.clusterInfo.CephVersion.CephVersionFormatted()},
@@ -249,6 +270,11 @@ func (c *Cluster) makeDeployment(nodeName string, selection rookalpha.Selection,
 		args = commonArgs
 	}
 
+	if osdProps.pvc.ClaimName != "" {
+		volumeMounts = append(volumeMounts, getPvcOSDBridgeMount(osdProps.pvc.ClaimName))
+		envVars = append(envVars, pvcBackedOSDEnvVar("true"))
+	}
+
 	privileged := true
 	runAsUser := int64(0)
 	readOnlyRootFilesystem := false
@@ -259,7 +285,7 @@ func (c *Cluster) makeDeployment(nodeName string, selection rookalpha.Selection,
 	}
 
 	// needed for luksOpen synchronization when devices are encrypted
-	hostIPC := storeConfig.EncryptedDevice
+	hostIPC := osdProps.storeConfig.EncryptedDevice
 
 	DNSPolicy := v1.DNSClusterFirst
 	if c.HostNetwork {
@@ -297,7 +323,6 @@ func (c *Cluster) makeDeployment(nodeName string, selection rookalpha.Selection,
 					},
 				},
 				Spec: v1.PodSpec{
-					NodeSelector:       map[string]string{v1.LabelHostname: nodeName},
 					RestartPolicy:      v1.RestartPolicyAlways,
 					ServiceAccountName: serviceAccountName,
 					HostNetwork:        c.HostNetwork,
@@ -323,7 +348,7 @@ func (c *Cluster) makeDeployment(nodeName string, selection rookalpha.Selection,
 							Image:           c.cephVersion.Image,
 							VolumeMounts:    volumeMounts,
 							Env:             envVars,
-							Resources:       resources,
+							Resources:       osdProps.resources,
 							SecurityContext: securityContext,
 							Lifecycle:       opspec.PodLifeCycle(osd.DataPath),
 						},
@@ -334,6 +359,12 @@ func (c *Cluster) makeDeployment(nodeName string, selection rookalpha.Selection,
 			Replicas: &replicaCount,
 		},
 	}
+	if osdProps.pvc.ClaimName == "" {
+		deployment.Spec.Template.Spec.NodeSelector = map[string]string{v1.LabelHostname: osdProps.crushHostname}
+	} else {
+		deployment.Spec.Template.Spec.InitContainers = append(deployment.Spec.Template.Spec.InitContainers, c.getPVCInitContainer(osdProps.pvc))
+		k8sutil.AddLabelToDeployement(OSDOverPVCLabelKey, osdProps.pvc.ClaimName, deployment)
+	}
 	k8sutil.AddRookVersionLabelToDeployment(deployment)
 	c.annotations.ApplyToObjectMeta(&deployment.ObjectMeta)
 	c.annotations.ApplyToObjectMeta(&deployment.Spec.Template.ObjectMeta)
@@ -361,15 +392,14 @@ func (c *Cluster) getCopyBinariesContainer() (v1.Volume, *v1.Container) {
 	}
 }
 
-func (c *Cluster) provisionPodTemplateSpec(devices []rookalpha.Device, selection rookalpha.Selection, resources v1.ResourceRequirements,
-	storeConfig config.StoreConfig, metadataDevice, nodeName, location string, restart v1.RestartPolicy) (*v1.PodTemplateSpec, error) {
+func (c *Cluster) provisionPodTemplateSpec(osdProps osdProperties, restart v1.RestartPolicy) (*v1.PodTemplateSpec, error) {
 
 	copyBinariesVolume, copyBinariesContainer := c.getCopyBinariesContainer()
 
 	volumes := append(opspec.PodVolumes(c.dataDirHostPath, c.Namespace), copyBinariesVolume)
 
 	// by default, don't define any volume config unless it is required
-	if len(devices) > 0 || selection.DeviceFilter != "" || selection.GetUseAllDevices() || metadataDevice != "" {
+	if len(osdProps.devices) > 0 || osdProps.selection.DeviceFilter != "" || osdProps.selection.GetUseAllDevices() || osdProps.metadataDevice != "" {
 		// create volume config for the data dir and /dev so the pod can access devices on the host
 		devVolume := v1.Volume{Name: "devices", VolumeSource: v1.VolumeSource{HostPath: &v1.HostPathVolumeSource{Path: "/dev"}}}
 		volumes = append(volumes, devVolume)
@@ -377,8 +407,13 @@ func (c *Cluster) provisionPodTemplateSpec(devices []rookalpha.Device, selection
 		volumes = append(volumes, udevVolume)
 	}
 
+	if osdProps.pvc.ClaimName != "" {
+		// Create volume config for PVCs
+		volumes = append(volumes, getPVCOSDVolumes(&osdProps)...)
+	}
+
 	// add each OSD directory as another host path volume source
-	for _, d := range selection.Directories {
+	for _, d := range osdProps.selection.Directories {
 		if c.skipVolumeForDirectory(d.Path) {
 			// the dataDirHostPath has already been added as a volume
 			continue
@@ -398,7 +433,7 @@ func (c *Cluster) provisionPodTemplateSpec(devices []rookalpha.Device, selection
 		ServiceAccountName: serviceAccountName,
 		Containers: []v1.Container{
 			*copyBinariesContainer,
-			c.provisionOSDContainer(devices, selection, resources, storeConfig, metadataDevice, nodeName, location, copyBinariesContainer.VolumeMounts[0]),
+			c.provisionOSDContainer(osdProps, copyBinariesContainer.VolumeMounts[0]),
 		},
 		RestartPolicy: restart,
 		Volumes:       volumes,
@@ -422,7 +457,7 @@ func (c *Cluster) provisionPodTemplateSpec(devices []rookalpha.Device, selection
 
 	// ceph-volume --dmcrypt uses cryptsetup that synchronizes with udev on
 	// host through semaphore
-	podSpec.HostIPC = storeConfig.EncryptedDevice
+	podSpec.HostIPC = osdProps.storeConfig.EncryptedDevice
 
 	return &v1.PodTemplateSpec{
 		ObjectMeta: podMeta,
@@ -430,6 +465,30 @@ func (c *Cluster) provisionPodTemplateSpec(devices []rookalpha.Device, selection
 	}, nil
 }
 
+// Currently we can't mount a block mode pv directly to a priviliged container
+// So we mount it to a non priviliged init container and then copy it to a common directory mounted inside init container
+// and the privileged provision container.
+func (c *Cluster) getPVCInitContainer(pvc v1.PersistentVolumeClaimVolumeSource) v1.Container {
+	return v1.Container{
+		Name:  blockPVCMapperInitContainer,
+		Image: c.cephVersion.Image,
+		Args:  []string{"cp", "-a", fmt.Sprintf("/%s", pvc.ClaimName), fmt.Sprintf("/mnt/%s", pvc.ClaimName)},
+		VolumeDevices: []v1.VolumeDevice{
+			{
+				Name:       pvc.ClaimName,
+				DevicePath: fmt.Sprintf("/%s", pvc.ClaimName),
+			},
+		},
+		VolumeMounts: []v1.VolumeMount{
+			{
+				MountPath: "/mnt",
+				Name:      fmt.Sprintf("%s-bridge", pvc.ClaimName),
+			},
+		},
+		SecurityContext: opmon.PodSecurityContext(),
+	}
+}
+
 func (c *Cluster) getConfigEnvVars(storeConfig config.StoreConfig, dataDir, nodeName, location string) []v1.EnvVar {
 	envVars := []v1.EnvVar{
 		nodeNameEnvVar(nodeName),
@@ -481,41 +540,40 @@ func (c *Cluster) getConfigEnvVars(storeConfig config.StoreConfig, dataDir, node
 	return envVars
 }
 
-func (c *Cluster) provisionOSDContainer(devices []rookalpha.Device, selection rookalpha.Selection, resources v1.ResourceRequirements,
-	storeConfig config.StoreConfig, metadataDevice, nodeName, location string, copyBinariesMount v1.VolumeMount) v1.Container {
+func (c *Cluster) provisionOSDContainer(osdProps osdProperties, copyBinariesMount v1.VolumeMount) v1.Container {
 
-	envVars := c.getConfigEnvVars(storeConfig, k8sutil.DataDir, nodeName, location)
+	envVars := c.getConfigEnvVars(osdProps.storeConfig, k8sutil.DataDir, osdProps.crushHostname, osdProps.location)
 	devMountNeeded := false
 	privileged := false
 
 	// only 1 of device list, device filter and use all devices can be specified.  We prioritize in that order.
-	if len(devices) > 0 {
-		deviceNames := make([]string, len(devices))
-		for i, device := range devices {
+	if len(osdProps.devices) > 0 {
+		deviceNames := make([]string, len(osdProps.devices))
+		for i, device := range osdProps.devices {
 			devSuffix := ""
 			count := "1"
 			if count, ok := device.Config[config.OSDsPerDeviceKey]; ok {
-				logger.Infof("%s osds requested on device %s (node %s)", count, device.Name, nodeName)
+				logger.Infof("%s osds requested on device %s (node %s)", count, device.Name, osdProps.crushHostname)
 			}
 			devSuffix += ":" + count
 			if md, ok := device.Config[config.MetadataDeviceKey]; ok {
-				logger.Infof("osd %s requested with metadataDevice %s (node %s)", device.Name, md, nodeName)
+				logger.Infof("osd %s requested with metadataDevice %s (node %s)", device.Name, md, osdProps.crushHostname)
 				devSuffix += ":" + md
 			}
 			deviceNames[i] = device.Name + devSuffix
 		}
 		envVars = append(envVars, dataDevicesEnvVar(strings.Join(deviceNames, ",")))
 		devMountNeeded = true
-	} else if selection.DeviceFilter != "" {
-		envVars = append(envVars, deviceFilterEnvVar(selection.DeviceFilter))
+	} else if osdProps.selection.DeviceFilter != "" {
+		envVars = append(envVars, deviceFilterEnvVar(osdProps.selection.DeviceFilter))
 		devMountNeeded = true
-	} else if selection.GetUseAllDevices() {
+	} else if osdProps.selection.GetUseAllDevices() {
 		envVars = append(envVars, deviceFilterEnvVar("all"))
 		devMountNeeded = true
 	}
 
-	if metadataDevice != "" {
-		envVars = append(envVars, metadataDeviceEnvVar(metadataDevice))
+	if osdProps.metadataDevice != "" {
+		envVars = append(envVars, metadataDeviceEnvVar(osdProps.metadataDevice))
 		devMountNeeded = true
 	}
 
@@ -527,11 +585,17 @@ func (c *Cluster) provisionOSDContainer(devices []rookalpha.Device, selection ro
 		volumeMounts = append(volumeMounts, udevMount)
 	}
 
-	if len(selection.Directories) > 0 {
+	if osdProps.pvc.ClaimName != "" {
+		volumeMounts = append(volumeMounts, getPvcOSDBridgeMount(osdProps.pvc.ClaimName))
+		envVars = append(envVars, dataDevicesEnvVar(strings.Join([]string{fmt.Sprintf("/mnt/%s", osdProps.pvc.ClaimName)}, ",")))
+		envVars = append(envVars, pvcBackedOSDEnvVar("true"))
+	}
+
+	if len(osdProps.selection.Directories) > 0 {
 		// for each directory the user has specified, create a volume mount and pass it to the pod via cmd line arg
-		dirPaths := make([]string, len(selection.Directories))
-		for i := range selection.Directories {
-			dpath := selection.Directories[i].Path
+		dirPaths := make([]string, len(osdProps.selection.Directories))
+		for i := range osdProps.selection.Directories {
+			dpath := osdProps.selection.Directories[i].Path
 			dirPaths[i] = dpath
 			if c.skipVolumeForDirectory(dpath) {
 				// the dataDirHostPath has already been added as a volume mount
@@ -540,20 +604,20 @@ func (c *Cluster) provisionOSDContainer(devices []rookalpha.Device, selection ro
 			volumeMounts = append(volumeMounts, v1.VolumeMount{Name: k8sutil.PathToVolumeName(dpath), MountPath: dpath})
 		}
 
-		if !IsRemovingNode(selection.DeviceFilter) {
+		if !IsRemovingNode(osdProps.selection.DeviceFilter) {
 			envVars = append(envVars, dataDirectoriesEnvVar(strings.Join(dirPaths, ",")))
 		}
 	}
 
 	// elevate to be privileged if it is going to mount devices or if running in a restricted environment such as openshift
-	if devMountNeeded || os.Getenv("ROOK_HOSTPATH_REQUIRES_PRIVILEGED") == "true" {
+	if devMountNeeded || os.Getenv("ROOK_HOSTPATH_REQUIRES_PRIVILEGED") == "true" || osdProps.pvc.ClaimName != "" {
 		privileged = true
 	}
 	runAsUser := int64(0)
 	runAsNonRoot := false
 	readOnlyRootFilesystem := false
 
-	return v1.Container{
+	osdProvisionContainer := v1.Container{
 		Command:      []string{path.Join(rookBinariesMountPath, "tini")},
 		Args:         []string{"--", path.Join(rookBinariesMountPath, "rook"), "ceph", "osd", "provision"},
 		Name:         "provision",
@@ -566,8 +630,14 @@ func (c *Cluster) provisionOSDContainer(devices []rookalpha.Device, selection ro
 			RunAsNonRoot:           &runAsNonRoot,
 			ReadOnlyRootFilesystem: &readOnlyRootFilesystem,
 		},
-		Resources: resources,
+		Resources: osdProps.resources,
 	}
+
+	return osdProvisionContainer
+}
+
+func getPvcOSDBridgeMount(claimName string) v1.VolumeMount {
+	return v1.VolumeMount{Name: fmt.Sprintf("%s-bridge", claimName), MountPath: "/mnt"}
 }
 
 func (c *Cluster) skipVolumeForDirectory(path string) bool {
@@ -576,6 +646,28 @@ func (c *Cluster) skipVolumeForDirectory(path string) bool {
 	return path == k8sutil.DataDir
 }
 
+func getPVCOSDVolumes(osdProps *osdProperties) []v1.Volume {
+	return []v1.Volume{
+		{
+			Name: osdProps.crushHostname,
+			VolumeSource: v1.VolumeSource{
+				PersistentVolumeClaim: &osdProps.pvc,
+			},
+		},
+		{
+			// We need a bridge mount which is basically a common volume mount between the non priviliged init container
+			// and the privileged provision container or osd daemon container
+			// The reason for this is mentioned in the comment for getPVCInitContainer() method
+			Name: fmt.Sprintf("%s-bridge", osdProps.pvc.ClaimName),
+			VolumeSource: v1.VolumeSource{
+				EmptyDir: &v1.EmptyDirVolumeSource{
+					Medium: "Memory",
+				},
+			},
+		},
+	}
+}
+
 func nodeNameEnvVar(name string) v1.EnvVar {
 	return v1.EnvVar{Name: "ROOK_NODE_NAME", Value: name}
 }
@@ -596,6 +688,10 @@ func dataDirectoriesEnvVar(dataDirectories string) v1.EnvVar {
 	return v1.EnvVar{Name: dataDirsEnvVarName, Value: dataDirectories}
 }
 
+func pvcBackedOSDEnvVar(pvcBacked string) v1.EnvVar {
+	return v1.EnvVar{Name: pvcBackedOSDVarName, Value: pvcBacked}
+}
+
 func getDirectoriesFromContainer(osdContainer v1.Container) []rookalpha.Directory {
 	var dirsArg string
 	for _, envVar := range osdContainer.Env {
@@ -650,3 +746,16 @@ func osdOnSDNFlag(hostnetwork bool, v cephver.CephVersion) []string {
 
 	return args
 }
+
+func makeStorageClassDeviceSetPVCID(storageClassDeviceSetName string, setIndex, pvcIndex int) (pvcId, pvcLabelSelector string) {
+	pvcStorageClassDeviceSetPVCId := fmt.Sprintf("%s-%v", storageClassDeviceSetName, setIndex)
+	return pvcStorageClassDeviceSetPVCId, fmt.Sprintf("%s=%s", CephDeviceSetPVCIDLabelKey, pvcStorageClassDeviceSetPVCId)
+}
+
+func makeStorageClassDeviceSetPVCLabel(storageClassDeviceSetName, pvcStorageClassDeviceSetPVCId string, pvcIndex, setIndex int) map[string]string {
+	return map[string]string{
+		CephDeviceSetLabelKey:      storageClassDeviceSetName,
+		CephSetIndexLabelKey:       fmt.Sprintf("%v", setIndex),
+		CephDeviceSetPVCIDLabelKey: pvcStorageClassDeviceSetPVCId,
+	}
+}

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

@@ -28,17 +28,24 @@ import (
 	cephver "github.com/rook/rook/pkg/operator/ceph/version"
 	"github.com/rook/rook/pkg/operator/k8sutil"
 	exectest "github.com/rook/rook/pkg/util/exec/test"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes/fake"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func TestPodContainer(t *testing.T) {
 	cluster := &Cluster{Namespace: "myosd", rookVersion: "23", cephVersion: cephv1.CephVersionSpec{}}
-	c, err := cluster.provisionPodTemplateSpec([]rookalpha.Device{}, rookalpha.Selection{}, v1.ResourceRequirements{}, config.StoreConfig{}, "", "node", "", v1.RestartPolicyAlways)
+	osdProps := osdProperties{
+		crushHostname: "node",
+		devices:       []rookalpha.Device{},
+		resources:     v1.ResourceRequirements{},
+		storeConfig:   config.StoreConfig{},
+	}
+	c, err := cluster.provisionPodTemplateSpec(osdProps, v1.RestartPolicyAlways)
 	assert.NotNil(t, c)
 	assert.Nil(t, err)
 	assert.Equal(t, 2, len(c.Spec.Containers))
@@ -88,7 +95,14 @@ func testPodDevices(t *testing.T, dataDir, deviceName string, allDevices bool) {
 		ID: 0,
 	}
 
-	deployment, err := c.makeDeployment(n.Name, n.Selection, v1.ResourceRequirements{}, config.StoreConfig{}, "", n.Location, osd)
+	osdProp := osdProperties{
+		crushHostname: n.Name,
+		selection:     n.Selection,
+		resources:     v1.ResourceRequirements{},
+		storeConfig:   config.StoreConfig{},
+		location:      n.Location,
+	}
+	deployment, err := c.makeDeployment(osdProp, osd)
 	assert.Nil(t, err)
 	assert.NotNil(t, deployment)
 	assert.Equal(t, "rook-ceph-osd-0", deployment.Name)
@@ -170,7 +184,16 @@ func TestStorageSpecDevicesAndDirectories(t *testing.T) {
 		IsDirectory: true,
 		DataPath:    "/my/root/path/osd1",
 	}
-	deployment, err := c.makeDeployment(n.Name, n.Selection, v1.ResourceRequirements{}, config.StoreConfig{}, "", n.Location, osd)
+
+	osdProp := osdProperties{
+		crushHostname: n.Name,
+		selection:     n.Selection,
+		resources:     v1.ResourceRequirements{},
+		storeConfig:   config.StoreConfig{},
+		location:      n.Location,
+	}
+
+	deployment, err := c.makeDeployment(osdProp, osd)
 	assert.NotNil(t, deployment)
 	assert.Nil(t, err)
 	// pod spec should have a volume for the given dir in the main container and the init container
@@ -199,7 +222,7 @@ func TestStorageSpecDevicesAndDirectories(t *testing.T) {
 		IsDirectory: true,
 		DataPath:    "/var/lib/rook/osd1",
 	}
-	deployment, err = c.makeDeployment(n.Name, n.Selection, v1.ResourceRequirements{}, config.StoreConfig{}, "", n.Location, osd)
+	deployment, err = c.makeDeployment(osdProp, osd)
 	assert.NotNil(t, deployment)
 	assert.Nil(t, err)
 	// pod spec should have a volume for the given dir in the main container and the init container
@@ -263,7 +286,16 @@ func TestStorageSpecConfig(t *testing.T) {
 	storeConfig := config.ToStoreConfig(storageSpec.Nodes[0].Config)
 	metadataDevice := config.MetadataDevice(storageSpec.Nodes[0].Config)
 
-	job, err := c.makeJob(n.Name, n.Devices, n.Selection, c.DesiredStorage.Nodes[0].Resources, storeConfig, metadataDevice, n.Location)
+	osdProp := osdProperties{
+		crushHostname:  n.Name,
+		devices:        n.Devices,
+		selection:      n.Selection,
+		resources:      c.DesiredStorage.Nodes[0].Resources,
+		storeConfig:    storeConfig,
+		metadataDevice: metadataDevice,
+		location:       n.Location,
+	}
+	job, err := c.makeJob(osdProp)
 	assert.NotNil(t, job)
 	assert.Nil(t, err)
 	assert.Equal(t, "rook-ceph-osd-prepare-node1", job.ObjectMeta.Name)
@@ -317,7 +349,17 @@ func TestHostNetwork(t *testing.T) {
 	osd := OSDInfo{
 		ID: 0,
 	}
-	r, err := c.makeDeployment(n.Name, n.Selection, v1.ResourceRequirements{}, config.StoreConfig{}, "", n.Location, osd)
+
+	osdProp := osdProperties{
+		crushHostname: n.Name,
+		devices:       n.Devices,
+		selection:     n.Selection,
+		resources:     c.DesiredStorage.Nodes[0].Resources,
+		storeConfig:   config.StoreConfig{},
+		location:      n.Location,
+	}
+
+	r, err := c.makeDeployment(osdProp, osd)
 	assert.NotNil(t, r)
 	assert.Nil(t, err)
 

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

@@ -58,7 +58,7 @@ func (c *provisionConfig) addError(message string, args ...interface{}) {
 	c.errorMessages = append(c.errorMessages, fmt.Sprintf(message, args...))
 }
 
-func (c *Cluster) updateNodeStatus(node string, status OrchestrationStatus) error {
+func (c *Cluster) updateOSDStatus(node string, status OrchestrationStatus) error {
 	return UpdateNodeStatus(c.kv, node, status)
 }
 
@@ -85,7 +85,7 @@ func UpdateNodeStatus(kv *k8sutil.ConfigMapKVStore, node string, status Orchestr
 func (c *Cluster) handleOrchestrationFailure(config *provisionConfig, nodeName, message string) {
 	config.addError(message)
 	status := OrchestrationStatus{Status: OrchestrationStatusFailed, Message: message}
-	if err := c.updateNodeStatus(nodeName, status); err != nil {
+	if err := c.updateOSDStatus(nodeName, status); err != nil {
 		config.addError("failed to update status for node %s. %+v", nodeName, err)
 	}
 }
@@ -253,10 +253,15 @@ func (c *Cluster) handleStatusConfigMapStatus(nodeName string, config *provision
 	logger.Infof("osd orchestration status for node %s is %s", nodeName, status.Status)
 	if status.Status == OrchestrationStatusCompleted {
 		if configOSDs {
-			c.startOSDDaemonsOnNode(nodeName, config, configMap, status)
+			if status.PvcBackedOSD {
+				c.startOSDDaemonsOnPVC(nodeName, config, configMap, status)
+			} else {
+				c.startOSDDaemonsOnNode(nodeName, config, configMap, status)
+			}
+			// remove the status configmap that indicated the progress
+			c.kv.ClearStore(fmt.Sprintf(orchestrationStatusMapName, nodeName))
 		}
-		// remove the status configmap that indicated the progress
-		c.kv.ClearStore(fmt.Sprintf(orchestrationStatusMapName, nodeName))
+
 		return true
 	}
 

pkg/operator/discover/discover_test.go

@@ -29,7 +29,7 @@ import (
 
 	"github.com/stretchr/testify/assert"
 
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
@@ -69,7 +69,7 @@ func TestStartDiscoveryDaemonset(t *testing.T) {
 
 func TestGetAvailableDevices(t *testing.T) {
 	clientset := test.New(3)
-
+	pvcBackedOSD := false
 	ns := "rook-system"
 	nodeName := "node123"
 	os.Setenv(k8sutil.PodNamespaceEnvVar, ns)
@@ -109,11 +109,11 @@ func TestGetAvailableDevices(t *testing.T) {
 	assert.Nil(t, err)
 	assert.Equal(t, 1, len(nodeDevices))
 
-	devices, err := GetAvailableDevices(context, nodeName, ns, d, "^sd.", false)
+	devices, err := GetAvailableDevices(context, nodeName, ns, d, "^sd.", pvcBackedOSD)
 	assert.Nil(t, err)
 	assert.Equal(t, 1, len(devices))
 	// devices should be in use now, 2nd try gets the same list
-	devices, err = GetAvailableDevices(context, nodeName, ns, d, "^sd.", false)
+	devices, err = GetAvailableDevices(context, nodeName, ns, d, "^sd.", pvcBackedOSD)
 	assert.Nil(t, err)
 	assert.Equal(t, 1, len(devices))
 }

pkg/operator/k8sutil/deployment.go

@@ -24,6 +24,7 @@ import (
 	"github.com/rook/rook/pkg/util"
 	apps "k8s.io/api/apps/v1"
 	v1 "k8s.io/api/apps/v1"
+	batchv1 "k8s.io/api/batch/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
 )
@@ -188,3 +189,27 @@ func AddRookVersionLabelToObjectMeta(meta *metav1.ObjectMeta) {
 	}
 	addRookVersionLabel(meta.Labels)
 }
+
+func AddLabelToDeployement(key, value string, d *v1.Deployment) {
+	if d == nil {
+		return
+	}
+	if d.Labels == nil {
+		d.Labels = map[string]string{}
+	}
+	addLabel(key, value, d.Labels)
+}
+
+func AddLabelToJob(key, value string, b *batchv1.Job) {
+	if b == nil {
+		return
+	}
+	if b.Labels == nil {
+		b.Labels = map[string]string{}
+	}
+	addLabel(key, value, b.Labels)
+}
+
+func addLabel(key, value string, labels map[string]string) {
+	labels[key] = value
+}

pkg/operator/k8sutil/node.go

@@ -88,6 +88,10 @@ func GetValidNodes(rookStorage rookalpha.StorageScopeSpec, clientset kubernetes.
 	return RookNodesMatchingKubernetesNodes(rookStorage, validK8sNodes)
 }
 
+func generateUniqueVolumeSourceName(sourceName, pvcName string) string {
+	return fmt.Sprintf("%s-%s", sourceName, pvcName)
+}
+
 // GetNodeNameFromHostname returns the name of the node resource looked up by the hostname label
 // Typically these will be the same name, but sometimes they are not such as when nodes have a longer
 // dns name, but the hostname is short.

pkg/operator/k8sutil/volume.go

@@ -22,7 +22,7 @@ import (
 	"os"
 	"strings"
 
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/util/validation"
 )
 

pkg/util/sys/device.go

@@ -92,9 +92,19 @@ func ListDevices(executor exec.Executor) ([]string, error) {
 }
 
 func GetDevicePartitions(device string, executor exec.Executor) (partitions []Partition, unusedSpace uint64, err error) {
-	cmd := fmt.Sprintf("lsblk /dev/%s", device)
-	output, err := executor.ExecuteCommandWithOutput(false, cmd, "lsblk", fmt.Sprintf("/dev/%s", device),
+
+	var devicePath string
+	splitDevicePath := strings.Split(device, "/")
+	if len(splitDevicePath) == 1 {
+		devicePath = fmt.Sprintf("/dev/%s", device) //device path for OSD on devices.
+	} else {
+		devicePath = device //use the exact device path (like /mnt/<pvc-name>) in case of PVC block device
+	}
+
+	cmd := fmt.Sprintf("lsblk %s", devicePath)
+	output, err := executor.ExecuteCommandWithOutput(false, cmd, "lsblk", devicePath,
 		"--bytes", "--pairs", "--output", "NAME,SIZE,TYPE,PKNAME")
+	logger.Infof("Output: %+v", output)
 	if err != nil {
 		return nil, 0, fmt.Errorf("failed to get device %s partitions. %+v", device, err)
 	}
@@ -106,6 +116,7 @@ func GetDevicePartitions(device string, executor exec.Executor) (partitions []Pa
 		name := props["NAME"]
 		if name == device {
 			// found the main device
+			logger.Info("Device found - ", name)
 			deviceSize, err = strconv.ParseUint(props["SIZE"], 10, 64)
 			if err != nil {
 				return nil, 0, fmt.Errorf("failed to get device %s size. %+v", device, err)
@@ -144,7 +155,14 @@ func GetDevicePartitions(device string, executor exec.Executor) (partitions []Pa
 }
 
 func GetDeviceProperties(device string, executor exec.Executor) (map[string]string, error) {
-	return GetDevicePropertiesFromPath(fmt.Sprintf("/dev/%s", device), executor)
+	// As we are mounting the block mode PVs on /mnt we use the entire path,
+	// e.g., if the device path is /mnt/example-pvc then its taken completely
+	// else if its just vdb then the following is used
+	devicePath := strings.Split(device, "/")
+	if len(devicePath) == 1 {
+		device = fmt.Sprintf("/dev/%s", device)
+	}
+	return GetDevicePropertiesFromPath(device, executor)
 }
 
 func GetDevicePropertiesFromPath(devicePath string, executor exec.Executor) (map[string]string, error) {
@@ -178,8 +196,12 @@ func GetUdevInfo(device string, executor exec.Executor) (map[string]string, erro
 
 // get the file systems available
 func GetDeviceFilesystems(device string, executor exec.Executor) (string, error) {
+	devicePath := strings.Split(device, "/")
+	if len(devicePath) == 1 {
+		device = fmt.Sprintf("/dev/%s", device)
+	}
 	cmd := fmt.Sprintf("get filesystem type for %s", device)
-	output, err := executor.ExecuteCommandWithOutput(false, cmd, "udevadm", "info", "--query=property", fmt.Sprintf("/dev/%s", device))
+	output, err := executor.ExecuteCommandWithOutput(false, cmd, "udevadm", "info", "--query=property", device)
 	if err != nil {
 		return "", fmt.Errorf("command %s failed: %+v", cmd, err)
 	}
@@ -224,9 +246,15 @@ func GetDiskUUID(device string, executor exec.Executor) (string, error) {
 		return "sgdiskNotFound", nil
 	}
 
+	devicePath := strings.Split(device, "/")
+	if len(devicePath) == 1 {
+		device = fmt.Sprintf("/dev/%s", device)
+	}
+
 	cmd := fmt.Sprintf("get disk %s uuid", device)
+
 	output, err := executor.ExecuteCommandWithOutput(false, cmd,
-		sgdisk, "--print", fmt.Sprintf("/dev/%s", device))
+		sgdisk, "--print", device)
 	if err != nil {
 		return "", err
 	}
@@ -295,7 +323,7 @@ func UnmountDevice(devicePath string, executor exec.Executor) error {
 // the number of partitions, whether Rook has created partitions on the device in the past
 // possibly from the same or a previous cluster, the filesystem found, or an err if failed
 // to retrieve the properties.
-func CheckIfDeviceAvailable(executor exec.Executor, name string) (int, bool, string, error) {
+func CheckIfDeviceAvailable(executor exec.Executor, name string, pvcBacked bool) (int, bool, string, error) {
 	ownPartitions := true
 	partitions, _, err := GetDevicePartitions(name, executor)
 	if err != nil {
@@ -306,12 +334,16 @@ func CheckIfDeviceAvailable(executor exec.Executor, name string) (int, bool, str
 		ownPartitions = false
 	}
 
-	// check if there is a file system on the device
-	devFS, err := GetDeviceFilesystems(name, executor)
-	if err != nil {
-		return 0, false, "", fmt.Errorf("failed to get device %s filesystem: %+v", name, err)
+	var devFS string
+	if !pvcBacked {
+		// check if there is a file system on the device
+		devFS, err = GetDeviceFilesystems(name, executor)
+		if err != nil {
+			return 0, false, "", fmt.Errorf("failed to get device %s filesystem: %+v", name, err)
+		}
+	} else {
+		devFS = "" //Not checking Filesystem in case of PVC block device.
 	}
-
 	return partCount, ownPartitions, devFS, nil
 }
 

pkg/util/sys/device_test.go

@@ -144,7 +144,7 @@ func TestGetPartitions(t *testing.T) {
 	executor := &exectest.MockExecutor{
 		MockExecuteCommandWithOutput: func(debug bool, actionName string, command string, arg ...string) (string, error) {
 			run++
-			logger.Infof("run %d action %s", run, actionName)
+			logger.Infof("run %d action %s command %s", run, actionName, command)
 			switch {
 			case run == 1:
 				return `NAME="sdc" SIZE="100000" TYPE="disk" PKNAME=""`, nil

tests/framework/installer/device.go

@@ -40,7 +40,8 @@ func IsAdditionalDeviceAvailableOnCluster() bool {
 			continue
 		}
 
-		_, ownPartitions, fs, err := sys.CheckIfDeviceAvailable(executor, device)
+		pvcBackedOSD := false
+		_, ownPartitions, fs, err := sys.CheckIfDeviceAvailable(executor, device, pvcBackedOSD)
 		if err != nil {
 			logger.Warningf("failed to detect device %s availability. %+v", device, err)
 			continue