Merge pull request #8028 from degorenko/device-classes-resource-limits

ceph: add ability to set resource limit for OSDs based on device classes

Documentation/ceph-cluster-crd.md

@@ -538,6 +538,10 @@ You can set resource requests/limits for Rook components through the [Resource R
 
 * `mon`: Set resource requests/limits for mons
 * `osd`: Set resource requests/limits for OSDs
+  This key applies for all OSDs regardless of their device classes. In case of need to apply resource requests/limits for OSDs with
+  particular device class use specific osd keys below.
+* `osd-<deviceClass>`: Set resource requests/limits for OSDs on a specific device class. Rook will automatically detect `hdd`,
+  `ssd`, or `nvme` device classes. Custom device classes can also be set.
 * `mgr`: Set resource requests/limits for MGRs
 * `mgr-sidecar`: Set resource requests/limits for the MGR sidecar, which is only created when `mgr.count: 2`.
   The sidecar requires very few resources since it only executes every 15 seconds to query Ceph for the active

cluster/examples/kubernetes/ceph/cluster.yaml

@@ -189,6 +189,10 @@ spec:
 # The above example requests/limits can also be added to the other components
 #    mon:
 #    osd:
+# For OSD it also is a possible to specify requests/limits based on device class
+#    osd-hdd:
+#    osd-ssd:
+#    osd-nvme:
 #    prepareosd:
 #    mgr-sidecar:
 #    crashcollector:

pkg/apis/ceph.rook.io/v1/resources.go

@@ -27,7 +27,7 @@ const (
 	ResourcesKeyMgr = "mgr"
 	// ResourcesKeyMgrSidecar represents the name of resource in the CR for a mgr
 	ResourcesKeyMgrSidecar = "mgr-sidecar"
-	// ResourcesKeyOSD represents the name of resource in the CR for an osd
+	// ResourcesKeyOSD represents the name of a resource in the CR for all OSDs
 	ResourcesKeyOSD = "osd"
 	// ResourcesKeyPrepareOSD represents the name of resource in the CR for the osd prepare job
 	ResourcesKeyPrepareOSD = "prepareosd"
@@ -60,11 +60,24 @@ func GetMonResources(p ResourceSpec) v1.ResourceRequirements {
 	return p[ResourcesKeyMon]
 }
 
-// GetOSDResources returns the placement for the OSDs
-func GetOSDResources(p ResourceSpec) v1.ResourceRequirements {
+// GetOSDResources returns the placement for all OSDs or for OSDs of specified device class (hdd, nvme, ssd)
+func GetOSDResources(p ResourceSpec, deviceClass string) v1.ResourceRequirements {
+	if deviceClass == "" {
+		return p[ResourcesKeyOSD]
+	}
+	// if device class specified, but not set in requirements return common osd requirements if present
+	r, ok := p[getOSDResourceKeyForDeviceClass(deviceClass)]
+	if ok {
+		return r
+	}
 	return p[ResourcesKeyOSD]
 }
 
+// getOSDResourceKeyForDeviceClass returns key name for device class in resources spec
+func getOSDResourceKeyForDeviceClass(deviceClass string) string {
+	return ResourcesKeyOSD + "-" + deviceClass
+}
+
 // GetPrepareOSDResources returns the placement for the OSDs prepare job
 func GetPrepareOSDResources(p ResourceSpec) v1.ResourceRequirements {
 	return p[ResourcesKeyPrepareOSD]

pkg/daemon/ceph/cleanup/disk.go

@@ -67,7 +67,7 @@ func (s *DiskSanitizer) StartSanitizeDisks() {
 	}
 
 	// Raw based OSDs
-	osdRawList, err := osd.GetCephVolumeRawOSDs(s.context, s.clusterInfo, s.clusterInfo.FSID, "", "", "", false)
+	osdRawList, err := osd.GetCephVolumeRawOSDs(s.context, s.clusterInfo, s.clusterInfo.FSID, "", "", "", false, true)
 	if err != nil {
 		logger.Errorf("failed to list raw osd(s). %v", err)
 	} else {

pkg/daemon/ceph/osd/daemon.go

@@ -346,7 +346,7 @@ func getAvailableDevices(context *clusterd.Context, agent *OsdAgent) (*DeviceOsd
 		rejectedReason := ""
 		if agent.pvcBacked {
 			block := fmt.Sprintf("/mnt/%s", agent.nodeName)
-			rawOsds, err := GetCephVolumeRawOSDs(context, agent.clusterInfo, agent.clusterInfo.FSID, block, agent.metadataDevice, "", false)
+			rawOsds, err := GetCephVolumeRawOSDs(context, agent.clusterInfo, agent.clusterInfo.FSID, block, agent.metadataDevice, "", false, true)
 			if err != nil {
 				isAvailable = false
 				rejectedReason = fmt.Sprintf("failed to detect if there is already an osd. %v", err)
@@ -422,6 +422,20 @@ func getAvailableDevices(context *clusterd.Context, agent *OsdAgent) (*DeviceOsd
 				}
 				matchedDevice = desiredDevice
 
+				if matchedDevice.DeviceClass == "" {
+					classNotSet := true
+					if agent.pvcBacked {
+						crushDeviceClass := os.Getenv(oposd.CrushDeviceClassVarName)
+						if crushDeviceClass != "" {
+							matchedDevice.DeviceClass = crushDeviceClass
+							classNotSet = false
+						}
+					}
+					if classNotSet {
+						matchedDevice.DeviceClass = sys.GetDiskDeviceClass(device)
+					}
+				}
+
 				if matched {
 					break
 				}

pkg/daemon/ceph/osd/volume.go

@@ -80,9 +80,10 @@ type osdInfo struct {
 }
 
 type osdTags struct {
-	OSDFSID     string `json:"ceph.osd_fsid"`
-	Encrypted   string `json:"ceph.encrypted"`
-	ClusterFSID string `json:"ceph.cluster_fsid"`
+	OSDFSID          string `json:"ceph.osd_fsid"`
+	Encrypted        string `json:"ceph.encrypted"`
+	ClusterFSID      string `json:"ceph.cluster_fsid"`
+	CrushDeviceClass string `json:"ceph.crush_device_class"`
 }
 
 type cephVolReport struct {
@@ -168,7 +169,7 @@ func (a *OsdAgent) configureCVDevices(context *clusterd.Context, devices *Device
 				// I'm leaving this code with an empty metadata device for now
 				metadataBlock, walBlock = "", ""
 
-				rawOsds, err = GetCephVolumeRawOSDs(context, a.clusterInfo, a.clusterInfo.FSID, block, metadataBlock, walBlock, lvBackedPV)
+				rawOsds, err = GetCephVolumeRawOSDs(context, a.clusterInfo, a.clusterInfo.FSID, block, metadataBlock, walBlock, lvBackedPV, false)
 				if err != nil {
 					logger.Infof("failed to get device already provisioned by ceph-volume raw. %v", err)
 				}
@@ -186,7 +187,7 @@ func (a *OsdAgent) configureCVDevices(context *clusterd.Context, devices *Device
 		osds = append(osds, lvmOsds...)
 
 		// List existing OSD(s) configured with ceph-volume raw mode
-		rawOsds, err = GetCephVolumeRawOSDs(context, a.clusterInfo, a.clusterInfo.FSID, block, "", "", false)
+		rawOsds, err = GetCephVolumeRawOSDs(context, a.clusterInfo, a.clusterInfo.FSID, block, "", "", false, false)
 		if err != nil {
 			logger.Infof("failed to get device already provisioned by ceph-volume raw. %v", err)
 		}
@@ -266,7 +267,7 @@ func (a *OsdAgent) configureCVDevices(context *clusterd.Context, devices *Device
 	if !a.pvcBacked {
 		block = ""
 	}
-	rawOsds, err = GetCephVolumeRawOSDs(context, a.clusterInfo, a.clusterInfo.FSID, block, metadataBlock, walBlock, lvBackedPV)
+	rawOsds, err = GetCephVolumeRawOSDs(context, a.clusterInfo, a.clusterInfo.FSID, block, metadataBlock, walBlock, lvBackedPV, false)
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to get devices already provisioned by ceph-volume raw")
 	}
@@ -865,13 +866,14 @@ func GetCephVolumeLVMOSDs(context *clusterd.Context, clusterInfo *client.Cluster
 			logger.Errorf("bad osd returned from ceph-volume %q", name)
 			continue
 		}
-		var osdFSID string
+		var osdFSID, osdDeviceClass string
 		for _, osd := range osdInfo {
 			if osd.Tags.ClusterFSID != cephfsid {
 				logger.Infof("skipping osd%d: %q running on a different ceph cluster %q", id, osd.Tags.OSDFSID, osd.Tags.ClusterFSID)
 				continue
 			}
 			osdFSID = osd.Tags.OSDFSID
+			osdDeviceClass = osd.Tags.CrushDeviceClass
 
 			// If no lv is specified let's take the one we discovered
 			if lv == "" {
@@ -900,6 +902,7 @@ func GetCephVolumeLVMOSDs(context *clusterd.Context, clusterInfo *client.Cluster
 			LVBackedPV:    lvBackedPV,
 			CVMode:        cvMode,
 			Store:         "bluestore",
+			DeviceClass:   osdDeviceClass,
 		}
 		osds = append(osds, osd)
 	}
@@ -929,7 +932,7 @@ func readCVLogContent(cvLogFilePath string) string {
 }
 
 // GetCephVolumeRawOSDs list OSD prepared with raw mode
-func GetCephVolumeRawOSDs(context *clusterd.Context, clusterInfo *client.ClusterInfo, cephfsid, block, metadataBlock, walBlock string, lvBackedPV bool) ([]oposd.OSDInfo, error) {
+func GetCephVolumeRawOSDs(context *clusterd.Context, clusterInfo *client.ClusterInfo, cephfsid, block, metadataBlock, walBlock string, lvBackedPV, skipDeviceClass bool) ([]oposd.OSDInfo, error) {
 	// lv can be a block device if raw mode is used
 	cvMode := "raw"
 
@@ -1007,6 +1010,15 @@ func GetCephVolumeRawOSDs(context *clusterd.Context, clusterInfo *client.Cluster
 			Store:         "bluestore",
 		}
 
+		if !skipDeviceClass {
+			diskInfo, err := clusterd.PopulateDeviceInfo(blockPath, context.Executor)
+			if err != nil {
+				return nil, errors.Wrapf(err, "failed to get device info for %q", blockPath)
+			}
+			osd.DeviceClass = sys.GetDiskDeviceClass(diskInfo)
+			logger.Infof("setting device class %q for device %q", osd.DeviceClass, diskInfo.Name)
+		}
+
 		// If this is an encrypted OSD
 		if os.Getenv(oposd.CephVolumeEncryptedKeyEnvVarName) != "" {
 			// // Set subsystem and label for recovery and detection

pkg/daemon/ceph/osd/volume_test.go

@@ -340,6 +340,9 @@ func TestConfigureCVDevices(t *testing.T) {
 			if command == "lsblk" && args[0] == mountedDev {
 				return fmt.Sprintf(`SIZE="17179869184" ROTA="1" RO="0" TYPE="lvm" PKNAME="" NAME="%s" KNAME="/dev/dm-1, a ...interface{})`, mapperDev), nil
 			}
+			if command == "sgdisk" {
+				return "Disk identifier (GUID): 18484D7E-5287-4CE9-AC73-D02FB69055CE", nil
+			}
 			if contains(args, "lvm") && contains(args, "list") {
 				return `{}`, nil
 			}
@@ -429,6 +432,9 @@ func TestConfigureCVDevices(t *testing.T) {
 			if command == "lsblk" && args[0] == mountedDev {
 				return fmt.Sprintf(`SIZE="17179869184" ROTA="1" RO="0" TYPE="lvm" PKNAME="" NAME="%s" KNAME="/dev/dm-1, a ...interface{})`, mapperDev), nil
 			}
+			if command == "sgdisk" {
+				return "Disk identifier (GUID): 18484D7E-5287-4CE9-AC73-D02FB69055CE", nil
+			}
 			if args[1] == "ceph-volume" && args[4] == "lvm" && args[5] == "list" && args[6] == mapperDev {
 				return `{}`, nil
 			}
@@ -549,8 +555,9 @@ func TestConfigureCVDevices(t *testing.T) {
 		executor := &exectest.MockExecutor{}
 		executor.MockExecuteCommandWithOutput = func(command string, args ...string) (string, error) {
 			logger.Infof("[MockExecuteCommandWithOutput] %s %v", command, args)
-			if command == "lsblk" && args[0] == mountedDev {
-				return fmt.Sprintf(`SIZE="17179869184" ROTA="1" RO="0" TYPE="lvm" PKNAME="" NAME="%s" KNAME="/dev/dm-1"`, mapperDev), nil
+			// get lsblk for disks from cephVolumeRAWTestResult var
+			if command == "lsblk" && (args[0] == "/dev/vdb" || args[0] == "/dev/vdc") {
+				return fmt.Sprintf(`SIZE="17179869184" ROTA="1" RO="0" TYPE="disk" PKNAME="" NAME="%s" KNAME="%s"`, args[0], args[0]), nil
 			}
 			if args[1] == "ceph-volume" && args[4] == "raw" && args[5] == "list" {
 				return cephVolumeRAWTestResult, nil
@@ -558,6 +565,9 @@ func TestConfigureCVDevices(t *testing.T) {
 			if args[1] == "ceph-volume" && args[4] == "lvm" && args[5] == "list" {
 				return `{}`, nil
 			}
+			if command == "sgdisk" {
+				return "Disk identifier (GUID): 18484D7E-5287-4CE9-AC73-D02FB69055CE", nil
+			}
 			return "", errors.Errorf("unknown command %s %s", command, args)
 		}
 		deviceClassSet := false
@@ -1112,12 +1122,19 @@ func TestParseCephVolumeRawResult(t *testing.T) {
 			}
 		}
 
+		// get lsblk for disks from cephVolumeRAWTestResult var
+		if command == "lsblk" && (args[0] == "/dev/vdb" || args[0] == "/dev/vdc") {
+			return fmt.Sprintf(`SIZE="17179869184" ROTA="1" RO="0" TYPE="disk" PKNAME="" NAME="%s" KNAME="%s"`, args[0], args[0]), nil
+		}
+		if command == "sgdisk" {
+			return "Disk identifier (GUID): 18484D7E-5287-4CE9-AC73-D02FB69055CE", nil
+		}
 		return "", errors.Errorf("unknown command: %s, args: %#v", command, args)
 	}
 	clusterInfo := &cephclient.ClusterInfo{Namespace: "name"}
 
 	context := &clusterd.Context{Executor: executor, Clientset: test.New(t, 3)}
-	osds, err := GetCephVolumeRawOSDs(context, clusterInfo, "4bfe8b72-5e69-4330-b6c0-4d914db8ab89", "", "", "", false)
+	osds, err := GetCephVolumeRawOSDs(context, clusterInfo, "4bfe8b72-5e69-4330-b6c0-4d914db8ab89", "", "", "", false, false)
 	assert.Nil(t, err)
 	require.NotNil(t, osds)
 	assert.Equal(t, 2, len(osds))

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

@@ -316,7 +316,8 @@ func (c *Cluster) startProvisioningOverNodes(config *provisionConfig, errs *prov
 		}
 
 		// fully resolve the storage config and resources for this node
-		n := c.resolveNode(node.Name)
+		// don't care about osd device class resources since it will be overwritten later for prepareosd resources
+		n := c.resolveNode(node.Name, "")
 		if n == nil {
 			logger.Warningf("node %q did not resolve", node.Name)
 			continue

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

@@ -31,6 +31,7 @@ const (
 	osdDatabaseSizeEnvVarName = "ROOK_OSD_DATABASE_SIZE"
 	osdWalSizeEnvVarName      = "ROOK_OSD_WAL_SIZE"
 	osdsPerDeviceEnvVarName   = "ROOK_OSDS_PER_DEVICE"
+	osdDeviceClassEnvVarName  = "ROOK_OSD_DEVICE_CLASS"
 	// EncryptedDeviceEnvVarName is used in the pod spec to indicate whether the OSD is encrypted or not
 	EncryptedDeviceEnvVarName = "ROOK_ENCRYPTED_DEVICE"
 	PVCNameEnvVarName         = "ROOK_PVC_NAME"
@@ -124,6 +125,10 @@ func devicePathFilterEnvVar(filter string) v1.EnvVar {
 	return v1.EnvVar{Name: "ROOK_DATA_DEVICE_PATH_FILTER", Value: filter}
 }
 
+func dataDeviceClassEnvVar(deviceClass string) v1.EnvVar {
+	return v1.EnvVar{Name: osdDeviceClassEnvVarName, Value: deviceClass}
+}
+
 func metadataDeviceEnvVar(metadataDevice string) v1.EnvVar {
 	return v1.EnvVar{Name: osdMetadataDeviceEnvVarName, Value: metadataDevice}
 }

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

@@ -97,6 +97,7 @@ type OSDInfo struct {
 	Cluster        string `json:"cluster"`
 	UUID           string `json:"uuid"`
 	DevicePartUUID string `json:"device-part-uuid"`
+	DeviceClass    string `json:"device-class"`
 	// 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"`
@@ -168,9 +169,13 @@ func (c *Cluster) Start() error {
 	errs := newProvisionErrors()
 
 	// Validate pod's memory if specified
-	err := controller.CheckPodMemory(cephv1.ResourcesKeyOSD, cephv1.GetOSDResources(c.spec.Resources), cephOsdPodMinimumMemory)
-	if err != nil {
-		return errors.Wrap(err, "failed to check pod memory")
+	for resourceKey, resourceValue := range c.spec.Resources {
+		if strings.HasPrefix(resourceKey, cephv1.ResourcesKeyOSD) {
+			err := controller.CheckPodMemory(resourceKey, resourceValue, cephOsdPodMinimumMemory)
+			if err != nil {
+				return errors.Wrap(err, "failed to check pod memory")
+			}
+		}
 	}
 	logger.Infof("start running osds in namespace %q", namespace)
 
@@ -256,7 +261,7 @@ func (c *Cluster) getExistingOSDDeploymentsOnPVCs() (*util.Set, error) {
 func deploymentOnNode(c *Cluster, osd OSDInfo, nodeName string, config *provisionConfig) (*appsv1.Deployment, error) {
 	osdLongName := fmt.Sprintf("OSD %d on node %q", osd.ID, nodeName)
 
-	osdProps, err := c.getOSDPropsForNode(nodeName)
+	osdProps, err := c.getOSDPropsForNode(nodeName, osd.DeviceClass)
 	if err != nil {
 		return nil, errors.Wrapf(err, "failed to generate config for %s", osdLongName)
 	}
@@ -277,7 +282,7 @@ func deploymentOnNode(c *Cluster, osd OSDInfo, nodeName string, config *provisio
 func deploymentOnPVC(c *Cluster, osd OSDInfo, pvcName string, config *provisionConfig) (*appsv1.Deployment, error) {
 	osdLongName := fmt.Sprintf("OSD %d on PVC %q", osd.ID, pvcName)
 
-	osdProps, err := c.getOSDPropsForPVC(pvcName)
+	osdProps, err := c.getOSDPropsForPVC(pvcName, osd.DeviceClass)
 	if err != nil {
 		return nil, errors.Wrapf(err, "failed to generate config for %s", osdLongName)
 	}
@@ -305,20 +310,20 @@ func setOSDProperties(c *Cluster, osdProps osdProperties, osd OSDInfo) error {
 	return nil
 }
 
-func (c *Cluster) resolveNode(nodeName string) *cephv1.Node {
+func (c *Cluster) resolveNode(nodeName, deviceClass string) *cephv1.Node {
 	// fully resolve the storage config and resources for this node
 	rookNode := c.ValidStorage.ResolveNode(nodeName)
 	if rookNode == nil {
 		return nil
 	}
-	rookNode.Resources = k8sutil.MergeResourceRequirements(rookNode.Resources, cephv1.GetOSDResources(c.spec.Resources))
+	rookNode.Resources = k8sutil.MergeResourceRequirements(rookNode.Resources, cephv1.GetOSDResources(c.spec.Resources, deviceClass))
 
 	return rookNode
 }
 
-func (c *Cluster) getOSDPropsForNode(nodeName string) (osdProperties, error) {
+func (c *Cluster) getOSDPropsForNode(nodeName, deviceClass string) (osdProperties, error) {
 	// fully resolve the storage config and resources for this node
-	n := c.resolveNode(nodeName)
+	n := c.resolveNode(nodeName, deviceClass)
 	if n == nil {
 		return osdProperties{}, errors.Errorf("failed to resolve node %q", nodeName)
 	}
@@ -337,7 +342,7 @@ func (c *Cluster) getOSDPropsForNode(nodeName string) (osdProperties, error) {
 	return osdProps, nil
 }
 
-func (c *Cluster) getOSDPropsForPVC(pvcName string) (osdProperties, error) {
+func (c *Cluster) getOSDPropsForPVC(pvcName, osdDeviceClass string) (osdProperties, error) {
 	for _, deviceSet := range c.deviceSets {
 		// The data PVC template is required.
 		dataSource, dataOK := deviceSet.PVCSources[bluestorePVCData]
@@ -360,7 +365,7 @@ func (c *Cluster) getOSDPropsForPVC(pvcName string) (osdProperties, error) {
 			}
 
 			if deviceSet.Resources.Limits == nil && deviceSet.Resources.Requests == nil {
-				deviceSet.Resources = cephv1.GetOSDResources(c.spec.Resources)
+				deviceSet.Resources = cephv1.GetOSDResources(c.spec.Resources, osdDeviceClass)
 			}
 
 			osdProps := osdProperties{
@@ -490,6 +495,9 @@ func (c *Cluster) getOSDInfo(d *appsv1.Deployment) (OSDInfo, error) {
 		if envVar.Name == osdWalDeviceEnvVarName {
 			osd.WalPath = envVar.Value
 		}
+		if envVar.Name == osdDeviceClassEnvVarName {
+			osd.DeviceClass = envVar.Value
+		}
 	}
 
 	// Needed for upgrade from v1.5 to v1.6. Rook v1.5 did not set ROOK_BLOCK_PATH for OSDs on nodes

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

@@ -390,6 +390,7 @@ func (c *Cluster) makeDeployment(osdProps osdProperties, osd OSDInfo, provisionC
 		{Name: "CEPH_ARGS", Value: "-m $(ROOK_CEPH_MON_HOST)"},
 		blockPathEnvVariable(osd.BlockPath),
 		cvModeEnvVariable(osd.CVMode),
+		dataDeviceClassEnvVar(osd.DeviceClass),
 	}...)
 	configEnvVars := append(c.getConfigEnvVars(osdProps, dataDir), []v1.EnvVar{
 		tiniEnvVar,

pkg/util/sys/device.go

@@ -284,6 +284,16 @@ func GetDiskUUID(device string, executor exec.Executor) (string, error) {
 	return parseUUID(device, output)
 }
 
+func GetDiskDeviceClass(disk *LocalDisk) string {
+	if disk.Rotational {
+		return "hdd"
+	}
+	if strings.Contains(disk.RealPath, "nvme") {
+		return "nvme"
+	}
+	return "ssd"
+}
+
 // CheckIfDeviceAvailable checks if a device is available for consumption. The caller
 // needs to decide based on the return values whether it is available.
 func CheckIfDeviceAvailable(executor exec.Executor, devicePath string, pvcBacked bool) (bool, string, error) {