diff --git a/Documentation/ceph-cluster-crd.md b/Documentation/ceph-cluster-crd.md
index d46233c0b..772c87b5a 100755
--- a/Documentation/ceph-cluster-crd.md
+++ b/Documentation/ceph-cluster-crd.md
@@ -81,6 +81,8 @@ spec:
             - ReadWriteOnce
 ```
 
+For a more advanced scenario, such as adding a dedicated device you can refer to the [dedicated metadata device for OSD on PVC section](#dedicated-metadata-device-for-osd-on-pvc).
+
 ## Settings
 
 Settings can be specified at the global level to apply to the cluster as a whole, while other settings can be specified at more fine-grained levels.  If any setting is unspecified, a suitable default will be used automatically.
@@ -696,6 +698,64 @@ spec:
             - ReadWriteOnce
 ```
 
+### Dedicated metadata device for OSD on PVC
+
+In the simplest case, Ceph OSD BlueStore consumes a single (primary) storage device.
+BlueStore is the engine used by the OSD to store data.
+
+The storage device is normally used as a whole, occupying the full device that is managed directly by BlueStore.
+It is also possible to deploy BlueStore across additional devices such as a DB device.
+This device can be used for storing BlueStore’s internal metadata.
+BlueStore (or rather, the embedded RocksDB) will put as much metadata as it can on the DB device to improve performance.
+If the DB device fills up, metadata will spill back onto the primary device (where it would have been otherwise).
+Again, it is only helpful to provision a DB device if it is faster than the primary device.
+
+You can have multiple `volumeClaimTemplates` where each might either represent a device or a metadata device.
+So just taking the `storage` section this will give something like:
+
+```yaml
+  storage:
+   storageClassDeviceSets:
+    - name: set1
+      count: 3
+      portable: false
+      tuneSlowDeviceClass: false
+      volumeClaimTemplates:
+      - metadata:
+          name: data
+        spec:
+          resources:
+            requests:
+              storage: 10Gi
+          # IMPORTANT: Change the storage class depending on your environment (e.g. local-storage, gp2)
+          storageClassName: gp2
+          volumeMode: Block
+          accessModes:
+            - ReadWriteOnce
+      - metadata:
+          name: metadata
+        spec:
+          resources:
+            requests:
+              # Find the right size https://docs.ceph.com/docs/mimic/rados/configuration/bluestore-config-ref/#sizing
+              storage: 5Gi
+          # IMPORTANT: Change the storage class depending on your environment (e.g. local-storage, gp2)
+          storageClassName: io1
+          volumeMode: Block
+          accessModes:
+            - ReadWriteOnce
+```
+
+> **NOTE**: Note that Rook only supports two naming convention for a given template:
+
+* "data": represents the main OSD block device, where your data are being stored
+* "metadata": represents the metadata device used to store the Ceph Bluestore database for an OSD.
+It is recommended to use a faster storage class for the metadata device, with a slower device for the data.
+Otherwise, having a separate metadata device will not improve the performance.
+To determine the size of the metadata block follow the [official Ceph sizing guide](https://docs.ceph.com/docs/mimic/rados/configuration/bluestore-config-ref/#sizing).
+
+With the present configuration, each OSD will have its main block allocated a 10GB device as well a 5GB device to act as a bluestore database.
+
 ### External cluster
 
 **The minimum supported Ceph version for the External Cluster is Luminous 12.2.x.**
diff --git a/PendingReleaseNotes.md b/PendingReleaseNotes.md
index 8754d64dc..813967254 100644
--- a/PendingReleaseNotes.md
+++ b/PendingReleaseNotes.md
@@ -9,6 +9,7 @@
 ### Ceph
 
 - OSD refactor: drop support for Rook legacy OSD, directory OSD and Filestore OSD. For more details refer to the [corresponding issue](https://github.com/rook/rook/issues/4724).
+- OSD on PVC now supports a metadata device, [refer to the cluster on PVC section](Documentation/ceph-cluster-crd.html#dedicated-metatada-device) or the [corresponding issue](https://github.com/rook/rook/issues/3852).
 
 ### EdgeFS
 
diff --git a/cluster/examples/kubernetes/ceph/cluster-on-pvc.yaml b/cluster/examples/kubernetes/ceph/cluster-on-pvc.yaml
index 12c505c22..d481a019e 100644
--- a/cluster/examples/kubernetes/ceph/cluster-on-pvc.yaml
+++ b/cluster/examples/kubernetes/ceph/cluster-on-pvc.yaml
@@ -100,6 +100,19 @@ spec:
           volumeMode: Block
           accessModes:
             - ReadWriteOnce
+      # dedicated block device to store bluestore database (block.db)
+      # - metadata:
+      #     name: metadata
+      #   spec:
+      #     resources:
+      #       requests:
+      #         # Find the right size https://docs.ceph.com/docs/master/rados/configuration/bluestore-config-ref/#sizing
+      #         storage: 5Gi
+      #     # IMPORTANT: Change the storage class depending on your environment (e.g. local-storage, gp2)
+      #     storageClassName: gp2
+      #     volumeMode: Block
+      #     accessModes:
+      #       - ReadWriteOnce
   disruptionManagement:
     managePodBudgets: false
     osdMaintenanceTimeout: 30
diff --git a/pkg/apis/rook.io/v1alpha2/types.go b/pkg/apis/rook.io/v1alpha2/types.go
index f5326f6c7..bcf4de21a 100644
--- a/pkg/apis/rook.io/v1alpha2/types.go
+++ b/pkg/apis/rook.io/v1alpha2/types.go
@@ -140,12 +140,14 @@ type StorageClassDeviceSet struct {
 	TuneSlowDeviceClass  bool                       `json:"tuneDeviceClass,omitempty"`      // TuneSlowDeviceClass Tune the OSD when running on a slow Device Class
 }
 
+// VolumeSource is a volume source spec for Rook
 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"`
-	Portable                    bool                                 `json:"portable,omitempty"`        // OSD portability across the hosts
-	TuneSlowDeviceClass         bool                                 `json:"tuneDeviceClass,omitempty"` // Tune the OSD when running on a slow Device Class
+	Portable                    bool                                 `json:"portable,omitempty"`        // Portable OSD portability across the hosts
+	TuneSlowDeviceClass         bool                                 `json:"tuneDeviceClass,omitempty"` // TuneSlowDeviceClass Tune the OSD when running on a slow Device Class
+	Type                        string                               `json:"type,omitempty"`            // Type represents the device type for an OSD, possible values are "data": the bluestore 'block' data and "metadata": the bluestore 'block.db' device
 }
diff --git a/pkg/daemon/ceph/osd/daemon.go b/pkg/daemon/ceph/osd/daemon.go
index 345ca6894..c491c73f3 100644
--- a/pkg/daemon/ceph/osd/daemon.go
+++ b/pkg/daemon/ceph/osd/daemon.go
@@ -35,6 +35,11 @@ import (
 	"github.com/rook/rook/pkg/util/sys"
 )
 
+const (
+	pvcDataTypeDevice     = "data"
+	pvcMetadataTypeDevice = "metadata"
+)
+
 var (
 	logger = capnslog.NewPackageLogger("github.com/rook/rook", "cephosd")
 )
@@ -169,16 +174,27 @@ func Provision(context *clusterd.Context, agent *OsdAgent, crushLocation string)
 	}
 
 	logger.Infof("discovering hardware")
+
 	var rawDevices []*sys.LocalDisk
 	if agent.pvcBacked {
-		if len(agent.devices) > 1 {
-			return errors.New("more than one desired device found in case of PVC backed OSDs. we expect exactly one device")
-		}
 		rawDevice, err := clusterd.PopulateDeviceInfo(agent.devices[0].Name, context.Executor)
 		if err != nil {
 			return errors.Wrapf(err, "failed to get device info for %q", agent.devices[0].Name)
 		}
+		rawDevice.Type = pvcDataTypeDevice
 		rawDevices = append(rawDevices, rawDevice)
+
+		// We have a metadata device
+		if len(agent.devices) > 1 {
+			rawMetadataDevice, err := clusterd.PopulateDeviceInfo(agent.devices[1].Name, context.Executor)
+			if err != nil {
+				return errors.Wrapf(err, "failed to get device info for %q", agent.devices[1].Name)
+			}
+
+			// set it's a metadata device
+			rawMetadataDevice.Type = pvcMetadataTypeDevice
+			rawDevices = append(rawDevices, rawMetadataDevice)
+		}
 	} else {
 		// We still need to use 'lsblk' as the underlaying way to discover devices
 		// Ideally, we would use the "ceph-volume inventory" command instead
@@ -269,6 +285,9 @@ func Provision(context *clusterd.Context, agent *OsdAgent, crushLocation string)
 
 func getAvailableDevices(context *clusterd.Context, desiredDevices []DesiredDevice, metadataDevice string, pvcBacked bool, cephVersion cephver.CephVersion) (*DeviceOsdMapping, error) {
 
+	logger.Debugf("desiredDevices are %+v", desiredDevices)
+	logger.Debugf("context.Devices are %+v", context.Devices)
+
 	available := &DeviceOsdMapping{Entries: map[string]*DeviceOsdIDEntry{}}
 	for _, device := range context.Devices {
 		// Ignore 'dm' device since they are not handled by c-v properly
@@ -372,6 +391,12 @@ func getAvailableDevices(context *clusterd.Context, desiredDevices []DesiredDevi
 				// the current device matches the user specifies filter/list, use it for data
 				logger.Infof("device %q is selected by the device filter/name %q", device.Name, matchedDevice.Name)
 				deviceInfo = &DeviceOsdIDEntry{Data: unassignedOSDID, Config: matchedDevice, PersistentDevicePaths: strings.Fields(device.DevLinks)}
+
+				// set that this is not an OSD but a metadata device
+				if device.Type == pvcMetadataTypeDevice {
+					logger.Infof("metadata device %q is selected by the device filter/name %q", device.Name, matchedDevice.Name)
+					deviceInfo = &DeviceOsdIDEntry{Config: matchedDevice, PersistentDevicePaths: strings.Fields(device.DevLinks), Metadata: []int{1}}
+				}
 			} else {
 				logger.Infof("skipping device %q that does not match the device filter/list (%v). %v", device.Name, desiredDevices, err)
 			}
@@ -380,7 +405,19 @@ func getAvailableDevices(context *clusterd.Context, desiredDevices []DesiredDevi
 		}
 
 		if deviceInfo != nil {
-			available.Entries[device.Name] = deviceInfo
+			// When running on PVC, we typically have a single device only
+			// So it's fine to name the first entry of the map "data" instead of the PVC name
+			// It is particularly useful when a metadata PVC is used because we need to identify it in the map
+			// So the entry must be named "metadata" so it can accessed later
+			if pvcBacked {
+				if device.Type == pvcDataTypeDevice {
+					available.Entries[pvcDataTypeDevice] = deviceInfo
+				} else if device.Type == pvcMetadataTypeDevice {
+					available.Entries[pvcMetadataTypeDevice] = deviceInfo
+				}
+			} else {
+				available.Entries[device.Name] = deviceInfo
+			}
 		}
 	}
 
diff --git a/pkg/daemon/ceph/osd/daemon_test.go b/pkg/daemon/ceph/osd/daemon_test.go
index c652cf0dd..6153d48f9 100644
--- a/pkg/daemon/ceph/osd/daemon_test.go
+++ b/pkg/daemon/ceph/osd/daemon_test.go
@@ -184,7 +184,7 @@ NAME="sdb1" SIZE="30" TYPE="part" PKNAME="sdb"`, nil
 		{Name: "nvme01", DevLinks: "/dev/disk/by-id/nvme-0246 /dev/disk/by-path/pci-0:2:4:6-nvme-1", RealName: "nvme01"},
 		{Name: "rda", RealName: "rda"},
 		{Name: "rdb", RealName: "rdb"},
-		{Name: "/mnt/set1-0-data-qfhfk", RealName: "xvdcy"},
+		{Name: "/mnt/set1-0-data-qfhfk", RealName: "xvdcy", Type: "data"},
 		{Name: "sdt1", RealName: "sdt1", Type: sys.PartType},
 	}
 
@@ -260,7 +260,7 @@ NAME="sdb1" SIZE="30" TYPE="part" PKNAME="sdb"`, nil
 	pvcBackedOSD = true
 	mapping, err = getAvailableDevices(context, []DesiredDevice{{Name: "all"}}, "", pvcBackedOSD, version)
 	assert.Nil(t, err)
-	assert.Equal(t, 7, len(mapping.Entries))
+	assert.Equal(t, 1, len(mapping.Entries), mapping)
 }
 
 func TestGetVolumeGroupName(t *testing.T) {
diff --git a/pkg/daemon/ceph/osd/device.go b/pkg/daemon/ceph/osd/device.go
index 46c1e2659..916a2df99 100644
--- a/pkg/daemon/ceph/osd/device.go
+++ b/pkg/daemon/ceph/osd/device.go
@@ -85,7 +85,6 @@ type DeviceOsdIDEntry struct {
 	Data                  int           // OSD ID that has data stored here
 	Metadata              []int         // OSD IDs (multiple) that have metadata stored here
 	Config                DesiredDevice // Device specific config options
-	LegacyPartitionsFound bool          // Whether legacy rook partitions were found
 	PersistentDevicePaths []string
 }
 
diff --git a/pkg/daemon/ceph/osd/volume.go b/pkg/daemon/ceph/osd/volume.go
index 4a13304ce..ae027b7aa 100644
--- a/pkg/daemon/ceph/osd/volume.go
+++ b/pkg/daemon/ceph/osd/volume.go
@@ -90,9 +90,11 @@ func (a *OsdAgent) configureCVDevices(context *clusterd.Context, devices *Device
 	var lvmOsds []oposd.OSDInfo
 	var rawOsds []oposd.OSDInfo
 	var lvBackedPV bool
-	var block, lvPath string
+	var block, lvPath, metadataBlock string
 	var err error
 
+	// Idempotency check, if the device list is empty devices have been prepared already
+	// In this case, just return the OSDInfo via a 'ceph-volume lvm|raw list' call
 	if len(devices.Entries) == 0 {
 		logger.Info("no new devices to configure. returning devices already configured with ceph-volume.")
 
@@ -115,7 +117,17 @@ func (a *OsdAgent) configureCVDevices(context *clusterd.Context, devices *Device
 			if a.cluster.CephVersion.IsAtLeast(cephVolumeRawModeMinCephVersion) && !lvBackedPV {
 				// For block mode
 				block = fmt.Sprintf("/mnt/%s", a.nodeName)
-				rawOsds, err = getCephVolumeRawOSDs(context, a.cluster.Name, a.cluster.FSID, block, lvBackedPV)
+
+				// This is hard to determine a potential metadata device here
+				// Also, I don't think (leseb) this code we have run in this condition
+				// I tried several things:
+				//    * evict a node, osd moves, the prepare job was never relaunched ever because we check for the osd deployment and skip the prepare
+				//    * restarted the operator, again the prepare job was not re-run
+				//
+				// I'm leaving this code with an empty metadata device for now
+				metadataBlock = ""
+
+				rawOsds, err = getCephVolumeRawOSDs(context, a.cluster.Name, a.cluster.FSID, block, metadataBlock, lvBackedPV)
 				if err != nil {
 					logger.Infof("failed to get device already provisioned by ceph-volume raw. %v", err)
 				}
@@ -132,15 +144,6 @@ func (a *OsdAgent) configureCVDevices(context *clusterd.Context, devices *Device
 		}
 		osds = append(osds, lvmOsds...)
 
-		// List existing OSD(s) configured with ceph-volume raw mode
-		if a.cluster.CephVersion.IsAtLeast(cephVolumeRawModeMinCephVersion) && !lvBackedPV {
-			rawOsds, err = getCephVolumeRawOSDs(context, a.cluster.Name, a.cluster.FSID, block, lvBackedPV)
-			if err != nil {
-				logger.Infof("failed to get devices already provisioned by ceph-volume. %v", err)
-			}
-			osds = append(osds, rawOsds...)
-		}
-
 		return osds, nil
 	}
 
@@ -152,13 +155,15 @@ func (a *OsdAgent) configureCVDevices(context *clusterd.Context, devices *Device
 
 	// Check if the PVC is an LVM block device (certain StorageClass do this)
 	if a.pvcBacked {
-		for name := range devices.Entries {
-			dev := name
+		for _, device := range devices.Entries {
+			dev := device.Config.Name
 			// When not using PV backend an LV
 			// Otherwise lsblk will fail since the block will be '/dev/mnt/set1-0-data-l6p5q'
 			// And thus won't be a block device
-			if !strings.HasPrefix(name, "/mnt") {
-				dev = path.Join("/dev", name)
+			//
+			// The same goes the metadata block device which is stored in /srv
+			if !strings.HasPrefix(dev, "/mnt") && !strings.HasPrefix(dev, "/srv") {
+				dev = path.Join("/dev", dev)
 			}
 			lvBackedPV, err = sys.IsLV(dev, context.Executor)
 			if err != nil {
@@ -180,7 +185,7 @@ func (a *OsdAgent) configureCVDevices(context *clusterd.Context, devices *Device
 
 	// If running on OSD on PVC
 	if a.pvcBacked {
-		if block, err = a.initializeBlockPVC(context, devices, lvBackedPV); err != nil {
+		if block, metadataBlock, err = a.initializeBlockPVC(context, devices, lvBackedPV); err != nil {
 			return nil, errors.Wrapf(err, "failed to initialize devices")
 		}
 	} else {
@@ -199,7 +204,7 @@ func (a *OsdAgent) configureCVDevices(context *clusterd.Context, devices *Device
 	// List THE configured OSD with ceph-volume raw mode
 	if a.cluster.CephVersion.IsAtLeast(cephVolumeRawModeMinCephVersion) && !lvBackedPV {
 		block = fmt.Sprintf("/mnt/%s", a.nodeName)
-		rawOsds, err = getCephVolumeRawOSDs(context, a.cluster.Name, a.cluster.FSID, block, lvBackedPV)
+		rawOsds, err = getCephVolumeRawOSDs(context, a.cluster.Name, a.cluster.FSID, block, metadataBlock, lvBackedPV)
 		if err != nil {
 			return nil, errors.Wrap(err, "failed to get devices already provisioned by ceph-volume raw")
 		}
@@ -209,7 +214,7 @@ func (a *OsdAgent) configureCVDevices(context *clusterd.Context, devices *Device
 	return osds, err
 }
 
-func (a *OsdAgent) initializeBlockPVC(context *clusterd.Context, devices *DeviceOsdMapping, lvBackedPV bool) (string, error) {
+func (a *OsdAgent) initializeBlockPVC(context *clusterd.Context, devices *DeviceOsdMapping, lvBackedPV bool) (string, string, error) {
 
 	// we need to return the block if raw mode is used and the lv if lvm mode
 	baseCommand := "stdbuf"
@@ -232,21 +237,40 @@ func (a *OsdAgent) initializeBlockPVC(context *clusterd.Context, devices *Device
 		baseArgs = []string{"-oL", cephVolumeCmd, "--log-path", cvLogDir, cephVolumeMode, "prepare", "--bluestore"}
 	}
 
-	var blockPath string
+	var metadataArg []string
+	var metadataDev bool
+	var blockPath, metadataBlockPath string
+
+	// Problem: map is an unordered collection
+	// therefore the iteration order of a map is not guaranteed to be the same every time you iterate over it.
+	// So we could first get the metadata device and then the main block in a scenario where a metadata PVC is present
 	for name, device := range devices.Entries {
-		if device.LegacyPartitionsFound {
-			logger.Infof("skipping device %q configured with legacy rook osd", name)
+		// If this is the metadata device there is nothing to do
+		// it'll be used in one of the iterations
+		if name == "metadata" {
+			logger.Debugf("device %q is a metadata device, skipping this iteration it will be used in the next one", device.Config.Name)
+			// Don't do this device
 			continue
 		}
+
+		// When running on PVC, the prepare job has a single OSD only so 1 disk
+		// However we can present a metadata device so we need to consume it
+		// This will make the devices.Entries larger than usual
+		if _, ok := devices.Entries["metadata"]; ok {
+			metadataDev = true
+			metadataArg = append(metadataArg, []string{"--block.db", devices.Entries["metadata"].Config.Name}...)
+			metadataBlockPath = devices.Entries["metadata"].Config.Name
+		}
+
 		if device.Data == -1 {
-			logger.Infof("configuring new device %q", name)
+			logger.Infof("configuring new device %q", device.Config.Name)
 			var err error
 			var deviceArg string
 			if lvBackedPV {
 				// pass 'vg/lv' to ceph-volume
 				deviceArg, err = getLVNameFromDevicePath(context, device.Config.Name)
 				if err != nil {
-					return "", errors.Wrapf(err, "failed to get lv name from device path %q", device.Config.Name)
+					return "", "", errors.Wrapf(err, "failed to get lv name from device path %q", device.Config.Name)
 				}
 			} else {
 				deviceArg = device.Config.Name
@@ -257,6 +281,11 @@ func (a *OsdAgent) initializeBlockPVC(context *clusterd.Context, devices *Device
 				deviceArg,
 			}...)
 
+			// Add the cli argument for the metadata device
+			if metadataDev {
+				immediateExecuteArgs = append(immediateExecuteArgs, metadataArg...)
+			}
+
 			// execute ceph-volume with the device
 			op, err := context.Executor.ExecuteCommandWithCombinedOutput(false, "", baseCommand, immediateExecuteArgs...)
 			if err != nil {
@@ -269,7 +298,7 @@ func (a *OsdAgent) initializeBlockPVC(context *clusterd.Context, devices *Device
 				}
 
 				// Return failure
-				return "", errors.Wrapf(err, "failed ceph-volume") // fail return here as validation provided by ceph-volume
+				return "", "", errors.Wrapf(err, "failed ceph-volume") // fail return here as validation provided by ceph-volume
 			}
 			logger.Infof("%v", op)
 			// if raw mode is used or PV on LV, let's return the path of the device
@@ -278,15 +307,15 @@ func (a *OsdAgent) initializeBlockPVC(context *clusterd.Context, devices *Device
 			} else {
 				blockPath = getLVPath(op)
 				if blockPath == "" {
-					return "", errors.New("failed to get lv path from ceph-volume lvm prepare output")
+					return "", "", errors.New("failed to get lv path from ceph-volume lvm prepare output")
 				}
 			}
 		} else {
-			logger.Infof("skipping device %q with osd %d already configured", name, device.Data)
+			logger.Infof("skipping device %q with osd %d already configured", device.Config.Name, device.Data)
 		}
 	}
 
-	return blockPath, nil
+	return blockPath, metadataBlockPath, nil
 }
 
 func getLVPath(op string) string {
@@ -376,11 +405,6 @@ func (a *OsdAgent) initializeDevices(context *clusterd.Context, devices *DeviceO
 
 	metadataDevices := make(map[string]map[string]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 := path.Join("/dev", name)
@@ -633,7 +657,7 @@ func readCVLogContent(cvLogFilePath string) string {
 	return string(b)
 }
 
-func getCephVolumeRawOSDs(context *clusterd.Context, clusterName string, cephfsid, block string, lvBackedPV bool) ([]oposd.OSDInfo, error) {
+func getCephVolumeRawOSDs(context *clusterd.Context, clusterName string, cephfsid, block, metadataBlock string, lvBackedPV bool) ([]oposd.OSDInfo, error) {
 	// lv can be a block device if raw mode is used
 	cvMode := "raw"
 
@@ -675,6 +699,7 @@ func getCephVolumeRawOSDs(context *clusterd.Context, clusterName string, cephfsi
 			// Thus in the activation sequence we might activate the wrong OSD and have OSDInfo messed up
 			// Hence, let's use the PVC name instead which will always remain consistent
 			BlockPath:     block,
+			MetadataPath:  metadataBlock,
 			SkipLVRelease: true,
 			LVBackedPV:    lvBackedPV,
 			CVMode:        cvMode,
diff --git a/pkg/daemon/ceph/osd/volume_test.go b/pkg/daemon/ceph/osd/volume_test.go
index 5b039aa4d..47758ca75 100644
--- a/pkg/daemon/ceph/osd/volume_test.go
+++ b/pkg/daemon/ceph/osd/volume_test.go
@@ -292,7 +292,7 @@ func TestParseCephVolumeRawResult(t *testing.T) {
 	}
 
 	context := &clusterd.Context{Executor: executor}
-	osds, err := getCephVolumeRawOSDs(context, "rook", "4bfe8b72-5e69-4330-b6c0-4d914db8ab89", "", false)
+	osds, err := getCephVolumeRawOSDs(context, "rook", "4bfe8b72-5e69-4330-b6c0-4d914db8ab89", "", "", false)
 	assert.Nil(t, err)
 	require.NotNil(t, osds)
 	assert.Equal(t, 2, len(osds))
diff --git a/pkg/operator/ceph/cluster/osd/deviceSet.go b/pkg/operator/ceph/cluster/osd/deviceSet.go
index 1c81fe847..05d29d27c 100644
--- a/pkg/operator/ceph/cluster/osd/deviceSet.go
+++ b/pkg/operator/ceph/cluster/osd/deviceSet.go
@@ -26,72 +26,94 @@ import (
 
 func (c *Cluster) prepareStorageClassDeviceSets(config *provisionConfig) []rookalpha.VolumeSource {
 	volumeSources := []rookalpha.VolumeSource{}
+
+	// Iterate over storageClassDeviceSet
 	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)
+			config.addError("cannot use storageClassDeviceSet %q 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)
+			// Check if the volume claim template has PVCs
+			if len(storageClassDeviceSet.VolumeClaimTemplates) == 0 {
+				logger.Warningf("no PVC available for storageClassDeviceSet %q", storageClassDeviceSet.Name)
 				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,
-				},
-				Portable:            storageClassDeviceSet.Portable,
-				TuneSlowDeviceClass: storageClassDeviceSet.TuneSlowDeviceClass,
-			})
-			logger.Infof("successfully provisioned osd for storageClassDeviceSet %s of set %v", storageClassDeviceSet.Name, i)
+			for _, pvcTemplate := range storageClassDeviceSet.VolumeClaimTemplates {
+				pvc, err := c.createStorageClassDeviceSetPVC(storageClassDeviceSet.Name, pvcTemplate, i)
+				if err != nil {
+					config.addError("failed to create osd for storageClassDeviceSet %q for count %d. %v", storageClassDeviceSet.Name, i, err)
+					continue
+				}
+				volumeSources = append(volumeSources, rookalpha.VolumeSource{
+					Name:      storageClassDeviceSet.Name,
+					Resources: storageClassDeviceSet.Resources,
+					Placement: storageClassDeviceSet.Placement,
+					Config:    storageClassDeviceSet.Config,
+					Type:      pvcTemplate.GetName(),
+					PersistentVolumeClaimSource: v1.PersistentVolumeClaimVolumeSource{
+						ClaimName: pvc.GetName(),
+						ReadOnly:  false,
+					},
+					Portable:            storageClassDeviceSet.Portable,
+					TuneSlowDeviceClass: storageClassDeviceSet.TuneSlowDeviceClass,
+				})
+				logger.Infof("successfully provisioned pvc %q for VolumeClaimTemplates %q for storageClassDeviceSet %q of set %v", pvc.GetName(), pvcTemplate.GetName(), storageClassDeviceSet.Name, i)
+			}
 		}
 	}
+
 	return volumeSources
 }
 
-func (c *Cluster) createStorageClassDeviceSetPVC(storageClassDeviceSet rookalpha.StorageClassDeviceSet, setIndex int) (*v1.PersistentVolumeClaim, error) {
-	if len(storageClassDeviceSet.VolumeClaimTemplates) == 0 {
-		return nil, errors.Errorf("no PVC available for storageClassDeviceSet %s", storageClassDeviceSet.Name)
+func (c *Cluster) createStorageClassDeviceSetPVC(storageClassDeviceSetName string, pvcTemplate v1.PersistentVolumeClaim, setIndex int) (*v1.PersistentVolumeClaim, error) {
+	// old labels and PVC ID
+	pvcStorageClassDeviceSetPVCId, pvcStorageClassDeviceSetPVCIdLabelSelector := makeStorageClassDeviceSetPVCID(storageClassDeviceSetName, setIndex)
+	pvc := makeStorageClassDeviceSetPVC(storageClassDeviceSetName, pvcStorageClassDeviceSetPVCId, setIndex, pvcTemplate)
+	oldPresentPVCs, err := c.context.Clientset.CoreV1().PersistentVolumeClaims(c.Namespace).List(metav1.ListOptions{LabelSelector: pvcStorageClassDeviceSetPVCIdLabelSelector})
+	if err != nil {
+		return nil, errors.Wrapf(err, "failed to create pvc %s for storageClassDeviceSet %s", pvc.GetGenerateName(), storageClassDeviceSetName)
 	}
-	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
+	// return old labelled pvc, if we find any
+	if len(oldPresentPVCs.Items) == 1 {
+		logger.Debugf("old labelled pvc %q found", oldPresentPVCs.Items[0].Name)
+		return &oldPresentPVCs.Items[0], nil
+	}
+
+	// check again with the new label for the presence of updated pvc
+	pvcStorageClassDeviceSetPVCId, pvcStorageClassDeviceSetPVCIdLabelSelector = makeStorageClassDeviceSetPVCIDNew(storageClassDeviceSetName, pvcTemplate.GetName(), setIndex)
+	pvc = makeStorageClassDeviceSetPVC(storageClassDeviceSetName, pvcStorageClassDeviceSetPVCId, setIndex, pvcTemplate)
 	presentPVCs, err := c.context.Clientset.CoreV1().PersistentVolumeClaims(c.Namespace).List(metav1.ListOptions{LabelSelector: pvcStorageClassDeviceSetPVCIdLabelSelector})
 	if err != nil {
-		return nil, errors.Wrapf(err, "failed to create pvc %s for storageClassDeviceSet %s", pvc.GetGenerateName(), storageClassDeviceSet.Name)
+		return nil, errors.Wrapf(err, "failed to create pvc %s for storageClassDeviceSet %s", pvc.GetGenerateName(), storageClassDeviceSetName)
 	}
-	if len(presentPVCs.Items) == 0 { // No PVC found, creating a new one
+
+	presentPVCsNum := len(presentPVCs.Items)
+	// No PVC found, creating a new one
+	if presentPVCsNum == 0 {
 		deployedPVC, err := c.context.Clientset.CoreV1().PersistentVolumeClaims(c.Namespace).Create(pvc)
 		if err != nil {
-			return nil, errors.Wrapf(err, "failed to create pvc %s for storageClassDeviceSet %s", pvc.GetGenerateName(), storageClassDeviceSet.Name)
+			return nil, errors.Wrapf(err, "failed to create pvc %q for storageClassDeviceSet %q", pvc.GetGenerateName(), storageClassDeviceSetName)
 		}
+		logger.Debugf("just created pvc %q", deployedPVC.Name)
 		return deployedPVC, nil
-	} else if len(presentPVCs.Items) == 1 { // The PVC is already present.
+		// The PVC is already present
+	} else if presentPVCsNum == 1 {
+		logger.Debugf("already present pvc %q", presentPVCs.Items[0].Name)
+		// Updating with the new label
 		return &presentPVCs.Items[0], nil
 	}
 	// More than one PVC exists with same labelSelector
-	return nil, errors.Errorf("more than one PVCs exists with label %s, pvcs %s", pvcStorageClassDeviceSetPVCIdLabelSelector, presentPVCs)
+	return nil, errors.Errorf("more than one PVCs exists with label %q, pvcs %q", pvcStorageClassDeviceSetPVCIdLabelSelector, presentPVCs)
 }
 
-func makeStorageClassDeviceSetPVC(storageClassDeviceSetName, pvcStorageClassDeviceSetPVCId string, pvcIndex, setIndex int, pvcTemplate v1.PersistentVolumeClaim) (pvcs *v1.PersistentVolumeClaim) {
-	pvcLabels := makeStorageClassDeviceSetPVCLabel(storageClassDeviceSetName, pvcStorageClassDeviceSetPVCId, pvcIndex, setIndex)
+func makeStorageClassDeviceSetPVC(storageClassDeviceSetName, pvcStorageClassDeviceSetPVCId string, setIndex int, pvcTemplate v1.PersistentVolumeClaim) (pvcs *v1.PersistentVolumeClaim) {
+	pvcLabels := makeStorageClassDeviceSetPVCLabel(storageClassDeviceSetName, pvcStorageClassDeviceSetPVCId, 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
diff --git a/pkg/operator/ceph/cluster/osd/osd.go b/pkg/operator/ceph/cluster/osd/osd.go
index 2629dc53e..9dd097457 100644
--- a/pkg/operator/ceph/cluster/osd/osd.go
+++ b/pkg/operator/ceph/cluster/osd/osd.go
@@ -69,6 +69,8 @@ const (
 	unknownID                           = -1
 	portableKey                         = "portable"
 	cephOsdPodMinimumMemory      uint64 = 2048 // minimum amount of memory in MB to run the pod
+	bluestorePVCMetadata                = "metadata"
+	bluestorePVCBlock                   = "data"
 )
 
 // Cluster keeps track of the OSDs
@@ -142,6 +144,7 @@ type OSDInfo struct {
 	DevicePartUUID string `json:"device-part-uuid"`
 	// BlockPath is the logical Volume path for an OSD created by Ceph-volume with format '/dev/<Volume Group>/<Logical Volume>' or simply /dev/vdb if block mode is used
 	BlockPath     string `json:"lv-path"`
+	MetadataPath  string `json:"metadata-path"`
 	SkipLVRelease bool   `json:"skip-lv-release"`
 	Location      string `json:"location"`
 	LVBackedPV    bool   `json:"lv-backed-pv"`
@@ -162,6 +165,7 @@ type osdProperties struct {
 	crushHostname       string
 	devices             []rookalpha.Device
 	pvc                 v1.PersistentVolumeClaimVolumeSource
+	metadataPVC         v1.PersistentVolumeClaimVolumeSource
 	selection           rookalpha.Selection
 	resources           v1.ResourceRequirements
 	storeConfig         osdconfig.StoreConfig
@@ -231,15 +235,51 @@ func (c *Cluster) startProvisioningOverPVCs(config *provisionConfig) {
 		return
 	}
 
-	for _, volume := range c.ValidStorage.VolumeSources {
+	for i, volume := range c.ValidStorage.VolumeSources {
+		// If the metadata template is first, we fail and assume we cannot build the data/metadata block relationship
+		if i == 0 && volume.Type == bluestorePVCMetadata {
+			config.addError("wrong template ordering, the %q device must be declared after the %q device.", bluestorePVCMetadata, bluestorePVCBlock)
+			return
+		}
+
+		metadataDevicePVCSource := v1.PersistentVolumeClaimVolumeSource{}
+
+		// If the volumeTemplate is unknown we do nothing
+		if volume.Type != bluestorePVCMetadata && volume.Type != bluestorePVCBlock {
+			logger.Errorf("unknown PVC template type %q, valid names are %q for the main OSD block and %q for a metadata device to back the OSD.", volume.Type, bluestorePVCBlock, bluestorePVCMetadata)
+			continue
+		}
+
+		// We don't need to use the metadata devices as OSDs
+		// They are just attached to OSD and field in their property
+		if volume.Type == bluestorePVCMetadata {
+			logger.Infof("PVC %q is not an OSD but a %q device", volume.PersistentVolumeClaimSource.ClaimName, volume.Type)
+			continue
+		}
+
+		// Let's see how the next PVC looks like
+		// If the next PVC has been identified as a PVC let's attached to this OSD property
+		//
+		// The logic will get a bit more complex if we plan support for a third device for "block.wal"
+		m := i + 1
+		if m < len(c.ValidStorage.VolumeSources) {
+			if c.ValidStorage.VolumeSources[m].Type == bluestorePVCMetadata {
+				logger.Infof("OSD will have its main bluestore block on %q and its metadata device on %q", volume.PersistentVolumeClaimSource.ClaimName, c.ValidStorage.VolumeSources[m].PersistentVolumeClaimSource.ClaimName)
+				metadataDevicePVCSource = c.ValidStorage.VolumeSources[m].PersistentVolumeClaimSource
+			}
+		}
+
 		osdProps := osdProperties{
 			crushHostname: volume.PersistentVolumeClaimSource.ClaimName,
 			pvc:           volume.PersistentVolumeClaimSource,
+			metadataPVC:   metadataDevicePVCSource,
 			resources:     volume.Resources,
 			placement:     volume.Placement,
 			portable:      volume.Portable,
 		}
 
+		logger.Debugf("osdProps are %+v", osdProps)
+
 		// 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 {
@@ -431,7 +471,15 @@ func (c *Cluster) startOSDDaemonsOnPVC(pvcName string, config *provisionConfig,
 
 		dp, err := c.makeDeployment(osdProps, osd, config)
 		if err != nil {
-			errMsg := fmt.Sprintf("failed to create deployment for pvc %q: %v", osdProps.crushHostname, err)
+			errMsg := fmt.Sprintf("failed to create deployment for pvc %q. %v", osdProps.crushHostname, err)
+			config.addError(errMsg)
+			continue
+		}
+
+		// Set the deployment hash as an annotation
+		err = patch.DefaultAnnotator.SetLastAppliedAnnotation(dp)
+		if err != nil {
+			errMsg := fmt.Sprintf("failed to set annotation for deployment %q. %v", dp.Name, err)
 			config.addError(errMsg)
 			continue
 		}
@@ -617,11 +665,23 @@ func (c *Cluster) resolveNode(nodeName string) *rookalpha.Node {
 }
 
 func (c *Cluster) getOSDPropsForPVC(pvcName string) (osdProperties, error) {
-	for _, volumeSource := range c.ValidStorage.VolumeSources {
+	var metadataDevicePVCSource v1.PersistentVolumeClaimVolumeSource
+
+	for i, volumeSource := range c.ValidStorage.VolumeSources {
+		// volumeSource should be consistent and the order always identical so doing the +1 thing shouldn't be too dangerous
 		if pvcName == volumeSource.PersistentVolumeClaimSource.ClaimName {
+			m := i + 1
+			if m < len(c.ValidStorage.VolumeSources) {
+				if c.ValidStorage.VolumeSources[m].Type == bluestorePVCMetadata {
+					logger.Infof("OSD will have its main bluestore block on %q and its metadata device on %q", volumeSource.PersistentVolumeClaimSource.ClaimName, c.ValidStorage.VolumeSources[m].PersistentVolumeClaimSource.ClaimName)
+					metadataDevicePVCSource = c.ValidStorage.VolumeSources[m].PersistentVolumeClaimSource
+				}
+			}
+
 			osdProps := osdProperties{
 				crushHostname:       volumeSource.PersistentVolumeClaimSource.ClaimName,
 				pvc:                 volumeSource.PersistentVolumeClaimSource,
+				metadataPVC:         metadataDevicePVCSource,
 				resources:           volumeSource.Resources,
 				placement:           volumeSource.Placement,
 				portable:            volumeSource.Portable,
@@ -678,6 +738,9 @@ func (c *Cluster) getOSDInfo(d *apps.Deployment) ([]OSDInfo, error) {
 		if envVar.Name == "ROOK_CV_MODE" {
 			osd.CVMode = envVar.Value
 		}
+		if envVar.Name == "ROOK_METADATA_DEVICE" {
+			osd.MetadataPath = envVar.Value
+		}
 	}
 
 	// If CVMode is empty, this likely means we upgraded Rook
diff --git a/pkg/operator/ceph/cluster/osd/spec.go b/pkg/operator/ceph/cluster/osd/spec.go
index 1d4caf871..b166beda0 100644
--- a/pkg/operator/ceph/cluster/osd/spec.go
+++ b/pkg/operator/ceph/cluster/osd/spec.go
@@ -40,23 +40,25 @@ import (
 )
 
 const (
-	osdStoreEnvVarName                  = "ROOK_OSD_STORE"
-	osdDatabaseSizeEnvVarName           = "ROOK_OSD_DATABASE_SIZE"
-	osdWalSizeEnvVarName                = "ROOK_OSD_WAL_SIZE"
-	osdJournalSizeEnvVarName            = "ROOK_OSD_JOURNAL_SIZE"
-	osdsPerDeviceEnvVarName             = "ROOK_OSDS_PER_DEVICE"
-	encryptedDeviceEnvVarName           = "ROOK_ENCRYPTED_DEVICE"
-	osdMetadataDeviceEnvVarName         = "ROOK_METADATA_DEVICE"
-	pvcBackedOSDVarName                 = "ROOK_PVC_BACKED_OSD"
-	blockPathVarName                    = "ROOK_BLOCK_PATH"
-	cvModeVarName                       = "ROOK_CV_MODE"
-	lvBackedPVVarName                   = "ROOK_LV_BACKED_PV"
-	rookBinariesMountPath               = "/rook"
-	rookBinariesVolumeName              = "rook-binaries"
-	activateOSDVolumeName               = "activate-osd"
-	activateOSDMountPath                = "/var/lib/ceph/osd/ceph-"
-	blockPVCMapperInitContainer         = "blkdevmapper"
-	osdMemoryTargetSafetyFactor float32 = 0.8
+	osdStoreEnvVarName                          = "ROOK_OSD_STORE"
+	osdDatabaseSizeEnvVarName                   = "ROOK_OSD_DATABASE_SIZE"
+	osdWalSizeEnvVarName                        = "ROOK_OSD_WAL_SIZE"
+	osdJournalSizeEnvVarName                    = "ROOK_OSD_JOURNAL_SIZE"
+	osdsPerDeviceEnvVarName                     = "ROOK_OSDS_PER_DEVICE"
+	encryptedDeviceEnvVarName                   = "ROOK_ENCRYPTED_DEVICE"
+	osdMetadataDeviceEnvVarName                 = "ROOK_METADATA_DEVICE"
+	pvcBackedOSDVarName                         = "ROOK_PVC_BACKED_OSD"
+	blockPathVarName                            = "ROOK_BLOCK_PATH"
+	cvModeVarName                               = "ROOK_CV_MODE"
+	lvBackedPVVarName                           = "ROOK_LV_BACKED_PV"
+	rookBinariesMountPath                       = "/rook"
+	rookBinariesVolumeName                      = "rook-binaries"
+	activateOSDVolumeName                       = "activate-osd"
+	activateOSDMountPath                        = "/var/lib/ceph/osd/ceph-"
+	blockPVCMapperInitContainer                 = "blkdevmapper"
+	blockPVCMetadataMapperInitContainer         = "blkdevmapper-metadata"
+	activatePVCOSDInitContainer                 = "activate"
+	osdMemoryTargetSafetyFactor         float32 = 0.8
 	// CephDeviceSetLabelKey is the Rook device set label key
 	CephDeviceSetLabelKey = "ceph.rook.io/DeviceSet"
 	// CephSetIndexLabelKey is the Rook label key index
@@ -77,6 +79,7 @@ OSD_STORE_FLAG="%s"
 OSD_DATA_DIR=/var/lib/ceph/osd/ceph-"$OSD_ID"
 CV_MODE=%s
 DEVICE=%s
+METADATA_DEVICE="$%s"
 
 # active the osd with ceph-volume
 if [[ "$CV_MODE" == "lvm" ]]; then
@@ -102,8 +105,12 @@ if [[ "$CV_MODE" == "lvm" ]]; then
 	# remove the temporary directory
 	rm --recursive --force "$TMP_DIR"
 else
+	ARGS=(--device ${DEVICE} --no-systemd --no-tmpfs)
+	if [ -n "$METADATA_DEVICE" ]; then
+		ARGS+=(--block.db ${METADATA_DEVICE})
+	fi
 	# ceph-volume raw mode only supports bluestore so we don't need to pass a store flag
-	ceph-volume "$CV_MODE" activate --device "$DEVICE" --no-systemd --no-tmpfs
+	ceph-volume "$CV_MODE" activate "${ARGS[@]}"
 fi
 
 `
@@ -117,16 +124,23 @@ const (
 )
 
 func (c *Cluster) makeJob(osdProps osdProperties, provisionConfig *provisionConfig) (*batch.Job, error) {
+	osdOnPVC := osdProps.pvc.ClaimName != ""
+	osdOnPVCMetadata := osdProps.metadataPVC.ClaimName != ""
 
 	podSpec, err := c.provisionPodTemplateSpec(osdProps, v1.RestartPolicyOnFailure, provisionConfig)
 	if err != nil {
 		return nil, err
 	}
 
-	if osdProps.pvc.ClaimName == "" {
+	if !osdOnPVC {
 		podSpec.Spec.NodeSelector = map[string]string{v1.LabelHostname: osdProps.crushHostname}
 	} else {
-		podSpec.Spec.InitContainers = append(podSpec.Spec.InitContainers, c.getPVCInitContainer(osdProps.pvc))
+		// This is not needed in raw mode and 14.2.8 brings it
+		// but we still want to do this not to lose backward compatibility with lvm based OSDs...
+		podSpec.Spec.InitContainers = append(podSpec.Spec.InitContainers, c.getPVCInitContainer(osdProps))
+		if osdOnPVCMetadata {
+			podSpec.Spec.InitContainers = append(podSpec.Spec.InitContainers, c.getPVCMetadataInitContainer("/srv", osdProps))
+		}
 	}
 
 	job := &batch.Job{
@@ -143,7 +157,7 @@ func (c *Cluster) makeJob(osdProps osdProperties, provisionConfig *provisionConf
 		},
 	}
 
-	if len(osdProps.pvc.ClaimName) > 0 {
+	if osdOnPVC {
 		k8sutil.AddLabelToJob(OSDOverPVCLabelKey, osdProps.pvc.ClaimName, job)
 	}
 
@@ -158,25 +172,34 @@ func (c *Cluster) makeDeployment(osdProps osdProperties, osd OSDInfo, provisionC
 	replicaCount := int32(1)
 	volumeMounts := opspec.CephVolumeMounts(provisionConfig.DataPathMap, false)
 	configVolumeMounts := opspec.RookVolumeMounts(provisionConfig.DataPathMap, false)
+	osdOnPVC := osdProps.pvc.ClaimName != ""
+	osdOnPVCMetadata := osdProps.metadataPVC.ClaimName != ""
+	// When running on PVC, OSDs don't need a bindmount on dataDirHostPath, only the monitors do
+	if osdOnPVC {
+		c.dataDirHostPath = ""
+	}
 	volumes := opspec.PodVolumes(provisionConfig.DataPathMap, c.dataDirHostPath, false)
 	failureDomainValue := osdProps.crushHostname
 	doConfigInit := true       // initialize ceph.conf in init container?
 	doBinaryCopyInit := true   // copy tini and rook binaries in an init container?
 	doActivateOSDInit := false // run an init container to activate the osd?
-	osdOnPVC := osdProps.pvc.ClaimName != ""
 
 	// If CVMode is empty, this likely means we upgraded Rook
 	// This property did not exist before so we need to initialize it
-	if osdOnPVC && osd.CVMode == "" {
+	// This property is used for both PVC and non-PVC use case
+	if osd.CVMode == "" {
 		osd.CVMode = "lvm"
 	}
 
 	dataDir := k8sutil.DataDir
 	// Create volume config for /dev so the pod can access devices on the host
-	devVolume := v1.Volume{Name: "devices", VolumeSource: v1.VolumeSource{HostPath: &v1.HostPathVolumeSource{Path: "/dev"}}}
-	volumes = append(volumes, devVolume)
-	devMount := v1.VolumeMount{Name: "devices", MountPath: "/dev"}
-	volumeMounts = append(volumeMounts, devMount)
+	// Only valid when running OSD with LVM mode
+	if osd.CVMode == "lvm" {
+		devVolume := v1.Volume{Name: "devices", VolumeSource: v1.VolumeSource{HostPath: &v1.HostPathVolumeSource{Path: "/dev"}}}
+		volumes = append(volumes, devVolume)
+		devMount := v1.VolumeMount{Name: "devices", MountPath: "/dev"}
+		volumeMounts = append(volumeMounts, devMount)
+	}
 
 	// If the OSD runs on PVC
 	if osdOnPVC {
@@ -266,6 +289,18 @@ func (c *Cluster) makeDeployment(osdProps osdProperties, osd OSDInfo, provisionC
 			Name:      "run-udev",
 			MountPath: "/run/udev"})
 
+	} else if osdOnPVC && osd.CVMode == "raw" {
+		doBinaryCopyInit = false
+		doConfigInit = false
+		command = []string{"ceph-osd"}
+		args = []string{
+			"--foreground",
+			"--id", osdID,
+			"--fsid", c.clusterInfo.FSID,
+			"--setuser", "ceph",
+			"--setgroup", "ceph",
+			fmt.Sprintf("--crush-location=%s", osd.Location),
+		}
 	} else {
 		doBinaryCopyInit = false
 		doConfigInit = false
@@ -300,7 +335,9 @@ func (c *Cluster) makeDeployment(osdProps osdProperties, osd OSDInfo, provisionC
 
 	args = append(args, commonArgs...)
 
-	if osdOnPVC {
+	osdDataDirPath := activateOSDMountPath + osdID
+	if osdOnPVC && osd.CVMode == "lvm" {
+		// Let's use the old bridge for these lvm based pvc osds
 		volumeMounts = append(volumeMounts, getPvcOSDBridgeMount(osdProps.pvc.ClaimName))
 		envVars = append(envVars, pvcBackedOSDEnvVar("true"))
 		envVars = append(envVars, blockPathEnvVariable(osd.BlockPath))
@@ -308,6 +345,17 @@ func (c *Cluster) makeDeployment(osdProps osdProperties, osd OSDInfo, provisionC
 		envVars = append(envVars, lvBackedPVEnvVar(strconv.FormatBool(osd.LVBackedPV)))
 	}
 
+	if osdOnPVC && osd.CVMode == "raw" {
+		volumeMounts = append(volumeMounts, getPvcOSDBridgeMountActivate(osdDataDirPath, osdProps.pvc.ClaimName))
+		envVars = append(envVars, pvcBackedOSDEnvVar("true"))
+		envVars = append(envVars, blockPathEnvVariable(osd.BlockPath))
+		envVars = append(envVars, cvModeEnvVariable(osd.CVMode))
+	}
+
+	// We cannot go un-privileged until we have a bindmount for logs and crash
+	// OpenShift requires privileged containers for that
+	// If we remove those OSD on PVC with raw mode won't need to be privileged
+	// We could try to run as ceph too, more investigations needed
 	privileged := true
 	runAsUser := int64(0)
 	readOnlyRootFilesystem := false
@@ -317,7 +365,7 @@ func (c *Cluster) makeDeployment(osdProps osdProperties, osd OSDInfo, provisionC
 		ReadOnlyRootFilesystem: &readOnlyRootFilesystem,
 	}
 
-	// needed for luksOpen synchronization when devices are encrypted
+	// needed for luksOpen synchronization when devices are encrypted and the osd is prepared with LVM
 	hostIPC := osdProps.storeConfig.EncryptedDevice
 
 	DNSPolicy := v1.DNSClusterFirst
@@ -340,19 +388,38 @@ func (c *Cluster) makeDeployment(osdProps osdProperties, osd OSDInfo, provisionC
 	if doBinaryCopyInit {
 		initContainers = append(initContainers, *copyBinariesContainer)
 	}
-	if osdOnPVC {
-		initContainers = append(initContainers, c.getPVCInitContainer(osdProps.pvc))
+	if osdOnPVC && osd.CVMode == "lvm" {
+		initContainers = append(initContainers, c.getPVCInitContainer(osdProps))
+	}
+
+	if osdOnPVC && osd.CVMode == "raw" {
+		initContainers = append(initContainers, c.getPVCInitContainerActivate(osdDataDirPath, osdProps))
+		if osdOnPVCMetadata {
+			initContainers = append(initContainers, c.getPVCMetadataInitContainerActivate(osdDataDirPath, osdProps))
+		}
+		initContainers = append(initContainers, c.getActivatePVCInitContainer(osdProps, osdID))
 	}
 	if doActivateOSDInit {
 		initContainers = append(initContainers, *activateOSDContainer)
 	}
 
+	// For OSD on PVC with LVM the directory does not exist yet
+	// It gets created by the 'ceph-volume lvm activate' command
+	//
+	// 	So OSD non-PVC the directory has been created by the 'activate' container already and has chown it
+	// So we don't need to chown it again
+	dataPath := ""
+
+	// Raw mode on PVC needs this path so that OSD's metadata files can be chown after 'ceph-bluestore-tool' ran
+	if osd.CVMode == "raw" && osdOnPVC {
+		dataPath = activateOSDMountPath + osdID
+	}
+
 	// Doing a chown in a post start lifecycle hook does not reliably complete before the OSD
 	// process starts, which can cause the pod to fail without the lifecycle hook's chown command
 	// completing. It can take an arbitrarily long time for a pod restart to successfully chown the
 	// directory. This is a race condition for all OSDs; therefore, do this in an init container.
 	// See more discussion here: https://github.com/rook/rook/pull/3594#discussion_r312279176
-	dataPath := ""
 	initContainers = append(initContainers,
 		opspec.ChownCephDataDirsInitContainer(
 			opconfig.DataPathMap{ContainerDataDir: dataPath},
@@ -457,6 +524,7 @@ func (c *Cluster) getCopyBinariesContainer() (v1.Volume, *v1.Container) {
 
 // This container runs all the actions needed to activate an OSD before we can run the OSD process
 func (c *Cluster) getActivateOSDInitContainer(osdID string, osdInfo OSDInfo, osdProps osdProperties) (v1.Volume, *v1.Container) {
+	osdOnPVC := osdProps.pvc.ClaimName != ""
 	volume := v1.Volume{Name: activateOSDVolumeName, VolumeSource: v1.VolumeSource{EmptyDir: &v1.EmptyDirVolumeSource{}}}
 	envVars := osdActivateEnvVar()
 	osdStore := "--bluestore"
@@ -470,7 +538,7 @@ func (c *Cluster) getActivateOSDInitContainer(osdID string, osdInfo OSDInfo, osd
 		{Name: k8sutil.ConfigOverrideName, ReadOnly: true, MountPath: opconfig.EtcCephDir},
 	}
 
-	if osdProps.pvc.ClaimName != "" {
+	if osdOnPVC {
 		volMounts = append(volMounts, getPvcOSDBridgeMount(osdProps.pvc.ClaimName))
 	}
 
@@ -479,23 +547,25 @@ func (c *Cluster) getActivateOSDInitContainer(osdID string, osdInfo OSDInfo, osd
 		Privileged: &privileged,
 	}
 
-	return volume, &v1.Container{
+	container := &v1.Container{
 		Command: []string{
 			"/bin/bash",
 			"-c",
-			fmt.Sprintf(activateOSDCode, osdID, osdInfo.UUID, osdStore, osdInfo.CVMode, osdInfo.BlockPath),
+			fmt.Sprintf(activateOSDCode, osdID, osdInfo.UUID, osdStore, osdInfo.CVMode, osdInfo.BlockPath, osdMetadataDeviceEnvVarName),
 		},
-		Name:            "activate-osd",
+		Name:            "activate",
 		Image:           c.cephVersion.Image,
 		VolumeMounts:    volMounts,
 		SecurityContext: securityContext,
 		Env:             envVars,
 		Resources:       osdProps.resources,
 	}
+
+	return volume, container
 }
 
 func (c *Cluster) provisionPodTemplateSpec(osdProps osdProperties, restart v1.RestartPolicy, provisionConfig *provisionConfig) (*v1.PodTemplateSpec, error) {
-
+	osdOnPVC := osdProps.pvc.ClaimName != ""
 	copyBinariesVolume, copyBinariesContainer := c.getCopyBinariesContainer()
 
 	// ceph-volume is currently set up to use /etc/ceph/ceph.conf; this means no user config
@@ -503,7 +573,7 @@ func (c *Cluster) provisionPodTemplateSpec(osdProps osdProperties, restart v1.Re
 	volumes := append(opspec.PodVolumes(provisionConfig.DataPathMap, c.dataDirHostPath, true), copyBinariesVolume)
 
 	// by default, don't define any volume config unless it is required
-	if len(osdProps.devices) > 0 || osdProps.selection.DeviceFilter != "" || osdProps.selection.DevicePathFilter != "" || osdProps.selection.GetUseAllDevices() || osdProps.metadataDevice != "" || osdProps.pvc.ClaimName != "" {
+	if len(osdProps.devices) > 0 || osdProps.selection.DeviceFilter != "" || osdProps.selection.DevicePathFilter != "" || osdProps.selection.GetUseAllDevices() || osdProps.metadataDevice != "" || osdOnPVC {
 		// 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)
@@ -511,7 +581,7 @@ func (c *Cluster) provisionPodTemplateSpec(osdProps osdProperties, restart v1.Re
 		volumes = append(volumes, udevVolume)
 	}
 
-	if osdProps.pvc.ClaimName != "" {
+	if osdOnPVC {
 		// Create volume config for PVCs
 		volumes = append(volumes, getPVCOSDVolumes(&osdProps)...)
 	}
@@ -536,7 +606,7 @@ func (c *Cluster) provisionPodTemplateSpec(osdProps osdProperties, restart v1.Re
 	if c.Network.IsHost() {
 		podSpec.DNSPolicy = v1.DNSClusterFirstWithHostNet
 	}
-	if len(osdProps.pvc.ClaimName) == 0 {
+	if !osdOnPVC {
 		c.placement.ApplyToPodSpec(&podSpec)
 	} else {
 		osdProps.placement.ApplyToPodSpec(&podSpec)
@@ -567,31 +637,136 @@ func (c *Cluster) provisionPodTemplateSpec(osdProps osdProperties, restart v1.Re
 // Currently we can't mount a block mode pv directly to a privileged container
 // So we mount it to a non privileged 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 {
+func (c *Cluster) getPVCInitContainer(osdProps osdProperties) v1.Container {
 	return v1.Container{
 		Name:  blockPVCMapperInitContainer,
 		Image: c.cephVersion.Image,
 		Command: []string{
 			"cp",
 		},
-		Args: []string{"-a", fmt.Sprintf("/%s", pvc.ClaimName), fmt.Sprintf("/mnt/%s", pvc.ClaimName)},
+		Args: []string{"-a", fmt.Sprintf("/%s", osdProps.pvc.ClaimName), fmt.Sprintf("/mnt/%s", osdProps.pvc.ClaimName)},
 		VolumeDevices: []v1.VolumeDevice{
 			{
-				Name:       pvc.ClaimName,
-				DevicePath: fmt.Sprintf("/%s", pvc.ClaimName),
+				Name:       osdProps.pvc.ClaimName,
+				DevicePath: fmt.Sprintf("/%s", osdProps.pvc.ClaimName),
 			},
 		},
 		VolumeMounts: []v1.VolumeMount{
 			{
 				MountPath: "/mnt",
-				Name:      fmt.Sprintf("%s-bridge", pvc.ClaimName),
+				Name:      fmt.Sprintf("%s-bridge", osdProps.pvc.ClaimName),
 			},
 		},
 		SecurityContext: opmon.PodSecurityContext(),
+		Resources:       osdProps.resources,
 	}
 }
 
+func (c *Cluster) getPVCInitContainerActivate(mountPath string, osdProps osdProperties) v1.Container {
+
+	return v1.Container{
+		Name:  blockPVCMapperInitContainer,
+		Image: c.cephVersion.Image,
+		Command: []string{
+			"cp",
+		},
+		Args: []string{"-a", fmt.Sprintf("/%s", osdProps.pvc.ClaimName), path.Join(mountPath, "block")},
+		VolumeDevices: []v1.VolumeDevice{
+			{
+				Name:       osdProps.pvc.ClaimName,
+				DevicePath: fmt.Sprintf("/%s", osdProps.pvc.ClaimName),
+			},
+		},
+		VolumeMounts: []v1.VolumeMount{
+			{
+				MountPath: mountPath,
+				SubPath:   path.Base(mountPath),
+				Name:      fmt.Sprintf("%s-bridge", osdProps.pvc.ClaimName),
+			},
+		},
+		SecurityContext: opmon.PodSecurityContext(),
+		Resources:       osdProps.resources,
+	}
+}
+
+// The reason why this is not part of getPVCInitContainer is that this will change the deployment spec object
+// and thus restart the osd deployment, so it is better to have it separated and only enable it
+// It will change the deployment spec because we must add a new argument to the method like 'mountPath' and use it in the container name
+// otherwise we will end up with a new conflict during the job/deployment initialization
+func (c *Cluster) getPVCMetadataInitContainer(mountPath string, osdProps osdProperties) v1.Container {
+	return v1.Container{
+		Name:  blockPVCMetadataMapperInitContainer,
+		Image: c.cephVersion.Image,
+		Command: []string{
+			"cp",
+		},
+		Args: []string{"-a", fmt.Sprintf("/%s", osdProps.metadataPVC.ClaimName), fmt.Sprintf("/srv/%s", osdProps.metadataPVC.ClaimName)},
+		VolumeDevices: []v1.VolumeDevice{
+			{
+				Name:       osdProps.metadataPVC.ClaimName,
+				DevicePath: fmt.Sprintf("/%s", osdProps.metadataPVC.ClaimName),
+			},
+		},
+		VolumeMounts: []v1.VolumeMount{
+			{
+				MountPath: "/srv",
+				Name:      fmt.Sprintf("%s-bridge", osdProps.metadataPVC.ClaimName),
+			},
+		},
+		SecurityContext: opmon.PodSecurityContext(),
+		Resources:       osdProps.resources,
+	}
+}
+
+func (c *Cluster) getPVCMetadataInitContainerActivate(mountPath string, osdProps osdProperties) v1.Container {
+	return v1.Container{
+		Name:  blockPVCMetadataMapperInitContainer,
+		Image: c.cephVersion.Image,
+		Command: []string{
+			"cp",
+		},
+		Args: []string{"-a", fmt.Sprintf("/%s", osdProps.metadataPVC.ClaimName), path.Join(mountPath, "block.db")},
+		VolumeDevices: []v1.VolumeDevice{
+			{
+				Name:       osdProps.metadataPVC.ClaimName,
+				DevicePath: fmt.Sprintf("/%s", osdProps.metadataPVC.ClaimName),
+			},
+		},
+		// We need to call getPvcOSDBridgeMountActivate() so that we can copy the metadata block into the "main" empty dir
+		// This empty dir is passed along every init container
+		VolumeMounts:    []v1.VolumeMount{getPvcOSDBridgeMountActivate(mountPath, osdProps.pvc.ClaimName)},
+		SecurityContext: opmon.PodSecurityContext(),
+		Resources:       osdProps.resources,
+	}
+}
+
+func (c *Cluster) getActivatePVCInitContainer(osdProps osdProperties, osdID string) v1.Container {
+	osdDataPath := activateOSDMountPath + osdID
+	osdDataBlockPath := path.Join(osdDataPath, "block")
+
+	container := v1.Container{
+		Name:  activatePVCOSDInitContainer,
+		Image: c.cephVersion.Image,
+		Command: []string{
+			"ceph-bluestore-tool",
+		},
+		Args: []string{"prime-osd-dir", "--dev", osdDataBlockPath, "--path", osdDataPath, "--no-mon-config"},
+		VolumeDevices: []v1.VolumeDevice{
+			{
+				Name:       osdProps.pvc.ClaimName,
+				DevicePath: osdDataBlockPath,
+			},
+		},
+		VolumeMounts:    []v1.VolumeMount{getPvcOSDBridgeMountActivate(osdDataPath, osdProps.pvc.ClaimName)},
+		SecurityContext: opmon.PodSecurityContext(),
+		Resources:       osdProps.resources,
+	}
+
+	return container
+}
+
 func (c *Cluster) getConfigEnvVars(osdProps osdProperties, dataDir string) []v1.EnvVar {
+	osdOnPVC := osdProps.pvc.ClaimName != ""
 	envVars := []v1.EnvVar{
 		nodeNameEnvVar(osdProps.crushHostname),
 		{Name: "ROOK_CLUSTER_ID", Value: string(c.ownerRef.UID)},
@@ -614,7 +789,7 @@ func (c *Cluster) getConfigEnvVars(osdProps osdProperties, dataDir string) []v1.
 
 	// Give a hint to the prepare pod for what the host in the CRUSH map should be
 	crushmapHostname := osdProps.crushHostname
-	if !osdProps.portable && osdProps.pvc.ClaimName != "" {
+	if !osdProps.portable && osdOnPVC {
 		// If it's a pvc that's not portable we only know what the host name should be when inside the osd prepare pod
 		crushmapHostname = ""
 	}
@@ -643,11 +818,12 @@ func (c *Cluster) getConfigEnvVars(osdProps osdProperties, dataDir string) []v1.
 }
 
 func (c *Cluster) provisionOSDContainer(osdProps osdProperties, copyBinariesMount v1.VolumeMount, provisionConfig *provisionConfig) v1.Container {
-
+	osdOnPVC := osdProps.pvc.ClaimName != ""
+	osdOnPVCMetadata := osdProps.metadataPVC.ClaimName != ""
 	envVars := c.getConfigEnvVars(osdProps, k8sutil.DataDir)
 
 	devMountNeeded := false
-	if osdProps.pvc.ClaimName != "" {
+	if osdOnPVC {
 		devMountNeeded = true
 	}
 	privileged := false
@@ -712,14 +888,21 @@ func (c *Cluster) provisionOSDContainer(osdProps osdProperties, copyBinariesMoun
 		volumeMounts = append(volumeMounts, udevMount)
 	}
 
-	if osdProps.pvc.ClaimName != "" {
+	// If the OSD runs on PVC
+	if osdOnPVC {
 		volumeMounts = append(volumeMounts, getPvcOSDBridgeMount(osdProps.pvc.ClaimName))
-		envVars = append(envVars, dataDevicesEnvVar(strings.Join([]string{fmt.Sprintf("/mnt/%s", osdProps.pvc.ClaimName)}, ",")))
+		// The device list is read by the Rook CLI via environment variables so let's add them
+		dev := []string{fmt.Sprintf("/mnt/%s", osdProps.pvc.ClaimName)}
+		if osdOnPVCMetadata {
+			volumeMounts = append(volumeMounts, getPvcMetadataOSDBridgeMount(osdProps.metadataPVC.ClaimName))
+			dev = append(dev, fmt.Sprintf("/srv/%s", osdProps.metadataPVC.ClaimName))
+		}
+		envVars = append(envVars, dataDevicesEnvVar(strings.Join(dev, ",")))
 		envVars = append(envVars, pvcBackedOSDEnvVar("true"))
 	}
 
 	// 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" || osdProps.pvc.ClaimName != "" {
+	if devMountNeeded || os.Getenv("ROOK_HOSTPATH_REQUIRES_PRIVILEGED") == "true" || osdOnPVC {
 		privileged = true
 	}
 	runAsUser := int64(0)
@@ -749,6 +932,14 @@ func getPvcOSDBridgeMount(claimName string) v1.VolumeMount {
 	return v1.VolumeMount{Name: fmt.Sprintf("%s-bridge", claimName), MountPath: "/mnt"}
 }
 
+func getPvcOSDBridgeMountActivate(mountPath, claimName string) v1.VolumeMount {
+	return v1.VolumeMount{Name: fmt.Sprintf("%s-bridge", claimName), MountPath: mountPath, SubPath: path.Base(mountPath)}
+}
+
+func getPvcMetadataOSDBridgeMount(claimName string) v1.VolumeMount {
+	return v1.VolumeMount{Name: fmt.Sprintf("%s-bridge", claimName), MountPath: "/srv"}
+}
+
 func (c *Cluster) skipVolumeForDirectory(path string) bool {
 	// If attempting to add a directory at /var/lib/rook, we need to skip the volume and volume mount
 	// since the dataDirHostPath is always mounting at /var/lib/rook
@@ -756,7 +947,7 @@ func (c *Cluster) skipVolumeForDirectory(path string) bool {
 }
 
 func getPVCOSDVolumes(osdProps *osdProperties) []v1.Volume {
-	return []v1.Volume{
+	volumes := []v1.Volume{
 		{
 			Name: osdProps.pvc.ClaimName,
 			VolumeSource: v1.VolumeSource{
@@ -775,6 +966,35 @@ func getPVCOSDVolumes(osdProps *osdProperties) []v1.Volume {
 			},
 		},
 	}
+
+	// If we have a metadata PVC let's add it
+	if osdProps.metadataPVC.ClaimName != "" {
+		metadataPVCVolume := []v1.Volume{
+			{
+				Name: osdProps.metadataPVC.ClaimName,
+				VolumeSource: v1.VolumeSource{
+					PersistentVolumeClaim: &osdProps.metadataPVC,
+				},
+			},
+			{
+				// We need a bridge mount which is basically a common volume mount between the non privileged 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.metadataPVC.ClaimName),
+				VolumeSource: v1.VolumeSource{
+					EmptyDir: &v1.EmptyDirVolumeSource{
+						Medium: "Memory",
+					},
+				},
+			},
+		}
+
+		volumes = append(volumes, metadataPVCVolume...)
+	}
+
+	logger.Debugf("volumes are %+v", volumes)
+
+	return volumes
 }
 
 func nodeNameEnvVar(name string) v1.EnvVar {
@@ -826,15 +1046,22 @@ func osdOnSDNFlag(network cephv1.NetworkSpec, v cephver.CephVersion) []string {
 	return args
 }
 
-func makeStorageClassDeviceSetPVCID(storageClassDeviceSetName string, setIndex, pvcIndex int) (pvcID, pvcLabelSelector string) {
-	pvcStorageClassDeviceSetPVCId := fmt.Sprintf("%s-%v", storageClassDeviceSetName, setIndex)
+func makeStorageClassDeviceSetPVCID(storageClassDeviceSetName string, setIndex int) (pvcID, pvcLabelSelector string) {
+	pvcStorageClassDeviceSetPVCId := fmt.Sprintf("%s-%d", storageClassDeviceSetName, setIndex)
 	return pvcStorageClassDeviceSetPVCId, fmt.Sprintf("%s=%s", CephDeviceSetPVCIDLabelKey, pvcStorageClassDeviceSetPVCId)
 }
 
-func makeStorageClassDeviceSetPVCLabel(storageClassDeviceSetName, pvcStorageClassDeviceSetPVCId string, pvcIndex, setIndex int) map[string]string {
+// This is the new function that generates the labels
+// It includes the pvcTemplateName in it
+func makeStorageClassDeviceSetPVCIDNew(storageClassDeviceSetName, pvcTemplateName string, setIndex int) (pvcID, pvcLabelSelector string) {
+	pvcStorageClassDeviceSetPVCId := fmt.Sprintf("%s-%s-%d", storageClassDeviceSetName, strings.Replace(pvcTemplateName, " ", "-", -1), setIndex)
+	return pvcStorageClassDeviceSetPVCId, fmt.Sprintf("%s=%s", CephDeviceSetPVCIDLabelKey, pvcStorageClassDeviceSetPVCId)
+}
+
+func makeStorageClassDeviceSetPVCLabel(storageClassDeviceSetName, pvcStorageClassDeviceSetPVCId string, setIndex int) map[string]string {
 	return map[string]string{
 		CephDeviceSetLabelKey:      storageClassDeviceSetName,
-		CephSetIndexLabelKey:       fmt.Sprintf("%v", setIndex),
+		CephSetIndexLabelKey:       fmt.Sprintf("%d", setIndex),
 		CephDeviceSetPVCIDLabelKey: pvcStorageClassDeviceSetPVCId,
 	}
 }
diff --git a/pkg/operator/ceph/cluster/osd/spec_test.go b/pkg/operator/ceph/cluster/osd/spec_test.go
index a51b160d9..64aa9e6f7 100644
--- a/pkg/operator/ceph/cluster/osd/spec_test.go
+++ b/pkg/operator/ceph/cluster/osd/spec_test.go
@@ -109,6 +109,7 @@ func testPodDevices(t *testing.T, dataDir, deviceName string, allDevices bool) {
 		DataPathMap: opconfig.NewDatalessDaemonDataPathMap(c.Namespace, "/var/lib/rook"),
 	}
 
+	// Test LVM based on OSD on bare metal
 	deployment, err := c.makeDeployment(osdProp, osd, dataPathMap)
 	assert.Nil(t, err)
 	assert.NotNil(t, deployment)
@@ -139,7 +140,7 @@ func testPodDevices(t *testing.T, dataDir, deviceName string, allDevices bool) {
 	assert.Equal(t, 2, len(deployment.Spec.Template.Spec.InitContainers))
 	initCont := deployment.Spec.Template.Spec.InitContainers[0]
 	assert.Equal(t, "ceph/ceph:v12.2.8", initCont.Image)
-	assert.Equal(t, "activate-osd", initCont.Name)
+	assert.Equal(t, "activate", initCont.Name)
 	assert.Equal(t, 3, len(initCont.VolumeMounts))
 
 	assert.Equal(t, 1, len(deployment.Spec.Template.Spec.Containers))
@@ -147,6 +148,74 @@ func testPodDevices(t *testing.T, dataDir, deviceName string, allDevices bool) {
 	assert.Equal(t, cephVersion.Image, cont.Image)
 	assert.Equal(t, 6, len(cont.VolumeMounts))
 	assert.Equal(t, "ceph-osd", cont.Command[0])
+
+	// Test OSD on PVC with LVM
+	osdProp = osdProperties{
+		crushHostname: n.Name,
+		selection:     n.Selection,
+		resources:     v1.ResourceRequirements{},
+		storeConfig:   config.StoreConfig{},
+		pvc:           v1.PersistentVolumeClaimVolumeSource{ClaimName: "mypvc"},
+	}
+	// Not needed when running on PVC
+	osd = OSDInfo{
+		ID:     0,
+		CVMode: "lvm",
+	}
+
+	deployment, err = c.makeDeployment(osdProp, osd, dataPathMap)
+	assert.Nil(t, err)
+	assert.NotNil(t, deployment)
+	assert.Equal(t, 4, len(deployment.Spec.Template.Spec.InitContainers), deployment.Spec.Template.Spec.InitContainers[2].Name)
+	assert.Equal(t, "config-init", deployment.Spec.Template.Spec.InitContainers[0].Name)
+	assert.Equal(t, "copy-bins", deployment.Spec.Template.Spec.InitContainers[1].Name)
+	assert.Equal(t, "blkdevmapper", deployment.Spec.Template.Spec.InitContainers[2].Name)
+	assert.Equal(t, "chown-container-data-dir", deployment.Spec.Template.Spec.InitContainers[3].Name)
+	assert.Equal(t, 1, len(deployment.Spec.Template.Spec.Containers))
+	initCont = deployment.Spec.Template.Spec.InitContainers[0]
+	assert.Equal(t, 4, len(initCont.VolumeMounts), initCont.VolumeMounts)
+	blkInitCont := deployment.Spec.Template.Spec.InitContainers[2]
+	assert.Equal(t, 1, len(blkInitCont.VolumeDevices))
+	cont = deployment.Spec.Template.Spec.Containers[0]
+	assert.Equal(t, 8, len(cont.VolumeMounts), cont.VolumeMounts)
+
+	// Test OSD on PVC with RAW
+	osd = OSDInfo{
+		ID:     0,
+		CVMode: "raw",
+	}
+	deployment, err = c.makeDeployment(osdProp, osd, dataPathMap)
+	assert.Nil(t, err)
+	assert.NotNil(t, deployment)
+	assert.Equal(t, 3, len(deployment.Spec.Template.Spec.InitContainers), deployment.Spec.Template.Spec.InitContainers[2].Name)
+	assert.Equal(t, "blkdevmapper", deployment.Spec.Template.Spec.InitContainers[0].Name)
+	assert.Equal(t, "activate", deployment.Spec.Template.Spec.InitContainers[1].Name)
+	assert.Equal(t, "chown-container-data-dir", deployment.Spec.Template.Spec.InitContainers[2].Name)
+	assert.Equal(t, 1, len(deployment.Spec.Template.Spec.Containers))
+	cont = deployment.Spec.Template.Spec.Containers[0]
+	assert.Equal(t, 5, len(cont.VolumeMounts), cont.VolumeMounts)
+
+	// Test OSD on PVC with RAW and metadata device
+	osd = OSDInfo{
+		ID:     0,
+		CVMode: "raw",
+	}
+	osdProp.metadataPVC = v1.PersistentVolumeClaimVolumeSource{ClaimName: "mypvc-metadata"}
+	deployment, err = c.makeDeployment(osdProp, osd, dataPathMap)
+	assert.Nil(t, err)
+	assert.NotNil(t, deployment)
+	assert.Equal(t, 4, len(deployment.Spec.Template.Spec.InitContainers))
+	assert.Equal(t, "blkdevmapper", deployment.Spec.Template.Spec.InitContainers[0].Name)
+	assert.Equal(t, "blkdevmapper-metadata", deployment.Spec.Template.Spec.InitContainers[1].Name)
+	assert.Equal(t, "activate", deployment.Spec.Template.Spec.InitContainers[2].Name)
+	assert.Equal(t, "chown-container-data-dir", deployment.Spec.Template.Spec.InitContainers[3].Name)
+	assert.Equal(t, 1, len(deployment.Spec.Template.Spec.Containers))
+	cont = deployment.Spec.Template.Spec.Containers[0]
+	assert.Equal(t, 5, len(cont.VolumeMounts), cont.VolumeMounts)
+	blkInitCont = deployment.Spec.Template.Spec.InitContainers[1]
+	assert.Equal(t, 1, len(blkInitCont.VolumeDevices))
+	blkMetaInitCont := deployment.Spec.Template.Spec.InitContainers[2]
+	assert.Equal(t, 1, len(blkMetaInitCont.VolumeDevices))
 }
 
 func verifyEnvVar(t *testing.T, envVars []v1.EnvVar, expectedName, expectedValue string, expectedFound bool) {
diff --git a/pkg/util/sys/device.go b/pkg/util/sys/device.go
index 33578fd2f..b238dd8ee 100644
--- a/pkg/util/sys/device.go
+++ b/pkg/util/sys/device.go
@@ -204,6 +204,11 @@ func GetDevicePropertiesFromPath(devicePath string, executor exec.Executor) (map
 	output, err := executor.ExecuteCommandWithOutput(false, cmd, "lsblk", devicePath,
 		"--bytes", "--nodeps", "--pairs", "--output", "SIZE,ROTA,RO,TYPE,PKNAME,NAME")
 	if err != nil {
+		// The "not a block device" error also returns code 32 so the ExitStatus() check hides this error
+		if strings.Contains(output, "not a block device") {
+			return nil, err
+		}
+
 		// try to get more information about the command error
 		cmdErr, ok := err.(*exec.CommandError)
 		if ok && cmdErr.ExitStatus() == 32 {