Merge pull request #4916 from SUSE/drive-groups

Ceph: provision OSDs using Drive Groups

Documentation/ceph-cluster-crd.md

@@ -357,6 +357,43 @@ The following storage selection settings are specific to Ceph and do not apply t
 * Mimic 13.2.3 or newer
 * Nautilus
 
+### Storage Selection Via Ceph Drive Groups
+
+Ceph Drive Groups allow for specifying highly advanced OSD layouts on nodes including
+non-homogeneous nodes. They are a way to describe a cluster layout using the properties of disks.
+It gives the user an abstract way tell Ceph which disks should turn into an OSD with which
+configuration without knowing the specifics of device names and paths. You can target specific disks
+by their device type, by vendor or model, by size, by whether they are rotational, and more. Disks
+with various properties can be specified to be data disks or wal/db disks.
+
+As a brief example, let's assume nodes with 20 SSDs and 4 NVMe devices. A Drive Group could specify
+that all SSD devices should be data disks and two of the NVMe devices should be wal/db disks for the
+SSDs. That would leave two NVMe devices remaining for other usage, either for Ceph or another
+application.
+
+Ceph supports adding devices as OSDs by Ceph Drive Group definitions in later versions of Ceph
+Octopus (v15.2.5+).
+See Ceph Drive Group docs for more info: https://docs.ceph.com/docs/master/cephadm/drivegroups/.
+Drive Groups cannot be used to provision OSDs on PVCs.
+
+> **IMPORTANT:** When managing a Rook/Ceph cluster's OSD layouts with Drive Groups, the `storage`
+> config is mostly ignored. `storageClassDeviceSets` can still be used to create OSDs on PVC, but
+> Rook will no longer use `storage` configs for creating OSDs on a node's devices. To avoid
+> confusion, we recommend using the `storage` config OR `driveGroups` and never both. Because
+> `storage` and `driveGroups` should not be used simultaneously, Rook only supports provisioning
+> OSDs with Drive Groups on new Rook-Ceph clusters.
+
+A Drive Group is defined by a name, a Ceph Drive Group spec, and a Rook placement
+* `name`: A name for the Drive Group.
+* `spec`: The Ceph Drive group spec. Some components of the spec are treated differently in the
+  context of Rook as noted below:
+  * Rook overrides Ceph's definition of `placement` in order to use Rook's `placement` below.
+  * Rook overrides Ceph's deprecated `host_pattern` in order to use Rook's `placement` below.
+  * Rook overrides Ceph's `service_id` field to be the same as the Drive Group `name` above.
+* `placement`: The placement criteria for nodes to provision with the Drive Group.
+  (Optional) Default is no placement criteria, which matches all untainted nodes.
+  The syntax is the same as for [other placement configuration](#placement-configuration-settings).
+
 ### Annotations Configuration Settings
 
 Annotations can be specified so that the Rook components will have those annotations added to them.

Documentation/ceph-examples.md

@@ -48,6 +48,9 @@ the cluster. These examples represent a very small set of the different ways to
 * `cluster.yaml`: This file contains common settings for a production storage cluster. Requires at least three nodes.
 * `cluster-test.yaml`: Settings for a test cluster where redundancy is not configured. Requires only a single node.
 * `cluster-on-pvc.yaml`: This file contains common settings for backing the Ceph Mons and OSDs by PVs. Useful when running in cloud environments or where local PVs have been created for Ceph to consume.
+* `cluster-with-drive-groups.yaml`: This file contains example configurations for creating advanced
+  OSD layouts on nodes using Ceph Drive Groups.
+  [See docs for more](Documentation/ceph-cluster-crd.md#storage-selection-via-ceph-drive-groups)
 * `cluster-external`: Connect to an [external Ceph cluster](ceph-cluster-crd.md#external-cluster) with minimal access to monitor the health of the cluster and connect to the storage.
 * `cluster-external-management`: Connect to an [external Ceph cluster](ceph-cluster-crd.md#external-cluster) with the admin key of the external cluster to enable
   remote creation of pools and configure services such as an [Object Store](ceph-object.md) or a [Shared Filesystem](ceph-filesystem.md).

PendingReleaseNotes.md

@@ -14,6 +14,8 @@
 - CephBlockPool CRD has a new field called `parameters` which allows to set any property on a given [pool](Documentation/ceph-pool-crd.html#add-specific-pool-properties)
 - OSD changes:
   - OSD on PVC now supports multipath device.
+  - OSDs can now be provisioned using Ceph's Drive Groups definitions for Ceph Octopus v15.2.5+.
+    [See docs for more](Documentation/ceph-cluster-crd.md#storage-selection-via-ceph-drive-groups)
 - Added [admission controller](Documentation/admission-controller-usage.md) support for CRD validations.
   - Support for Ceph CRDs is provided. Some validations for CephClusters are included and additional validations can be added for other CRDs
   - Can be extended to add support for other providers

cluster/charts/rook-ceph/templates/resources.yaml

@@ -138,6 +138,18 @@ spec:
                   type: string
                 config: {}
                 storageClassDeviceSets: {}
+            driveGroups:
+              type: array
+              nullable: true
+              items:
+                properties:
+                  name:
+                    type: string
+                  spec: {}
+                  placement: {}
+                required:
+                - name
+                - spec
             monitoring:
               properties:
                 enabled:

cluster/examples/kubernetes/ceph/cluster-with-drive-groups.yaml

@@ -0,0 +1,100 @@
+#################################################################################################################
+# Define the settings for the rook-ceph cluster with common settings for a production cluster.
+# All nodes with available raw devices will be used for the Ceph cluster. At least three nodes are required
+# in this example. See the documentation for more details on storage settings available.
+
+# For example, to create the cluster:
+#   kubectl create -f common.yaml
+#   kubectl create -f operator.yaml
+#   kubectl create -f cluster-with-drive-groups.yaml
+#################################################################################################################
+
+apiVersion: ceph.rook.io/v1
+kind: CephCluster
+metadata:
+  name: rook-ceph
+  namespace: rook-ceph
+spec:
+  cephVersion:
+      image: ceph/ceph:v15.2.5
+      allowUnsupported: false
+  skipUpgradeChecks: false
+  dataDirHostPath: /var/lib/rook
+  mon:
+    count: 3
+    allowMultiplePerNode: false
+  dashboard:
+    enabled: true
+    ssl: true
+  crashCollector:
+    disable: false
+  annotations:
+  resources:
+  removeOSDsIfOutAndSafeToRemove: false
+  # When managing a Rook/Ceph cluster's OSD layouts with Drive Groups, the `storage` config is
+  # mostly ignored. `storageClassDeviceSets` can still be used to create OSDs on PVC, but Rook
+  # will no longer use `storage` configs for creating OSDs on a node's devices. To avoid
+  # confusion, we recommend using the `storage` config OR `driveGroups` and never both.
+
+  # Ceph supports adding devices as OSDs by Ceph Drive Group definitions in Ceph Octopus (v15.2.5+).
+  # See Drive Group docs for more info: https://docs.ceph.com/docs/master/cephadm/drivegroups/
+  # Drive Groups cannot be used to configure provisioning of OSDs on PVCs.
+  # A Drive Group is defined by a name, a Ceph Drive Group spec, and a Rook placement
+  driveGroups:
+    # Default example: use all nodes and all devices on those nodes
+    # This is equivalent to `storage:useAllNodes: true` and `storage:useAllDevices: true`
+    - name: all_nodes_all_devices
+      spec:
+        data_devices:
+          all: true
+        # The Ceph Drive Group spec's 'placement' and 'host_pattern' fields are ignored in Rook.
+        # Rook-based 'placement' below is used instead. The Ceph Drive Group Spec's 'service_id'
+        # field is also ignored; it is set by Rook to be equal to the Drive Group name.
+        # placement: # is ignored here in favor of Rook's placement below
+        # host_pattern: # (deprecated in Ceph) is similarly ignored here in favor of Rook's placement
+        # service_id: # is always set to the Drive Group name ("all_nodes" here)
+      placement: # example: omitting 'placement' or leaving it empty matches all untainted nodes
+
+    # # Further examples
+    # - name: nodes_by_hostname
+    #   spec:
+    #     # example: use rotational devices for data and non-rotational devices for metadata
+    #     data_devices:
+    #       rotational: 1
+    #     db_devices:
+    #       rotational: 0
+    #   placement:
+    #     # example: select nodes by individual names similarly to specifying 'storage:nodes'
+    #     nodeAffinity:
+    #       requiredDuringSchedulingIgnoredDuringExecution:
+    #         nodeSelectorTerms:
+    #         - matchExpressions:
+    #           - key: kubernetes.io/hostname
+    #             operator: In
+    #             values:
+    #             - k8s-master-0
+    #             - k8s-worker-0
+    #             - k8s-worker-1
+    # - name: nodes_by_label
+    #   spec:
+    #     # example: max of 6 devices between 10-50TB, and separate db and wal devices
+    #     data_devices:
+    #       limit: 6
+    #       size: "10TB:50TB"
+    #     db_devices:
+    #       model: SSD-123-foo
+    #     wal_devices:
+    #       model: NVME-QQQQ-987
+    #   placement:
+    #     # example: select nodes by label with a toleration
+    #     nodeAffinity:
+    #       requiredDuringSchedulingIgnoredDuringExecution:
+    #         nodeSelectorTerms:
+    #         - matchExpressions:
+    #           - key: role
+    #             operator: In
+    #             values:
+    #             - storage-node
+    #     tolerations:
+    #       - key: storage-node
+    #         operator: Exists

cluster/examples/kubernetes/ceph/common.yaml

@@ -157,6 +157,18 @@ spec:
                   type: string
                 config: {}
                 storageClassDeviceSets: {}
+            driveGroups:
+              type: array
+              nullable: true
+              items:
+                properties:
+                  name:
+                    type: string
+                  spec: {}
+                  placement: {}
+                required:
+                - name
+                - spec
             monitoring:
               properties:
                 enabled:

cmd/rook/ceph/osd.go

@@ -17,6 +17,7 @@ limitations under the License.
 package ceph
 
 import (
+	"encoding/json"
 	"os"
 	"strconv"
 	"strings"
@@ -67,6 +68,7 @@ var (
 	pvcBackedOSD            bool
 	blockPath               string
 	lvBackedPV              bool
+	driveGroups             string
 )
 
 func addOSDFlags(command *cobra.Command) {
@@ -74,6 +76,7 @@ func addOSDFlags(command *cobra.Command) {
 	addOSDConfigFlags(provisionCmd)
 
 	// flags specific to provisioning
+	provisionCmd.Flags().StringVar(&driveGroups, "drive-groups", "", "JSON marshalled specification of Ceph Drive Groups")
 	provisionCmd.Flags().StringVar(&cfg.devices, "data-devices", "", "comma separated list of devices to use for storage")
 	provisionCmd.Flags().StringVar(&osdDataDeviceFilter, "data-device-filter", "", "a regex filter for the device names to use, or \"all\"")
 	provisionCmd.Flags().StringVar(&osdDataDevicePathFilter, "data-device-path-filter", "", "a regex filter for the device path names to use")
@@ -175,6 +178,14 @@ func prepareOSD(cmd *cobra.Command, args []string) error {
 		return err
 	}
 
+	var dgs osdcfg.DriveGroupBlobs
+	if driveGroups != "" {
+		err := json.Unmarshal([]byte(driveGroups), &dgs)
+		if err != nil {
+			return errors.Wrap(err, "failed to unmarshal Ceph Drive Groups spec")
+		}
+	}
+
 	var dataDevices []osddaemon.DesiredDevice
 	if osdDataDeviceFilter != "" {
 		if cfg.devices != "" || osdDataDevicePathFilter != "" {
@@ -211,7 +222,7 @@ func prepareOSD(cmd *cobra.Command, args []string) error {
 	forceFormat := false
 	ownerRef := opcontroller.ClusterOwnerRef(clusterInfo.Name, ownerRefID)
 	kv := k8sutil.NewConfigMapKVStore(clusterInfo.Name, context.Clientset, ownerRef)
-	agent := osddaemon.NewAgent(context, dataDevices, cfg.metadataDevice, forceFormat,
+	agent := osddaemon.NewAgent(context, dgs, dataDevices, cfg.metadataDevice, forceFormat,
 		cfg.storeConfig, &clusterInfo, cfg.nodeName, kv, cfg.pvcBacked)
 
 	err = osddaemon.Provision(context, agent, crushLocation)

design/ceph/orchestrator-osd-add-remove.md

@@ -80,9 +80,8 @@ internal `Go` spec for a drive group is given below.
 // DriveGroups is a map of Drive Group names to Drive Group specifications.
 type DriveGroups map[string]DriveGroupSpec
 
-// DriveGroupSpec is a mapping from keys to values
-// Rook must use the "host_pattern" key to determine the host on which to run the Drive Group
-type DriveGroupSpec map[string]interface{}
+// DriveGroupSpec is a YAML or JSON parsable spec.
+type DriveGroupSpec interface{}
 ```
 
 `ceph-volume` feature request #1 will be used here to allow Rook to pass along the Drive Group YAML

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

@@ -0,0 +1,28 @@
+/*
+Copyright 2020 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 v1
+
+import (
+	"k8s.io/apimachinery/pkg/runtime"
+)
+
+// DeepCopy allows the map[string]interface{} type of DriveGroupSpec to have other DeepCopy-related
+// code generated for it. The generator does not know how to copy arbitrary interface{} types
+// without this.
+func (d DriveGroupSpec) DeepCopy() DriveGroupSpec {
+	return (DriveGroupSpec)(runtime.DeepCopyJSON(d))
+}

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

@@ -0,0 +1,93 @@
+/*
+Copyright 2020 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 v1
+
+import (
+	"encoding/json"
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+// TestDriveGroupSpec_DeepCopy tests that DriveGroupSpec's DeepCopy method does indeed create a deep
+// copy and not merely do a copy that results in any part of the copy having pointer references to
+// the original.
+func TestDriveGroupSpec_DeepCopy(t *testing.T) {
+	dgJSON := `{"string": "test", "num": 1, "bool": true, "array": ["array-string", 2, false], "map": {"map-string": "tested", "map-num": 3, "map-bool": false, "map-array": ["mas", 4, true]}}`
+
+	dg := DriveGroupSpec{}
+	err := json.Unmarshal([]byte(dgJSON), &dg)
+	assert.NoError(t, err)
+	fmt.Printf("%+v\n", dg)
+
+	copy := dg.DeepCopy()
+
+	// type casting helper func to get stuff from DriveGroupSpec's so go compiler doesn't complain
+	// get map item as array
+	getArr := func(m map[string]interface{}, arrayKey string) []interface{} {
+		i, ok := (m[arrayKey]).([]interface{})
+		if !ok {
+			panic("did not convert array")
+		}
+		return i
+	}
+	// get map item as another map
+	getMap := func(m map[string]interface{}, mapKey string) map[string]interface{} {
+		i, ok := (m[mapKey]).(map[string]interface{})
+		if !ok {
+			panic("did not convert map")
+		}
+		return i
+	}
+
+	// After copy, the two should be equal
+	assert.Equal(t, dg, copy)
+	assert.Equal(t, 5, len(copy))
+	assert.Equal(t, "test", copy["string"])
+	assert.Equal(t, float64(1), copy["num"]) // json umarshals nums as float64
+	assert.Equal(t, true, copy["bool"])
+	a := getArr(copy, "array")
+	assert.Equal(t, 3, len(a))
+	assert.Equal(t, "array-string", a[0])
+	assert.Equal(t, float64(2), a[1])
+	assert.Equal(t, false, a[2])
+	m := getMap(copy, "map")
+	assert.Equal(t, 4, len(m))
+	assert.Equal(t, "tested", m["map-string"])
+	assert.Equal(t, float64(3), m["map-num"])
+	assert.Equal(t, false, m["map-bool"])
+	a2 := getArr(m, "map-array")
+	assert.Equal(t, 3, len(a2))
+	assert.Equal(t, "mas", a2[0])
+	assert.Equal(t, float64(4), a2[1])
+	assert.Equal(t, true, a2[2])
+
+	// If the copy's elements are changed, it should NOT change the original
+	copy["string"] = "changed"
+	assert.NotEqual(t, dg, copy)
+	assert.Equal(t, "test", dg["string"])
+	assert.Equal(t, "changed", copy["string"])
+
+	// Create a new copy for testing more deeply nested changes
+	copy = dg.DeepCopy()
+	assert.Equal(t, dg, copy)                         // copy should be equal to orig again
+	getArr(getMap(copy, "map"), "map-array")[1] = "4" // change array nested in map
+	assert.NotEqual(t, dg, copy)                      // copy should no longer be equal
+	assert.Equal(t, "4", getArr(getMap(copy, "map"), "map-array")[1])
+	assert.Equal(t, float64(4), getArr(getMap(dg, "map"), "map-array")[1])
+}

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

@@ -64,6 +64,10 @@ type ClusterSpec struct {
 	// The version information that instructs Rook to orchestrate a particular version of Ceph.
 	CephVersion CephVersionSpec `json:"cephVersion,omitempty"`
 
+	// Ceph Drive Groups specification for how storage should be used in the cluster, given
+	// precedent over the Storage spec.
+	DriveGroups DriveGroupsSpec `json:"driveGroups,omitempty"`
+
 	// A spec for available storage in the cluster and how it should be used
 	Storage rookv1.StorageScopeSpec `json:"storage,omitempty"`
 
@@ -133,6 +137,25 @@ type CephVersionSpec struct {
 	AllowUnsupported bool `json:"allowUnsupported,omitempty"`
 }
 
+// DriveGroupsSpec is a list Ceph Drive Group specifications.
+type DriveGroupsSpec []DriveGroup
+
+// DriveGroup specifies a Ceph Drive Group and where Rook should apply the Drive Group.
+type DriveGroup struct {
+	// Name is a unique name used to identify the Drive Group.
+	Name string `json:"name"`
+
+	// Spec is the JSON or YAML definition of a Ceph Drive Group. Placement information contained
+	// within this spec is ignored, as placement is specified via the Rook placement mechanism below.
+	Spec DriveGroupSpec `json:"spec"`
+
+	// Placement defines which nodes the Drive Group should be applied to.
+	Placement rookv1.Placement `json:"placement,omitempty"`
+}
+
+// DriveGroupSpec is a YAML or JSON definition of a Ceph Drive Group.
+type DriveGroupSpec map[string]interface{}
+
 // DashboardSpec represents the settings for the Ceph dashboard
 type DashboardSpec struct {
 	// Whether to enable the dashboard

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

@@ -17,10 +17,12 @@ limitations under the License.
 package v1
 
 import (
+	"strconv"
+	"strings"
+
 	"github.com/pkg/errors"
 	"k8s.io/apimachinery/pkg/runtime"
 	"sigs.k8s.io/controller-runtime/pkg/webhook"
-	"strconv"
 )
 
 // compile-time assertions ensures CephCluster implements webhook.Validator so a webhook builder
@@ -28,7 +30,12 @@ import (
 var _ webhook.Validator = &CephCluster{}
 
 func (c *CephCluster) ValidateCreate() error {
-	logger.Info("validate create cephcluster")
+	logger.Infof("validate create cephcluster %q", c.ObjectMeta.Name)
+
+	if err := validateCommon(*c); err != nil {
+		return err
+	}
+
 	//If external mode enabled, then check if other fields are empty
 	if c.Spec.External.Enable {
 		if c.Spec.Mon != (MonSpec{}) || c.Spec.Dashboard != (DashboardSpec{}) || c.Spec.Monitoring != (MonitoringSpec{}) || c.Spec.DisruptionManagement != (DisruptionManagementSpec{}) || len(c.Spec.Mgr.Modules) > 0 || len(c.Spec.Network.Provider) > 0 || len(c.Spec.Network.Selectors) > 0 {
@@ -39,7 +46,12 @@ func (c *CephCluster) ValidateCreate() error {
 }
 
 func (c *CephCluster) ValidateUpdate(old runtime.Object) error {
-	logger.Info("validate update cephcluster")
+	logger.Info("validate update cephcluster %q", c.ObjectMeta.Name)
+
+	if err := validateCommon(*c); err != nil {
+		return err
+	}
+
 	occ := old.(*CephCluster)
 	return validateUpdatedCephCluster(c, occ)
 }
@@ -63,3 +75,42 @@ func validateUpdatedCephCluster(updatedCephCluster *CephCluster, found *CephClus
 
 	return nil
 }
+
+// Validate resources that need validated for both creates and updates
+func validateCommon(cluster CephCluster) error {
+	// If drive groups are set, only storage for OSDs on PVCs can be used simultaneously
+	if len(cluster.Spec.DriveGroups) > 0 {
+		invalidConfigs := []string{}
+		if cluster.Spec.Storage.UseAllNodes {
+			invalidConfigs = append(invalidConfigs, "storage:useAllNodes is true")
+		}
+		if cluster.Spec.Storage.NodeCount > 0 {
+			invalidConfigs = append(invalidConfigs, "storage:nodeCount is set")
+		}
+		if *cluster.Spec.Storage.UseAllDevices {
+			invalidConfigs = append(invalidConfigs, "storage:useAllDevices is true")
+		}
+		if len(cluster.Spec.Storage.Nodes) > 0 {
+			invalidConfigs = append(invalidConfigs, "storage:nodes are set")
+		}
+		if cluster.Spec.Storage.DeviceFilter != "" {
+			invalidConfigs = append(invalidConfigs, "storage:deviceFilter is set")
+		}
+		if cluster.Spec.Storage.DevicePathFilter != "" {
+			invalidConfigs = append(invalidConfigs, "storage:devicePathFilter is set")
+		}
+		if len(cluster.Spec.Storage.Devices) > 0 {
+			invalidConfigs = append(invalidConfigs, "storage:devices are set")
+		}
+		if len(cluster.Spec.Storage.Config) > 0 {
+			invalidConfigs = append(invalidConfigs, "storage:config is set")
+		}
+
+		if len(invalidConfigs) > 0 {
+			j := strings.Join(invalidConfigs, ", ")
+			return errors.Errorf("invalid config : cannot specify driveGroups along with any of the following current configs: %s", j)
+		}
+	}
+
+	return nil
+}

pkg/apis/ceph.rook.io/v1/zz_generated.deepcopy.go

@@ -895,6 +895,13 @@ func (in *ClientSpec) DeepCopy() *ClientSpec {
 func (in *ClusterSpec) DeepCopyInto(out *ClusterSpec) {
 	*out = *in
 	out.CephVersion = in.CephVersion
+	if in.DriveGroups != nil {
+		in, out := &in.DriveGroups, &out.DriveGroups
+		*out = make(DriveGroupsSpec, len(*in))
+		for i := range *in {
+			(*in)[i].DeepCopyInto(&(*out)[i])
+		}
+	}
 	in.Storage.DeepCopyInto(&out.Storage)
 	if in.Annotations != nil {
 		in, out := &in.Annotations, &out.Annotations
@@ -1091,6 +1098,55 @@ func (in *DisruptionManagementSpec) DeepCopy() *DisruptionManagementSpec {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *DriveGroup) DeepCopyInto(out *DriveGroup) {
+	*out = *in
+	out.Spec = in.Spec.DeepCopy()
+	in.Placement.DeepCopyInto(&out.Placement)
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new DriveGroup.
+func (in *DriveGroup) DeepCopy() *DriveGroup {
+	if in == nil {
+		return nil
+	}
+	out := new(DriveGroup)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in DriveGroupSpec) DeepCopyInto(out *DriveGroupSpec) {
+	{
+		in := &in
+		*out = in.DeepCopy()
+		return
+	}
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in DriveGroupsSpec) DeepCopyInto(out *DriveGroupsSpec) {
+	{
+		in := &in
+		*out = make(DriveGroupsSpec, len(*in))
+		for i := range *in {
+			(*in)[i].DeepCopyInto(&(*out)[i])
+		}
+		return
+	}
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new DriveGroupsSpec.
+func (in DriveGroupsSpec) DeepCopy() DriveGroupsSpec {
+	if in == nil {
+		return nil
+	}
+	out := new(DriveGroupsSpec)
+	in.DeepCopyInto(out)
+	return *out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *ErasureCodedSpec) DeepCopyInto(out *ErasureCodedSpec) {
 	*out = *in

pkg/daemon/ceph/osd/agent.go

@@ -32,6 +32,7 @@ type OsdAgent struct {
 	cluster        *cephconfig.ClusterInfo
 	nodeName       string
 	forceFormat    bool
+	driveGroups    config.DriveGroupBlobs
 	devices        []DesiredDevice
 	metadataDevice string
 	storeConfig    config.StoreConfig
@@ -47,10 +48,11 @@ type device struct {
 }
 
 // NewAgent is the instantiation of the OSD agent
-func NewAgent(context *clusterd.Context, devices []DesiredDevice, metadataDevice string, forceFormat bool,
+func NewAgent(context *clusterd.Context, driveGroups config.DriveGroupBlobs, devices []DesiredDevice, metadataDevice string, forceFormat bool,
 	storeConfig config.StoreConfig, cluster *cephconfig.ClusterInfo, nodeName string, kv *k8sutil.ConfigMapKVStore, pvcBacked bool) *OsdAgent {
 
 	return &OsdAgent{
+		driveGroups:    driveGroups,
 		devices:        devices,
 		metadataDevice: metadataDevice,
 		forceFormat:    forceFormat,

pkg/daemon/ceph/osd/daemon.go

@@ -212,16 +212,21 @@ func Provision(context *clusterd.Context, agent *OsdAgent, crushLocation string)
 
 	logger.Info("creating and starting the osds")
 
+	// orchestration is about to start, update the status
+	status = oposd.OrchestrationStatus{Status: oposd.OrchestrationStatusOrchestrating, PvcBackedOSD: agent.pvcBacked}
+	oposd.UpdateNodeStatus(agent.kv, agent.nodeName, status)
+
+	// Run Drive Group configuration
+	if err := agent.configureDriveGroups(context); err != nil {
+		return err
+	}
+
 	// determine the set of devices that can/should be used for OSDs.
 	devices, err := getAvailableDevices(context, agent)
 	if err != nil {
 		return errors.Wrap(err, "failed to get available devices")
 	}
 
-	// orchestration is about to start, update the status
-	status = oposd.OrchestrationStatus{Status: oposd.OrchestrationStatusOrchestrating, PvcBackedOSD: agent.pvcBacked}
-	oposd.UpdateNodeStatus(agent.kv, agent.nodeName, status)
-
 	// start the desired OSDs on devices
 	logger.Infof("configuring osd devices: %+v", devices)
 

pkg/daemon/ceph/osd/drivegroups.go

@@ -0,0 +1,64 @@
+/*
+Copyright 2020 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 (
+	"github.com/pkg/errors"
+	"github.com/rook/rook/pkg/clusterd"
+)
+
+const (
+	cvDriveGroupsCommand = "drive-group"
+)
+
+func (a *OsdAgent) configureDriveGroups(context *clusterd.Context) error {
+	if len(a.driveGroups) == 0 {
+		logger.Debug("No Drive Groups configured.")
+		return nil
+	}
+
+	if !a.driveGroupsAreSupported(context) {
+		return errors.New("failed to configure OSDs via Drive Groups. " +
+			"Rook will fail OSD creation; update Ceph to a newer version or remove Drive Groups from the CephCluster resource to correct this. " +
+			"the current version of ceph-volume does not support creating OSDs via Drive Groups; this is only available in later versions of Ceph Octopus (v15).")
+	}
+
+	// Create OSD bootstrap keyring
+	err := createOSDBootstrapKeyring(context, a.cluster.Name, cephConfigDir)
+	if err != nil {
+		return errors.Wrapf(err, "failed to generate OSD bootstrap keyring")
+	}
+
+	for group, spec := range a.driveGroups {
+		logger.Infof("configuring Drive Group %q: %+v", group, spec)
+		_, err := callCephVolume(context, cvDriveGroupsCommand, "--spec", spec)
+		if err != nil {
+			return errors.Wrapf(err, "failed to configure Drive Group %q", group)
+		}
+	}
+	return nil
+}
+
+func (a *OsdAgent) driveGroupsAreSupported(context *clusterd.Context) bool {
+	// Call `ceph-volume drive-group` command with no args:
+	// if supported, will return 0 (and help message which we ingore), else returns nonzero (error)
+	_, err := callCephVolume(context, cvDriveGroupsCommand)
+	if err != nil {
+		return false
+	}
+	return true
+}

pkg/daemon/ceph/osd/volume.go

@@ -724,3 +724,32 @@ func GetCephVolumeRawOSDs(context *clusterd.Context, clusterName string, cephfsi
 
 	return osds, nil
 }
+
+func callCephVolume(context *clusterd.Context, args ...string) (string, error) {
+	// Use stdbuf to capture the python output buffer such that we can write to the pod log as the
+	// logging happens instead of using the default buffering that will log everything after
+	// ceph-volume exits
+	baseCommand := "stdbuf"
+
+	// Send the log to a temp location that isn't persisted to disk so that we can print out the
+	// failure log later without also printing out past failures
+	// TODO: does this mess up expectations from the ceph log collector daemon?
+	logPath := "/tmp/ceph-log"
+	os.MkdirAll(logPath, 0644)
+	baseArgs := []string{"-oL", cephVolumeCmd, "--log-path", logPath}
+
+	co, err := context.Executor.ExecuteCommandWithCombinedOutput(baseCommand, append(baseArgs, args...)...)
+	if err != nil {
+		// Print c-v log before exiting with failure
+		cvLog := readCVLogContent("/tmp/ceph-log/ceph-volume.log")
+		logger.Errorf("%s", co)
+		if cvLog != "" {
+			logger.Errorf("%s", cvLog)
+		}
+
+		return "", errors.Wrapf(err, "failed ceph-volume call (see ceph-volume log above for more details)")
+	}
+	logger.Debugf("%v", co)
+
+	return co, nil
+}

pkg/operator/ceph/cluster/cluster.go

@@ -180,6 +180,7 @@ func (c *cluster) doOrchestration(rookImage string, cephVersion cephver.CephVers
 		rookImage,
 		spec.CephVersion,
 		spec.Storage,
+		spec.DriveGroups,
 		spec.DataDirHostPath,
 		cephv1.GetOSDPlacement(spec.Placement),
 		cephv1.GetOSDAnnotations(spec.Annotations),

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

@@ -97,3 +97,10 @@ func convertToIntIgnoreErr(raw string) int {
 
 	return val
 }
+
+// A DriveGroupBlob is a simple JSON blob defining a Ceph Drive Group. Drive Group blobs are passed
+// to ceph-volume for node disk configuration.
+type DriveGroupBlob string
+
+// DriveGroupBlobs is a mapping from Ceph Drive Group names to JSON blobs of Drive Group specs.
+type DriveGroupBlobs map[string]string

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

@@ -0,0 +1,114 @@
+/*
+Copyright 2020 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 (
+	"encoding/json"
+
+	"github.com/pkg/errors"
+	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
+	"github.com/rook/rook/pkg/operator/ceph/cluster/osd/config"
+	"github.com/rook/rook/pkg/operator/k8sutil"
+	v1 "k8s.io/api/core/v1"
+)
+
+// MarshalAsDriveGroupBlobs converts a Ceph CRD <Drive Group Name> => <Drive Group Spec> mapping
+// into a JSON-marshalled <Drive Group Name> => <Drive Group JSON blob> mapping.
+func MarshalAsDriveGroupBlobs(dgs cephv1.DriveGroupsSpec) (string, error) {
+	var blobs config.DriveGroupBlobs = map[string]string{}
+
+	for _, dg := range dgs {
+		b, err := json.Marshal(dg.Spec)
+		if err != nil {
+			return "", errors.Wrapf(err, "failed to marshal Drive Group %q into a JSON blob", dg.Name)
+		}
+		blobs[dg.Name] = string(b)
+	}
+
+	b, err := json.Marshal(blobs)
+	if err != nil {
+		return "", errors.Wrapf(err, "failed to marshal Drive Group JSON blobs after marshalling each spec into a blob: %+v", blobs)
+	}
+
+	return string(b), nil
+}
+
+// SanitizeDriveGroups processes the drive groups to remove or correct invalid specs.
+func SanitizeDriveGroups(dgs cephv1.DriveGroupsSpec) cephv1.DriveGroupsSpec {
+	out := cephv1.DriveGroupsSpec{}
+	for _, dg := range dgs {
+		group := *dg.DeepCopy()
+		if len(group.Spec) == 0 {
+			logger.Warningf("drive group %q spec is empty. skipping this drive group: %+v", group.Name, group.Spec)
+			continue
+		}
+
+		// We want to do placement only based on Rook's definition of placement, so remove any
+		// placement the Drive Group specifies by removing any existing placements from the spec.
+		if p, ok := group.Spec["placement"]; ok {
+			logger.Warningf("Rook will ignore 'placement' spec (%+v) within Drive Group %q by removing it from the Drive Group spec; Rook will control placement instead", p, group.Name)
+			delete(group.Spec, "placement")
+		}
+		if h, ok := group.Spec["host_pattern"]; ok {
+			logger.Warningf("Rook will ignore deprecated 'host_pattern' spec (\"%+v\") within Drive Group %q by removing it from the Drive Group spec; Rook will control placement instead", h, group.Name)
+			delete(group.Spec, "host_pattern")
+		}
+
+		// Ceph requires a placement to be set, so we set a host pattern to match any host
+		// use 'interface{}' because DeepCopyJSON only supports maps of map[string]interface{}
+		group.Spec["placement"] = map[string]interface{}{
+			"host_pattern": "*",
+		}
+		// Ceph requires the service ID to be set, which is the name of the drive group; in Rook, we
+		// force this value to be the same as the name given in the CephCluster CRD
+		if s, ok := group.Spec["service_id"]; ok {
+			logger.Warningf("Rook will ignore the 'service_id' spec (\"%+v\") within Drive Group %q by overwriting it with the Rook-defined name (%q) instead", s, group.Name, group.Name)
+		}
+		group.Spec["service_id"] = group.Name
+
+		out = append(out, group)
+	}
+	return out
+}
+
+// DriveGroupPlacementMatchesNode returns true if the Drive Group's placement matches the given node.
+// It returns false if the placement does not match the given node.
+// It returns an error if placement match cannot be determined.
+func DriveGroupPlacementMatchesNode(dg cephv1.DriveGroup, n *v1.Node) (bool, error) {
+	valid, err := k8sutil.ValidNode(*n, dg.Placement)
+	if err != nil {
+		return false, errors.Wrapf(err, "failed to determine if node %q is valid for drive group %q", n.Name, dg.Name)
+	}
+	return valid, nil
+}
+
+// DriveGroupsWithPlacementMatchingNode returns a subset of the Drive Groups Spec with placement
+// that matches the given node.
+// It returns an error if placement cannot be determined for the node for any Drive Group in the spec.
+func DriveGroupsWithPlacementMatchingNode(dgs cephv1.DriveGroupsSpec, n *v1.Node) (cephv1.DriveGroupsSpec, error) {
+	groups := cephv1.DriveGroupsSpec{}
+	for _, dg := range dgs {
+		match, err := DriveGroupPlacementMatchesNode(dg, n)
+		if err != nil {
+			return cephv1.DriveGroupsSpec{}, errors.Wrapf(err, "failed to find all drive groups matching node %q", n.Name)
+		}
+		if match {
+			groups = append(groups, dg)
+		}
+	}
+	return groups, nil
+}

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

@@ -0,0 +1,229 @@
+/*
+Copyright 2020 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"
+	"testing"
+
+	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
+	rookv1 "github.com/rook/rook/pkg/apis/rook.io/v1"
+	"github.com/stretchr/testify/assert"
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func TestSanitizeDriveGroups(t *testing.T) {
+	// this should look the same in input and output
+	dgSanitized := cephv1.DriveGroup{
+		Name: "sanitized",
+		Spec: cephv1.DriveGroupSpec{
+			"placement": map[string]interface{}{
+				"host_pattern": "*",
+			},
+			"data_devices": map[string]interface{}{
+				"all": "true",
+			},
+		},
+		Placement: rookv1.Placement{},
+	}
+	// this should not make it to output
+	dgEmptySpec := cephv1.DriveGroup{
+		Name:      "empty-spec",
+		Spec:      cephv1.DriveGroupSpec{},
+		Placement: rookv1.Placement{},
+	}
+	// when sanitized, this should look the same as dgSanitized except the name
+	dgHostPattern := cephv1.DriveGroup{
+		Name: "host-pattern",
+		Spec: cephv1.DriveGroupSpec{
+			"host_pattern": "*",
+			"data_devices": map[string]interface{}{
+				"all": "true",
+			},
+		},
+		Placement: rookv1.Placement{},
+	}
+	// when sanitized this should look the same as dgSanitized except the name
+	dgPlacement := cephv1.DriveGroup{
+		Name: "placement-host-pattern",
+		Spec: cephv1.DriveGroupSpec{
+			"placement": map[string]interface{}{
+				"nodes": []interface{}{"node1", "node2"},
+			},
+			"data_devices": map[string]interface{}{
+				"all": "true",
+			},
+		},
+		Placement: rookv1.Placement{},
+	}
+	// when sanitized, the service_id should be replaced by 'different-service-id'
+	dgServiceID := cephv1.DriveGroup{
+		Name: "different-service-id",
+		Spec: cephv1.DriveGroupSpec{
+			"data_devices": map[string]interface{}{
+				"all": "true",
+			},
+			"service_id": "custom_id",
+		},
+		Placement: rookv1.Placement{},
+	}
+
+	// assert that the drive groups are "functionally" equal and that the actual has the expected name
+	// The expected drive group's name information will not be used to inform "functional" equality
+	assertDGsFunctionallyEqual := func(expectedName string, expected, actual cephv1.DriveGroup) {
+		t.Helper()
+		newExpected := *expected.DeepCopy()
+		newExpected.Name = expectedName
+		newExpected.Spec["service_id"] = expectedName
+		// also assert that the actual service_id given is the expected name
+		assert.Equal(t, expectedName, actual.Spec["service_id"])
+		assert.Equal(t, newExpected, actual)
+	}
+
+	// The first drive group already has placement.host_pattern set to the sanitized value, so both
+	// input and output should be the same
+	dgs := cephv1.DriveGroupsSpec{dgSanitized}
+	out := SanitizeDriveGroups(dgs)
+	fmt.Println(out)
+	assert.Len(t, out, 1)
+	assertDGsFunctionallyEqual(dgSanitized.Name, dgSanitized, out[0])
+
+	// An empty spec should not make it into the output
+	dgs = cephv1.DriveGroupsSpec{dgEmptySpec}
+	out = SanitizeDriveGroups(dgs)
+	assert.Empty(t, out)
+
+	// Groups with deprecated top-level host_pattern or top-level placement defined should have
+	// those replaced with a glob-anything host pattern
+	dgs = cephv1.DriveGroupsSpec{dgHostPattern, dgPlacement, dgEmptySpec}
+	out = SanitizeDriveGroups(dgs)
+	assert.Len(t, out, 2)
+	assertDGsFunctionallyEqual(dgHostPattern.Name, dgSanitized, out[0])
+	assertDGsFunctionallyEqual(dgPlacement.Name, dgSanitized, out[1])
+
+	// A group with a preexisting service_id should have it replaced
+	dgs = cephv1.DriveGroupsSpec{dgServiceID}
+	out = SanitizeDriveGroups(dgs)
+	assert.Len(t, out, 1)
+	assert.Equal(t, dgServiceID.Name, out[0].Spec["service_id"])
+	assertDGsFunctionallyEqual(dgServiceID.Name, dgSanitized, out[0])
+}
+
+func TestDriveGroupsWithPlacementMatchingNode(t *testing.T) {
+	//
+	// Make a bunch of drive groups to test
+	//
+
+	// dg1 is a basic drive group with no placement
+	dg1 := cephv1.DriveGroup{
+		Name: "dg1",
+		Spec: cephv1.DriveGroupSpec{
+			"data_devices": map[string]interface{}{
+				"all": "true",
+			},
+		},
+		Placement: rookv1.Placement{},
+	}
+
+	// dg2 has affinity for label role=storage
+	dg2 := *dg1.DeepCopy()
+	dg2.Name = "dg2"
+	dg2.Placement.NodeAffinity = &v1.NodeAffinity{
+		RequiredDuringSchedulingIgnoredDuringExecution: &v1.NodeSelector{
+			NodeSelectorTerms: []v1.NodeSelectorTerm{
+				{
+					MatchExpressions: []v1.NodeSelectorRequirement{
+						{Key: "role", Operator: v1.NodeSelectorOpIn, Values: []string{"storage"}},
+					},
+				},
+			},
+		},
+	}
+
+	// dg3 tolerates taint role=storage
+	dg3 := *dg1.DeepCopy()
+	dg3.Name = "dg3"
+	dg3.Placement.Tolerations = []v1.Toleration{
+		{Key: "role", Operator: v1.TolerationOpEqual, Value: "storage"},
+	}
+
+	// dg4 has affinity for label role=storage and tolerates taint role=storage
+	dg4 := *dg2.DeepCopy()
+	dg4.Name = "dg4"
+	dg4.Placement.Tolerations = []v1.Toleration{
+		{Key: "role", Operator: v1.TolerationOpEqual, Value: "storage"},
+	}
+
+	dgs := cephv1.DriveGroupsSpec{dg1, dg2, dg3, dg4}
+
+	assertGroupsApplyToNode := func(n v1.Node, groups ...string) {
+		t.Helper()
+		gs, err := DriveGroupsWithPlacementMatchingNode(dgs, &n)
+		assert.NoError(t, err)
+		actualGroups := []string{}
+		for _, g := range gs {
+			actualGroups = append(actualGroups, g.Name)
+		}
+		assert.ElementsMatch(t, groups, actualGroups)
+	}
+
+	//
+	// Make a bunch of nodes with different stuff going on
+	//
+
+	// node1 is a basic node that is ready without labels or taints
+	node1 := v1.Node{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:   "node1",
+			Labels: map[string]string{},
+		},
+		Spec: v1.NodeSpec{},
+		Status: v1.NodeStatus{
+			Conditions: []v1.NodeCondition{
+				{Type: v1.NodeReady, Status: v1.ConditionTrue},
+			},
+		},
+	}
+	assertGroupsApplyToNode(node1, "dg1", "dg3")
+
+	// node2 is not ready; no nodes should apply
+	node2 := *node1.DeepCopy()
+	node2.ObjectMeta.Name = "node2"
+	node2.Status.Conditions[0].Status = v1.ConditionFalse
+	assertGroupsApplyToNode(node2)
+
+	// node3 has a label role=storage
+	node3 := *node1.DeepCopy()
+	node3.ObjectMeta.Name = "node3"
+	node3.ObjectMeta.Labels["role"] = "storage"
+	assertGroupsApplyToNode(node3, "dg1", "dg2", "dg3", "dg4")
+
+	// node4 has taint role=storage:NoExecute but no label role=storage
+	node4 := *node1.DeepCopy()
+	node4.ObjectMeta.Name = "node4"
+	node4.Spec.Taints = []v1.Taint{
+		{Key: "role", Value: "storage", Effect: v1.TaintEffectNoExecute},
+	}
+	assertGroupsApplyToNode(node4, "dg3")
+
+	// node5 has both taint and label from node3 and node4
+	node5 := *node4.DeepCopy()
+	node5.ObjectMeta.Name = "node5"
+	node5.ObjectMeta.Labels["role"] = "storage"
+	assertGroupsApplyToNode(node5, "dg3", "dg4")
+}

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

@@ -85,6 +85,7 @@ type Cluster struct {
 	cephVersion                                cephv1.CephVersionSpec
 	DesiredStorage                             rookv1.StorageScopeSpec // user-defined storage scope spec
 	ValidStorage                               rookv1.StorageScopeSpec // valid subset of `Storage`, computed at runtime
+	DriveGroups                                cephv1.DriveGroupsSpec
 	dataDirHostPath                            string
 	Network                                    cephv1.NetworkSpec
 	resources                                  v1.ResourceRequirements
@@ -105,6 +106,7 @@ func New(
 	rookVersion string,
 	cephVersion cephv1.CephVersionSpec,
 	storageSpec rookv1.StorageScopeSpec,
+	driveGroups cephv1.DriveGroupsSpec,
 	dataDirHostPath string,
 	placement rookv1.Placement,
 	annotations rookv1.Annotations,
@@ -126,6 +128,7 @@ func New(
 		rookVersion:       rookVersion,
 		cephVersion:       cephVersion,
 		DesiredStorage:    storageSpec,
+		DriveGroups:       driveGroups,
 		dataDirHostPath:   dataDirHostPath,
 		Network:           network,
 		resources:         resources,
@@ -180,6 +183,9 @@ type osdProperties struct {
 	tuneSlowDeviceClass bool
 	schedulerName       string
 	crushDeviceClass    string
+
+	// Drive Groups which apply to the node
+	driveGroups cephv1.DriveGroupsSpec
 }
 
 func (osdProps osdProperties) onPVC() bool {
@@ -201,8 +207,8 @@ 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 && 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 == false && len(c.DesiredStorage.Nodes) == 0 && len(c.DesiredStorage.VolumeSources) == 0 && len(c.DesiredStorage.StorageClassDeviceSets) == 0 && len(c.DriveGroups) == 0 {
+		logger.Warningf("useAllNodes is set to false and no nodes, Drive Groups, storageClassDevicesets or volumeSources are specified, no OSD pods are going to be created")
 	}
 
 	// start the jobs to provision the OSD devices
@@ -334,6 +340,22 @@ func (c *Cluster) startProvisioningOverNodes(config *provisionConfig) {
 		return
 	}
 
+	// Only provision nodes with either the 'storage' config or the 'driveGroups'; not both
+	// If Drive Groups are configured, always use those
+	// This should not apply to OSDs on PVCs; those can be configured alongside Drive Groups
+	if len(c.DriveGroups) > 0 {
+		c.startNodeDriveGroupProvisioners(config)
+	} else {
+		c.startNodeStorageProvisioners(config)
+	}
+
+	logger.Infof("start osds after provisioning is completed, if needed")
+	c.completeProvision(config)
+}
+
+func (c *Cluster) startNodeStorageProvisioners(config *provisionConfig) {
+	logger.Debug("starting provisioning on nodes using storage config")
+
 	if c.DesiredStorage.UseAllNodes {
 		if len(c.DesiredStorage.Nodes) > 0 {
 			logger.Warningf("useAllNodes is TRUE, but nodes are specified. NODES in the cluster CR will be IGNORED unless useAllNodes is FALSE.")
@@ -382,10 +404,6 @@ func (c *Cluster) startProvisioningOverNodes(config *provisionConfig) {
 			continue
 		}
 
-		// update the orchestration status of this node to the starting state
-		status := OrchestrationStatus{Status: OrchestrationStatusStarting}
-		c.updateOSDStatus(n.Name, status)
-
 		// create the job that prepares osds on the node
 		storeConfig := osdconfig.ToStoreConfig(n.Config)
 		metadataDevice := osdconfig.MetadataDevice(n.Config)
@@ -397,21 +415,84 @@ func (c *Cluster) startProvisioningOverNodes(config *provisionConfig) {
 			storeConfig:    storeConfig,
 			metadataDevice: metadataDevice,
 		}
-		job, err := c.makeJob(osdProps, config)
+		c.makeAndRunJob(n.Name, "provision", osdProps, config)
+	}
+}
+
+func (c *Cluster) startNodeDriveGroupProvisioners(config *provisionConfig) {
+	logger.Debug("starting provisioning on nodes using Drive Groups config")
+
+	if c.DesiredStorage.UseAllNodes {
+		logger.Warningf("The user has specified useAllNodes=true in the storage config. This will be ignored because driveGroups are configured.")
+	}
+	if len(c.DesiredStorage.Nodes) > 0 {
+		logger.Warningf("The user has specified nodes in the storage config. This will be ignored because driveGroups are configured.")
+	}
+
+	nodes, err := c.context.Clientset.CoreV1().Nodes().List(metav1.ListOptions{})
+	if err != nil {
+		config.addError("failed to get all nodes: %v", err)
+		return
+	}
+
+	// With Drive Groups, we effectively treat it as though the user has specified
+	// 'useAllNodes: true` because we want the Drive Groups' placements to determine whether the
+	// DGroups are configured on a node
+	c.DesiredStorage.Nodes = nil
+
+	sanitizedDGs := SanitizeDriveGroups(c.DriveGroups)
+
+	// Drive Groups should considered on every node in the k8s cluster; each drive group's
+	// 'placement' should be the selector for placement across all of K8s' nodes and not be affected
+	// by node selections in the 'storage' config
+	for _, n := range nodes.Items {
+		normalizedHostname := k8sutil.GetNormalizedHostname(n)
+		storageNode := rookv1.Node{
+			Name: normalizedHostname,
+		}
+		c.DesiredStorage.Nodes = append(c.DesiredStorage.Nodes, storageNode)
+
+		groups, err := DriveGroupsWithPlacementMatchingNode(sanitizedDGs, &n)
 		if err != nil {
-			message := fmt.Sprintf("failed to create prepare job node %q. %v", n.Name, err)
-			config.addError(message)
-			status := OrchestrationStatus{Status: OrchestrationStatusCompleted, Message: message}
-			c.updateOSDStatus(n.Name, status)
+			config.addError("failed to determine drive groups with placement matching node %q (hostname: %q): %+v", n.Name, normalizedHostname, err)
+			continue
 		}
 
-		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)}
-			c.updateOSDStatus(n.Name, status)
+		if len(groups) == 0 {
+			logger.Debugf("skipping drive group provisioning on node %q (hostname: %q). no drive groups match node or node is not ready/schedulable", n.Name, normalizedHostname)
+			continue
 		}
+
+		osdProps := osdProperties{
+			crushHostname: normalizedHostname,
+			driveGroups:   groups,
+		}
+		c.makeAndRunJob(normalizedHostname, "provision drive groups", osdProps, config)
+	}
+
+	// With Drive Groups, any node *could* be valid, and we need to do this so nodes resolve when
+	// starting OSD daemons. Each DGroup's individual placement will determine if the group is valid
+	// for a given node.
+	c.ValidStorage = *c.DesiredStorage.DeepCopy()
+}
+
+func (c *Cluster) makeAndRunJob(nodeName, action string, osdProps osdProperties, config *provisionConfig) {
+	// update the orchestration status of this node to the starting state
+	status := OrchestrationStatus{Status: OrchestrationStatusStarting}
+	c.updateOSDStatus(nodeName, status)
+
+	job, err := c.makeJob(osdProps, config)
+	if err != nil {
+		message := fmt.Sprintf("failed to create OSD %s job for node %q. %v", action, nodeName, err)
+		config.addError(message)
+		status := OrchestrationStatus{Status: OrchestrationStatusCompleted, Message: message}
+		c.updateOSDStatus(nodeName, status)
+	}
+
+	if !c.runJob(job, nodeName, config, action) {
+		status := OrchestrationStatus{Status: OrchestrationStatusCompleted, Message: fmt.Sprintf("failed to start osd %s job on node %s", action, nodeName)}
+		c.updateOSDStatus(nodeName, status)
 	}
-	logger.Infof("start osds after provisioning is completed, if needed")
-	c.completeProvision(config)
 }
 
 func (c *Cluster) runJob(job *batch.Job, nodeName string, config *provisionConfig, action string) bool {

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

@@ -64,7 +64,7 @@ func TestStart(t *testing.T) {
 		CephVersion: cephver.Nautilus,
 	}
 	c := New(clusterInfo, &clusterd.Context{Clientset: clientset, ConfigDir: "/var/lib/rook", Executor: &exectest.MockExecutor{}}, "ns", "myversion", cephv1.CephVersionSpec{},
-		rookv1.StorageScopeSpec{}, "", rookv1.Placement{}, rookv1.Annotations{}, cephv1.NetworkSpec{}, v1.ResourceRequirements{}, v1.ResourceRequirements{}, "my-priority-class", metav1.OwnerReference{}, false, false, cephv1.CephClusterHealthCheckSpec{})
+		rookv1.StorageScopeSpec{}, cephv1.DriveGroupsSpec{}, "", rookv1.Placement{}, rookv1.Annotations{}, cephv1.NetworkSpec{}, v1.ResourceRequirements{}, v1.ResourceRequirements{}, "my-priority-class", metav1.OwnerReference{}, false, false, cephv1.CephClusterHealthCheckSpec{})
 
 	// Start the first time
 	err := c.Start()
@@ -148,7 +148,7 @@ func TestAddRemoveNode(t *testing.T) {
 	}
 
 	c := New(clusterInfo, &clusterd.Context{Clientset: clientset, ConfigDir: "/var/lib/rook", Executor: executor}, "ns-add-remove", "myversion", cephv1.CephVersionSpec{},
-		storageSpec, "/foo", rookv1.Placement{}, rookv1.Annotations{}, cephv1.NetworkSpec{}, v1.ResourceRequirements{}, v1.ResourceRequirements{}, "my-priority-class", metav1.OwnerReference{}, false, false, cephv1.CephClusterHealthCheckSpec{})
+		storageSpec, cephv1.DriveGroupsSpec{}, "/foo", rookv1.Placement{}, rookv1.Annotations{}, cephv1.NetworkSpec{}, v1.ResourceRequirements{}, v1.ResourceRequirements{}, "my-priority-class", metav1.OwnerReference{}, false, false, cephv1.CephClusterHealthCheckSpec{})
 
 	// kick off the start of the orchestration in a goroutine
 	var startErr error
@@ -227,7 +227,7 @@ func TestAddRemoveNode(t *testing.T) {
 	// modify the storage spec to remove the node from the cluster
 	storageSpec.Nodes = []rookv1.Node{}
 	c = New(clusterInfo, &clusterd.Context{Clientset: clientset, ConfigDir: "/var/lib/rook", Executor: mockExec}, "ns-add-remove", "myversion", cephv1.CephVersionSpec{},
-		storageSpec, "", rookv1.Placement{}, rookv1.Annotations{}, cephv1.NetworkSpec{}, v1.ResourceRequirements{}, v1.ResourceRequirements{}, "my-priority-class", metav1.OwnerReference{}, false, false, cephv1.CephClusterHealthCheckSpec{})
+		storageSpec, cephv1.DriveGroupsSpec{}, "", rookv1.Placement{}, rookv1.Annotations{}, cephv1.NetworkSpec{}, v1.ResourceRequirements{}, v1.ResourceRequirements{}, "my-priority-class", metav1.OwnerReference{}, false, false, cephv1.CephClusterHealthCheckSpec{})
 
 	// reset the orchestration status watcher
 	statusMapWatcher = watch.NewFake()
@@ -284,7 +284,7 @@ func TestAddNodeFailure(t *testing.T) {
 		CephVersion: cephver.Nautilus,
 	}
 	c := New(clusterInfo, &clusterd.Context{Clientset: clientset, ConfigDir: "/var/lib/rook", Executor: &exectest.MockExecutor{}}, "ns-add-remove", "myversion", cephv1.CephVersionSpec{},
-		storageSpec, "/foo", rookv1.Placement{}, rookv1.Annotations{}, cephv1.NetworkSpec{}, v1.ResourceRequirements{}, v1.ResourceRequirements{}, "my-priority-class", metav1.OwnerReference{}, false, false, cephv1.CephClusterHealthCheckSpec{})
+		storageSpec, cephv1.DriveGroupsSpec{}, "/foo", rookv1.Placement{}, rookv1.Annotations{}, cephv1.NetworkSpec{}, v1.ResourceRequirements{}, v1.ResourceRequirements{}, "my-priority-class", metav1.OwnerReference{}, false, false, cephv1.CephClusterHealthCheckSpec{})
 
 	// kick off the start of the orchestration in a goroutine
 	var startErr error
@@ -353,7 +353,7 @@ func TestGetPVCHostName(t *testing.T) {
 
 func TestGetOSDInfo(t *testing.T) {
 	c := New(&cephconfig.ClusterInfo{}, &clusterd.Context{}, "ns", "myversion", cephv1.CephVersionSpec{},
-		rookv1.StorageScopeSpec{}, "", rookv1.Placement{}, rookv1.Annotations{}, cephv1.NetworkSpec{},
+		rookv1.StorageScopeSpec{}, cephv1.DriveGroupsSpec{}, "", rookv1.Placement{}, rookv1.Annotations{}, cephv1.NetworkSpec{},
 		v1.ResourceRequirements{}, v1.ResourceRequirements{}, "my-priority-class", metav1.OwnerReference{}, false, false, cephv1.CephClusterHealthCheckSpec{})
 
 	node := "n1"

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

@@ -570,13 +570,18 @@ func (c *Cluster) provisionPodTemplateSpec(osdProps osdProperties, restart v1.Re
 		return nil, errors.New("empty volumes")
 	}
 
+	provisionContainer, err := c.provisionOSDContainer(osdProps, copyBinariesContainer.VolumeMounts[0], provisionConfig)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to generate OSD provisioning container")
+	}
+
 	podSpec := v1.PodSpec{
 		ServiceAccountName: serviceAccountName,
 		InitContainers: []v1.Container{
 			*copyBinariesContainer,
 		},
 		Containers: []v1.Container{
-			c.provisionOSDContainer(osdProps, copyBinariesContainer.VolumeMounts[0], provisionConfig),
+			provisionContainer,
 		},
 		RestartPolicy:     restart,
 		Volumes:           volumes,
@@ -814,12 +819,29 @@ func (c *Cluster) getConfigEnvVars(osdProps osdProperties, dataDir string) []v1.
 	return envVars
 }
 
-func (c *Cluster) provisionOSDContainer(osdProps osdProperties, copyBinariesMount v1.VolumeMount, provisionConfig *provisionConfig) v1.Container {
+func (c *Cluster) provisionOSDContainer(osdProps osdProperties, copyBinariesMount v1.VolumeMount, provisionConfig *provisionConfig) (v1.Container, error) {
 	envVars := c.getConfigEnvVars(osdProps, k8sutil.DataDir)
 
 	// enable debug logging in the prepare job
 	envVars = append(envVars, setDebugLogLevelEnvVar(true))
 
+	// Drive Groups cannot be used to configure OSDs on PVCs, so ignore if this is a PVC config
+	// This shouldn't ever happen, but do the PVC check to be sure
+	if len(osdProps.driveGroups) > 0 && !osdProps.onPVC() {
+		v, err := c.getDriveGroupEnvVar(osdProps)
+		if err != nil {
+			// Because OSD creation via drive groups should take precedent over other types of drive
+			// creation, if there is an error here, we should fail. Allowing OSD creation to proceed
+			// without drive group information could result in OSD configs being created which the
+			// user does not want.
+			return v1.Container{}, errors.Wrapf(err, "failed to get drive group info for OSD provisioning container")
+		}
+		// An env var with no name means there are no groups, don't add the var
+		if v.Name != "" {
+			envVars = append(envVars, v)
+		}
+	}
+
 	// only 1 of device list, device filter, device path filter and use all devices can be specified.  We prioritize in that order.
 	if len(osdProps.devices) > 0 {
 		deviceNames := make([]string, len(osdProps.devices))
@@ -911,7 +933,7 @@ func (c *Cluster) provisionOSDContainer(osdProps osdProperties, copyBinariesMoun
 		Resources: c.prepareResources,
 	}
 
-	return osdProvisionContainer
+	return osdProvisionContainer, nil
 }
 
 func getPvcOSDBridgeMount(claimName string) v1.VolumeMount {
@@ -983,10 +1005,26 @@ func getPVCOSDVolumes(osdProps *osdProperties) []v1.Volume {
 	return volumes
 }
 
+func (c *Cluster) getDriveGroupEnvVar(osdProps osdProperties) (v1.EnvVar, error) {
+	if len(osdProps.driveGroups) == 0 {
+		return v1.EnvVar{}, nil
+	}
+
+	b, err := MarshalAsDriveGroupBlobs(osdProps.driveGroups)
+	if err != nil {
+		return v1.EnvVar{}, errors.Wrap(err, "failed to marshal drive groups into an env var")
+	}
+	return driveGroupsEnvVar(b), nil
+}
+
 func nodeNameEnvVar(name string) v1.EnvVar {
 	return v1.EnvVar{Name: "ROOK_NODE_NAME", Value: name}
 }
 
+func driveGroupsEnvVar(driveGroups string) v1.EnvVar {
+	return v1.EnvVar{Name: "ROOK_DRIVE_GROUPS", Value: driveGroups}
+}
+
 func dataDevicesEnvVar(dataDevices string) v1.EnvVar {
 	return v1.EnvVar{Name: "ROOK_DATA_DEVICES", Value: dataDevices}
 }

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

@@ -81,6 +81,7 @@ func testPodDevices(t *testing.T, dataDir, deviceName string, allDevices bool) {
 	devices := []rookv1.Device{
 		{Name: deviceName},
 	}
+	driveGroups := cephv1.DriveGroupsSpec{}
 
 	clientset := fake.NewSimpleClientset()
 	cephVersion := cephv1.CephVersionSpec{Image: "ceph/ceph:v15"}
@@ -88,7 +89,7 @@ func testPodDevices(t *testing.T, dataDir, deviceName string, allDevices bool) {
 		CephVersion: cephver.Nautilus,
 	}
 	c := New(clusterInfo, &clusterd.Context{Clientset: clientset, ConfigDir: "/var/lib/rook", Executor: &exectest.MockExecutor{}}, "ns", "rook/rook:myversion", cephVersion,
-		storageSpec, dataDir, rookv1.Placement{}, rookv1.Annotations{}, cephv1.NetworkSpec{}, v1.ResourceRequirements{}, v1.ResourceRequirements{}, "my-priority-class", metav1.OwnerReference{}, false, false, cephv1.CephClusterHealthCheckSpec{})
+		storageSpec, driveGroups, dataDir, rookv1.Placement{}, rookv1.Annotations{}, cephv1.NetworkSpec{}, v1.ResourceRequirements{}, v1.ResourceRequirements{}, "my-priority-class", metav1.OwnerReference{}, false, false, cephv1.CephClusterHealthCheckSpec{})
 
 	devMountNeeded := deviceName != "" || allDevices
 
@@ -261,13 +262,14 @@ func TestStorageSpecConfig(t *testing.T) {
 			},
 		},
 	}
+	driveGroups := cephv1.DriveGroupsSpec{}
 
 	clientset := fake.NewSimpleClientset()
 	clusterInfo := &cephconfig.ClusterInfo{
 		CephVersion: cephver.Nautilus,
 	}
 	c := New(clusterInfo, &clusterd.Context{Clientset: clientset, ConfigDir: "/var/lib/rook", Executor: &exectest.MockExecutor{}}, "ns", "rook/rook:myversion", cephv1.CephVersionSpec{},
-		storageSpec, "", rookv1.Placement{}, rookv1.Annotations{}, cephv1.NetworkSpec{}, v1.ResourceRequirements{}, v1.ResourceRequirements{}, "my-priority-class", metav1.OwnerReference{}, false, false, cephv1.CephClusterHealthCheckSpec{})
+		storageSpec, driveGroups, "", rookv1.Placement{}, rookv1.Annotations{}, cephv1.NetworkSpec{}, v1.ResourceRequirements{}, v1.ResourceRequirements{}, "my-priority-class", metav1.OwnerReference{}, false, false, cephv1.CephClusterHealthCheckSpec{})
 
 	n := c.DesiredStorage.ResolveNode(storageSpec.Nodes[0].Name)
 	storeConfig := config.ToStoreConfig(storageSpec.Nodes[0].Config)
@@ -311,13 +313,14 @@ func TestHostNetwork(t *testing.T) {
 			},
 		},
 	}
+	driveGroups := cephv1.DriveGroupsSpec{}
 
 	clientset := fake.NewSimpleClientset()
 	clusterInfo := &cephconfig.ClusterInfo{
 		CephVersion: cephver.Nautilus,
 	}
 	c := New(clusterInfo, &clusterd.Context{Clientset: clientset, ConfigDir: "/var/lib/rook", Executor: &exectest.MockExecutor{}}, "ns", "myversion", cephv1.CephVersionSpec{},
-		storageSpec, "", rookv1.Placement{}, rookv1.Annotations{}, cephv1.NetworkSpec{HostNetwork: true}, v1.ResourceRequirements{}, v1.ResourceRequirements{}, "my-priority-class", metav1.OwnerReference{}, false, false, cephv1.CephClusterHealthCheckSpec{})
+		storageSpec, driveGroups, "", rookv1.Placement{}, rookv1.Annotations{}, cephv1.NetworkSpec{HostNetwork: true}, v1.ResourceRequirements{}, v1.ResourceRequirements{}, "my-priority-class", metav1.OwnerReference{}, false, false, cephv1.CephClusterHealthCheckSpec{})
 
 	n := c.DesiredStorage.ResolveNode(storageSpec.Nodes[0].Name)
 	osd := OSDInfo{
@@ -359,7 +362,7 @@ func TestOsdOnSDNFlag(t *testing.T) {
 func TestOsdPrepareResources(t *testing.T) {
 	clientset := fake.NewSimpleClientset()
 	c := New(&cephconfig.ClusterInfo{}, &clusterd.Context{Clientset: clientset, ConfigDir: "/var/lib/rook", Executor: &exectest.MockExecutor{}}, "ns", "myversion", cephv1.CephVersionSpec{},
-		rookv1.StorageScopeSpec{}, "", rookv1.Placement{}, rookv1.Annotations{}, cephv1.NetworkSpec{}, v1.ResourceRequirements{}, v1.ResourceRequirements{}, "my-priority-class", metav1.OwnerReference{}, false, false, cephv1.CephClusterHealthCheckSpec{})
+		rookv1.StorageScopeSpec{}, cephv1.DriveGroupsSpec{}, "", rookv1.Placement{}, rookv1.Annotations{}, cephv1.NetworkSpec{}, v1.ResourceRequirements{}, v1.ResourceRequirements{}, "my-priority-class", metav1.OwnerReference{}, false, false, cephv1.CephClusterHealthCheckSpec{})
 
 	// TEST 2: NOT running on PVC and some prepareResources are specificied
 	rr := v1.ResourceRequirements{
@@ -402,3 +405,78 @@ func TestOsdActivateEnvVar(t *testing.T) {
 	assert.Equal(t, "CEPH_ARGS", osdActivateEnv[4].Name)
 	assert.Equal(t, "-m $(ROOK_CEPH_MON_HOST)", osdActivateEnv[4].Value)
 }
+
+func TestDriveGroups(t *testing.T) {
+	storageSpec := rookv1.StorageScopeSpec{
+		UseAllNodes: true,
+		Selection:   rookv1.Selection{},
+		Nodes:       []rookv1.Node{{Name: "node1"}, {Name: "node2"}},
+	}
+	driveGroups := cephv1.DriveGroupsSpec{
+		cephv1.DriveGroup{
+			Name: "group1",
+			Spec: cephv1.DriveGroupSpec{
+				"data_devices": map[string]string{
+					"all": "true",
+				},
+			},
+			Placement: rookv1.Placement{},
+		},
+	}
+
+	clientset := fake.NewSimpleClientset()
+
+	cephVersion := cephv1.CephVersionSpec{Image: "test-image"}
+
+	clusterInfo := &cephconfig.ClusterInfo{}
+
+	dataPathMap := &provisionConfig{
+		DataPathMap: opconfig.NewDatalessDaemonDataPathMap("ns", "/var/lib/rook"),
+	}
+
+	c := New(clusterInfo, &clusterd.Context{Clientset: clientset, ConfigDir: "/var/lib/rook", Executor: &exectest.MockExecutor{}}, "ns", "rook/rook:myversion", cephVersion,
+		storageSpec, driveGroups, "/var/lib/rook", rookv1.Placement{}, rookv1.Annotations{}, cephv1.NetworkSpec{}, v1.ResourceRequirements{}, v1.ResourceRequirements{}, "my-priority-class", metav1.OwnerReference{}, false, false, cephv1.CephClusterHealthCheckSpec{})
+
+	n := c.DesiredStorage.ResolveNode("node1")
+
+	osdProp := osdProperties{
+		crushHostname: n.Name,
+	}
+
+	assertNoDriveGroups := func(e []v1.EnvVar) {
+		for _, v := range e {
+			if v.Name == "ROOK_DRIVE_GROUPS" {
+				t.Helper()
+				t.Errorf("ROOK_DRIVE_GROUPS should not be specified to provisioning pod but is: %+v", v)
+			}
+		}
+	}
+
+	assertDriveGroups := func(e []v1.EnvVar) {
+		for _, v := range e {
+			if v.Name == "ROOK_DRIVE_GROUPS" {
+				return
+			}
+		}
+		t.Helper()
+		t.Errorf("ROOK_DRIVE_GROUPS should be specified to provisioning pod but is not: %+v", e)
+	}
+
+	// Drive groups are not specified; should NOT have env var specified
+	clusterInfo.CephVersion = cephver.CephVersion{Major: 15, Minor: 2, Extra: 5}
+	c = New(clusterInfo, &clusterd.Context{Clientset: clientset, ConfigDir: "/var/lib/rook", Executor: &exectest.MockExecutor{}}, "ns", "rook/rook:myversion", cephVersion,
+		storageSpec, cephv1.DriveGroupsSpec{}, "/var/lib/rook", rookv1.Placement{}, rookv1.Annotations{}, cephv1.NetworkSpec{}, v1.ResourceRequirements{}, v1.ResourceRequirements{}, "my-priority-class", metav1.OwnerReference{}, false, false, cephv1.CephClusterHealthCheckSpec{})
+	job, err := c.makeJob(osdProp, dataPathMap)
+	assert.NotNil(t, job)
+	assert.NoError(t, err)
+	assertNoDriveGroups(job.Spec.Template.Spec.Containers[0].Env)
+
+	// Drive groups are specified; should have env var specified
+	osdProp.driveGroups = driveGroups
+	c = New(clusterInfo, &clusterd.Context{Clientset: clientset, ConfigDir: "/var/lib/rook", Executor: &exectest.MockExecutor{}}, "ns", "rook/rook:myversion", cephVersion,
+		storageSpec, driveGroups, "/var/lib/rook", rookv1.Placement{}, rookv1.Annotations{}, cephv1.NetworkSpec{}, v1.ResourceRequirements{}, v1.ResourceRequirements{}, "my-priority-class", metav1.OwnerReference{}, false, false, cephv1.CephClusterHealthCheckSpec{})
+	job, err = c.makeJob(osdProp, dataPathMap)
+	assert.NotNil(t, job)
+	assert.NoError(t, err)
+	assertDriveGroups(job.Spec.Template.Spec.Containers[0].Env)
+}

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

@@ -42,7 +42,7 @@ func TestOrchestrationStatus(t *testing.T) {
 		CephVersion: cephver.Nautilus,
 	}
 	c := New(clusterInfo, &clusterd.Context{Clientset: clientset, ConfigDir: "/var/lib/rook", Executor: &exectest.MockExecutor{}}, "ns", "myversion", cephv1.CephVersionSpec{},
-		rookv1.StorageScopeSpec{}, "", rookv1.Placement{}, rookv1.Annotations{}, cephv1.NetworkSpec{}, v1.ResourceRequirements{}, v1.ResourceRequirements{}, "my-priority-class", metav1.OwnerReference{}, false, false, cephv1.CephClusterHealthCheckSpec{})
+		rookv1.StorageScopeSpec{}, cephv1.DriveGroupsSpec{}, "", rookv1.Placement{}, rookv1.Annotations{}, cephv1.NetworkSpec{}, v1.ResourceRequirements{}, v1.ResourceRequirements{}, "my-priority-class", metav1.OwnerReference{}, false, false, cephv1.CephClusterHealthCheckSpec{})
 	kv := k8sutil.NewConfigMapKVStore(c.Namespace, clientset, metav1.OwnerReference{})
 	nodeName := "mynode"
 	cmName := fmt.Sprintf(orchestrationStatusMapName, nodeName)

pkg/operator/k8sutil/node.go

@@ -237,6 +237,12 @@ func normalizeHostname(kubernetesNode v1.Node) string {
 	return hostname
 }
 
+// GetNormalizedHostname returns the normalized hostname of the Kubernetes node. Rook uses
+// normalized hostnames as its internal reference for nodes.
+func GetNormalizedHostname(kubernetesNode v1.Node) string {
+	return normalizeHostname(kubernetesNode)
+}
+
 // GetKubernetesNodesMatchingRookNodes lists all the nodes in Kubernetes and returns all the
 // Kubernetes nodes that have a corresponding match in the list of Rook nodes.
 func GetKubernetesNodesMatchingRookNodes(rookNodes []rookv1.Node, clientset kubernetes.Interface) ([]v1.Node, error) {

tests/framework/installer/ceph_manifests.go

@@ -221,6 +221,18 @@ spec:
                 deviceFilter: {}
                 config: {}
                 storageClassDeviceSets: {}
+            driveGroups:
+              type: array
+              nullable: true
+              items:
+                properties:
+                  name:
+                    type: string
+                  spec: {}
+                  placement: {}
+                required:
+                - name
+                - spec
             monitoring:
               properties:
                 enabled: