diff --git a/Documentation/ceph-cluster-crd.md b/Documentation/ceph-cluster-crd.md index ee5cc8805..a7ea6d603 100755 --- a/Documentation/ceph-cluster-crd.md +++ b/Documentation/ceph-cluster-crd.md @@ -6,10 +6,18 @@ indent: true # Ceph Cluster CRD Rook allows creation and customization of storage clusters through the custom resource definitions (CRDs). +There are two different modes to create your cluster, depending on whether storage can be dynamically provisioned on which to base the Ceph cluster. -## Sample +1. Specify host paths and raw devices +2. Specify the storage class Rook should use to consume storage via PVCs -To get you started, here is a simple example of a CRD to configure a Ceph cluster with all nodes and all devices. More examples are included [later in this doc](#samples). +Following is an example for each of these approaches. + + +## Host-based Cluster + +To get you started, here is a simple example of a CRD to configure a Ceph cluster with all nodes and all devices. Next example is where Mons and OSDs are backed by PVCs. +More examples are included [later in this doc](#samples). **NOTE** In addition to your CephCluster object, you need to create the namespace, service accounts, and RBAC rules for the namespace you are going to create the CephCluster in. These resources are defined in the example `common.yaml`. @@ -33,6 +41,42 @@ spec: useAllDevices: true ``` +## PVC-based Cluster + +```yaml +apiVersion: ceph.rook.io/v1 +kind: CephCluster +metadata: + name: rook-ceph + namespace: rook-ceph +spec: + dataDirHostPath: /var/lib/rook + mon: + count: 3 + volumeClaimTemplate: + spec: + storageClassName: local-storage + resources: + requests: + storage: 10Gi + storage: + storageClassDeviceSets: + - name: set1 + count: 3 + volumeClaimTemplates: + - metadata: + name: data + spec: + resources: + requests: + storage: 10Gi + # IMPORTANT: Change the storage class depending on your environment (e.g. local-storage, gp2) + storageClassName: local-storage + volumeMode: Block + accessModes: + - ReadWriteOnce +``` + ## 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. @@ -79,6 +123,7 @@ For more details on the mons and when to choose a number other than `3`, see the See [node settings](#node-settings) below. - `config`: Config settings applied to all OSDs on the node unless overridden by `devices` or `directories`. See the [config settings](#osd-configuration-settings) below. - [storage selection settings](#storage-selection-settings) + - [Storage Class Device Sets](#storage-class-device-sets) ### Mon Settings @@ -151,7 +196,20 @@ Below are the settings available, both at the cluster and individual node level, - `path`: The path on disk of the directory (e.g., `/rook/storage-dir`) - `config`: Directory-specific config settings. See the [config settings](#osd-configuration-settings) below - `location`: Location information about the cluster to help with data placement, such as region or data center. This is directly fed into the underlying Ceph CRUSH map. The type of this field is `string`. For example, to add data center location information, set this field to `rack=rack1`. More information on CRUSH maps can be found in the [ceph docs](http://docs.ceph.com/docs/master/rados/operations/crush-map/) +- `storageClassDeviceSets`: Explained in [Storage Class Device Sets](#storage-class-device-sets) +### Storage Class Device Sets +The following are the settings for Storage Class Device Sets which can be configured to create OSDs that are backed by block mode PVs. + +- `name`: A name for the set. +- `count`: The number of devices in the set. +- `resources`: The CPU and RAM requests/limits for the devices.(Optional) +- `placement`: The placement criteria for the devices. Default is no placement criteria.(Optional) +- `volumeClaimTemplates`: A list of PVC templates to use for provisioning the underlying storage devices. + - `resources.requests.storage`: The desired capacity for the underlying storage devices. + - `storageClassName`: The StorageClass to provision PVCs from. Default would be to use the cluster-default StorageClass. + - `volumeMode`: The volume mode to be set for the PVC. Which should be Block + - `accessModes`: The access mode for the PVC to be bound by OSD. ### OSD Configuration Settings The following storage selection settings are specific to Ceph and do not apply to other backends. All variables are key-value pairs represented as strings. @@ -490,7 +548,7 @@ racks in the data center setup. ### Using PVC storage for monitors In the CRD specification below three monitors are created each using a 10Gi PVC -created by Rook using the `fast-disks` storage class. +created by Rook using the `local-storage` storage class. ```yaml apiVersion: ceph.rook.io/v1 @@ -507,7 +565,7 @@ spec: allowMultiplePerNode: false volumeClaimTemplate: spec: - storageClassName: fast-disks + storageClassName: local-storage resources: requests: storage: 10Gi @@ -525,3 +583,66 @@ spec: osdsPerDevice: "1" ``` +### Using StorageClassDeviceSets + +In the CRD specification below, 3 OSDs (having specific placement and resource values) and 3 mons with each using a 10Gi PVC, are created by Rook using the `local-storage` storage class. + +```yaml +apiVersion: ceph.rook.io/v1 +kind: CephCluster +metadata: + name: rook-ceph + namespace: rook-ceph +spec: + dataDirHostPath: /var/lib/rook + mon: + count: 3 + allowMultiplePerNode: false + volumeClaimTemplate: + spec: + storageClassName: local-storage + resources: + requests: + storage: 10Gi + cephVersion: + image: ceph/ceph:v14.2.1-20190430 + allowUnsupported: false + dashboard: + enabled: true + network: + hostNetwork: false + storage: + storageClassDeviceSets: + - name: set1 + count: 3 + resources: + limits: + cpu: "500m" + memory: "4Gi" + requests: + cpu: "500m" + memory: "4Gi" + placement: + podAntiAffinity: + preferredDuringSchedulingIgnoredDuringExecution: + - weight: 100 + podAffinityTerm: + labelSelector: + matchExpressions: + - key: "rook.io/cluster" + operator: In + values: + - cluster1 + topologyKey: "failure-domain.beta.kubernetes.io/zone" + volumeClaimTemplates: + - metadata: + name: data + spec: + resources: + requests: + storage: 10Gi + storageClassName: local-storage + volumeMode: Block + accessModes: + - ReadWriteOnce +``` diff --git a/PendingReleaseNotes.md b/PendingReleaseNotes.md index 1df861c41..b99d1e60d 100644 --- a/PendingReleaseNotes.md +++ b/PendingReleaseNotes.md @@ -41,6 +41,7 @@ an example usage - Ceph daemons now run under 'ceph' user and not 'root' anymore (monitor or osd store already owned by 'root' will keep running under 'root') - Ceph monitors have initial support for running on PVC storage. See docs on [monitor settings for more detail](Documentation/ceph-cluster-crd.md#mon-settings). +- Ceph OSDs can be created by using StorageClassDeviceSet. See docs on [Storage Class Device Sets](Documentation/ceph-cluster-crd.md#storage-class-device-sets). ## Breaking Changes diff --git a/cluster/examples/kubernetes/ceph/cluster-on-pvc.yaml b/cluster/examples/kubernetes/ceph/cluster-on-pvc.yaml index 3b24b1080..fd6a0b587 100644 --- a/cluster/examples/kubernetes/ceph/cluster-on-pvc.yaml +++ b/cluster/examples/kubernetes/ceph/cluster-on-pvc.yaml @@ -1,8 +1,14 @@ -# -# Hi, it looks like you just had a conflict merging #3341. Please integrate the -# `mon.volumeClaimTemplate` configuration and associated comment below, and -# remove this note. Thanks :) -# +################################################################################################################# +# 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-on-pvc.yaml +################################################################################################################# + apiVersion: ceph.rook.io/v1 kind: CephCluster metadata: @@ -11,6 +17,8 @@ metadata: spec: dataDirHostPath: /var/lib/rook mon: + count: 3 + allowMultiplePerNode: false # A volume claim template can be specified in which case new monitors (and # monitors created during fail over) will construct a PVC based on the # template for the monitor's primary storage. Changes to the template do not @@ -23,3 +31,45 @@ spec: resources: requests: storage: 10Gi + cephVersion: + image: ceph/ceph:v14.2.1-20190430 + allowUnsupported: false + dashboard: + enabled: true + network: + hostNetwork: false + storage: + storageClassDeviceSets: + - name: set1 + count: 3 + resources: + # limits: + # cpu: "500m" + # memory: "4Gi" + # requests: + # cpu: "500m" + # memory: "4Gi" + # placement: + # podAntiAffinity: + # preferredDuringSchedulingIgnoredDuringExecution: + # - weight: 100 + # podAffinityTerm: + # labelSelector: + # matchExpressions: + # - key: "rook.io/cluster" + # operator: In + # values: + # - cluster1 + # topologyKey: "failure-domain.beta.kubernetes.io/zone" + volumeClaimTemplates: + - metadata: + name: data + spec: + resources: + requests: + storage: 10Gi + # IMPORTANT: Change the storage class depending on your environment (e.g. local-storage, gp2) + storageClassName: generic + volumeMode: Block + accessModes: + - ReadWriteOnce diff --git a/cmd/rook/ceph/ceph.go b/cmd/rook/ceph/ceph.go index 6ea0a6170..ec8d49826 100644 --- a/cmd/rook/ceph/ceph.go +++ b/cmd/rook/ceph/ceph.go @@ -51,6 +51,7 @@ type config struct { networkInfo clusterd.NetworkInfo monEndpoints string nodeName string + pvcBacked bool } func init() { diff --git a/cmd/rook/ceph/osd.go b/cmd/rook/ceph/osd.go index ee22f2106..38315fdba 100644 --- a/cmd/rook/ceph/osd.go +++ b/cmd/rook/ceph/osd.go @@ -66,6 +66,7 @@ var ( osdStringID string osdUUID string osdIsDevice bool + pvcBackedOSD bool ) func addOSDFlags(command *cobra.Command) { @@ -79,6 +80,7 @@ func addOSDFlags(command *cobra.Command) { provisionCmd.Flags().StringVar(&cfg.metadataDevice, "metadata-device", "", "device to use for metadata (e.g. a high performance SSD/NVMe device)") provisionCmd.Flags().BoolVar(&cfg.forceFormat, "force-format", false, "true to force the format of any specified devices, even if they already have a filesystem. BE CAREFUL!") + provisionCmd.Flags().BoolVar(&cfg.pvcBacked, "pvc-backed-osd", false, "true to specify a block mode pvc is backing the OSD") // flags for generating the osd config osdConfigCmd.Flags().IntVar(&osdID, "osd-id", -1, "osd id for which to generate config") @@ -92,6 +94,7 @@ func addOSDFlags(command *cobra.Command) { osdStartCmd.Flags().StringVar(&osdStringID, "osd-id", "", "the osd ID") osdStartCmd.Flags().StringVar(&osdUUID, "osd-uuid", "", "the osd UUID") osdStartCmd.Flags().StringVar(&osdStoreType, "osd-store-type", "", "whether the osd is bluestore or filestore") + osdStartCmd.Flags().BoolVar(&pvcBackedOSD, "pvc-backed-osd", false, "Whether the OSD backing store in PVC or not") // add the subcommands to the parent osd command osdCmd.AddCommand(osdConfigCmd, @@ -139,7 +142,7 @@ func startOSD(cmd *cobra.Command, args []string) error { commonOSDInit(osdStartCmd) context := createContext() - err := osddaemon.StartOSD(context, osdStoreType, osdStringID, osdUUID, args) + err := osddaemon.StartOSD(context, osdStoreType, osdStringID, osdUUID, pvcBackedOSD, args) if err != nil { rook.TerminateFatal(err) } @@ -244,14 +247,15 @@ func prepareOSD(cmd *cobra.Command, args []string) error { ownerRef := cluster.ClusterOwnerRef(clusterInfo.Name, ownerRefID) kv := k8sutil.NewConfigMapKVStore(clusterInfo.Name, context.Clientset, ownerRef) agent := osddaemon.NewAgent(context, dataDevices, cfg.metadataDevice, cfg.directories, forceFormat, - crushLocation, cfg.storeConfig, &clusterInfo, cfg.nodeName, kv) + crushLocation, cfg.storeConfig, &clusterInfo, cfg.nodeName, kv, cfg.pvcBacked) err = osddaemon.Provision(context, agent) if err != nil { // something failed in the OSD orchestration, update the status map with failure details status := oposd.OrchestrationStatus{ - Status: oposd.OrchestrationStatusFailed, - Message: err.Error(), + Status: oposd.OrchestrationStatusFailed, + Message: err.Error(), + PvcBackedOSD: cfg.pvcBacked, } oposd.UpdateNodeStatus(kv, cfg.nodeName, status) diff --git a/pkg/apis/rook.io/v1alpha2/types.go b/pkg/apis/rook.io/v1alpha2/types.go index a17a9eccf..cf07b489a 100644 --- a/pkg/apis/rook.io/v1alpha2/types.go +++ b/pkg/apis/rook.io/v1alpha2/types.go @@ -16,7 +16,7 @@ limitations under the License. package v1alpha2 import ( - "k8s.io/api/core/v1" + v1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ) @@ -34,6 +34,8 @@ type StorageScopeSpec struct { Location string `json:"location,omitempty"` Config map[string]string `json:"config"` Selection + VolumeSources []VolumeSource `json:"volumeSources,omitempty"` + StorageClassDeviceSets []StorageClassDeviceSet `json:"storageClassDeviceSets"` } type Node struct { @@ -127,3 +129,20 @@ type KeyType string type AnnotationsSpec map[KeyType]Annotations type Annotations map[string]string + +type StorageClassDeviceSet struct { + Name string `json:"name,omitempty"` // A unique identifier for the set + Count int `json:"count,omitemity"` // Number of devices in this set + Resources v1.ResourceRequirements `json:"resources,omitempty"` // Requests/limits for the devices + Placement Placement `json:"placement,omitempty"` // Placement constraints for the devices + Config map[string]string `json:"config,omitempty"` // Provider-specific device configuration + VolumeClaimTemplates []v1.PersistentVolumeClaim `json:"volumeClaimTemplates,omitempty"` // List of PVC templates for the underlying storage devices +} + +type VolumeSource struct { + Name string `json:"name,omitempty"` + PersistentVolumeClaimSource v1.PersistentVolumeClaimVolumeSource `json:"persistentVolumeClaimSource,omitempty"` + Resources v1.ResourceRequirements `json:"resources,omitempty"` + Placement Placement `json:"placement,omitempty"` + Config map[string]string `json:"config,omitempty"` +} diff --git a/pkg/apis/rook.io/v1alpha2/zz_generated.deepcopy.go b/pkg/apis/rook.io/v1alpha2/zz_generated.deepcopy.go index 71d2a5f2d..2711edb03 100644 --- a/pkg/apis/rook.io/v1alpha2/zz_generated.deepcopy.go +++ b/pkg/apis/rook.io/v1alpha2/zz_generated.deepcopy.go @@ -350,6 +350,38 @@ func (in *Selection) DeepCopy() *Selection { return out } +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *StorageClassDeviceSet) DeepCopyInto(out *StorageClassDeviceSet) { + *out = *in + in.Resources.DeepCopyInto(&out.Resources) + in.Placement.DeepCopyInto(&out.Placement) + if in.Config != nil { + in, out := &in.Config, &out.Config + *out = make(map[string]string, len(*in)) + for key, val := range *in { + (*out)[key] = val + } + } + if in.VolumeClaimTemplates != nil { + in, out := &in.VolumeClaimTemplates, &out.VolumeClaimTemplates + *out = make([]v1.PersistentVolumeClaim, len(*in)) + for i := range *in { + (*in)[i].DeepCopyInto(&(*out)[i]) + } + } + return +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new StorageClassDeviceSet. +func (in *StorageClassDeviceSet) DeepCopy() *StorageClassDeviceSet { + if in == nil { + return nil + } + out := new(StorageClassDeviceSet) + in.DeepCopyInto(out) + return out +} + // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *StorageScopeSpec) DeepCopyInto(out *StorageScopeSpec) { *out = *in @@ -368,6 +400,20 @@ func (in *StorageScopeSpec) DeepCopyInto(out *StorageScopeSpec) { } } in.Selection.DeepCopyInto(&out.Selection) + if in.VolumeSources != nil { + in, out := &in.VolumeSources, &out.VolumeSources + *out = make([]VolumeSource, len(*in)) + for i := range *in { + (*in)[i].DeepCopyInto(&(*out)[i]) + } + } + if in.StorageClassDeviceSets != nil { + in, out := &in.StorageClassDeviceSets, &out.StorageClassDeviceSets + *out = make([]StorageClassDeviceSet, len(*in)) + for i := range *in { + (*in)[i].DeepCopyInto(&(*out)[i]) + } + } return } @@ -444,3 +490,29 @@ func (in *VolumeList) DeepCopyObject() runtime.Object { } return nil } + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *VolumeSource) DeepCopyInto(out *VolumeSource) { + *out = *in + out.PersistentVolumeClaimSource = in.PersistentVolumeClaimSource + in.Resources.DeepCopyInto(&out.Resources) + in.Placement.DeepCopyInto(&out.Placement) + if in.Config != nil { + in, out := &in.Config, &out.Config + *out = make(map[string]string, len(*in)) + for key, val := range *in { + (*out)[key] = val + } + } + return +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new VolumeSource. +func (in *VolumeSource) DeepCopy() *VolumeSource { + if in == nil { + return nil + } + out := new(VolumeSource) + in.DeepCopyInto(out) + return out +} diff --git a/pkg/clusterd/disk.go b/pkg/clusterd/disk.go index 48f9a3fd1..0509fb2f4 100644 --- a/pkg/clusterd/disk.go +++ b/pkg/clusterd/disk.go @@ -54,93 +54,104 @@ func DiscoverDevices(executor exec.Executor) ([]*sys.LocalDisk, error) { // skip device continue } - - diskProps, err := sys.GetDeviceProperties(d, executor) - if err != nil { - logger.Warningf("skipping device %s: %+v", d, err) + disk := PopulateDeviceInfo(d, executor) + if disk == nil { continue } - - diskType, ok := diskProps["TYPE"] - if !ok || (diskType != sys.SSDType && diskType != sys.CryptType && diskType != sys.DiskType && diskType != sys.PartType && diskType != sys.LinearType) { - if !ok { - logger.Warningf("skipping device %s: diskType is empty", d) - } else { - logger.Warningf("skipping device %s: unsupported diskType %+s", d, diskType) - } - // unsupported disk type, just continue - continue - } - - // get the UUID for disks - var diskUUID string - if diskType != sys.PartType { - diskUUID, err = sys.GetDiskUUID(d, executor) - if err != nil { - logger.Warningf("device %s has an unknown uuid. %+v", d, err) - continue - } - } - - udevInfo, err := sys.GetUdevInfo(d, executor) - if err != nil { - logger.Warningf("failed to get udev info for device %s: %+v", d, err) - continue - } - - disk := &sys.LocalDisk{Name: d, UUID: diskUUID} - - if val, ok := diskProps["TYPE"]; ok { - disk.Type = val - } - if val, ok := diskProps["SIZE"]; ok { - if size, err := strconv.ParseUint(val, 10, 64); err == nil { - disk.Size = size - } - } - if val, ok := diskProps["ROTA"]; ok { - if rotates, err := strconv.ParseBool(val); err == nil { - disk.Rotational = rotates - } - } - if val, ok := diskProps["RO"]; ok { - if ro, err := strconv.ParseBool(val); err == nil { - disk.Readonly = ro - } - } - if val, ok := diskProps["PKNAME"]; ok { - disk.Parent = val - } - - // parse udev info output - if val, ok := udevInfo["DEVLINKS"]; ok { - disk.DevLinks = val - } - if val, ok := udevInfo["ID_FS_TYPE"]; ok { - disk.Filesystem = val - } - if val, ok := udevInfo["ID_SERIAL"]; ok { - disk.Serial = val - } - - if val, ok := udevInfo["ID_VENDOR"]; ok { - disk.Vendor = val - } - - if val, ok := udevInfo["ID_MODEL"]; ok { - disk.Model = val - } - - if val, ok := udevInfo["ID_WWN_WITH_EXTENSION"]; ok { - disk.WWNVendorExtension = val - } - - if val, ok := udevInfo["ID_WWN"]; ok { - disk.WWN = val - } - + disk = PopulateDeviceUdevInfo(d, executor, disk) disks = append(disks, disk) } return disks, nil } + +func PopulateDeviceInfo(d string, executor exec.Executor) *sys.LocalDisk { + diskProps, err := sys.GetDeviceProperties(d, executor) + if err != nil { + logger.Warningf("skipping device %s: %+v", d, err) + return nil + } + + diskType, ok := diskProps["TYPE"] + if !ok || (diskType != sys.SSDType && diskType != sys.CryptType && diskType != sys.DiskType && diskType != sys.PartType && diskType != sys.LinearType) { + if !ok { + logger.Warningf("skipping device %s: diskType is empty", d) + } else { + logger.Warningf("skipping device %s: unsupported diskType %+s", d, diskType) + } + // unsupported disk type, just continue + return nil + } + + // get the UUID for disks + var diskUUID string + if diskType != sys.PartType { + diskUUID, err = sys.GetDiskUUID(d, executor) + if err != nil { + logger.Warningf("device %s has an unknown uuid. %+v", d, err) + return nil + } + } + + disk := &sys.LocalDisk{Name: d, UUID: diskUUID} + + if val, ok := diskProps["TYPE"]; ok { + disk.Type = val + } + if val, ok := diskProps["SIZE"]; ok { + if size, err := strconv.ParseUint(val, 10, 64); err == nil { + disk.Size = size + } + } + if val, ok := diskProps["ROTA"]; ok { + if rotates, err := strconv.ParseBool(val); err == nil { + disk.Rotational = rotates + } + } + if val, ok := diskProps["RO"]; ok { + if ro, err := strconv.ParseBool(val); err == nil { + disk.Readonly = ro + } + } + if val, ok := diskProps["PKNAME"]; ok { + disk.Parent = val + } + + return disk +} + +func PopulateDeviceUdevInfo(d string, executor exec.Executor, disk *sys.LocalDisk) *sys.LocalDisk { + udevInfo, err := sys.GetUdevInfo(d, executor) + if err != nil { + logger.Warningf("failed to get udev info for device %s: %+v", d, err) + return disk + } + // parse udev info output + if val, ok := udevInfo["DEVLINKS"]; ok { + disk.DevLinks = val + } + if val, ok := udevInfo["ID_FS_TYPE"]; ok { + disk.Filesystem = val + } + if val, ok := udevInfo["ID_SERIAL"]; ok { + disk.Serial = val + } + + if val, ok := udevInfo["ID_VENDOR"]; ok { + disk.Vendor = val + } + + if val, ok := udevInfo["ID_MODEL"]; ok { + disk.Model = val + } + + if val, ok := udevInfo["ID_WWN_WITH_EXTENSION"]; ok { + disk.WWNVendorExtension = val + } + + if val, ok := udevInfo["ID_WWN"]; ok { + disk.WWN = val + } + + return disk +} diff --git a/pkg/daemon/ceph/osd/agent.go b/pkg/daemon/ceph/osd/agent.go index 90b1c1b61..121da5df9 100644 --- a/pkg/daemon/ceph/osd/agent.go +++ b/pkg/daemon/ceph/osd/agent.go @@ -56,6 +56,7 @@ type OsdAgent struct { procMan *proc.ProcManager storeConfig config.StoreConfig kv *k8sutil.ConfigMapKVStore + pvcBacked bool configCounter int32 osdsCompleted chan struct{} } @@ -66,7 +67,7 @@ type device struct { } func NewAgent(context *clusterd.Context, devices []DesiredDevice, metadataDevice, directories string, forceFormat bool, - location string, storeConfig config.StoreConfig, cluster *cephconfig.ClusterInfo, nodeName string, kv *k8sutil.ConfigMapKVStore) *OsdAgent { + location string, storeConfig config.StoreConfig, cluster *cephconfig.ClusterInfo, nodeName string, kv *k8sutil.ConfigMapKVStore, pvcBacked bool) *OsdAgent { return &OsdAgent{ devices: devices, @@ -78,6 +79,7 @@ func NewAgent(context *clusterd.Context, devices []DesiredDevice, metadataDevice cluster: cluster, nodeName: nodeName, kv: kv, + pvcBacked: pvcBacked, procMan: proc.New(context.Executor), osdProc: make(map[int]*proc.MonitoredProc), } @@ -147,10 +149,17 @@ func (a *OsdAgent) removeDirs(context *clusterd.Context, removedDirs map[string] func (a *OsdAgent) configureDevices(context *clusterd.Context, devices *DeviceOsdMapping) ([]oposd.OSDInfo, error) { - cvSupported, err := getCephVolumeSupported(context) - if err != nil { - logger.Errorf("failed to detect if ceph-volume is available. %+v", err) + var cvSupported bool + var err error + if a.pvcBacked { + cvSupported = true //ceph-volume is always supported when OSD is backed by PVC + } else { + cvSupported, err = getCephVolumeSupported(context) + if err != nil { + logger.Errorf("failed to detect if ceph-volume is available. %+v", err) + } } + if a.metadataDevice != "" { // ceph-volume still is work in progress for accepting fast devices for the metadata logger.Warningf("ceph-volume metadata support is experimental. osd provision might fail if vg on %s does not have enough space", a.metadataDevice) @@ -160,7 +169,7 @@ func (a *OsdAgent) configureDevices(context *clusterd.Context, devices *DeviceOs if devices == nil || len(devices.Entries) == 0 { logger.Infof("no more devices to configure") if cvSupported { - return getCephVolumeOSDs(context, a.cluster.Name, a.cluster.FSID) + return getCephVolumeOSDs(context, a.cluster.Name, a.cluster.FSID, "") } return osds, nil } diff --git a/pkg/daemon/ceph/osd/agent_test.go b/pkg/daemon/ceph/osd/agent_test.go index 51b99d269..77d8c1834 100644 --- a/pkg/daemon/ceph/osd/agent_test.go +++ b/pkg/daemon/ceph/osd/agent_test.go @@ -247,6 +247,8 @@ func testOSDAgentWithDevicesHelper(t *testing.T, storeConfig config.StoreConfig, "sdx": {Data: -1}, "sdy": {Data: -1}, }} + + agent.pvcBacked = false _, err = agent.configureDevices(context, devices) assert.Nil(t, err) @@ -398,7 +400,7 @@ func createTestAgent(t *testing.T, devices, configDir, nodeName string, storeCon cluster := &cephconfig.ClusterInfo{Name: "myclust"} context := &clusterd.Context{ConfigDir: configDir, Executor: executor, Clientset: testop.New(1)} agent := NewAgent(context, desiredDevices, "", "", forceFormat, location, *storeConfig, - cluster, nodeName, mockKVStore()) + cluster, nodeName, mockKVStore(), false) return agent, executor, context } @@ -465,7 +467,8 @@ func TestGetPartitionPerfScheme(t *testing.T) { } context.Executor = executor - devices, err := getAvailableDevices(context, []DesiredDevice{{Name: "sda"}, {Name: "sdb"}}, "sdc") + pvcBackedOSD := false + devices, err := getAvailableDevices(context, []DesiredDevice{{Name: "sda"}, {Name: "sdb"}}, "sdc", pvcBackedOSD) assert.Nil(t, err) scheme, _, err := a.getPartitionPerfScheme(context, devices, false) assert.Nil(t, err) @@ -542,7 +545,8 @@ func TestGetPartitionSchemeDiskInUse(t *testing.T) { // get the partition scheme based on the available devices. Since sda is already in use, the partition // scheme returned should reflect that. - devices, err := getAvailableDevices(context, []DesiredDevice{{Name: "sda"}}, "") + pvcBackedOSD := false + devices, err := getAvailableDevices(context, []DesiredDevice{{Name: "sda"}}, "", pvcBackedOSD) scheme, _, err := a.getPartitionPerfScheme(context, devices, false) assert.Nil(t, err) @@ -610,7 +614,8 @@ func TestGetPartitionSchemeDiskNameChanged(t *testing.T) { // get the current partition scheme. This should notice that the device names changed and update the // partition scheme to have the latest device names - devices, err := getAvailableDevices(context, []DesiredDevice{{Name: "sda-changed"}}, "nvme01") + pvcBackedOSD := false + devices, err := getAvailableDevices(context, []DesiredDevice{{Name: "sda-changed"}}, "nvme01", pvcBackedOSD) scheme, _, err := a.getPartitionPerfScheme(context, devices, false) assert.Nil(t, err) require.NotNil(t, scheme) diff --git a/pkg/daemon/ceph/osd/daemon.go b/pkg/daemon/ceph/osd/daemon.go index beefe0254..29d0ebc4d 100644 --- a/pkg/daemon/ceph/osd/daemon.go +++ b/pkg/daemon/ceph/osd/daemon.go @@ -39,7 +39,7 @@ var ( ) // StartOSD starts an OSD on a device that was provisioned by ceph-volume -func StartOSD(context *clusterd.Context, osdType, osdID, osdUUID string, cephArgs []string) error { +func StartOSD(context *clusterd.Context, osdType, osdID, osdUUID string, pvcBackedOSD bool, cephArgs []string) error { // ensure the config mount point exists configDir := fmt.Sprintf("/var/lib/ceph/osd/ceph-%s", osdID) @@ -48,6 +48,16 @@ func StartOSD(context *clusterd.Context, osdType, osdID, osdUUID string, cephArg logger.Errorf("failed to create config dir %s. %+v", configDir, err) } + if pvcBackedOSD { + if err := updateLVMConfig(context); err != nil { + return fmt.Errorf("sed failure, %+v", err) // fail return here as validation provided by ceph-volume + } + + context.Executor.ExecuteCommand(false, "", "/sbin/vgchange", "-an") + + context.Executor.ExecuteCommand(false, "", "/sbin/vgchange", "-ay") + } + // activate the osd with ceph-volume storeFlag := "--" + osdType if err := context.Executor.ExecuteCommand(false, "", "stdbuf", "-oL", "ceph-volume", "lvm", "activate", "--no-systemd", storeFlag, osdID, osdUUID); err != nil { @@ -100,16 +110,25 @@ func Provision(context *clusterd.Context, agent *OsdAgent) error { } logger.Infof("discovering hardware") - rawDevices, err := clusterd.DiscoverDevices(context.Executor) - if err != nil { - return fmt.Errorf("failed initial hardware discovery. %+v", err) + var rawDevices []*sys.LocalDisk + if agent.pvcBacked { + if len(agent.devices) > 1 { + return fmt.Errorf("more than one desired device found in case of PVC backed OSDs. we expect exactly one device") + } + rawDevices = append(rawDevices, clusterd.PopulateDeviceInfo(agent.devices[0].Name, context.Executor)) + } else { + rawDevices, err = clusterd.DiscoverDevices(context.Executor) + if err != nil { + return fmt.Errorf("failed initial hardware discovery. %+v", err) + } } + context.Devices = rawDevices logger.Infof("creating and starting the osds") // determine the set of devices that can/should be used for OSDs. - devices, err := getAvailableDevices(context, agent.devices, agent.metadataDevice) + devices, err := getAvailableDevices(context, agent.devices, agent.metadataDevice, agent.pvcBacked) if err != nil { return fmt.Errorf("failed to get available devices. %+v", err) } @@ -121,7 +140,7 @@ func Provision(context *clusterd.Context, agent *OsdAgent) error { } // orchestration is about to start, update the status - status = oposd.OrchestrationStatus{Status: oposd.OrchestrationStatusOrchestrating} + status = oposd.OrchestrationStatus{Status: oposd.OrchestrationStatusOrchestrating, PvcBackedOSD: agent.pvcBacked} if err := oposd.UpdateNodeStatus(agent.kv, agent.nodeName, status); err != nil { return err } @@ -168,7 +187,7 @@ func Provision(context *clusterd.Context, agent *OsdAgent) error { osds := append(deviceOSDs, dirOSDs...) // orchestration is completed, update the status - status = oposd.OrchestrationStatus{OSDs: osds, Status: oposd.OrchestrationStatusCompleted} + status = oposd.OrchestrationStatus{OSDs: osds, Status: oposd.OrchestrationStatusCompleted, PvcBackedOSD: agent.pvcBacked} if err := oposd.UpdateNodeStatus(agent.kv, agent.nodeName, status); err != nil { return err } @@ -176,7 +195,7 @@ func Provision(context *clusterd.Context, agent *OsdAgent) error { return nil } -func getAvailableDevices(context *clusterd.Context, desiredDevices []DesiredDevice, metadataDevice string) (*DeviceOsdMapping, error) { +func getAvailableDevices(context *clusterd.Context, desiredDevices []DesiredDevice, metadataDevice string, pvcBacked bool) (*DeviceOsdMapping, error) { available := &DeviceOsdMapping{Entries: map[string]*DeviceOsdIDEntry{}} @@ -189,7 +208,7 @@ func getAvailableDevices(context *clusterd.Context, desiredDevices []DesiredDevi if device.Type == sys.PartType { continue } - partCount, ownPartitions, fs, err := sys.CheckIfDeviceAvailable(context.Executor, device.Name) + partCount, ownPartitions, fs, err := sys.CheckIfDeviceAvailable(context.Executor, device.Name, pvcBacked) if err != nil { return nil, fmt.Errorf("failed to get device %s info. %+v", device.Name, err) } diff --git a/pkg/daemon/ceph/osd/daemon_test.go b/pkg/daemon/ceph/osd/daemon_test.go index 43ac9ac8e..656d41be5 100644 --- a/pkg/daemon/ceph/osd/daemon_test.go +++ b/pkg/daemon/ceph/osd/daemon_test.go @@ -70,6 +70,8 @@ func TestRunDaemon(t *testing.T) { agent, _, context := createTestAgent(t, "none", configDir, "node5375", &config.StoreConfig{StoreType: config.Bluestore}) agent.devices[0].IsFilter = true + agent.pvcBacked = false + logger.Infof("Agent %+v", agent) err := Provision(context, agent) assert.Nil(t, err) } @@ -209,7 +211,8 @@ NAME="sdb1" SIZE="30" TYPE="part" PKNAME="sdb"`, nil } // select all devices, including nvme01 for metadata - mapping, err := getAvailableDevices(context, []DesiredDevice{{Name: "all"}}, "nvme01") + pvcBackedOSD := false + mapping, err := getAvailableDevices(context, []DesiredDevice{{Name: "all"}}, "nvme01", pvcBackedOSD) assert.Nil(t, err) assert.Equal(t, 5, len(mapping.Entries)) assert.Equal(t, -1, mapping.Entries["sda"].Data) @@ -221,29 +224,29 @@ NAME="sdb1" SIZE="30" TYPE="part" PKNAME="sdb"`, nil assert.Equal(t, 0, len(mapping.Entries["nvme01"].Metadata)) // select no devices both using and not using a filter - mapping, err = getAvailableDevices(context, nil, "") + mapping, err = getAvailableDevices(context, nil, "", pvcBackedOSD) assert.Nil(t, err) assert.Equal(t, 0, len(mapping.Entries)) - mapping, err = getAvailableDevices(context, nil, "") + mapping, err = getAvailableDevices(context, nil, "", pvcBackedOSD) assert.Nil(t, err) assert.Equal(t, 0, len(mapping.Entries)) // select the sd* devices - mapping, err = getAvailableDevices(context, []DesiredDevice{{Name: "^sd.$", IsFilter: true}}, "") + mapping, err = getAvailableDevices(context, []DesiredDevice{{Name: "^sd.$", IsFilter: true}}, "", pvcBackedOSD) assert.Nil(t, err) assert.Equal(t, 2, len(mapping.Entries)) assert.Equal(t, -1, mapping.Entries["sda"].Data) assert.Equal(t, -1, mapping.Entries["sdd"].Data) // select an exact device - mapping, err = getAvailableDevices(context, []DesiredDevice{{Name: "sdd"}}, "") + mapping, err = getAvailableDevices(context, []DesiredDevice{{Name: "sdd"}}, "", pvcBackedOSD) assert.Nil(t, err) assert.Equal(t, 1, len(mapping.Entries)) assert.Equal(t, -1, mapping.Entries["sdd"].Data) // select all devices except those that have a prefix of "s" - mapping, err = getAvailableDevices(context, []DesiredDevice{{Name: "^[^s]", IsFilter: true}}, "") + mapping, err = getAvailableDevices(context, []DesiredDevice{{Name: "^[^s]", IsFilter: true}}, "", pvcBackedOSD) assert.Nil(t, err) assert.Equal(t, 3, len(mapping.Entries)) assert.Equal(t, -1, mapping.Entries["rda"].Data) diff --git a/pkg/daemon/ceph/osd/device.go b/pkg/daemon/ceph/osd/device.go index 63639dc75..9eda86b72 100644 --- a/pkg/daemon/ceph/osd/device.go +++ b/pkg/daemon/ceph/osd/device.go @@ -111,7 +111,8 @@ func formatDevice(context *clusterd.Context, config *osdConfig, forceFormat bool } // check if partitions belong to rook - _, ownPartitions, devFS, err := sys.CheckIfDeviceAvailable(context.Executor, dataDetails.Device) + pvcBackedOSD := false + _, ownPartitions, devFS, err := sys.CheckIfDeviceAvailable(context.Executor, dataDetails.Device, pvcBackedOSD) if err != nil { return nil, fmt.Errorf("failed to format device. %+v", err) } @@ -171,7 +172,8 @@ func partitionMetadata(context *clusterd.Context, info *config.MetadataDeviceInf } // check one last time to make sure it's OK for us to format this metadata device - _, ownPartitions, fs, err := sys.CheckIfDeviceAvailable(context.Executor, info.Device) + pvcBackedOSD := false + _, ownPartitions, fs, err := sys.CheckIfDeviceAvailable(context.Executor, info.Device, pvcBackedOSD) if err != nil { return fmt.Errorf("failed to get metadata device %s info: %+v", info.Device, err) } else if fs != "" || !ownPartitions { diff --git a/pkg/daemon/ceph/osd/device_test.go b/pkg/daemon/ceph/osd/device_test.go index 12cbfaeef..7e14c9a92 100644 --- a/pkg/daemon/ceph/osd/device_test.go +++ b/pkg/daemon/ceph/osd/device_test.go @@ -117,7 +117,7 @@ NAME="sda3" SIZE="20" TYPE="part" PKNAME="sda"` assert.Equal(t, "lsblk /dev/sda", name) output = "" case 5: - assert.Equal(t, "get filesystem type for sda", name) + assert.Equal(t, "get filesystem type for /dev/sda", name) output = "" } outputExecCount++ diff --git a/pkg/daemon/ceph/osd/volume.go b/pkg/daemon/ceph/osd/volume.go index 1a1fa0d23..df84223af 100644 --- a/pkg/daemon/ceph/osd/volume.go +++ b/pkg/daemon/ceph/osd/volume.go @@ -21,6 +21,7 @@ import ( "fmt" "path" "strconv" + "strings" "syscall" "github.com/rook/rook/pkg/clusterd" @@ -28,6 +29,7 @@ import ( "github.com/rook/rook/pkg/operator/ceph/cluster/osd/config" "github.com/rook/rook/pkg/util/display" "github.com/rook/rook/pkg/util/exec" + "github.com/rook/rook/pkg/util/sys" ) var cephConfigDir = "/var/lib/ceph" @@ -42,11 +44,12 @@ const ( func (a *OsdAgent) configureCVDevices(context *clusterd.Context, devices *DeviceOsdMapping) ([]oposd.OSDInfo, error) { var osds []oposd.OSDInfo + var lv string var err error if len(devices.Entries) == 0 { logger.Infof("no new devices to configure. returning devices already configured with ceph-volume.") - osds, err = getCephVolumeOSDs(context, a.cluster.Name, a.cluster.FSID) + osds, err = getCephVolumeOSDs(context, a.cluster.Name, a.cluster.FSID, lv) if err != nil { logger.Infof("failed to get devices already provisioned by ceph-volume. %+v", err) } @@ -57,15 +60,97 @@ func (a *OsdAgent) configureCVDevices(context *clusterd.Context, devices *Device if err != nil { return nil, fmt.Errorf("failed to generate osd keyring. %+v", err) } - - if err = a.initializeDevices(context, devices); err != nil { - return nil, fmt.Errorf("failed to initialize devices. %+v", err) + if a.pvcBacked { + if lv, err = a.initializeBlockPVC(context, devices); err != nil { + return nil, fmt.Errorf("failed to initialize devices. %+v", err) + } + } else { + if err = a.initializeDevices(context, devices); err != nil { + return nil, fmt.Errorf("failed to initialize devices. %+v", err) + } } - osds, err = getCephVolumeOSDs(context, a.cluster.Name, a.cluster.FSID) + osds, err = getCephVolumeOSDs(context, a.cluster.Name, a.cluster.FSID, lv) return osds, err } +func (a *OsdAgent) initializeBlockPVC(context *clusterd.Context, devices *DeviceOsdMapping) (string, error) { + if err := updateLVMConfig(context); err != nil { + return "", fmt.Errorf("sed failure, %+v", err) // fail return here as validation provided by ceph-volume + } + baseCommand := "stdbuf" + baseArgs := []string{"-oL", cephVolumeCmd, "lvm", "prepare"} + var lvpath string + for name, device := range devices.Entries { + if device.LegacyPartitionsFound { + logger.Infof("skipping device %s configured with legacy rook osd", name) + continue + } + if device.Data == -1 { + logger.Infof("configuring new device %s", name) + deviceArg := device.Config.Name + immediateExecuteArgs := append(baseArgs, []string{ + "--data", + deviceArg, + }...) + // execute ceph-volume with the device + + if op, err := context.Executor.ExecuteCommandWithOutput(false, "", baseCommand, immediateExecuteArgs...); err != nil { + return "", fmt.Errorf("failed ceph-volume. %+v", err) // fail return here as validation provided by ceph-volume + } else { + logger.Infof("%v", op) + lvpath = getLVPath(op) + if lvpath == "" { + return "", fmt.Errorf("failed to get lvpath from ceph-volume lvm prepare output") + } + } + } else { + logger.Infof("skipping device %s with osd %d already configured", name, device.Data) + } + } + + return lvpath, nil +} + +func getLVPath(op string) string { + tmp := sys.Grep(op, "Volume group") + vgtmp := strings.Split(tmp, "\"") + + tmp = sys.Grep(op, "Logical volume") + lvtmp := strings.Split(tmp, "\"") + if len(vgtmp) > 0 && len(lvtmp) > 0 { + if sys.Grep(vgtmp[1], "ceph") != "" && sys.Grep(lvtmp[1], "osd-block") != "" { + return fmt.Sprintf("/dev/%s/%s", vgtmp[1], lvtmp[1]) + } + } + return "" +} + +func updateLVMConfig(context *clusterd.Context) error { + testArgs := []string{"-oL", "sed", "-i", "-e", "s#udev_sync = 1#udev_sync = 0#", "/etc/lvm/lvm.conf"} + if err := context.Executor.ExecuteCommand(false, "", "stdbuf", testArgs...); err != nil { + return fmt.Errorf("failed ceph-volume. %+v", err) // fail return here as validation provided by ceph-volume + } + testArgs = []string{"-oL", "sed", "-i", "-e", "s#udev_rules = 1#udev_rules = 0#", "/etc/lvm/lvm.conf"} + if err := context.Executor.ExecuteCommand(false, "", "stdbuf", testArgs...); err != nil { + return fmt.Errorf("failed ceph-volume. %+v", err) // fail return here as validation provided by ceph-volume + } + testArgs = []string{"-oL", "sed", "-i", "-e", "s#use_lvmetad = 1#use_lvmetad = 0#", "/etc/lvm/lvm.conf"} + if err := context.Executor.ExecuteCommand(false, "", "stdbuf", testArgs...); err != nil { + return fmt.Errorf("failed ceph-volume. %+v", err) // fail return here as validation provided by ceph-volume + } + testArgs = []string{"-oL", "sed", "-i", "-e", "s#scan = \\[ \"/dev\" \\]#scan = [ \"/dev\", \"/mnt\" ]#", "/etc/lvm/lvm.conf"} + if err := context.Executor.ExecuteCommand(false, "", "stdbuf", testArgs...); err != nil { + return fmt.Errorf("failed ceph-volume. %+v", err) // fail return here as validation provided by ceph-volume + } + testArgs = []string{"-oL", "sed", "-i", "-e", "0,/# filter =.*/{s%# filter =.*% filter = [ \"a|^/mnt/.*| r|.*/|\" ]%}", "/etc/lvm/lvm.conf"} + if err := context.Executor.ExecuteCommand(false, "", "stdbuf", testArgs...); err != nil { + return fmt.Errorf("failed ceph-volume. %+v", err) // fail return here as validation provided by ceph-volume + } + + return nil +} + func (a *OsdAgent) initializeDevices(context *clusterd.Context, devices *DeviceOsdMapping) error { storeFlag := "--bluestore" if a.storeConfig.StoreType == config.Filestore { @@ -133,6 +218,9 @@ func (a *OsdAgent) initializeDevices(context *clusterd.Context, devices *DeviceO "--report", }...) + logger.Infof("Base command - %+v", baseCommand) + logger.Infof("immediateReportArgs - %+v", baseCommand) + logger.Infof("immediateExecuteArgs - %+v", immediateExecuteArgs) if err := context.Executor.ExecuteCommand(false, "", baseCommand, immediateReportArgs...); err != nil { return fmt.Errorf("failed ceph-volume report. %+v", err) // fail return here as validation provided by ceph-volume } @@ -200,7 +288,9 @@ func sanitizeOSDsPerDevice(count int) string { } func getCephVolumeSupported(context *clusterd.Context) (bool, error) { + _, err := context.Executor.ExecuteCommandWithOutput(false, "", cephVolumeCmd, "lvm", "batch", "--prepare") + if err != nil { if cmdErr, ok := err.(*exec.CommandError); ok { exitStatus := cmdErr.ExitStatus() @@ -216,9 +306,9 @@ func getCephVolumeSupported(context *clusterd.Context) (bool, error) { return true, nil } -func getCephVolumeOSDs(context *clusterd.Context, clusterName string, cephfsid string) ([]oposd.OSDInfo, error) { +func getCephVolumeOSDs(context *clusterd.Context, clusterName string, cephfsid string, lv string) ([]oposd.OSDInfo, error) { - result, err := context.Executor.ExecuteCommandWithOutput(false, "", cephVolumeCmd, "lvm", "list", "--format", "json") + result, err := context.Executor.ExecuteCommandWithOutput(false, "", cephVolumeCmd, "lvm", "list", lv, "--format", "json") if err != nil { return nil, fmt.Errorf("failed to retrieve ceph-volume results. %+v", err) } diff --git a/pkg/daemon/ceph/osd/volume_test.go b/pkg/daemon/ceph/osd/volume_test.go index 8020c9038..5de99401e 100644 --- a/pkg/daemon/ceph/osd/volume_test.go +++ b/pkg/daemon/ceph/osd/volume_test.go @@ -252,7 +252,7 @@ func TestParseCephVolumeResult(t *testing.T) { } context := &clusterd.Context{Executor: executor} - osds, err := getCephVolumeOSDs(context, "rook", "4bfe8b72-5e69-4330-b6c0-4d914db8ab89") + osds, err := getCephVolumeOSDs(context, "rook", "4bfe8b72-5e69-4330-b6c0-4d914db8ab89", "") assert.Nil(t, err) require.NotNil(t, osds) assert.Equal(t, 2, len(osds)) @@ -272,7 +272,7 @@ func TestCephVolumeResultMultiClusterSingleOSD(t *testing.T) { } context := &clusterd.Context{Executor: executor} - osds, err := getCephVolumeOSDs(context, "rook", "451267e6-883f-4936-8dff-080d781c67d5") + osds, err := getCephVolumeOSDs(context, "rook", "451267e6-883f-4936-8dff-080d781c67d5", "") assert.Nil(t, err) require.NotNil(t, osds) assert.Equal(t, 1, len(osds)) @@ -293,7 +293,7 @@ func TestCephVolumeResultMultiClusterMultiOSD(t *testing.T) { } context := &clusterd.Context{Executor: executor} - osds, err := getCephVolumeOSDs(context, "rook", "451267e6-883f-4936-8dff-080d781c67d5") + osds, err := getCephVolumeOSDs(context, "rook", "451267e6-883f-4936-8dff-080d781c67d5", "") assert.Nil(t, err) require.NotNil(t, osds) assert.Equal(t, 1, len(osds)) diff --git a/pkg/daemon/discover/discover.go b/pkg/daemon/discover/discover.go index 3b7a65564..0380f7ae1 100644 --- a/pkg/daemon/discover/discover.go +++ b/pkg/daemon/discover/discover.go @@ -34,7 +34,7 @@ import ( "github.com/rook/rook/pkg/operator/k8sutil" "github.com/rook/rook/pkg/util/sys" - "k8s.io/api/core/v1" + v1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ) @@ -344,6 +344,7 @@ func updateDeviceCM(context *clusterd.Context) error { func probeDevices(context *clusterd.Context) ([]sys.LocalDisk, error) { devices := make([]sys.LocalDisk, 0) localDevices, err := clusterd.DiscoverDevices(context.Executor) + logger.Infof("localdevices - %+v", localDevices) if err != nil { return devices, fmt.Errorf("failed initial hardware discovery. %+v", err) } diff --git a/pkg/daemon/discover/discover_test.go b/pkg/daemon/discover/discover_test.go index 9c567a36d..b1ca5fb60 100644 --- a/pkg/daemon/discover/discover_test.go +++ b/pkg/daemon/discover/discover_test.go @@ -78,7 +78,7 @@ func TestProbeDevices(t *testing.T) { output = "testa" case "lsblk /dev/testa": output = `SIZE="249510756352" ROTA="1" RO="0" TYPE="disk" PKNAME=""` - case "get filesystem type for testa": + case "get filesystem type for /dev/testa": output = udevOutput case "get parent for device testa": output = ` testa @@ -88,7 +88,7 @@ testa2 centos_host13-root testa2 centos_host13-swap testa2 centos_host13-home ` - case "get disk testa uuid": + case "get disk /dev/testa uuid": output = sgdiskOutput case "get disk testa fs serial": diff --git a/pkg/operator/ceph/cluster/osd/deviceSet.go b/pkg/operator/ceph/cluster/osd/deviceSet.go new file mode 100644 index 000000000..0f5a432e1 --- /dev/null +++ b/pkg/operator/ceph/cluster/osd/deviceSet.go @@ -0,0 +1,129 @@ +/* +Copyright 2016 The Rook Authors. All rights reserved. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package osd + +import ( + "fmt" + + rookalpha "github.com/rook/rook/pkg/apis/rook.io/v1alpha2" + opspec "github.com/rook/rook/pkg/operator/ceph/spec" + v1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/kubernetes" +) + +func (c *Cluster) prepareStorageClassDeviceSets(config *provisionConfig) []rookalpha.VolumeSource { + volumeSources := []rookalpha.VolumeSource{} + for _, storageClassDeviceSet := range c.DesiredStorage.StorageClassDeviceSets { + if err := opspec.CheckPodMemory(storageClassDeviceSet.Resources, cephOsdPodMinimumMemory); err != nil { + config.addError("cannot use storageClassDeviceSet %s for creating osds %v", storageClassDeviceSet.Name, err) + continue + } + for i := 0; i < storageClassDeviceSet.Count; i++ { + pvc, err := c.createStorageClassDeviceSetPVC(storageClassDeviceSet, i) + if err != nil { + config.addError("%+v", err) + config.addError("OSD creation for storageClassDeviceSet %v failed for count %v", storageClassDeviceSet.Name, i) + continue + } + volumeSources = append(volumeSources, rookalpha.VolumeSource{ + Name: storageClassDeviceSet.Name, + Resources: storageClassDeviceSet.Resources, + Placement: storageClassDeviceSet.Placement, + Config: storageClassDeviceSet.Config, + PersistentVolumeClaimSource: v1.PersistentVolumeClaimVolumeSource{ + ClaimName: pvc.GetName(), + ReadOnly: false, + }, + }) + logger.Infof("successfully provisioned osd for storageClassDeviceSet %s of set %v", storageClassDeviceSet.Name, i) + } + } + return volumeSources +} + +func (c *Cluster) createStorageClassDeviceSetPVC(storageClassDeviceSet rookalpha.StorageClassDeviceSet, setIndex int) (*v1.PersistentVolumeClaim, error) { + if len(storageClassDeviceSet.VolumeClaimTemplates) == 0 { + return nil, fmt.Errorf("No PVC available for storageClassDeviceSet %s", storageClassDeviceSet.Name) + } + pvcStorageClassDeviceSetPVCId, pvcStorageClassDeviceSetPVCIdLabelSelector := makeStorageClassDeviceSetPVCID(storageClassDeviceSet.Name, setIndex, 0) + + pvc := makeStorageClassDeviceSetPVC(storageClassDeviceSet.Name, pvcStorageClassDeviceSetPVCId, 0, setIndex, storageClassDeviceSet.VolumeClaimTemplates[0]) + // Check if a PVC already exists with same StorageClassDeviceSet label + presentPVCs, err := c.context.Clientset.CoreV1().PersistentVolumeClaims(c.Namespace).List(metav1.ListOptions{LabelSelector: pvcStorageClassDeviceSetPVCIdLabelSelector}) + if err != nil { + return nil, fmt.Errorf("failed to create pvc %v for storageClassDeviceSet %v, err %+v", pvc.GetGenerateName(), storageClassDeviceSet.Name, err) + } + if len(presentPVCs.Items) == 0 { // No PVC found, creating a new one + deployedPVC, err := c.context.Clientset.CoreV1().PersistentVolumeClaims(c.Namespace).Create(pvc) + if err != nil { + return nil, fmt.Errorf("failed to create pvc %v for storageClassDeviceSet %v, err %+v", pvc.GetGenerateName(), storageClassDeviceSet.Name, err) + } + return deployedPVC, nil + } else if len(presentPVCs.Items) == 1 { // The PVC is already present. + return &presentPVCs.Items[0], nil + } + // More than one PVC exists with same labelSelector + return nil, fmt.Errorf("more than one PVCs exists with label %v, pvcs %+v", pvcStorageClassDeviceSetPVCIdLabelSelector, presentPVCs) +} + +// GetValidVolumeSources returns all volumes that has an OSDDeployment +func GetValidVolumeSources(clientset kubernetes.Interface, namespace string, volumeSources []rookalpha.VolumeSource) (validVolumes []rookalpha.VolumeSource, err error) { + if clientset == nil { + return validVolumes, fmt.Errorf("nil clientset received") + } + for _, vol := range volumeSources { + osdDeployments, err := clientset.AppsV1().Deployments(namespace).List(metav1.ListOptions{LabelSelector: OSDOverPVCLabelKey + "=" + vol.PersistentVolumeClaimSource.ClaimName}) + if err != nil { + return validVolumes, err + } + osdPrepareJob, err := clientset.BatchV1().Jobs(namespace).List(metav1.ListOptions{LabelSelector: OSDOverPVCLabelKey + "=" + vol.PersistentVolumeClaimSource.ClaimName}) + if err != nil { + return validVolumes, err + } + if len(osdDeployments.Items) == 0 && len(osdPrepareJob.Items) == 0 { + validVolumes = append(validVolumes, vol) + } + } + return +} + +func makeStorageClassDeviceSetPVC(storageClassDeviceSetName, pvcStorageClassDeviceSetPVCId string, pvcIndex, setIndex int, pvcTemplate v1.PersistentVolumeClaim) (pvcs *v1.PersistentVolumeClaim) { + pvcLabels := makeStorageClassDeviceSetPVCLabel(storageClassDeviceSetName, pvcStorageClassDeviceSetPVCId, pvcIndex, setIndex) + + // pvc naming format -- + pvcGenerateName := pvcStorageClassDeviceSetPVCId + "-" + + // Add pvcTemplate name to pvc name. i.e --- + if len(pvcTemplate.GetName()) != 0 { + pvcGenerateName = pvcGenerateName + pvcTemplate.GetName() + "-" + } + + // Add user provided labels to pvcTemplates + for k, v := range pvcTemplate.GetLabels() { + pvcLabels[k] = v + } + + return &v1.PersistentVolumeClaim{ + ObjectMeta: metav1.ObjectMeta{ + GenerateName: pvcGenerateName, + Labels: pvcLabels, + Annotations: pvcTemplate.Annotations, + }, + Spec: pvcTemplate.Spec, + } +} diff --git a/pkg/operator/ceph/cluster/osd/osd.go b/pkg/operator/ceph/cluster/osd/osd.go index b5b80de70..0ff684923 100644 --- a/pkg/operator/ceph/cluster/osd/osd.go +++ b/pkg/operator/ceph/cluster/osd/osd.go @@ -126,13 +126,29 @@ type OSDInfo struct { } type OrchestrationStatus struct { - OSDs []OSDInfo `json:"osds"` - Status string `json:"status"` - Message string `json:"message"` + OSDs []OSDInfo `json:"osds"` + Status string `json:"status"` + PvcBackedOSD bool `json:"pvc-backed-osd"` + Message string `json:"message"` +} + +type osdProperties struct { + //crushHostname refers to the hostname or PVC name when the OSD is provisioned on Nodes or PVC block device, respectively. + crushHostname string + devices []rookalpha.Device + pvc v1.PersistentVolumeClaimVolumeSource + selection rookalpha.Selection + resources v1.ResourceRequirements + storeConfig osdconfig.StoreConfig + placement rookalpha.Placement + metadataDevice string + location string } // Start the osd management func (c *Cluster) Start() error { + config := newProvisionConfig() + // Validate pod's memory if specified // This is valid for both Filestore and Bluestore err := opspec.CheckPodMemory(c.resources, cephOsdPodMinimumMemory) @@ -142,52 +158,17 @@ func (c *Cluster) Start() error { logger.Infof("start running osds in namespace %s", c.Namespace) - if c.DesiredStorage.UseAllNodes == false && len(c.DesiredStorage.Nodes) == 0 { - logger.Warningf("useAllNodes is set to false and no nodes are specified, no OSD pods are going to be created") + if c.DesiredStorage.UseAllNodes == false && len(c.DesiredStorage.Nodes) == 0 && len(c.DesiredStorage.VolumeSources) == 0 && len(c.DesiredStorage.StorageClassDeviceSets) == 0 { + logger.Warningf("useAllNodes is set to false and no nodes, storageClassDevicesets or volumeSources are specified, no OSD pods are going to be created") } - if c.DesiredStorage.UseAllNodes { - // Get the list of all nodes in the cluster. The placement settings will be applied below. - hostnameMap, err := k8sutil.GetNodeHostNames(c.context.Clientset) - if err != nil { - logger.Warningf("failed to get node hostnames: %v", err) - return err - } - c.DesiredStorage.Nodes = nil - for _, hostname := range hostnameMap { - storageNode := rookalpha.Node{ - Name: hostname, - } - c.DesiredStorage.Nodes = append(c.DesiredStorage.Nodes, storageNode) - } - logger.Debugf("storage nodes: %+v", c.DesiredStorage.Nodes) - } - // generally speaking, this finds nodes which are capable of running new osds - validNodes := k8sutil.GetValidNodes(c.DesiredStorage, c.context.Clientset, c.placement) - - // no valid node is ready to run an osd - if len(validNodes) == 0 { - logger.Warningf("no valid node available to run an osd in namespace %s. "+ - "Rook will not create any new OSD nodes and will skip checking for removed nodes since "+ - "removing all OSD nodes without destroying the Rook cluster is unlikely to be intentional", c.Namespace) - return nil - } - logger.Infof("%d of the %d storage nodes are valid", len(validNodes), len(c.DesiredStorage.Nodes)) - c.ValidStorage = *c.DesiredStorage.DeepCopy() - c.ValidStorage.Nodes = validNodes - // start the jobs to provision the OSD devices and directories - config := newProvisionConfig() + + logger.Infof("start provisioning the osds on pvcs, if needed") + c.startProvisioningOverPVCs(config) + logger.Infof("start provisioning the osds on nodes, if needed") - c.startProvisioning(config) - - // start the OSD pods, waiting for the provisioning to be completed - logger.Infof("start osds after provisioning is completed, if needed") - c.completeProvision(config) - - // handle the removed nodes and rebalance the PGs - logger.Infof("checking if any nodes were removed") - c.handleRemovedNodes(config) + c.startProvisioningOverNodes(config) if len(config.errorMessages) > 0 { return fmt.Errorf("%d failures encountered while running osds in namespace %s: %+v", @@ -225,12 +206,105 @@ func (c *Cluster) Start() error { return nil } -func (c *Cluster) startProvisioning(config *provisionConfig) { +func (c *Cluster) startProvisioningOverPVCs(config *provisionConfig) { + // Parsing storageClassDeviceSets and parsing it to volume sources + c.DesiredStorage.VolumeSources = append(c.DesiredStorage.VolumeSources, c.prepareStorageClassDeviceSets(config)...) + + // Filtering out valid volume sources + validVolumeSources, err := GetValidVolumeSources(c.context.Clientset, c.Namespace, c.DesiredStorage.VolumeSources) + if err != nil { + config.addError(fmt.Sprintf("%v", err)) + return + } + c.ValidStorage.VolumeSources = validVolumeSources + + // no validVolumeSource is ready to run an osd + if len(c.DesiredStorage.VolumeSources) == 0 && len(c.DesiredStorage.StorageClassDeviceSets) == 0 { + logger.Warningf("no valid volumeSource available to run an osd in namespace %s. "+ + "Rook will not create any new OSD nodes and will skip checking for removed pvcs "+ + "removing all OSD nodes without destroying the Rook cluster is unlikely to be intentional", c.Namespace) + return + } + + logger.Infof("%d of the %d volumeSources are valid", len(validVolumeSources), len(c.DesiredStorage.VolumeSources)) + for _, volume := range c.ValidStorage.VolumeSources { + osdProps := osdProperties{ + crushHostname: volume.PersistentVolumeClaimSource.ClaimName, + pvc: volume.PersistentVolumeClaimSource, + resources: volume.Resources, + placement: volume.Placement, + } + + // update the orchestration status of this pvc to the starting state + status := OrchestrationStatus{Status: OrchestrationStatusStarting, PvcBackedOSD: true} + if err := c.updateOSDStatus(osdProps.crushHostname, status); err != nil { + config.addError("failed to set orchestration starting status for pvc %s: %+v", osdProps.crushHostname, err) + continue + } + + job, err := c.makeJob(osdProps) + if err != nil { + message := fmt.Sprintf("failed to create prepare job for pvc %s: %v", osdProps.crushHostname, err) + config.addError(message) + status := OrchestrationStatus{Status: OrchestrationStatusCompleted, Message: message, PvcBackedOSD: true} + if err := c.updateOSDStatus(osdProps.crushHostname, status); err != nil { + config.addError("failed to update pvc %s status. %+v", osdProps.crushHostname, err) + continue + } + } + + if !c.runJob(job, osdProps.crushHostname, config, "provision") { + status := OrchestrationStatus{ + Status: OrchestrationStatusCompleted, + Message: fmt.Sprintf("failed to start osd provisioning on pvc %s", osdProps.crushHostname), + PvcBackedOSD: true, + } + if err := c.updateOSDStatus(osdProps.crushHostname, status); err != nil { + config.addError("failed to update osd %s status. %+v", osdProps.crushHostname, err) + } + } + } + logger.Infof("start osds after provisioning is completed, if needed") + c.completeProvision(config) +} + +func (c *Cluster) startProvisioningOverNodes(config *provisionConfig) { if len(c.dataDirHostPath) == 0 { logger.Warningf("skipping osd provisioning where no dataDirHostPath is set") return } + if c.DesiredStorage.UseAllNodes { + // Get the list of all nodes in the cluster. The placement settings will be applied below. + hostnameMap, err := k8sutil.GetNodeHostNames(c.context.Clientset) + if err != nil { + config.addError("failed to get node hostnames: %v", err) + return + } + c.DesiredStorage.Nodes = nil + for _, hostname := range hostnameMap { + storageNode := rookalpha.Node{ + Name: hostname, + } + c.DesiredStorage.Nodes = append(c.DesiredStorage.Nodes, storageNode) + } + logger.Debugf("storage nodes: %+v", c.DesiredStorage.Nodes) + } + // generally speaking, this finds nodes which are capable of running new osds + validNodes := k8sutil.GetValidNodes(c.DesiredStorage, c.context.Clientset, c.placement) + + logger.Infof("%d of the %d storage nodes are valid", len(validNodes), len(c.DesiredStorage.Nodes)) + + c.ValidStorage = *c.DesiredStorage.DeepCopy() + c.ValidStorage.Nodes = validNodes + + // no valid node is ready to run an osd + if len(validNodes) == 0 { + logger.Warningf("no valid nodes available to run an osd in namespace %s. "+ + "Rook will not create any new OSD nodes and will skip checking for removed nodes since "+ + "removing all OSD nodes without destroying the Rook cluster is unlikely to be intentional", c.Namespace) + return + } // start with nodes currently in the storage spec for _, node := range c.ValidStorage.Nodes { // fully resolve the storage config and resources for this node @@ -247,7 +321,7 @@ func (c *Cluster) startProvisioning(config *provisionConfig) { // update the orchestration status of this node to the starting state status := OrchestrationStatus{Status: OrchestrationStatusStarting} - if err := c.updateNodeStatus(n.Name, status); err != nil { + if err := c.updateOSDStatus(n.Name, status); err != nil { config.addError("failed to set orchestration starting status for node %s: %+v", n.Name, err) continue } @@ -255,12 +329,21 @@ func (c *Cluster) startProvisioning(config *provisionConfig) { // create the job that prepares osds on the node storeConfig := osdconfig.ToStoreConfig(n.Config) metadataDevice := osdconfig.MetadataDevice(n.Config) - job, err := c.makeJob(n.Name, n.Devices, n.Selection, n.Resources, storeConfig, metadataDevice, n.Location) + osdProps := osdProperties{ + crushHostname: n.Name, + devices: n.Devices, + selection: n.Selection, + resources: n.Resources, + storeConfig: storeConfig, + metadataDevice: metadataDevice, + location: n.Location, + } + job, err := c.makeJob(osdProps) if err != nil { message := fmt.Sprintf("failed to create prepare job node %s: %v", n.Name, err) config.addError(message) status := OrchestrationStatus{Status: OrchestrationStatusCompleted, Message: message} - if err := c.updateNodeStatus(n.Name, status); err != nil { + if err := c.updateOSDStatus(n.Name, status); err != nil { config.addError("failed to update node %s status. %+v", n.Name, err) continue } @@ -268,11 +351,19 @@ func (c *Cluster) startProvisioning(config *provisionConfig) { if !c.runJob(job, n.Name, config, "provision") { status := OrchestrationStatus{Status: OrchestrationStatusCompleted, Message: fmt.Sprintf("failed to start osd provisioning on node %s", n.Name)} - if err := c.updateNodeStatus(n.Name, status); err != nil { + if err := c.updateOSDStatus(n.Name, status); err != nil { config.addError("failed to update node %s status. %+v", n.Name, err) } } } + logger.Infof("start osds after provisioning is completed, if needed") + c.completeProvision(config) + + // start the OSD pods, waiting for the provisioning to be completed + // handle the removed nodes and rebalance the PGs + logger.Infof("checking if any nodes were removed") + c.handleRemovedNodes(config) + } func (c *Cluster) runJob(job *batch.Job, nodeName string, config *provisionConfig, action string) bool { @@ -291,6 +382,56 @@ func (c *Cluster) runJob(job *batch.Job, nodeName string, config *provisionConfi return true } +func (c *Cluster) startOSDDaemonsOnPVC(pvcName string, config *provisionConfig, configMap *v1.ConfigMap, status *OrchestrationStatus) { + osds := status.OSDs + logger.Infof("starting %d osd daemons on pvc %s", len(osds), pvcName) + conf := make(map[string]string) + storeConfig := osdconfig.ToStoreConfig(conf) + osdProps := osdProperties{ + crushHostname: pvcName, + pvc: v1.PersistentVolumeClaimVolumeSource{ + ClaimName: pvcName, + }, + } + + // start osds + for _, osd := range osds { + logger.Debugf("start osd %v", osd) + dp, err := c.makeDeployment(osdProps, osd) + if err != nil { + errMsg := fmt.Sprintf("failed to create deployment for pvc %s: %v", osdProps.crushHostname, err) + config.addError(errMsg) + continue + } + + _, err = c.context.Clientset.AppsV1().Deployments(c.Namespace).Create(dp) + if err != nil { + if !errors.IsAlreadyExists(err) { + // we failed to create job, update the orchestration status for this pvc + logger.Warningf("failed to create osd deployment for pvc %s, osd %v: %+v", osdProps.crushHostname, osd, err) + continue + } + logger.Infof("deployment for osd %d already exists. updating if needed", osd.ID) + // Always invoke ceph version before an upgrade so we are sure to be up-to-date + daemon := string(opconfig.OsdType) + var cephVersionToUse cephver.CephVersion + currentCephVersion, err := client.LeastUptodateDaemonVersion(c.context, c.clusterInfo.Name, daemon) + if err != nil { + logger.Warningf("failed to retrieve current ceph %s version. %+v", daemon, err) + logger.Debug("could not detect ceph version during update, this is likely an initial bootstrap, proceeding with c.clusterInfo.CephVersion") + cephVersionToUse = c.clusterInfo.CephVersion + } else { + logger.Debugf("current cluster version for osds before upgrading is: %+v", currentCephVersion) + cephVersionToUse = currentCephVersion + } + if err = updateDeploymentAndWait(c.context, dp, c.Namespace, daemon, strconv.Itoa(osd.ID), cephVersionToUse); err != nil { + config.addError(fmt.Sprintf("failed to update osd deployment %d. %+v", osd.ID, err)) + } + } + logger.Infof("started deployment for osd %d (dir=%t, type=%s)", osd.ID, osd.IsDirectory, storeConfig.StoreType) + } +} + func (c *Cluster) startOSDDaemonsOnNode(nodeName string, config *provisionConfig, configMap *v1.ConfigMap, status *OrchestrationStatus) { osds := status.OSDs @@ -302,14 +443,23 @@ func (c *Cluster) startOSDDaemonsOnNode(nodeName string, config *provisionConfig config.addError("node %s did not resolve to start osds", nodeName) return } - storeConfig := osdconfig.ToStoreConfig(n.Config) metadataDevice := osdconfig.MetadataDevice(n.Config) + osdProps := osdProperties{ + crushHostname: n.Name, + devices: n.Devices, + selection: n.Selection, + resources: n.Resources, + storeConfig: storeConfig, + metadataDevice: metadataDevice, + location: n.Location, + } + // start osds for _, osd := range osds { logger.Debugf("start osd %v", osd) - dp, err := c.makeDeployment(n.Name, n.Selection, n.Resources, storeConfig, metadataDevice, n.Location, osd) + dp, err := c.makeDeployment(osdProps, osd) if err != nil { errMsg := fmt.Sprintf("failed to create deployment for node %s: %v", n.Name, err) config.addError(errMsg) @@ -349,6 +499,7 @@ func (c *Cluster) handleRemovedNodes(config *provisionConfig) { removedNodes, err := c.findRemovedNodes() if err != nil { config.addError("failed to find removed nodes: %+v", err) + return } logger.Infof("processing %d removed nodes", len(removedNodes)) @@ -387,7 +538,7 @@ func (c *Cluster) handleRemovedNodes(config *provisionConfig) { } } - if err := c.updateNodeStatus(removedNode, OrchestrationStatus{Status: OrchestrationStatusCompleted}); err != nil { + if err := c.updateOSDStatus(removedNode, OrchestrationStatus{Status: OrchestrationStatusCompleted}); err != nil { config.addError("failed to set orchestration starting status for removed node %s: %+v", removedNode, err) } @@ -403,20 +554,26 @@ func (c *Cluster) handleRemovedNodes(config *provisionConfig) { func (c *Cluster) cleanupRemovedNode(config *provisionConfig, nodeName, crushName string) { // update the orchestration status of this removed node to the starting state - if err := c.updateNodeStatus(nodeName, OrchestrationStatus{Status: OrchestrationStatusStarting}); err != nil { + if err := c.updateOSDStatus(nodeName, OrchestrationStatus{Status: OrchestrationStatusStarting}); err != nil { config.addError("failed to set orchestration starting status for removed node %s: %+v", nodeName, err) return } // trigger orchestration on the removed node by telling it not to use any storage at all. note that the directories are still passed in // so that the pod will be able to mount them and migrate data from them. - job, err := c.makeJob(nodeName, []rookalpha.Device{}, rookalpha.Selection{DeviceFilter: "none"}, - v1.ResourceRequirements{}, osdconfig.StoreConfig{}, "", "") + osdProps := osdProperties{ + crushHostname: nodeName, + devices: []rookalpha.Device{}, + selection: rookalpha.Selection{DeviceFilter: "none"}, + resources: v1.ResourceRequirements{}, + storeConfig: osdconfig.StoreConfig{}, + } + job, err := c.makeJob(osdProps) if err != nil { message := fmt.Sprintf("failed to create prepare job node %s: %v", nodeName, err) config.addError(message) status := OrchestrationStatus{Status: OrchestrationStatusCompleted, Message: message} - if err := c.updateNodeStatus(nodeName, status); err != nil { + if err := c.updateOSDStatus(nodeName, status); err != nil { config.addError("failed to update node %s status. %+v", nodeName, err) } return @@ -424,7 +581,7 @@ func (c *Cluster) cleanupRemovedNode(config *provisionConfig, nodeName, crushNam if !c.runJob(job, nodeName, config, "remove") { status := OrchestrationStatus{Status: OrchestrationStatusCompleted, Message: fmt.Sprintf("failed to cleanup osd config on node %s", nodeName)} - if err := c.updateNodeStatus(nodeName, status); err != nil { + if err := c.updateOSDStatus(nodeName, status); err != nil { config.addError("failed to update node %s status. %+v", nodeName, err) } return @@ -458,7 +615,8 @@ func (c *Cluster) discoverStorageNodes() (map[string][]*apps.Deployment, error) // get the node name from the node selector nodeName, ok := osdPodSpec.NodeSelector[v1.LabelHostname] if !ok || nodeName == "" { - return nil, fmt.Errorf("osd deployment %s doesn't have a node name on its node selector: %+v", osdDeployment.Name, osdPodSpec.NodeSelector) + logger.Debugf("skipping osd %s because osd deployment %s doesn't have a node name on its node selector: %+v", osdDeployment.Name, osdDeployment.Name, osdPodSpec.NodeSelector) + continue } if _, ok := discoveredNodes[nodeName]; !ok { diff --git a/pkg/operator/ceph/cluster/osd/osd_test.go b/pkg/operator/ceph/cluster/osd/osd_test.go index 2023031aa..3eb0a1f96 100644 --- a/pkg/operator/ceph/cluster/osd/osd_test.go +++ b/pkg/operator/ceph/cluster/osd/osd_test.go @@ -197,7 +197,7 @@ func TestAddRemoveNode(t *testing.T) { return `{"stats":{"total_bytes":0,"total_used_bytes":0,"total_avail_bytes":3072}}`, nil } if args[0] == "pg" && args[1] == "dump" { - return `[]`, nil + return `{}`, nil } if args[0] == "auth" && args[1] == "del" { assert.Equal(t, "osd.1", args[2]) @@ -257,17 +257,29 @@ func TestDiscoverOSDs(t *testing.T) { node2 := "n2" osd1 := OSDInfo{ID: 0, IsDirectory: true, IsFileStore: true, DataPath: "/rook/path"} - d1, err := c.makeDeployment(node1, rookalpha.Selection{}, v1.ResourceRequirements{}, config.StoreConfig{}, "", "", osd1) + osdProp := osdProperties{ + crushHostname: node1, + selection: rookalpha.Selection{}, + resources: v1.ResourceRequirements{}, + storeConfig: config.StoreConfig{}, + } + d1, err := c.makeDeployment(osdProp, osd1) assert.Nil(t, err) assert.NotNil(t, d1) osd2 := OSDInfo{ID: 101, IsDirectory: true, IsFileStore: true, DataPath: "/rook/path"} - d2, err := c.makeDeployment(node1, rookalpha.Selection{}, v1.ResourceRequirements{}, config.StoreConfig{}, "", "", osd2) + d2, err := c.makeDeployment(osdProp, osd2) assert.Nil(t, err) assert.NotNil(t, d2) + osdProp2 := osdProperties{ + crushHostname: node2, + selection: rookalpha.Selection{}, + resources: v1.ResourceRequirements{}, + storeConfig: config.StoreConfig{}, + } osd3 := OSDInfo{ID: 23, IsDirectory: true, IsFileStore: true, DataPath: "/rook/path"} - d3, err := c.makeDeployment(node2, rookalpha.Selection{}, v1.ResourceRequirements{}, config.StoreConfig{}, "", "", osd3) + d3, err := c.makeDeployment(osdProp2, osd3) assert.Nil(t, err) assert.NotNil(t, d3) diff --git a/pkg/operator/ceph/cluster/osd/spec.go b/pkg/operator/ceph/cluster/osd/spec.go index 109f28e45..49776b90a 100644 --- a/pkg/operator/ceph/cluster/osd/spec.go +++ b/pkg/operator/ceph/cluster/osd/spec.go @@ -47,23 +47,35 @@ const ( osdsPerDeviceEnvVarName = "ROOK_OSDS_PER_DEVICE" encryptedDeviceEnvVarName = "ROOK_ENCRYPTED_DEVICE" osdMetadataDeviceEnvVarName = "ROOK_METADATA_DEVICE" + pvcBackedOSDVarName = "ROOK_PVC_BACKED_OSD" rookBinariesMountPath = "/rook" rookBinariesVolumeName = "rook-binaries" + blockPVCMapperInitContainer = "blkdevmapper" osdMemoryTargetSafetyFactor float32 = 0.8 + CephDeviceSetLabelKey = "ceph.rook.io/DeviceSet" + CephSetIndexLabelKey = "ceph.rook.io/setIndex" + CephDeviceSetPVCIDLabelKey = "ceph.rook.io/DeviceSetPVCId" + OSDOverPVCLabelKey = "ceph.rook.io/pvc" ) -func (c *Cluster) makeJob(nodeName string, devices []rookalpha.Device, - selection rookalpha.Selection, resources v1.ResourceRequirements, storeConfig config.StoreConfig, metadataDevice, location string) (*batch.Job, error) { +func (c *Cluster) makeJob(osdProps osdProperties) (*batch.Job, error) { - podSpec, err := c.provisionPodTemplateSpec(devices, selection, resources, storeConfig, metadataDevice, nodeName, location, v1.RestartPolicyOnFailure) + podSpec, err := c.provisionPodTemplateSpec(osdProps, v1.RestartPolicyOnFailure) if err != nil { return nil, err } - podSpec.Spec.NodeSelector = map[string]string{v1.LabelHostname: nodeName} + + if osdProps.pvc.ClaimName == "" { + podSpec.Spec.NodeSelector = map[string]string{v1.LabelHostname: osdProps.crushHostname} + } else { + podSpec.Spec.InitContainers = []v1.Container{ + c.getPVCInitContainer(osdProps.pvc), + } + } job := &batch.Job{ ObjectMeta: metav1.ObjectMeta{ - Name: k8sutil.TruncateNodeName(prepareAppNameFmt, nodeName), + Name: k8sutil.TruncateNodeName(prepareAppNameFmt, osdProps.crushHostname), Namespace: c.Namespace, Labels: map[string]string{ k8sutil.AppAttr: prepareAppName, @@ -74,14 +86,18 @@ func (c *Cluster) makeJob(nodeName string, devices []rookalpha.Device, Template: *podSpec, }, } + + if len(osdProps.pvc.ClaimName) > 0 { + k8sutil.AddLabelToJob(OSDOverPVCLabelKey, osdProps.pvc.ClaimName, job) + } + k8sutil.AddRookVersionLabelToJob(job) opspec.AddCephVersionLabelToJob(c.clusterInfo.CephVersion, job) k8sutil.SetOwnerRef(&job.ObjectMeta, &c.ownerRef) return job, nil } -func (c *Cluster) makeDeployment(nodeName string, selection rookalpha.Selection, resources v1.ResourceRequirements, - storeConfig config.StoreConfig, metadataDevice, location string, osd OSDInfo) (*apps.Deployment, error) { +func (c *Cluster) makeDeployment(osdProps osdProperties, osd OSDInfo) (*apps.Deployment, error) { replicaCount := int32(1) volumeMounts := opspec.CephVolumeMounts() @@ -113,6 +129,11 @@ func (c *Cluster) makeDeployment(nodeName string, selection rookalpha.Selection, volumeMounts = append(volumeMounts, devMount) } + if osdProps.pvc.ClaimName != "" { + // Create volume config for PVCs + volumes = append(volumes, getPVCOSDVolumes(&osdProps)...) + } + if len(volumes) == 0 { return nil, fmt.Errorf("empty volumes") } @@ -125,7 +146,7 @@ func (c *Cluster) makeDeployment(nodeName string, selection rookalpha.Selection, osdID := strconv.Itoa(osd.ID) tiniEnvVar := v1.EnvVar{Name: "TINI_SUBREAPER", Value: ""} envVars := []v1.EnvVar{ - nodeNameEnvVar(nodeName), + nodeNameEnvVar(osdProps.crushHostname), k8sutil.PodIPEnvVar(k8sutil.PrivateIPEnvVar), k8sutil.PodIPEnvVar(k8sutil.PublicIPEnvVar), tiniEnvVar, @@ -136,7 +157,7 @@ func (c *Cluster) makeDeployment(nodeName string, selection rookalpha.Selection, {Name: "ROOK_OSD_ID", Value: osdID}, {Name: "ROOK_OSD_STORE_TYPE", Value: storeType}, }...) - configEnvVars := append(c.getConfigEnvVars(storeConfig, dataDir, nodeName, location), []v1.EnvVar{ + configEnvVars := append(c.getConfigEnvVars(osdProps.storeConfig, dataDir, osdProps.crushHostname, osdProps.location), []v1.EnvVar{ tiniEnvVar, {Name: "ROOK_OSD_ID", Value: osdID}, {Name: "ROOK_CEPH_VERSION", Value: c.clusterInfo.CephVersion.CephVersionFormatted()}, @@ -249,6 +270,11 @@ func (c *Cluster) makeDeployment(nodeName string, selection rookalpha.Selection, args = commonArgs } + if osdProps.pvc.ClaimName != "" { + volumeMounts = append(volumeMounts, getPvcOSDBridgeMount(osdProps.pvc.ClaimName)) + envVars = append(envVars, pvcBackedOSDEnvVar("true")) + } + privileged := true runAsUser := int64(0) readOnlyRootFilesystem := false @@ -259,7 +285,7 @@ func (c *Cluster) makeDeployment(nodeName string, selection rookalpha.Selection, } // needed for luksOpen synchronization when devices are encrypted - hostIPC := storeConfig.EncryptedDevice + hostIPC := osdProps.storeConfig.EncryptedDevice DNSPolicy := v1.DNSClusterFirst if c.HostNetwork { @@ -297,7 +323,6 @@ func (c *Cluster) makeDeployment(nodeName string, selection rookalpha.Selection, }, }, Spec: v1.PodSpec{ - NodeSelector: map[string]string{v1.LabelHostname: nodeName}, RestartPolicy: v1.RestartPolicyAlways, ServiceAccountName: serviceAccountName, HostNetwork: c.HostNetwork, @@ -323,7 +348,7 @@ func (c *Cluster) makeDeployment(nodeName string, selection rookalpha.Selection, Image: c.cephVersion.Image, VolumeMounts: volumeMounts, Env: envVars, - Resources: resources, + Resources: osdProps.resources, SecurityContext: securityContext, Lifecycle: opspec.PodLifeCycle(osd.DataPath), }, @@ -334,6 +359,12 @@ func (c *Cluster) makeDeployment(nodeName string, selection rookalpha.Selection, Replicas: &replicaCount, }, } + if osdProps.pvc.ClaimName == "" { + deployment.Spec.Template.Spec.NodeSelector = map[string]string{v1.LabelHostname: osdProps.crushHostname} + } else { + deployment.Spec.Template.Spec.InitContainers = append(deployment.Spec.Template.Spec.InitContainers, c.getPVCInitContainer(osdProps.pvc)) + k8sutil.AddLabelToDeployement(OSDOverPVCLabelKey, osdProps.pvc.ClaimName, deployment) + } k8sutil.AddRookVersionLabelToDeployment(deployment) c.annotations.ApplyToObjectMeta(&deployment.ObjectMeta) c.annotations.ApplyToObjectMeta(&deployment.Spec.Template.ObjectMeta) @@ -361,15 +392,14 @@ func (c *Cluster) getCopyBinariesContainer() (v1.Volume, *v1.Container) { } } -func (c *Cluster) provisionPodTemplateSpec(devices []rookalpha.Device, selection rookalpha.Selection, resources v1.ResourceRequirements, - storeConfig config.StoreConfig, metadataDevice, nodeName, location string, restart v1.RestartPolicy) (*v1.PodTemplateSpec, error) { +func (c *Cluster) provisionPodTemplateSpec(osdProps osdProperties, restart v1.RestartPolicy) (*v1.PodTemplateSpec, error) { copyBinariesVolume, copyBinariesContainer := c.getCopyBinariesContainer() volumes := append(opspec.PodVolumes(c.dataDirHostPath, c.Namespace), copyBinariesVolume) // by default, don't define any volume config unless it is required - if len(devices) > 0 || selection.DeviceFilter != "" || selection.GetUseAllDevices() || metadataDevice != "" { + if len(osdProps.devices) > 0 || osdProps.selection.DeviceFilter != "" || osdProps.selection.GetUseAllDevices() || osdProps.metadataDevice != "" { // create volume config for the data dir and /dev so the pod can access devices on the host devVolume := v1.Volume{Name: "devices", VolumeSource: v1.VolumeSource{HostPath: &v1.HostPathVolumeSource{Path: "/dev"}}} volumes = append(volumes, devVolume) @@ -377,8 +407,13 @@ func (c *Cluster) provisionPodTemplateSpec(devices []rookalpha.Device, selection volumes = append(volumes, udevVolume) } + if osdProps.pvc.ClaimName != "" { + // Create volume config for PVCs + volumes = append(volumes, getPVCOSDVolumes(&osdProps)...) + } + // add each OSD directory as another host path volume source - for _, d := range selection.Directories { + for _, d := range osdProps.selection.Directories { if c.skipVolumeForDirectory(d.Path) { // the dataDirHostPath has already been added as a volume continue @@ -398,7 +433,7 @@ func (c *Cluster) provisionPodTemplateSpec(devices []rookalpha.Device, selection ServiceAccountName: serviceAccountName, Containers: []v1.Container{ *copyBinariesContainer, - c.provisionOSDContainer(devices, selection, resources, storeConfig, metadataDevice, nodeName, location, copyBinariesContainer.VolumeMounts[0]), + c.provisionOSDContainer(osdProps, copyBinariesContainer.VolumeMounts[0]), }, RestartPolicy: restart, Volumes: volumes, @@ -422,7 +457,7 @@ func (c *Cluster) provisionPodTemplateSpec(devices []rookalpha.Device, selection // ceph-volume --dmcrypt uses cryptsetup that synchronizes with udev on // host through semaphore - podSpec.HostIPC = storeConfig.EncryptedDevice + podSpec.HostIPC = osdProps.storeConfig.EncryptedDevice return &v1.PodTemplateSpec{ ObjectMeta: podMeta, @@ -430,6 +465,30 @@ func (c *Cluster) provisionPodTemplateSpec(devices []rookalpha.Device, selection }, nil } +// Currently we can't mount a block mode pv directly to a priviliged container +// So we mount it to a non priviliged init container and then copy it to a common directory mounted inside init container +// and the privileged provision container. +func (c *Cluster) getPVCInitContainer(pvc v1.PersistentVolumeClaimVolumeSource) v1.Container { + return v1.Container{ + Name: blockPVCMapperInitContainer, + Image: c.cephVersion.Image, + Args: []string{"cp", "-a", fmt.Sprintf("/%s", pvc.ClaimName), fmt.Sprintf("/mnt/%s", pvc.ClaimName)}, + VolumeDevices: []v1.VolumeDevice{ + { + Name: pvc.ClaimName, + DevicePath: fmt.Sprintf("/%s", pvc.ClaimName), + }, + }, + VolumeMounts: []v1.VolumeMount{ + { + MountPath: "/mnt", + Name: fmt.Sprintf("%s-bridge", pvc.ClaimName), + }, + }, + SecurityContext: opmon.PodSecurityContext(), + } +} + func (c *Cluster) getConfigEnvVars(storeConfig config.StoreConfig, dataDir, nodeName, location string) []v1.EnvVar { envVars := []v1.EnvVar{ nodeNameEnvVar(nodeName), @@ -481,41 +540,40 @@ func (c *Cluster) getConfigEnvVars(storeConfig config.StoreConfig, dataDir, node return envVars } -func (c *Cluster) provisionOSDContainer(devices []rookalpha.Device, selection rookalpha.Selection, resources v1.ResourceRequirements, - storeConfig config.StoreConfig, metadataDevice, nodeName, location string, copyBinariesMount v1.VolumeMount) v1.Container { +func (c *Cluster) provisionOSDContainer(osdProps osdProperties, copyBinariesMount v1.VolumeMount) v1.Container { - envVars := c.getConfigEnvVars(storeConfig, k8sutil.DataDir, nodeName, location) + envVars := c.getConfigEnvVars(osdProps.storeConfig, k8sutil.DataDir, osdProps.crushHostname, osdProps.location) devMountNeeded := false privileged := false // only 1 of device list, device filter and use all devices can be specified. We prioritize in that order. - if len(devices) > 0 { - deviceNames := make([]string, len(devices)) - for i, device := range devices { + if len(osdProps.devices) > 0 { + deviceNames := make([]string, len(osdProps.devices)) + for i, device := range osdProps.devices { devSuffix := "" count := "1" if count, ok := device.Config[config.OSDsPerDeviceKey]; ok { - logger.Infof("%s osds requested on device %s (node %s)", count, device.Name, nodeName) + logger.Infof("%s osds requested on device %s (node %s)", count, device.Name, osdProps.crushHostname) } devSuffix += ":" + count if md, ok := device.Config[config.MetadataDeviceKey]; ok { - logger.Infof("osd %s requested with metadataDevice %s (node %s)", device.Name, md, nodeName) + logger.Infof("osd %s requested with metadataDevice %s (node %s)", device.Name, md, osdProps.crushHostname) devSuffix += ":" + md } deviceNames[i] = device.Name + devSuffix } envVars = append(envVars, dataDevicesEnvVar(strings.Join(deviceNames, ","))) devMountNeeded = true - } else if selection.DeviceFilter != "" { - envVars = append(envVars, deviceFilterEnvVar(selection.DeviceFilter)) + } else if osdProps.selection.DeviceFilter != "" { + envVars = append(envVars, deviceFilterEnvVar(osdProps.selection.DeviceFilter)) devMountNeeded = true - } else if selection.GetUseAllDevices() { + } else if osdProps.selection.GetUseAllDevices() { envVars = append(envVars, deviceFilterEnvVar("all")) devMountNeeded = true } - if metadataDevice != "" { - envVars = append(envVars, metadataDeviceEnvVar(metadataDevice)) + if osdProps.metadataDevice != "" { + envVars = append(envVars, metadataDeviceEnvVar(osdProps.metadataDevice)) devMountNeeded = true } @@ -527,11 +585,17 @@ func (c *Cluster) provisionOSDContainer(devices []rookalpha.Device, selection ro volumeMounts = append(volumeMounts, udevMount) } - if len(selection.Directories) > 0 { + if osdProps.pvc.ClaimName != "" { + volumeMounts = append(volumeMounts, getPvcOSDBridgeMount(osdProps.pvc.ClaimName)) + envVars = append(envVars, dataDevicesEnvVar(strings.Join([]string{fmt.Sprintf("/mnt/%s", osdProps.pvc.ClaimName)}, ","))) + envVars = append(envVars, pvcBackedOSDEnvVar("true")) + } + + if len(osdProps.selection.Directories) > 0 { // for each directory the user has specified, create a volume mount and pass it to the pod via cmd line arg - dirPaths := make([]string, len(selection.Directories)) - for i := range selection.Directories { - dpath := selection.Directories[i].Path + dirPaths := make([]string, len(osdProps.selection.Directories)) + for i := range osdProps.selection.Directories { + dpath := osdProps.selection.Directories[i].Path dirPaths[i] = dpath if c.skipVolumeForDirectory(dpath) { // the dataDirHostPath has already been added as a volume mount @@ -540,20 +604,20 @@ func (c *Cluster) provisionOSDContainer(devices []rookalpha.Device, selection ro volumeMounts = append(volumeMounts, v1.VolumeMount{Name: k8sutil.PathToVolumeName(dpath), MountPath: dpath}) } - if !IsRemovingNode(selection.DeviceFilter) { + if !IsRemovingNode(osdProps.selection.DeviceFilter) { envVars = append(envVars, dataDirectoriesEnvVar(strings.Join(dirPaths, ","))) } } // elevate to be privileged if it is going to mount devices or if running in a restricted environment such as openshift - if devMountNeeded || os.Getenv("ROOK_HOSTPATH_REQUIRES_PRIVILEGED") == "true" { + if devMountNeeded || os.Getenv("ROOK_HOSTPATH_REQUIRES_PRIVILEGED") == "true" || osdProps.pvc.ClaimName != "" { privileged = true } runAsUser := int64(0) runAsNonRoot := false readOnlyRootFilesystem := false - return v1.Container{ + osdProvisionContainer := v1.Container{ Command: []string{path.Join(rookBinariesMountPath, "tini")}, Args: []string{"--", path.Join(rookBinariesMountPath, "rook"), "ceph", "osd", "provision"}, Name: "provision", @@ -566,8 +630,14 @@ func (c *Cluster) provisionOSDContainer(devices []rookalpha.Device, selection ro RunAsNonRoot: &runAsNonRoot, ReadOnlyRootFilesystem: &readOnlyRootFilesystem, }, - Resources: resources, + Resources: osdProps.resources, } + + return osdProvisionContainer +} + +func getPvcOSDBridgeMount(claimName string) v1.VolumeMount { + return v1.VolumeMount{Name: fmt.Sprintf("%s-bridge", claimName), MountPath: "/mnt"} } func (c *Cluster) skipVolumeForDirectory(path string) bool { @@ -576,6 +646,28 @@ func (c *Cluster) skipVolumeForDirectory(path string) bool { return path == k8sutil.DataDir } +func getPVCOSDVolumes(osdProps *osdProperties) []v1.Volume { + return []v1.Volume{ + { + Name: osdProps.crushHostname, + VolumeSource: v1.VolumeSource{ + PersistentVolumeClaim: &osdProps.pvc, + }, + }, + { + // We need a bridge mount which is basically a common volume mount between the non priviliged init container + // and the privileged provision container or osd daemon container + // The reason for this is mentioned in the comment for getPVCInitContainer() method + Name: fmt.Sprintf("%s-bridge", osdProps.pvc.ClaimName), + VolumeSource: v1.VolumeSource{ + EmptyDir: &v1.EmptyDirVolumeSource{ + Medium: "Memory", + }, + }, + }, + } +} + func nodeNameEnvVar(name string) v1.EnvVar { return v1.EnvVar{Name: "ROOK_NODE_NAME", Value: name} } @@ -596,6 +688,10 @@ func dataDirectoriesEnvVar(dataDirectories string) v1.EnvVar { return v1.EnvVar{Name: dataDirsEnvVarName, Value: dataDirectories} } +func pvcBackedOSDEnvVar(pvcBacked string) v1.EnvVar { + return v1.EnvVar{Name: pvcBackedOSDVarName, Value: pvcBacked} +} + func getDirectoriesFromContainer(osdContainer v1.Container) []rookalpha.Directory { var dirsArg string for _, envVar := range osdContainer.Env { @@ -650,3 +746,16 @@ func osdOnSDNFlag(hostnetwork bool, v cephver.CephVersion) []string { return args } + +func makeStorageClassDeviceSetPVCID(storageClassDeviceSetName string, setIndex, pvcIndex int) (pvcId, pvcLabelSelector string) { + pvcStorageClassDeviceSetPVCId := fmt.Sprintf("%s-%v", storageClassDeviceSetName, setIndex) + return pvcStorageClassDeviceSetPVCId, fmt.Sprintf("%s=%s", CephDeviceSetPVCIDLabelKey, pvcStorageClassDeviceSetPVCId) +} + +func makeStorageClassDeviceSetPVCLabel(storageClassDeviceSetName, pvcStorageClassDeviceSetPVCId string, pvcIndex, setIndex int) map[string]string { + return map[string]string{ + CephDeviceSetLabelKey: storageClassDeviceSetName, + CephSetIndexLabelKey: fmt.Sprintf("%v", setIndex), + CephDeviceSetPVCIDLabelKey: pvcStorageClassDeviceSetPVCId, + } +} diff --git a/pkg/operator/ceph/cluster/osd/spec_test.go b/pkg/operator/ceph/cluster/osd/spec_test.go index cb6248f62..ae4361ec9 100644 --- a/pkg/operator/ceph/cluster/osd/spec_test.go +++ b/pkg/operator/ceph/cluster/osd/spec_test.go @@ -28,17 +28,24 @@ import ( cephver "github.com/rook/rook/pkg/operator/ceph/version" "github.com/rook/rook/pkg/operator/k8sutil" exectest "github.com/rook/rook/pkg/util/exec/test" - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" v1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/client-go/kubernetes/fake" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" ) func TestPodContainer(t *testing.T) { cluster := &Cluster{Namespace: "myosd", rookVersion: "23", cephVersion: cephv1.CephVersionSpec{}} - c, err := cluster.provisionPodTemplateSpec([]rookalpha.Device{}, rookalpha.Selection{}, v1.ResourceRequirements{}, config.StoreConfig{}, "", "node", "", v1.RestartPolicyAlways) + osdProps := osdProperties{ + crushHostname: "node", + devices: []rookalpha.Device{}, + resources: v1.ResourceRequirements{}, + storeConfig: config.StoreConfig{}, + } + c, err := cluster.provisionPodTemplateSpec(osdProps, v1.RestartPolicyAlways) assert.NotNil(t, c) assert.Nil(t, err) assert.Equal(t, 2, len(c.Spec.Containers)) @@ -88,7 +95,14 @@ func testPodDevices(t *testing.T, dataDir, deviceName string, allDevices bool) { ID: 0, } - deployment, err := c.makeDeployment(n.Name, n.Selection, v1.ResourceRequirements{}, config.StoreConfig{}, "", n.Location, osd) + osdProp := osdProperties{ + crushHostname: n.Name, + selection: n.Selection, + resources: v1.ResourceRequirements{}, + storeConfig: config.StoreConfig{}, + location: n.Location, + } + deployment, err := c.makeDeployment(osdProp, osd) assert.Nil(t, err) assert.NotNil(t, deployment) assert.Equal(t, "rook-ceph-osd-0", deployment.Name) @@ -170,7 +184,16 @@ func TestStorageSpecDevicesAndDirectories(t *testing.T) { IsDirectory: true, DataPath: "/my/root/path/osd1", } - deployment, err := c.makeDeployment(n.Name, n.Selection, v1.ResourceRequirements{}, config.StoreConfig{}, "", n.Location, osd) + + osdProp := osdProperties{ + crushHostname: n.Name, + selection: n.Selection, + resources: v1.ResourceRequirements{}, + storeConfig: config.StoreConfig{}, + location: n.Location, + } + + deployment, err := c.makeDeployment(osdProp, osd) assert.NotNil(t, deployment) assert.Nil(t, err) // pod spec should have a volume for the given dir in the main container and the init container @@ -199,7 +222,7 @@ func TestStorageSpecDevicesAndDirectories(t *testing.T) { IsDirectory: true, DataPath: "/var/lib/rook/osd1", } - deployment, err = c.makeDeployment(n.Name, n.Selection, v1.ResourceRequirements{}, config.StoreConfig{}, "", n.Location, osd) + deployment, err = c.makeDeployment(osdProp, osd) assert.NotNil(t, deployment) assert.Nil(t, err) // pod spec should have a volume for the given dir in the main container and the init container @@ -263,7 +286,16 @@ func TestStorageSpecConfig(t *testing.T) { storeConfig := config.ToStoreConfig(storageSpec.Nodes[0].Config) metadataDevice := config.MetadataDevice(storageSpec.Nodes[0].Config) - job, err := c.makeJob(n.Name, n.Devices, n.Selection, c.DesiredStorage.Nodes[0].Resources, storeConfig, metadataDevice, n.Location) + osdProp := osdProperties{ + crushHostname: n.Name, + devices: n.Devices, + selection: n.Selection, + resources: c.DesiredStorage.Nodes[0].Resources, + storeConfig: storeConfig, + metadataDevice: metadataDevice, + location: n.Location, + } + job, err := c.makeJob(osdProp) assert.NotNil(t, job) assert.Nil(t, err) assert.Equal(t, "rook-ceph-osd-prepare-node1", job.ObjectMeta.Name) @@ -317,7 +349,17 @@ func TestHostNetwork(t *testing.T) { osd := OSDInfo{ ID: 0, } - r, err := c.makeDeployment(n.Name, n.Selection, v1.ResourceRequirements{}, config.StoreConfig{}, "", n.Location, osd) + + osdProp := osdProperties{ + crushHostname: n.Name, + devices: n.Devices, + selection: n.Selection, + resources: c.DesiredStorage.Nodes[0].Resources, + storeConfig: config.StoreConfig{}, + location: n.Location, + } + + r, err := c.makeDeployment(osdProp, osd) assert.NotNil(t, r) assert.Nil(t, err) diff --git a/pkg/operator/ceph/cluster/osd/status.go b/pkg/operator/ceph/cluster/osd/status.go index 336792164..8ecff21e7 100644 --- a/pkg/operator/ceph/cluster/osd/status.go +++ b/pkg/operator/ceph/cluster/osd/status.go @@ -58,7 +58,7 @@ func (c *provisionConfig) addError(message string, args ...interface{}) { c.errorMessages = append(c.errorMessages, fmt.Sprintf(message, args...)) } -func (c *Cluster) updateNodeStatus(node string, status OrchestrationStatus) error { +func (c *Cluster) updateOSDStatus(node string, status OrchestrationStatus) error { return UpdateNodeStatus(c.kv, node, status) } @@ -85,7 +85,7 @@ func UpdateNodeStatus(kv *k8sutil.ConfigMapKVStore, node string, status Orchestr func (c *Cluster) handleOrchestrationFailure(config *provisionConfig, nodeName, message string) { config.addError(message) status := OrchestrationStatus{Status: OrchestrationStatusFailed, Message: message} - if err := c.updateNodeStatus(nodeName, status); err != nil { + if err := c.updateOSDStatus(nodeName, status); err != nil { config.addError("failed to update status for node %s. %+v", nodeName, err) } } @@ -253,10 +253,15 @@ func (c *Cluster) handleStatusConfigMapStatus(nodeName string, config *provision logger.Infof("osd orchestration status for node %s is %s", nodeName, status.Status) if status.Status == OrchestrationStatusCompleted { if configOSDs { - c.startOSDDaemonsOnNode(nodeName, config, configMap, status) + if status.PvcBackedOSD { + c.startOSDDaemonsOnPVC(nodeName, config, configMap, status) + } else { + c.startOSDDaemonsOnNode(nodeName, config, configMap, status) + } + // remove the status configmap that indicated the progress + c.kv.ClearStore(fmt.Sprintf(orchestrationStatusMapName, nodeName)) } - // remove the status configmap that indicated the progress - c.kv.ClearStore(fmt.Sprintf(orchestrationStatusMapName, nodeName)) + return true } diff --git a/pkg/operator/discover/discover_test.go b/pkg/operator/discover/discover_test.go index 20b963384..9c7df596a 100644 --- a/pkg/operator/discover/discover_test.go +++ b/pkg/operator/discover/discover_test.go @@ -29,7 +29,7 @@ import ( "github.com/stretchr/testify/assert" - "k8s.io/api/core/v1" + v1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ) @@ -69,7 +69,7 @@ func TestStartDiscoveryDaemonset(t *testing.T) { func TestGetAvailableDevices(t *testing.T) { clientset := test.New(3) - + pvcBackedOSD := false ns := "rook-system" nodeName := "node123" os.Setenv(k8sutil.PodNamespaceEnvVar, ns) @@ -109,11 +109,11 @@ func TestGetAvailableDevices(t *testing.T) { assert.Nil(t, err) assert.Equal(t, 1, len(nodeDevices)) - devices, err := GetAvailableDevices(context, nodeName, ns, d, "^sd.", false) + devices, err := GetAvailableDevices(context, nodeName, ns, d, "^sd.", pvcBackedOSD) assert.Nil(t, err) assert.Equal(t, 1, len(devices)) // devices should be in use now, 2nd try gets the same list - devices, err = GetAvailableDevices(context, nodeName, ns, d, "^sd.", false) + devices, err = GetAvailableDevices(context, nodeName, ns, d, "^sd.", pvcBackedOSD) assert.Nil(t, err) assert.Equal(t, 1, len(devices)) } diff --git a/pkg/operator/k8sutil/deployment.go b/pkg/operator/k8sutil/deployment.go index 832edae5a..a35596823 100644 --- a/pkg/operator/k8sutil/deployment.go +++ b/pkg/operator/k8sutil/deployment.go @@ -24,6 +24,7 @@ import ( "github.com/rook/rook/pkg/util" apps "k8s.io/api/apps/v1" v1 "k8s.io/api/apps/v1" + batchv1 "k8s.io/api/batch/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/client-go/kubernetes" ) @@ -188,3 +189,27 @@ func AddRookVersionLabelToObjectMeta(meta *metav1.ObjectMeta) { } addRookVersionLabel(meta.Labels) } + +func AddLabelToDeployement(key, value string, d *v1.Deployment) { + if d == nil { + return + } + if d.Labels == nil { + d.Labels = map[string]string{} + } + addLabel(key, value, d.Labels) +} + +func AddLabelToJob(key, value string, b *batchv1.Job) { + if b == nil { + return + } + if b.Labels == nil { + b.Labels = map[string]string{} + } + addLabel(key, value, b.Labels) +} + +func addLabel(key, value string, labels map[string]string) { + labels[key] = value +} diff --git a/pkg/operator/k8sutil/node.go b/pkg/operator/k8sutil/node.go index 1102be478..08cfa73f6 100644 --- a/pkg/operator/k8sutil/node.go +++ b/pkg/operator/k8sutil/node.go @@ -88,6 +88,10 @@ func GetValidNodes(rookStorage rookalpha.StorageScopeSpec, clientset kubernetes. return RookNodesMatchingKubernetesNodes(rookStorage, validK8sNodes) } +func generateUniqueVolumeSourceName(sourceName, pvcName string) string { + return fmt.Sprintf("%s-%s", sourceName, pvcName) +} + // GetNodeNameFromHostname returns the name of the node resource looked up by the hostname label // Typically these will be the same name, but sometimes they are not such as when nodes have a longer // dns name, but the hostname is short. diff --git a/pkg/operator/k8sutil/volume.go b/pkg/operator/k8sutil/volume.go index 255bc6590..8ebda3f08 100644 --- a/pkg/operator/k8sutil/volume.go +++ b/pkg/operator/k8sutil/volume.go @@ -22,7 +22,7 @@ import ( "os" "strings" - "k8s.io/api/core/v1" + v1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/util/validation" ) diff --git a/pkg/util/sys/device.go b/pkg/util/sys/device.go index 0c9de3355..0e97f10cc 100644 --- a/pkg/util/sys/device.go +++ b/pkg/util/sys/device.go @@ -92,9 +92,19 @@ func ListDevices(executor exec.Executor) ([]string, error) { } func GetDevicePartitions(device string, executor exec.Executor) (partitions []Partition, unusedSpace uint64, err error) { - cmd := fmt.Sprintf("lsblk /dev/%s", device) - output, err := executor.ExecuteCommandWithOutput(false, cmd, "lsblk", fmt.Sprintf("/dev/%s", device), + + var devicePath string + splitDevicePath := strings.Split(device, "/") + if len(splitDevicePath) == 1 { + devicePath = fmt.Sprintf("/dev/%s", device) //device path for OSD on devices. + } else { + devicePath = device //use the exact device path (like /mnt/) in case of PVC block device + } + + cmd := fmt.Sprintf("lsblk %s", devicePath) + output, err := executor.ExecuteCommandWithOutput(false, cmd, "lsblk", devicePath, "--bytes", "--pairs", "--output", "NAME,SIZE,TYPE,PKNAME") + logger.Infof("Output: %+v", output) if err != nil { return nil, 0, fmt.Errorf("failed to get device %s partitions. %+v", device, err) } @@ -106,6 +116,7 @@ func GetDevicePartitions(device string, executor exec.Executor) (partitions []Pa name := props["NAME"] if name == device { // found the main device + logger.Info("Device found - ", name) deviceSize, err = strconv.ParseUint(props["SIZE"], 10, 64) if err != nil { return nil, 0, fmt.Errorf("failed to get device %s size. %+v", device, err) @@ -144,7 +155,14 @@ func GetDevicePartitions(device string, executor exec.Executor) (partitions []Pa } func GetDeviceProperties(device string, executor exec.Executor) (map[string]string, error) { - return GetDevicePropertiesFromPath(fmt.Sprintf("/dev/%s", device), executor) + // As we are mounting the block mode PVs on /mnt we use the entire path, + // e.g., if the device path is /mnt/example-pvc then its taken completely + // else if its just vdb then the following is used + devicePath := strings.Split(device, "/") + if len(devicePath) == 1 { + device = fmt.Sprintf("/dev/%s", device) + } + return GetDevicePropertiesFromPath(device, executor) } func GetDevicePropertiesFromPath(devicePath string, executor exec.Executor) (map[string]string, error) { @@ -178,8 +196,12 @@ func GetUdevInfo(device string, executor exec.Executor) (map[string]string, erro // get the file systems available func GetDeviceFilesystems(device string, executor exec.Executor) (string, error) { + devicePath := strings.Split(device, "/") + if len(devicePath) == 1 { + device = fmt.Sprintf("/dev/%s", device) + } cmd := fmt.Sprintf("get filesystem type for %s", device) - output, err := executor.ExecuteCommandWithOutput(false, cmd, "udevadm", "info", "--query=property", fmt.Sprintf("/dev/%s", device)) + output, err := executor.ExecuteCommandWithOutput(false, cmd, "udevadm", "info", "--query=property", device) if err != nil { return "", fmt.Errorf("command %s failed: %+v", cmd, err) } @@ -224,9 +246,15 @@ func GetDiskUUID(device string, executor exec.Executor) (string, error) { return "sgdiskNotFound", nil } + devicePath := strings.Split(device, "/") + if len(devicePath) == 1 { + device = fmt.Sprintf("/dev/%s", device) + } + cmd := fmt.Sprintf("get disk %s uuid", device) + output, err := executor.ExecuteCommandWithOutput(false, cmd, - sgdisk, "--print", fmt.Sprintf("/dev/%s", device)) + sgdisk, "--print", device) if err != nil { return "", err } @@ -295,7 +323,7 @@ func UnmountDevice(devicePath string, executor exec.Executor) error { // the number of partitions, whether Rook has created partitions on the device in the past // possibly from the same or a previous cluster, the filesystem found, or an err if failed // to retrieve the properties. -func CheckIfDeviceAvailable(executor exec.Executor, name string) (int, bool, string, error) { +func CheckIfDeviceAvailable(executor exec.Executor, name string, pvcBacked bool) (int, bool, string, error) { ownPartitions := true partitions, _, err := GetDevicePartitions(name, executor) if err != nil { @@ -306,12 +334,16 @@ func CheckIfDeviceAvailable(executor exec.Executor, name string) (int, bool, str ownPartitions = false } - // check if there is a file system on the device - devFS, err := GetDeviceFilesystems(name, executor) - if err != nil { - return 0, false, "", fmt.Errorf("failed to get device %s filesystem: %+v", name, err) + var devFS string + if !pvcBacked { + // check if there is a file system on the device + devFS, err = GetDeviceFilesystems(name, executor) + if err != nil { + return 0, false, "", fmt.Errorf("failed to get device %s filesystem: %+v", name, err) + } + } else { + devFS = "" //Not checking Filesystem in case of PVC block device. } - return partCount, ownPartitions, devFS, nil } diff --git a/pkg/util/sys/device_test.go b/pkg/util/sys/device_test.go index 9c1102a2e..0526a3fd7 100644 --- a/pkg/util/sys/device_test.go +++ b/pkg/util/sys/device_test.go @@ -144,7 +144,7 @@ func TestGetPartitions(t *testing.T) { executor := &exectest.MockExecutor{ MockExecuteCommandWithOutput: func(debug bool, actionName string, command string, arg ...string) (string, error) { run++ - logger.Infof("run %d action %s", run, actionName) + logger.Infof("run %d action %s command %s", run, actionName, command) switch { case run == 1: return `NAME="sdc" SIZE="100000" TYPE="disk" PKNAME=""`, nil diff --git a/tests/framework/installer/device.go b/tests/framework/installer/device.go index 1ca6f772e..6be822f68 100644 --- a/tests/framework/installer/device.go +++ b/tests/framework/installer/device.go @@ -40,7 +40,8 @@ func IsAdditionalDeviceAvailableOnCluster() bool { continue } - _, ownPartitions, fs, err := sys.CheckIfDeviceAvailable(executor, device) + pvcBackedOSD := false + _, ownPartitions, fs, err := sys.CheckIfDeviceAvailable(executor, device, pvcBackedOSD) if err != nil { logger.Warningf("failed to detect device %s availability. %+v", device, err) continue