From d2f52aebe56fe16bf2bac29c8e9f3d47ae05cfd8 Mon Sep 17 00:00:00 2001 From: rohan47 Date: Tue, 2 Jul 2019 10:42:58 +0530 Subject: [PATCH] Adds support for storageClassDeviceSet in rook-ceph operator MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Added code to support StorageClassDeviceSet spec provided in the cluster-on-pvc.yaml - The code reads the StorageClassDeviceSet spec and creates pvc based on the ‘count’ field for each device set. - OSD prepare job is started for each PVC which activates the ceph-volume on each PVC - Finally OSD is started on each of the PVC device. Co-authored-by: rohan47 Co-authored-by: Ashish Ranjan Signed-off-by: Santosh Pillai --- cluster/examples/kubernetes/ceph/cluster.yaml | 8 +- cmd/rook/ceph/ceph.go | 1 + cmd/rook/ceph/osd.go | 12 +- pkg/apis/rook.io/v1alpha2/types.go | 21 +- .../rook.io/v1alpha2/zz_generated.deepcopy.go | 72 +++++ pkg/clusterd/disk.go | 177 +++++------ pkg/daemon/ceph/osd/agent.go | 19 +- pkg/daemon/ceph/osd/agent_test.go | 13 +- pkg/daemon/ceph/osd/daemon.go | 37 ++- pkg/daemon/ceph/osd/daemon_test.go | 15 +- pkg/daemon/ceph/osd/device.go | 6 +- pkg/daemon/ceph/osd/device_test.go | 2 +- pkg/daemon/ceph/osd/volume.go | 104 ++++++- pkg/daemon/ceph/osd/volume_test.go | 6 +- pkg/daemon/discover/discover.go | 3 +- pkg/daemon/discover/discover_test.go | 4 +- pkg/operator/ceph/cluster/osd/deviceSet.go | 129 ++++++++ pkg/operator/ceph/cluster/osd/osd.go | 276 ++++++++++++++---- pkg/operator/ceph/cluster/osd/osd_test.go | 20 +- pkg/operator/ceph/cluster/osd/spec.go | 187 +++++++++--- pkg/operator/ceph/cluster/osd/spec_test.go | 58 +++- pkg/operator/ceph/cluster/osd/status.go | 15 +- pkg/operator/discover/discover_test.go | 8 +- pkg/operator/k8sutil/deployment.go | 25 ++ pkg/operator/k8sutil/node.go | 4 + pkg/operator/k8sutil/volume.go | 2 +- pkg/util/sys/device.go | 54 +++- pkg/util/sys/device_test.go | 2 +- tests/framework/installer/device.go | 3 +- 29 files changed, 1020 insertions(+), 263 deletions(-) create mode 100644 pkg/operator/ceph/cluster/osd/deviceSet.go diff --git a/cluster/examples/kubernetes/ceph/cluster.yaml b/cluster/examples/kubernetes/ceph/cluster.yaml index 8c9c02789..8989fae0b 100644 --- a/cluster/examples/kubernetes/ceph/cluster.yaml +++ b/cluster/examples/kubernetes/ceph/cluster.yaml @@ -99,8 +99,12 @@ spec: # mon: # osd: storage: # cluster level storage configuration and selection - useAllNodes: true - useAllDevices: true + persistentVolumeClaim: + - claimName: "example-pvc" + - claimName: "example-pvc1" + - claimName: "example-pvc4" + useAllNodes: false + useAllDevices: false deviceFilter: location: config: 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