From 82425aafcc96b6b5e2673fe82dff86a157f806bd Mon Sep 17 00:00:00 2001 From: Huamin Chen Date: Tue, 1 May 2018 09:01:29 -0400 Subject: [PATCH] prepare osd in a job per node. Once all osds are prepared, store osd info in orchestration configmap. Operator watches the configmap, starts one osd replica set per osd. Signed-off-by: Huamin Chen --- .gitignore | 2 + Jenkinsfile | 2 + .../rook-ceph/templates/clusterrole.yaml | 11 + .../examples/kubernetes/ceph/operator.yaml | 11 + cmd/rook/ceph/osd.go | 12 +- pkg/clusterd/disk.go | 8 +- pkg/daemon/ceph/osd/agent.go | 132 +++++--- pkg/daemon/ceph/osd/agent_test.go | 6 +- pkg/daemon/ceph/osd/daemon.go | 48 ++- pkg/daemon/ceph/osd/daemon_test.go | 10 +- pkg/daemon/ceph/osd/device.go | 13 +- pkg/daemon/ceph/osd/device_test.go | 29 +- pkg/daemon/ceph/osd/health_test.go | 2 +- pkg/daemon/discover/discover.go | 16 +- pkg/operator/ceph/cluster/controller.go | 3 + pkg/operator/ceph/cluster/osd/osd.go | 283 ++++++++++++------ pkg/operator/ceph/cluster/osd/osd_test.go | 60 +++- pkg/operator/ceph/cluster/osd/pod.go | 183 ++++++++--- pkg/operator/ceph/cluster/osd/pod_test.go | 134 ++++----- pkg/operator/discover/discover.go | 177 ++++++++++- pkg/operator/discover/discover_test.go | 23 ++ pkg/util/sys/device.go | 14 +- pkg/util/sys/device_test.go | 17 +- tests/framework/installer/install_data.go | 40 +++ tests/framework/installer/install_helper.go | 45 ++- tests/integration/base_deploy_test.go | 12 + .../integration/block_createBlock_k8s_test.go | 2 +- tests/integration/helm_test.go | 4 +- tests/integration/smoke_test.go | 2 +- .../integration/zblock_mount_unmount_test.go | 3 +- tests/scripts/minikube.sh | 1 + 31 files changed, 949 insertions(+), 356 deletions(-) diff --git a/.gitignore b/.gitignore index 05a1df911..1d7e67947 100644 --- a/.gitignore +++ b/.gitignore @@ -6,3 +6,5 @@ /_output /vendor/ /vendor.orig/ + +/tests/integration/rook-test/ \ No newline at end of file diff --git a/Jenkinsfile b/Jenkinsfile index c7b80566a..1fc1fed0e 100644 --- a/Jenkinsfile +++ b/Jenkinsfile @@ -166,6 +166,8 @@ def RunIntegrationTest(k, v) { } finally{ archive '_output/tests/*.log' + sh 'sudo rm -rf ${PWD}/rook-test' + sh 'sudo ls -l ${PWD}' deleteDir() } } diff --git a/cluster/charts/rook-ceph/templates/clusterrole.yaml b/cluster/charts/rook-ceph/templates/clusterrole.yaml index b13e1214a..cb494acec 100644 --- a/cluster/charts/rook-ceph/templates/clusterrole.yaml +++ b/cluster/charts/rook-ceph/templates/clusterrole.yaml @@ -81,6 +81,17 @@ rules: - get - list - watch +- apiGroups: + - batch + resources: + - jobs + verbs: + - get + - list + - watch + - create + - update + - delete - apiGroups: - ceph.rook.io resources: diff --git a/cluster/examples/kubernetes/ceph/operator.yaml b/cluster/examples/kubernetes/ceph/operator.yaml index 5f6c12fab..e167783a9 100644 --- a/cluster/examples/kubernetes/ceph/operator.yaml +++ b/cluster/examples/kubernetes/ceph/operator.yaml @@ -200,6 +200,17 @@ rules: - get - list - watch +- apiGroups: + - batch + resources: + - jobs + verbs: + - get + - list + - watch + - create + - update + - delete - apiGroups: - ceph.rook.io resources: diff --git a/cmd/rook/ceph/osd.go b/cmd/rook/ceph/osd.go index ca8510532..1fa148dd9 100644 --- a/cmd/rook/ceph/osd.go +++ b/cmd/rook/ceph/osd.go @@ -18,7 +18,9 @@ package ceph import ( "fmt" "os" + "os/signal" "strings" + "syscall" "github.com/rook/rook/cmd/rook/rook" "github.com/rook/rook/pkg/daemon/ceph/client" @@ -40,6 +42,7 @@ var osdCmd = &cobra.Command{ var ( osdDataDeviceFilter string ownerRefID string + prepareOnly bool ) func addOSDFlags(command *cobra.Command) { @@ -58,6 +61,9 @@ func addOSDFlags(command *cobra.Command) { command.Flags().IntVar(&cfg.storeConfig.DatabaseSizeMB, "osd-database-size", osdcfg.DBDefaultSizeMB, "default size (MB) for OSD database (bluestore)") command.Flags().IntVar(&cfg.storeConfig.JournalSizeMB, "osd-journal-size", osdcfg.JournalDefaultSizeMB, "default size (MB) for OSD journal (filestore)") command.Flags().StringVar(&cfg.storeConfig.StoreType, "osd-store", "", "type of backing OSD store to use (bluestore or filestore)") + + // only prepare devices but not start ceph-osd daemon + command.Flags().BoolVar(&prepareOnly, "osd-prepare-only", true, "true to only prepare ceph osd directories or devices but not start ceph-osd daemon") } func init() { @@ -115,9 +121,11 @@ func startOSD(cmd *cobra.Command, args []string) error { ownerRef := cluster.ClusterOwnerRef(clusterInfo.Name, ownerRefID) kv := k8sutil.NewConfigMapKVStore(clusterInfo.Name, clientset, ownerRef) agent := osd.NewAgent(context, dataDevices, usingDeviceFilter, cfg.metadataDevice, cfg.directories, forceFormat, - crushLocation, cfg.storeConfig, &clusterInfo, cfg.nodeName, kv) + crushLocation, cfg.storeConfig, &clusterInfo, cfg.nodeName, kv, prepareOnly) - err = osd.Run(context, agent, nil) + sigc := make(chan os.Signal, 1) + signal.Notify(sigc, syscall.SIGTERM) + err = osd.Run(context, agent, sigc) if err != nil { // something failed in the OSD orchestration, update the status map with failure details status := oposd.OrchestrationStatus{ diff --git a/pkg/clusterd/disk.go b/pkg/clusterd/disk.go index 62c2a6837..96c476887 100644 --- a/pkg/clusterd/disk.go +++ b/pkg/clusterd/disk.go @@ -35,7 +35,7 @@ func GetAvailableDevices(devices []*sys.LocalDisk) []string { var available []string for _, device := range devices { logger.Debugf("Evaluating device %+v", device) - if getDeviceEmpty(device) { + if GetDeviceEmpty(device) { logger.Debugf("Available device: %s", device.Name) available = append(available, device.Name) } @@ -45,8 +45,8 @@ func GetAvailableDevices(devices []*sys.LocalDisk) []string { } // check whether a device is completely empty -func getDeviceEmpty(device *sys.LocalDisk) bool { - return device.Parent == "" && (device.Type == sys.DiskType || device.Type == sys.SSDType || device.Type == sys.CryptType) && device.Filesystem == "" +func GetDeviceEmpty(device *sys.LocalDisk) bool { + return device.Parent == "" && (device.Type == sys.DiskType || device.Type == sys.SSDType || device.Type == sys.CryptType || device.Type == sys.LVMType) && len(device.Partitions) == 0 && device.Filesystem == "" } func ignoreDevice(d string) bool { @@ -121,8 +121,6 @@ func DiscoverDevices(executor exec.Executor) ([]*sys.LocalDisk, error) { disk.Parent = val } - disk.Empty = getDeviceEmpty(disk) - // parse udev info output if val, ok := udevInfo["DEVLINKS"]; ok { disk.DevLinks = val diff --git a/pkg/daemon/ceph/osd/agent.go b/pkg/daemon/ceph/osd/agent.go index 0ede555e1..83e864520 100644 --- a/pkg/daemon/ceph/osd/agent.go +++ b/pkg/daemon/ceph/osd/agent.go @@ -30,6 +30,7 @@ import ( "github.com/rook/rook/pkg/clusterd" "github.com/rook/rook/pkg/daemon/ceph/client" "github.com/rook/rook/pkg/daemon/ceph/mon" + oposd "github.com/rook/rook/pkg/operator/ceph/cluster/osd" "github.com/rook/rook/pkg/operator/ceph/cluster/osd/config" "github.com/rook/rook/pkg/operator/k8sutil" "github.com/rook/rook/pkg/util" @@ -58,21 +59,33 @@ type OsdAgent struct { kv *k8sutil.ConfigMapKVStore configCounter int32 osdsCompleted chan struct{} + prepareOnly bool } func NewAgent(context *clusterd.Context, devices string, usingDeviceFilter bool, metadataDevice, directories string, forceFormat bool, - location string, storeConfig config.StoreConfig, cluster *mon.ClusterInfo, nodeName string, kv *k8sutil.ConfigMapKVStore) *OsdAgent { + location string, storeConfig config.StoreConfig, cluster *mon.ClusterInfo, nodeName string, kv *k8sutil.ConfigMapKVStore, prepareOnly bool) *OsdAgent { - return &OsdAgent{devices: devices, usingDeviceFilter: usingDeviceFilter, metadataDevice: metadataDevice, - directories: directories, forceFormat: forceFormat, location: location, storeConfig: storeConfig, - cluster: cluster, nodeName: nodeName, kv: kv, - procMan: proc.New(context.Executor), osdProc: make(map[int]*proc.MonitoredProc), + return &OsdAgent{ + devices: devices, + usingDeviceFilter: usingDeviceFilter, + metadataDevice: metadataDevice, + directories: directories, + forceFormat: forceFormat, + location: location, + storeConfig: storeConfig, + cluster: cluster, + nodeName: nodeName, + kv: kv, + procMan: proc.New(context.Executor), + osdProc: make(map[int]*proc.MonitoredProc), + prepareOnly: prepareOnly, } } -func (a *OsdAgent) configureDirs(context *clusterd.Context, dirs map[string]int) error { +func (a *OsdAgent) configureDirs(context *clusterd.Context, dirs map[string]int) ([]oposd.OSDInfo, error) { + var osds []oposd.OSDInfo if len(dirs) == 0 { - return nil + return osds, nil } succeeded := 0 @@ -85,7 +98,7 @@ func (a *OsdAgent) configureDirs(context *clusterd.Context, dirs map[string]int) // the osd hasn't been registered with ceph yet, do so now to give it a cluster wide ID osdID, osdUUID, err := registerOSD(context, a.cluster.Name) if err != nil { - return err + return osds, err } dirs[dirPath] = *osdID @@ -93,17 +106,18 @@ func (a *OsdAgent) configureDirs(context *clusterd.Context, dirs map[string]int) config.uuid = *osdUUID } - err := a.startOSD(context, config) + osd, err := a.startOSD(context, config) if err != nil { logger.Errorf("failed to config osd in path %s. %+v", dirPath, err) lastErr = err } else { succeeded++ + osds = append(osds, *osd) } } logger.Infof("%d/%d osd dirs succeeded on this node", succeeded, len(dirs)) - return lastErr + return osds, lastErr } func (a *OsdAgent) removeDirs(context *clusterd.Context, removedDirs map[string]int) error { @@ -134,22 +148,23 @@ func (a *OsdAgent) removeDirs(context *clusterd.Context, removedDirs map[string] return nil } -func (a *OsdAgent) configureDevices(context *clusterd.Context, devices *DeviceOsdMapping) error { +func (a *OsdAgent) configureDevices(context *clusterd.Context, devices *DeviceOsdMapping) ([]oposd.OSDInfo, error) { + var osds []oposd.OSDInfo if devices == nil || len(devices.Entries) == 0 { - return nil + return osds, nil } // compute an OSD layout scheme that will optimize performance scheme, err := a.getPartitionPerfScheme(context, devices) logger.Debugf("partition scheme: %+v, err: %+v", scheme, err) if err != nil { - return fmt.Errorf("failed to get OSD partition scheme: %+v", err) + return osds, fmt.Errorf("failed to get OSD partition scheme: %+v", err) } if scheme.Metadata != nil { // partition the dedicated metadata device if err := partitionMetadata(context, scheme.Metadata, a.kv, config.GetConfigStoreName(a.nodeName)); err != nil { - return fmt.Errorf("failed to partition metadata %+v: %+v", scheme.Metadata, err) + return osds, fmt.Errorf("failed to partition metadata %+v: %+v", scheme.Metadata, err) } } @@ -158,16 +173,17 @@ func (a *OsdAgent) configureDevices(context *clusterd.Context, devices *DeviceOs for _, entry := range scheme.Entries { config := &osdConfig{id: entry.ID, uuid: entry.OsdUUID, configRoot: context.ConfigDir, partitionScheme: entry, storeConfig: a.storeConfig, kv: a.kv, storeName: config.GetConfigStoreName(a.nodeName)} - err := a.startOSD(context, config) + osd, err := a.startOSD(context, config) if err != nil { - return fmt.Errorf("failed to config osd %d. %+v", entry.ID, err) + return osds, fmt.Errorf("failed to config osd %d. %+v", entry.ID, err) } else { succeeded++ + osds = append(osds, *osd) } } logger.Infof("%d/%d osd devices succeeded on this node", succeeded, len(scheme.Entries)) - return nil + return osds, nil } func (a *OsdAgent) removeDevices(context *clusterd.Context, removedDevicesScheme *config.PerfScheme) error { @@ -370,20 +386,20 @@ func refreshDeviceInfo(name string, nameToUUID map[string]string, scheme *config } } -func (a *OsdAgent) startOSD(context *clusterd.Context, cfg *osdConfig) error { +func (a *OsdAgent) startOSD(context *clusterd.Context, cfg *osdConfig) (*oposd.OSDInfo, error) { cfg.rootPath = getOSDRootDir(cfg.configRoot, cfg.id) // if the osd is using filestore on a device and it's previously been formatted/partitioned, // go ahead and remount the device now. if err := remountFilestoreDeviceIfNeeded(context, cfg); err != nil { - return err + return nil, err } // prepare the osd root dir, which will tell us if it's a new osd newOSD, err := prepareOSDRoot(cfg) if err != nil { - return err + return nil, err } if newOSD { @@ -391,7 +407,7 @@ func (a *OsdAgent) startOSD(context *clusterd.Context, cfg *osdConfig) error { // format and partition the device if needed savedScheme, err := config.LoadScheme(a.kv, config.GetConfigStoreName(a.nodeName)) if err != nil { - return fmt.Errorf("failed to load the saved partition scheme from %s: %+v", cfg.configRoot, err) + return nil, fmt.Errorf("failed to load the saved partition scheme from %s: %+v", cfg.configRoot, err) } skipFormat := false @@ -406,7 +422,7 @@ func (a *OsdAgent) startOSD(context *clusterd.Context, cfg *osdConfig) error { if !skipFormat { err = formatDevice(context, cfg, a.forceFormat, a.storeConfig) if err != nil { - return fmt.Errorf("failed format/partition of osd %d. %+v", cfg.id, err) + return nil, fmt.Errorf("failed format/partition of osd %d. %+v", cfg.id, err) } logger.Notice("waiting after partition/format...") @@ -415,13 +431,13 @@ func (a *OsdAgent) startOSD(context *clusterd.Context, cfg *osdConfig) error { } // osd_data_dir/ready does not exist yet, create/initialize the OSD - err := initializeOSD(cfg, context, a.cluster, a.location) + err := initializeOSD(cfg, context, a.cluster, a.location, a.prepareOnly) if err != nil { - return fmt.Errorf("failed to initialize OSD at %s: %+v", cfg.rootPath, err) + return nil, fmt.Errorf("failed to initialize OSD at %s: %+v", cfg.rootPath, err) } } else { // update the osd config file - err := writeConfigFile(cfg, context, a.cluster, a.location) + err := writeConfigFile(cfg, context, a.cluster, a.location, a.prepareOnly) if err != nil { logger.Warningf("failed to update config file. %+v", err) } @@ -430,17 +446,21 @@ func (a *OsdAgent) startOSD(context *clusterd.Context, cfg *osdConfig) error { // look up some basic information about it so we can run it. err = loadOSDInfo(cfg) if err != nil { - return fmt.Errorf("failed to get OSD information from %s: %+v", cfg.rootPath, err) + return nil, fmt.Errorf("failed to get OSD information from %s: %+v", cfg.rootPath, err) } } - + osdInfo := getOSDInfo(a.cluster.Name, cfg) + if a.prepareOnly { + logger.Infof("done with preparing osd %v", osdInfo) + return osdInfo, nil + } // run the OSD in a child process now that it is fully initialized and ready to go - err = a.runOSD(context, a.cluster.Name, cfg) + err = a.runOSD(osdInfo) if err != nil { - return fmt.Errorf("failed to run osd %d: %+v", cfg.id, err) + return osdInfo, fmt.Errorf("failed to run osd %d: %+v", cfg.id, err) } - return nil + return osdInfo, nil } func prepareOSDRoot(cfg *osdConfig) (newOSD bool, err error) { @@ -464,41 +484,57 @@ func prepareOSDRoot(cfg *osdConfig) (newOSD bool, err error) { return newOSD, nil } -// runs an OSD with the given config in a child process -func (a *OsdAgent) runOSD(context *clusterd.Context, clusterName string, config *osdConfig) error { - // start the OSD daemon in the foreground with the given config - logger.Infof("starting osd %d at %s", config.id, config.rootPath) - +func getOSDInfo(clusterName string, config *osdConfig) *oposd.OSDInfo { confFile := getOSDConfFilePath(config.rootPath, clusterName) util.WriteFileToLog(logger, confFile) - - osdUUIDArg := fmt.Sprintf("--osd-uuid=%s", config.uuid.String()) - params := []string{"--foreground", - fmt.Sprintf("--id=%d", config.id), - fmt.Sprintf("--cluster=%s", clusterName), - fmt.Sprintf("--osd-data=%s", config.rootPath), - fmt.Sprintf("--conf=%s", confFile), - fmt.Sprintf("--keyring=%s", getOSDKeyringPath(config.rootPath)), - osdUUIDArg, + osd := &oposd.OSDInfo{ + ID: config.id, + DataPath: config.rootPath, + Config: confFile, + Cluster: clusterName, + KeyringPath: getOSDKeyringPath(config.rootPath), + UUID: config.uuid.String(), + IsFileStore: isFilestore(config), } if isFilestore(config) { - params = append(params, fmt.Sprintf("--osd-journal=%s", getOSDJournalPath(config.rootPath))) + osd.Journal = getOSDJournalPath(config.rootPath) + } + return osd +} + +// runs an OSD with the given config in a child process +func (a *OsdAgent) runOSD(osdInfo *oposd.OSDInfo) error { + // start the OSD daemon in the foreground with the given config + logger.Infof("starting osd %s at %s", osdInfo.ID, osdInfo.DataPath) + + osdUUIDArg := fmt.Sprintf("--osd-uuid=%s", osdInfo.UUID) + params := []string{"--foreground", + fmt.Sprintf("--id=%d", osdInfo.ID), + fmt.Sprintf("--cluster=%s", osdInfo.Cluster), + fmt.Sprintf("--osd-data=%s", osdInfo.DataPath), + fmt.Sprintf("--conf=%s", osdInfo.Config), + fmt.Sprintf("--keyring=%s", osdInfo.KeyringPath), + osdUUIDArg, + } + + if osdInfo.IsFileStore { + params = append(params, fmt.Sprintf("--osd-journal=%s", osdInfo.Journal)) } process, err := a.procMan.Start( - fmt.Sprintf("osd%d", config.id), + fmt.Sprintf("osd%d", osdInfo.ID), "ceph-osd", regexp.QuoteMeta(osdUUIDArg), proc.ReuseExisting, params...) if err != nil { - return fmt.Errorf("failed to start osd %d: %+v", config.id, err) + return fmt.Errorf("failed to start osd %d: %+v", osdInfo.ID, err) } if process != nil { // if the process was already running Start will return nil in which case we don't want to overwrite it - a.osdProc[config.id] = process + a.osdProc[osdInfo.ID] = process } return nil diff --git a/pkg/daemon/ceph/osd/agent_test.go b/pkg/daemon/ceph/osd/agent_test.go index 577fe54ff..94b47513d 100644 --- a/pkg/daemon/ceph/osd/agent_test.go +++ b/pkg/daemon/ceph/osd/agent_test.go @@ -226,7 +226,7 @@ func testOSDAgentWithDevicesHelper(t *testing.T, storeConfig config.StoreConfig) "sdx": {Data: -1}, "sdy": {Data: -1}, }} - err = agent.configureDevices(context, devices) + _, err = agent.configureDevices(context, devices) assert.Nil(t, err) assert.Equal(t, int32(0), agent.configCounter) @@ -293,7 +293,7 @@ func TestOSDAgentNoDevices(t *testing.T) { filepath.Join(configDir, "sdx"): -1, filepath.Join(configDir, "sdy"): -1, } - err = agent.configureDirs(context, dirs) + _, err = agent.configureDirs(context, dirs) assert.Nil(t, err) assert.Equal(t, 2, runCount) assert.Equal(t, 2, startCount) @@ -353,7 +353,7 @@ func createTestAgent(t *testing.T, devices, configDir, nodeName string, storeCon cluster := &mon.ClusterInfo{Name: "myclust"} context := &clusterd.Context{ConfigDir: configDir, Executor: executor, Clientset: testop.New(1)} agent := NewAgent(context, devices, false, "", "", forceFormat, location, *storeConfig, - cluster, nodeName, mockKVStore()) + cluster, nodeName, mockKVStore(), false /* prepareOnly */) return agent, executor, context } diff --git a/pkg/daemon/ceph/osd/daemon.go b/pkg/daemon/ceph/osd/daemon.go index 8b42878d7..f6fd10e12 100644 --- a/pkg/daemon/ceph/osd/daemon.go +++ b/pkg/daemon/ceph/osd/daemon.go @@ -17,9 +17,9 @@ package osd import ( "fmt" + "os" "path" "regexp" - "time" "strings" @@ -34,9 +34,12 @@ import ( "k8s.io/apimachinery/pkg/api/errors" ) -var logger = capnslog.NewPackageLogger("github.com/rook/rook", "cephosd") +var ( + logger = capnslog.NewPackageLogger("github.com/rook/rook", "cephosd") + mountPoints = []string{} +) -func Run(context *clusterd.Context, agent *OsdAgent, done chan struct{}) error { +func Run(context *clusterd.Context, agent *OsdAgent, done chan os.Signal) error { // set the initial orchestration status status := oposd.OrchestrationStatus{Status: oposd.OrchestrationStatusComputingDiff} @@ -47,7 +50,11 @@ func Run(context *clusterd.Context, agent *OsdAgent, done chan struct{}) error { // set the crush location in the osd config file cephConfig := mon.CreateDefaultCephConfig(context, agent.cluster, path.Join(context.ConfigDir, agent.cluster.Name)) cephConfig.GlobalConfig.CrushLocation = agent.location - + // don't set public or cluster addr if prepare only + if agent.prepareOnly { + cephConfig.PublicAddr = "" + cephConfig.ClusterAddr = "" + } // write the latest config to the config dir if err := mon.GenerateAdminConnectionConfigWithSettings(context, agent.cluster, cephConfig); err != nil { return fmt.Errorf("failed to write connection config. %+v", err) @@ -97,27 +104,29 @@ func Run(context *clusterd.Context, agent *OsdAgent, done chan struct{}) error { // start the desired OSDs on devices logger.Infof("configuring osd devices: %+v", devices) - if err := agent.configureDevices(context, devices); err != nil { + deviceOSDs, err := agent.configureDevices(context, devices) + if err != nil { return fmt.Errorf("failed to configure devices. %+v", err) } // also start OSDs for the devices that will be removed. In order to remove devices, we need the // OSDs to first be running so they can participate in the rebalancing logger.Infof("configuring removed osd devices: %+v", removedDevicesMapping) - if err := agent.configureDevices(context, removedDevicesMapping); err != nil { + if _, err := agent.configureDevices(context, removedDevicesMapping); err != nil { // some devices that will be removed may be legitimately dead, let's try to remove them even if they can't start up logger.Warningf("failed to configure removed devices, but proceeding with removal attempts. %+v", err) } // start up the OSDs for directories logger.Infof("configuring osd dirs: %+v", dirs) - if err := agent.configureDirs(context, dirs); err != nil { + dirOSDs, err := agent.configureDirs(context, dirs) + if err != nil { return fmt.Errorf("failed to configure dirs %v. %+v", dirs, err) } // start up the OSDs for directories that will be removed. logger.Infof("configuring removed osd dirs: %+v", removedDirs) - if err := agent.configureDirs(context, removedDirs); err != nil { + if _, err := agent.configureDirs(context, removedDirs); err != nil { // some dirs that will be removed may be legitimately dead, let's try to remove them even if they can't start up logger.Warningf("failed to configure removed dirs, but proceeding with removal attempts. %+v", err) } @@ -143,26 +152,37 @@ func Run(context *clusterd.Context, agent *OsdAgent, done chan struct{}) error { logger.Warningf("failed to clean up node resources, they may need to be cleaned up manually: %+v", err) } } - + logger.Infof("device osds:%v\ndir osds: %v", deviceOSDs, dirOSDs) + osds := append(deviceOSDs, dirOSDs...) // orchestration is completed, update the status - status = oposd.OrchestrationStatus{Status: oposd.OrchestrationStatusCompleted} + status = oposd.OrchestrationStatus{OSDs: osds, Status: oposd.OrchestrationStatusCompleted} if err := oposd.UpdateOrchestrationStatusMap(context.Clientset, agent.cluster.Name, agent.nodeName, status); err != nil { return err } + if len(mountPoints) == 0 { + // no device mounted, exit now + return nil + } + + // if there are devices mounted for filestore, wait and unmount them before exit // OSD processes monitoring mon := NewMonitor(context, agent) go mon.Run() - // FIX - logger.Infof("sleeping a while to let the osds run...") select { - case <-time.After(1000000 * time.Second): - logger.Warning("OSD sleep has expired") case <-done: logger.Infof("done channel signaled") } + // umount devices + for m := range mountPoints { + logger.Infof("unmount %s", mountPoints[m]) + if err = sys.UnmountDevice(mountPoints[m], context.Executor); err != nil { + logger.Warningf("failed to unmount %s", mountPoints[m]) + } + + } return nil } diff --git a/pkg/daemon/ceph/osd/daemon_test.go b/pkg/daemon/ceph/osd/daemon_test.go index 74aa33f83..fa1eba01b 100644 --- a/pkg/daemon/ceph/osd/daemon_test.go +++ b/pkg/daemon/ceph/osd/daemon_test.go @@ -19,7 +19,9 @@ import ( "fmt" "io/ioutil" "os" + "os/signal" "strings" + "syscall" "testing" "github.com/rook/rook/pkg/clusterd" @@ -69,13 +71,13 @@ func TestRunDaemon(t *testing.T) { agent, _, context := createTestAgent(t, "none", configDir, "node5375", &config.StoreConfig{StoreType: config.Bluestore}) agent.usingDeviceFilter = true - - done := make(chan struct{}) + sigc := make(chan os.Signal, 1) + signal.Notify(sigc, syscall.SIGTERM) go func() { - done <- struct{}{} + sigc <- syscall.SIGTERM }() - err := Run(context, agent, done) + err := Run(context, agent, sigc) assert.Nil(t, err) } diff --git a/pkg/daemon/ceph/osd/device.go b/pkg/daemon/ceph/osd/device.go index 37763bc96..687094fd6 100644 --- a/pkg/daemon/ceph/osd/device.go +++ b/pkg/daemon/ceph/osd/device.go @@ -257,6 +257,7 @@ func prepareFilestoreDevice(context *clusterd.Context, cfg *osdConfig, doFormat if err = sys.MountDevice(dataPartPath, cfg.rootPath, context.Executor); err != nil { return fmt.Errorf("failed to mount %s at %s: %+v", dataPartPath, cfg.rootPath, context.Executor) } + mountPoints = append(mountPoints, cfg.rootPath) return nil } @@ -425,7 +426,7 @@ func getStoreSettings(cfg *osdConfig) (map[string]string, error) { return settings, nil } -func writeConfigFile(cfg *osdConfig, context *clusterd.Context, cluster *mon.ClusterInfo, location string) error { +func writeConfigFile(cfg *osdConfig, context *clusterd.Context, cluster *mon.ClusterInfo, location string, prepareOnly bool) error { cephConfig := mon.CreateDefaultCephConfig(context, cluster, cfg.rootPath) if isBluestore(cfg) { cephConfig.GlobalConfig.OsdObjectStore = config.Bluestore @@ -433,6 +434,11 @@ func writeConfigFile(cfg *osdConfig, context *clusterd.Context, cluster *mon.Clu cephConfig.GlobalConfig.OsdObjectStore = config.Filestore } cephConfig.CrushLocation = location + // don't set public or cluster addr if prepare only + if prepareOnly { + cephConfig.PublicAddr = "" + cephConfig.ClusterAddr = "" + } if cfg.dir || isFilestoreDevice(cfg) { // using the local file system requires some config overrides @@ -457,9 +463,8 @@ func writeConfigFile(cfg *osdConfig, context *clusterd.Context, cluster *mon.Clu return nil } -func initializeOSD(config *osdConfig, context *clusterd.Context, cluster *mon.ClusterInfo, location string) error { - - err := writeConfigFile(config, context, cluster, location) +func initializeOSD(config *osdConfig, context *clusterd.Context, cluster *mon.ClusterInfo, location string, prepareOnly bool) error { + err := writeConfigFile(config, context, cluster, location, prepareOnly) if err != nil { return fmt.Errorf("failed to write config file: %+v", err) } diff --git a/pkg/daemon/ceph/osd/device_test.go b/pkg/daemon/ceph/osd/device_test.go index 1d8dd7a60..342c06163 100644 --- a/pkg/daemon/ceph/osd/device_test.go +++ b/pkg/daemon/ceph/osd/device_test.go @@ -100,29 +100,26 @@ func TestOverwriteRookOwnedPartitions(t *testing.T) { logger.Infof("OUTPUT %d for %s. %s %+v", outputExecCount, name, command, args) var output string switch outputExecCount { - case 0, 7: // we'll call this twice, once explicitly below to verify rook owns the partitions and a 2nd time within formatDevice + case 0: // we'll call this twice, once explicitly below to verify rook owns the partitions and a 2nd time within formatDevice assert.Equal(t, command, "lsblk") output = `NAME="sda" SIZE="65" TYPE="disk" PKNAME="" NAME="sda1" SIZE="30" TYPE="part" PKNAME="sda" NAME="sda2" SIZE="10" TYPE="part" PKNAME="sda" NAME="sda3" SIZE="20" TYPE="part" PKNAME="sda"` - case 1, 8: - assert.Equal(t, "udevadm /dev/sda1", name) + case 1: + assert.Equal(t, "udevadm info sda1", name) output = "ID_PART_ENTRY_NAME=ROOK-OSD0-WAL" - case 2, 9: - assert.Equal(t, "get filesystem type for sda1", name) - output = "" - case 3, 10: - assert.Equal(t, "udevadm /dev/sda2", name) + case 2: + assert.Equal(t, "udevadm info sda2", name) output = "ID_PART_ENTRY_NAME=ROOK-OSD0-DB" - case 4, 11: - assert.Equal(t, "get filesystem type for sda2", name) - output = "" - case 5, 12: - assert.Equal(t, "udevadm /dev/sda3", name) + case 3: + assert.Equal(t, "udevadm info sda3", name) output = "ID_PART_ENTRY_NAME=ROOK-OSD0-BLOCK" - case 6, 13: - assert.Equal(t, "get filesystem type for sda3", name) + case 4: + assert.Equal(t, "lsblk /dev/sda", name) + output = "" + case 5: + assert.Equal(t, "get filesystem type for sda", name) output = "" } outputExecCount++ @@ -153,7 +150,7 @@ NAME="sda3" SIZE="20" TYPE="part" PKNAME="sda"` // to format and the format/partitioning will happen. err = formatDevice(context, config, false, storeConfig) assert.Nil(t, err) - assert.Equal(t, 15, outputExecCount) + assert.Equal(t, 6, outputExecCount) } func TestPartitionBluestoreMetadata(t *testing.T) { diff --git a/pkg/daemon/ceph/osd/health_test.go b/pkg/daemon/ceph/osd/health_test.go index 175d54161..ebaf7bfba 100644 --- a/pkg/daemon/ceph/osd/health_test.go +++ b/pkg/daemon/ceph/osd/health_test.go @@ -73,7 +73,7 @@ func TestOSDStatus(t *testing.T) { "sdx": {Data: -1}, }} // Creating an OSD for which Down status will be returned - err = agent.configureDevices(context, devices) + _, err = agent.configureDevices(context, devices) assert.Nil(t, err) // Initializing an OSD monitoring diff --git a/pkg/daemon/discover/discover.go b/pkg/daemon/discover/discover.go index b5ca5da9e..24f4a55f3 100644 --- a/pkg/daemon/discover/discover.go +++ b/pkg/daemon/discover/discover.go @@ -53,22 +53,24 @@ func Run(context *clusterd.Context) error { nodeName = os.Getenv(k8sutil.NodeNameEnvVar) namespace = os.Getenv(k8sutil.PodNamespaceEnvVar) cmName = LocalDiskCMName + nodeName - tick := time.NewTicker(probeInterval).C sigc := make(chan os.Signal, 1) signal.Notify(sigc, syscall.SIGTERM) + err := updateDeviceCM(context) + if err != nil { + logger.Infof("failed to update device configmap: %v", err) + return err + } for { select { case <-sigc: logger.Infof("shutdown signal received, exiting...") return nil - case <-tick: - err := updateDeviceCM(context) - if err != nil { - logger.Infof("failed to update device configmap: %v", err) - } + case <-time.After(probeInterval): + updateDeviceCM(context) } } } + func updateDeviceCM(context *clusterd.Context) error { logger.Infof("updating device configmap") devices, err := probeDevices(context) @@ -156,6 +158,8 @@ func probeDevices(context *clusterd.Context) ([]sys.LocalDisk, error) { } device.Partitions = partitions device.Filesystem = fs + device.Empty = clusterd.GetDeviceEmpty(device) + devices = append(devices, *device) } diff --git a/pkg/operator/ceph/cluster/controller.go b/pkg/operator/ceph/cluster/controller.go index 3d5e9e4aa..d1c33f9a1 100644 --- a/pkg/operator/ceph/cluster/controller.go +++ b/pkg/operator/ceph/cluster/controller.go @@ -32,12 +32,14 @@ import ( "github.com/rook/rook/pkg/clusterd" "github.com/rook/rook/pkg/daemon/ceph/agent/flexvolume/attachment" "github.com/rook/rook/pkg/daemon/ceph/client" + "github.com/rook/rook/pkg/operator/ceph/cluster/mgr" "github.com/rook/rook/pkg/operator/ceph/cluster/mon" "github.com/rook/rook/pkg/operator/ceph/cluster/osd" "github.com/rook/rook/pkg/operator/ceph/file" "github.com/rook/rook/pkg/operator/ceph/object" "github.com/rook/rook/pkg/operator/ceph/pool" + "github.com/rook/rook/pkg/operator/discover" "github.com/rook/rook/pkg/operator/k8sutil" "k8s.io/api/core/v1" apiextensionsv1beta1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1" @@ -381,6 +383,7 @@ func (c *ClusterController) onDelete(obj interface{}) { if clust.Spec.Storage.AnyUseAllDevices() { c.devicesInUse = false } + discover.FreeDevicesByCluster(c.context, clust.Name) } func (c *ClusterController) handleDelete(cluster *cephv1alpha1.Cluster, retryInterval time.Duration) error { diff --git a/pkg/operator/ceph/cluster/osd/osd.go b/pkg/operator/ceph/cluster/osd/osd.go index cce64db9f..5cf73a0af 100644 --- a/pkg/operator/ceph/cluster/osd/osd.go +++ b/pkg/operator/ceph/cluster/osd/osd.go @@ -20,6 +20,7 @@ package osd import ( "encoding/json" "fmt" + "os" "strings" "time" @@ -52,7 +53,11 @@ const ( OrchestrationStatusCompleted = "completed" OrchestrationStatusFailed = "failed" appName = "rook-ceph-osd" + prepareAppName = "rook-ceph-osd-prepare" + prepareAppNameFmt = "rook-ceph-osd-prepare-%s" + osdAppNameFmt = "rook-ceph-osd-id-%d" appNameFmt = "rook-ceph-osd-%s" + osdLabelKey = "ceph-osd-id" clusterAvailableSpaceReserve = 0.05 defaultServiceAccountName = "rook-ceph-cluster" ) @@ -97,9 +102,26 @@ func New(context *clusterd.Context, namespace, version, serviceAccount string, s } } +type OSDInfo struct { + ID int `json:"id"` + DataPath string `json:"data-path"` + Config string `json:"conf"` + Cluster string `json:"cluster"` + KeyringPath string `json:"keyring-path"` + UUID string `json:"uuid"` + Journal string `json:"journal"` + IsFileStore bool `json:"is-file-store"` +} + type OrchestrationStatus struct { - Status string `json:"status"` - Message string `json:"message"` + OSDs []OSDInfo `json:"osds"` + Status string `json:"status"` + Message string `json:"message"` +} + +type deploymentPerNode struct { + node rookalpha.Node + deployments []*extensions.Deployment } // Start the osd management @@ -126,34 +148,35 @@ func (c *Cluster) Start() error { }() if c.Storage.UseAllNodes { - // make a daemonset for all nodes in the cluster - storeConfig := config.ToStoreConfig(c.Storage.Config) - metadataDevice := config.MetadataDevice(c.Storage.Config) - ds := c.makeDaemonSet(c.Storage.Selection, storeConfig, metadataDevice, c.Storage.Location) - _, err := c.context.Clientset.Extensions().DaemonSets(c.Namespace).Create(ds) + // resolve all storage nodes + c.Storage.Nodes = nil + rookSystemNS := os.Getenv(k8sutil.PodNamespaceEnvVar) + allNodeDevices, err := discover.ListDevices(c.context, rookSystemNS, "" /* all nodes */) if err != nil { - if !errors.IsAlreadyExists(err) { - return fmt.Errorf("failed to create osd daemon set. %+v", err) - } - logger.Infof("osd daemon set already exists") - } else { - logger.Infof("osd daemon set started") + logger.Warningf("failed to get storage nodes from namespace %s: %v", rookSystemNS, err) + return err } - - return nil + for nodeName := range allNodeDevices { + storageNode := rookalpha.Node{ + Name: nodeName, + } + c.Storage.Nodes = append(c.Storage.Nodes, storageNode) + } + logger.Debugf("storage nodes: %+v", c.Storage.Nodes) } // orchestrate individual nodes, starting with any that are still ongoing (in the case that we // are resuming a previous orchestration attempt) if inProgressNode, status := c.findInProgressNode(); inProgressNode != "" { logger.Infof("resuming orchestration of in progress node %s, status: %+v", inProgressNode, status) - if err := c.waitForCompletion(inProgressNode); err != nil { + if _, err := c.waitForCompletion(inProgressNode); err != nil { logger.Warningf("failed waiting for in progress node %s, will continue with orchestration. %+v", inProgressNode, err) } } errorMessages := make([]string, 0) - + clusterName := c.Namespace + devicesToUse := make(map[string][]rookalpha.Device, len(c.Storage.Nodes)) // start with nodes currently in the storage spec for i := range c.Storage.Nodes { // fully resolve the storage config and resources for this node @@ -167,41 +190,99 @@ func (c *Cluster) Start() error { errorMessages = append(errorMessages, fmt.Sprintf("failed to set orchestration starting status for node %s: %+v", n.Name, err)) continue } - devicesToUse := n.Devices - availDev, deviceErr := discover.GetAvailableDevices(c.context, n.Name, c.Namespace, n.Devices, n.Selection.DeviceFilter, n.Selection.GetUseAllDevices()) + devicesToUse[n.Name] = n.Devices + availDev, deviceErr := discover.GetAvailableDevices(c.context, n.Name, clusterName, n.Devices, n.Selection.DeviceFilter, n.Selection.GetUseAllDevices()) if deviceErr != nil { - logger.Warningf("failed to get devices for node %s cluster %s: %v", n.Name, c.Namespace, deviceErr) + logger.Warningf("failed to get devices for node %s cluster %s: %v", n.Name, clusterName, deviceErr) } else { - devicesToUse = availDev + devicesToUse[n.Name] = availDev logger.Infof("avail devices for node %s: %+v", n.Name, availDev) } - // create the replicaSet that will run the OSDs for this node - rs := c.makeReplicaSet(n.Name, devicesToUse, n.Selection, n.Resources, storeConfig, metadataDevice, n.Location) - _, err := c.context.Clientset.Extensions().ReplicaSets(c.Namespace).Create(rs) + if len(availDev) == 0 && len(c.dataDirHostPath) == 0 { + errorMessages = append(errorMessages, fmt.Sprintf("empty volumes for node %s", n.Name)) + continue + } + // create the job that prepares osds on the node + job, err := c.makeJob(n.Name, devicesToUse[n.Name], n.Selection, n.Resources, storeConfig, metadataDevice, n.Location) + if err != nil { + message := fmt.Sprintf("failed to create prepare job node %s: %v", n.Name, err) + logger.Info(message) + status := OrchestrationStatus{Status: OrchestrationStatusCompleted, Message: message} + if err := UpdateOrchestrationStatusMap(c.context.Clientset, c.Namespace, n.Name, status); err != nil { + errorMessages = append(errorMessages, message) + continue + } + } + _, err = c.context.Clientset.Batch().Jobs(c.Namespace).Create(job) if err != nil { if !errors.IsAlreadyExists(err) { - // we failed to create the replica set, update the orchestration status for this node - message := fmt.Sprintf("failed to create osd replica set for node %s. %+v", n.Name, err) + // we failed to create job, update the orchestration status for this node + message := fmt.Sprintf("failed to create osd prepare job for node %s. %+v", n.Name, err) c.handleOrchestrationFailure(*n, message, &errorMessages) + err = discover.FreeDevices(c.context, n.Name, clusterName) + if err != nil { + logger.Warningf("failed to free devices: %s", err) + } continue } else { - // TODO: if the replica set already exists, we may need to edit the pod template spec, for example if device filter has changed - message := fmt.Sprintf("osd replica set already exists for node %s", n.Name) + // TODO: if the job already exists, we may need to edit the pod template spec, for example if device filter has changed + message := fmt.Sprintf("failed to set orchestration status for node %s, status: %+v: %+v", n.Name, status, err) logger.Info(message) status := OrchestrationStatus{Status: OrchestrationStatusCompleted, Message: message} if err := UpdateOrchestrationStatusMap(c.context.Clientset, c.Namespace, n.Name, status); err != nil { - errorMessages = append(errorMessages, fmt.Sprintf("failed to set orchestration status for node %s, status: %+v: %+v", n.Name, status, err)) + errorMessages = append(errorMessages, message) continue } } } else { - logger.Infof("osd replica set started for node %s", n.Name) + logger.Infof("osd prepare job started for node %s", n.Name) } + } + for i := range c.Storage.Nodes { + // fully resolve the storage config and resources for this node + n := c.resolveNode(c.Storage.Nodes[i]) + storeConfig := config.ToStoreConfig(n.Config) + metadataDevice := config.MetadataDevice(n.Config) // wait for the current node's orchestration to be completed - if err := c.waitForCompletion(n.Name); err != nil { - errorMessages = append(errorMessages, err.Error()) + if status, err := c.waitForCompletion(n.Name); err != nil { + logger.Warningf("failed to prepare node %s: %v", n.Name, err) + err = discover.FreeDevices(c.context, n.Name, clusterName) + if err != nil { + logger.Warningf("failed to free devices: %s", err) + errorMessages = append(errorMessages, err.Error()) + } continue + } else { + // start osds + osds := status.OSDs + logger.Debugf("osds prepared on node %s: %+v", n.Name, osds) + for _, osd := range osds { + logger.Debugf("start osd %v", osd) + dp, err := c.makeDeployment(n.Name, devicesToUse[n.Name], n.Selection, n.Resources, storeConfig, metadataDevice, n.Location, osd) + if err != nil { + errMsg := fmt.Sprintf("nil deployment for node %s: %v", n.Name, err) + logger.Warningf(errMsg) + errorMessages = append(errorMessages, errMsg) + err = discover.FreeDevices(c.context, n.Name, clusterName) + if err != nil { + logger.Warningf("failed to free devices: %s", err) + } + continue + } + dp, err = c.context.Clientset.Extensions().Deployments(c.Namespace).Create(dp) + if err != nil { + if !errors.IsAlreadyExists(err) { + // we failed to create job, update the orchestration status for this node + logger.Warningf("failed to create osd deployment for node %s, osd %v: %+v", n.Name, osd, err) + err = discover.FreeDevices(c.context, n.Name, clusterName) + if err != nil { + logger.Warningf("failed to free devices: %s", err) + } + continue + } + } + } } } @@ -213,53 +294,60 @@ func (c *Cluster) Start() error { for i := range removedNodes { n := removedNodes[i] - storeConfig := config.ToStoreConfig(n.Config) - metadataDevice := config.MetadataDevice(n.Config) + storeConfig := config.ToStoreConfig(n.node.Config) + metadataDevice := config.MetadataDevice(n.node.Config) - if err := c.isSafeToRemoveNode(n); err != nil { - message := fmt.Sprintf("skipping the removal of node %s because it is not safe to do so: %+v", n.Name, err) - c.handleOrchestrationFailure(n, message, &errorMessages) + if err := c.isSafeToRemoveNode(n.node); err != nil { + message := fmt.Sprintf("skipping the removal of node %s because it is not safe to do so: %+v", n.node.Name, err) + c.handleOrchestrationFailure(n.node, message, &errorMessages) continue } - - logger.Infof("removing node %s from the cluster", n.Name) + if len(n.node.Name) == 0 { + continue + } + logger.Infof("removing node %s from the cluster", n.node.Name) // update the orchestration status of this removed node to the starting state - if err := UpdateOrchestrationStatusMap(c.context.Clientset, c.Namespace, n.Name, OrchestrationStatus{Status: OrchestrationStatusStarting}); err != nil { - errorMessages = append(errorMessages, fmt.Sprintf("failed to set orchestration starting status for removed node %s: %+v", n.Name, err)) + if err := UpdateOrchestrationStatusMap(c.context.Clientset, c.Namespace, n.node.Name, OrchestrationStatus{Status: OrchestrationStatusStarting}); err != nil { + errorMessages = append(errorMessages, fmt.Sprintf("failed to set orchestration starting status for removed node %s: %+v", n.node.Name, err)) continue } // 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. - rs := c.makeReplicaSet(n.Name, nil, rookalpha.Selection{DeviceFilter: "none", Directories: n.Directories}, - v1.ResourceRequirements{}, storeConfig, metadataDevice, n.Location) - rs, err := c.context.Clientset.Extensions().ReplicaSets(c.Namespace).Update(rs) + job, err := c.makeJob(n.node.Name, nil, rookalpha.Selection{DeviceFilter: "none", Directories: n.node.Directories}, v1.ResourceRequirements{}, storeConfig, metadataDevice, "") if err != nil { - message := fmt.Sprintf("failed to update osd replica set for removed node %s. %+v", n.Name, err) - c.handleOrchestrationFailure(n, message, &errorMessages) + message := fmt.Sprintf("failed to create osd job for removed node %s. %+v", n.node.Name, err) + c.handleOrchestrationFailure(n.node, message, &errorMessages) + continue + } + job, err = c.context.Clientset.Batch().Jobs(c.Namespace).Update(job) + if err != nil { + message := fmt.Sprintf("failed to update osd job for removed node %s. %+v", n.node.Name, err) + c.handleOrchestrationFailure(n.node, message, &errorMessages) continue } else { - logger.Infof("osd replica set updated for node %s", n.Name) + logger.Infof("osd job updated for node %s", n.node.Name) } + for _, dp := range n.deployments { + // delete the pod associated with the deployment so that it will be restarted with the new template + if err := c.deleteOSDPod(dp); err != nil { + message := fmt.Sprintf("failed to find and delete OSD pod for deployments %s. %+v", dp.Name, err) + c.handleOrchestrationFailure(n.node, message, &errorMessages) + continue + } - // delete the pod associated with the replica set so that it will be restarted with the new template - if err := c.deleteOSDPod(rs); err != nil { - message := fmt.Sprintf("failed to find and delete OSD pod for replica set %s. %+v", rs.Name, err) - c.handleOrchestrationFailure(n, message, &errorMessages) - continue - } + // wait for the removed node's orchestration to be completed + if _, err := c.waitForCompletion(n.node.Name); err != nil { + errorMessages = append(errorMessages, err.Error()) + continue + } - // wait for the removed node's orchestration to be completed - if err := c.waitForCompletion(n.Name); err != nil { - errorMessages = append(errorMessages, err.Error()) - continue - } - - // orchestration of the removed node completed, we can delete the replica set now - if err := c.context.Clientset.Extensions().ReplicaSets(c.Namespace).Delete(rs.Name, &metav1.DeleteOptions{}); err != nil { - errorMessages = append(errorMessages, fmt.Sprintf("failed to delete replica set %s: %+v", rs.Name, err)) - continue + // orchestration of the removed node completed, we can delete the deployment now + if err := c.context.Clientset.Extensions().Deployments(c.Namespace).Delete(dp.Name, &metav1.DeleteOptions{}); err != nil { + errorMessages = append(errorMessages, fmt.Sprintf("failed to delete deployment %s: %+v", dp.Name, err)) + continue + } } } @@ -393,12 +481,12 @@ func (c *Cluster) findInProgressNode() (string, *OrchestrationStatus) { return "", nil } -func (c *Cluster) waitForCompletion(node string) error { +func (c *Cluster) waitForCompletion(node string) (*OrchestrationStatus, error) { // check the status map to see if the node is already completed before we start watching cm, err := c.context.Clientset.CoreV1().ConfigMaps(c.Namespace).Get(OrchestrationStatusMapName, metav1.GetOptions{}) if err != nil { if !errors.IsNotFound(err) { - return err + return nil, err } // the status map doesn't exist yet, watching below is still an OK thing to do } else { @@ -406,27 +494,28 @@ func (c *Cluster) waitForCompletion(node string) error { status := parseOrchestrationStatus(cm.Data, node) if status != nil { if status.Status == OrchestrationStatusCompleted { - return nil + return status, nil } else if status.Status == OrchestrationStatusFailed { - return fmt.Errorf("orchestration for node %s failed: %+v", node, status) + return nil, fmt.Errorf("orchestration for node %s failed: %+v", node, status) } } } // start watching for changes on the orchestration status map - var startingVersion string + startingVersion := "0" if cm != nil { startingVersion = cm.ResourceVersion } opts := metav1.ListOptions{ FieldSelector: fields.OneTermEqualSelector(api.ObjectNameField, OrchestrationStatusMapName).String(), ResourceVersion: startingVersion, + Watch: true, } for { w, err := c.context.Clientset.CoreV1().ConfigMaps(c.Namespace).Watch(opts) if err != nil { - return fmt.Errorf("failed to start watch on %s: %+v", OrchestrationStatusMapName, err) + return nil, fmt.Errorf("failed to start watch on %s: %+v", OrchestrationStatusMapName, err) } defer w.Stop() @@ -440,7 +529,6 @@ func (c *Cluster) waitForCompletion(node string) error { <-time.After(100 * time.Millisecond) break ResultLoop } - if e.Type == watch.Modified { statusMap := e.Object.(*v1.ConfigMap) status := parseOrchestrationStatus(statusMap.Data, node) @@ -449,9 +537,9 @@ func (c *Cluster) waitForCompletion(node string) error { } if status.Status == OrchestrationStatusCompleted { - return nil + return status, nil } else if status.Status == OrchestrationStatusFailed { - return fmt.Errorf("orchestration for node %s failed: %+v", node, status) + return nil, fmt.Errorf("orchestration for node %s failed: %+v", node, status) } } @@ -467,8 +555,8 @@ func IsRemovingNode(devices string) bool { return devices == "none" } -func (c *Cluster) findRemovedNodes() ([]rookalpha.Node, error) { - var removedNodes []rookalpha.Node +func (c *Cluster) findRemovedNodes() ([]deploymentPerNode, error) { + var removedNodes []deploymentPerNode // first discover the storage nodes that are still running discoveredNodes, err := c.discoverStorageNodes() @@ -480,7 +568,7 @@ func (c *Cluster) findRemovedNodes() ([]rookalpha.Node, error) { found := false for _, newNode := range c.Storage.Nodes { // discovered storage node still exists in the current storage spec, move on to next discovered node - if discoveredNode.Name == newNode.Name { + if discoveredNode.node.Name == newNode.Name { found = true break } @@ -495,25 +583,23 @@ func (c *Cluster) findRemovedNodes() ([]rookalpha.Node, error) { return removedNodes, nil } -func (c *Cluster) discoverStorageNodes() ([]rookalpha.Node, error) { - var discoveredNodes []rookalpha.Node +func (c *Cluster) discoverStorageNodes() ([]deploymentPerNode, error) { + var discoveredNodes []deploymentPerNode listOpts := metav1.ListOptions{LabelSelector: fmt.Sprintf("app=%s", appName)} - osdReplicaSets, err := c.context.Clientset.Extensions().ReplicaSets(c.Namespace).List(listOpts) + osdDeployments, err := c.context.Clientset.Extensions().Deployments(c.Namespace).List(listOpts) if err != nil { - return nil, fmt.Errorf("failed to list osd replica sets: %+v", err) + return nil, fmt.Errorf("failed to list osd deployment: %+v", err) } - - discoveredNodes = make([]rookalpha.Node, len(osdReplicaSets.Items)) - for i, osdReplicaSet := range osdReplicaSets.Items { - osdPodSpec := osdReplicaSet.Spec.Template.Spec + discoveredNodes = make([]deploymentPerNode, len(osdDeployments.Items)) + for i, osdDeployment := range osdDeployments.Items { + osdPodSpec := osdDeployment.Spec.Template.Spec // get the node name from the node selector nodeName, ok := osdPodSpec.NodeSelector[apis.LabelHostname] if !ok || nodeName == "" { - return nil, fmt.Errorf("osd replicaset %s doesn't have a node name on its node selector: %+v", osdReplicaSet.Name, osdPodSpec.NodeSelector) + return nil, fmt.Errorf("osd deployment %s doesn't have a node name on its node selector: %+v", osdDeployment.Name, osdPodSpec.NodeSelector) } - // get the osd container osdContainer, err := k8sutil.GetMatchingContainer(osdPodSpec.Containers, appName) if err != nil { @@ -532,8 +618,18 @@ func (c *Cluster) discoverStorageNodes() ([]rookalpha.Node, error) { Location: rookalpha.GetLocationFromContainer(osdContainer), Config: getConfigFromContainer(osdContainer), } - - discoveredNodes[i] = node + found := false + for _, n := range discoveredNodes { + if nodeName == n.node.Name { + n.deployments = append(n.deployments, &osdDeployment) + found = true + break + } + } + if !found { + discoveredNodes[i].node = node + discoveredNodes[i].deployments = append(discoveredNodes[i].deployments, &osdDeployment) + } } return discoveredNodes, nil @@ -597,7 +693,10 @@ func (c *Cluster) isSafeToRemoveNode(node rookalpha.Node) error { return nil } -func (c *Cluster) deleteOSDPod(rs *extensions.ReplicaSet) error { +func (c *Cluster) deleteOSDPod(dp *extensions.Deployment) error { + if dp == nil { + return nil + } // list all OSD pods first opts := metav1.ListOptions{LabelSelector: fields.OneTermEqualSelector(k8sutil.AppAttr, appName).String()} pods, err := c.context.Clientset.CoreV1().Pods(c.Namespace).List(opts) @@ -605,13 +704,13 @@ func (c *Cluster) deleteOSDPod(rs *extensions.ReplicaSet) error { return err } - // iterate over all the OSD pods, looking for a match to the given replica set and the pod's owner + // iterate over all the OSD pods, looking for a match to the given deployment and the pod's owner var pod *v1.Pod for i := range pods.Items { p := &pods.Items[i] for _, owner := range p.OwnerReferences { - if owner.Name == rs.Name && owner.UID == rs.UID { - // the owner of this pod matches the name and UID of the given replica set + if owner.Name == dp.Name && owner.UID == dp.UID { + // the owner of this pod matches the name and UID of the given deployment pod = p break } @@ -623,7 +722,7 @@ func (c *Cluster) deleteOSDPod(rs *extensions.ReplicaSet) error { } if pod == nil { - return fmt.Errorf("pod for replica set %s not found", rs.Name) + return fmt.Errorf("pod for deployment %s not found", dp.Name) } err = c.context.Clientset.CoreV1().Pods(c.Namespace).Delete(pod.Name, &metav1.DeleteOptions{}) diff --git a/pkg/operator/ceph/cluster/osd/osd_test.go b/pkg/operator/ceph/cluster/osd/osd_test.go index 2006f2c4d..ac0d3893f 100644 --- a/pkg/operator/ceph/cluster/osd/osd_test.go +++ b/pkg/operator/ceph/cluster/osd/osd_test.go @@ -18,11 +18,13 @@ package osd import ( "encoding/json" "fmt" + "os" "testing" "time" rookalpha "github.com/rook/rook/pkg/apis/rook.io/v1alpha2" "github.com/rook/rook/pkg/clusterd" + discoverDaemon "github.com/rook/rook/pkg/daemon/discover" "github.com/rook/rook/pkg/operator/ceph/cluster/osd/config" "github.com/rook/rook/pkg/operator/k8sutil" exectest "github.com/rook/rook/pkg/util/exec/test" @@ -51,6 +53,24 @@ func TestStart(t *testing.T) { assert.Nil(t, err) } +func createDiscoverConfigmap(nodeName, ns string, clientset *fake.Clientset) error { + data := make(map[string]string, 1) + data[discoverDaemon.LocalDiskCMData] = `[{"name":"sdx","parent":"","hasChildren":false,"devLinks":"/dev/disk/by-id/scsi-36001405f826bd553d8c4dbf9f41c18be /dev/disk/by-id/wwn-0x6001405f826bd553d8c4dbf9f41c18be /dev/disk/by-path/ip-127.0.0.1:3260-iscsi-iqn.2016-06.world.srv:storage.target01-lun-1","size":10737418240,"uuid":"","serial":"36001405f826bd553d8c4dbf9f41c18be","type":"disk","rotational":true,"readOnly":false,"ownPartition":true,"filesystem":"","vendor":"LIO-ORG","model":"disk02","wwn":"0x6001405f826bd553","wwnVendorExtension":"0x6001405f826bd553d8c4dbf9f41c18be","empty":true}]` + cm := &v1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{ + Name: "local-device-" + nodeName, + Namespace: ns, + Labels: map[string]string{ + k8sutil.AppAttr: discoverDaemon.AppName, + discoverDaemon.NodeAttr: nodeName, + }, + }, + Data: data, + } + _, err := clientset.CoreV1().ConfigMaps(ns).Create(cm) + return err +} + func TestAddRemoveNode(t *testing.T) { // create a storage spec with the given nodes/devices/dirs nodeName := "node8230" @@ -68,6 +88,12 @@ func TestAddRemoveNode(t *testing.T) { // set up a fake k8s client set and watcher to generate events that the operator will listen to clientset := fake.NewSimpleClientset() + os.Setenv(k8sutil.PodNamespaceEnvVar, "rook-system") + defer os.Unsetenv(k8sutil.PodNamespaceEnvVar) + + cmErr := createDiscoverConfigmap(nodeName, "rook-system", clientset) + assert.Nil(t, cmErr) + statusMapWatcher := watch.NewFake() clientset.PrependWatchReactor("configmaps", k8stesting.DefaultWatchReactor(statusMapWatcher, nil)) @@ -102,8 +128,7 @@ func TestAddRemoveNode(t *testing.T) { osdPod := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{ Name: "osdPod", Labels: map[string]string{k8sutil.AppAttr: appName}, - OwnerReferences: []metav1.OwnerReference{{Name: "rook-ceph-osd-node8230"}}, - }} + OwnerReferences: []metav1.OwnerReference{{Name: "rook-ceph-osd-id-1"}}}} c.context.Clientset.CoreV1().Pods(c.Namespace).Create(osdPod) // mock the ceph calls that will be called during remove node @@ -169,12 +194,18 @@ func TestAddNodeFailure(t *testing.T) { }, } - // create a fake clientset that will return an error when the operator tries to create a replica set + // create a fake clientset that will return an error when the operator tries to create a job clientset := fake.NewSimpleClientset() - clientset.PrependReactor("create", "replicasets", func(action k8stesting.Action) (handled bool, ret runtime.Object, err error) { - return true, nil, fmt.Errorf("mock failed to create replica set") + clientset.PrependReactor("create", "jobs", func(action k8stesting.Action) (handled bool, ret runtime.Object, err error) { + return true, nil, fmt.Errorf("mock failed to create jobs") }) + os.Setenv(k8sutil.PodNamespaceEnvVar, "rook-system") + defer os.Unsetenv(k8sutil.PodNamespaceEnvVar) + + cmErr := createDiscoverConfigmap(nodeName, "rook-system", clientset) + assert.Nil(t, cmErr) + c := New(&clusterd.Context{Clientset: clientset, ConfigDir: "/var/lib/rook", Executor: &exectest.MockExecutor{}}, "ns-add-remove", "myversion", "", storageSpec, "", rookalpha.Placement{}, false, v1.ResourceRequirements{}, metav1.OwnerReference{}) @@ -189,7 +220,7 @@ func TestAddNodeFailure(t *testing.T) { // wait for orchestration to complete waitForOrchestrationCompletion(c, nodeName, &startCompleted) - // verify orchestration failed (because the operator failed to create a replica set) + // verify orchestration failed (because the operator failed to create a job) assert.True(t, startCompleted) assert.NotNil(t, startErr) } @@ -236,7 +267,18 @@ func mockNodeOrchestrationCompletion(c *Cluster, nodeName string, statusMapWatch if status != nil && status.Status == OrchestrationStatusStarting { // the node has started orchestration, simulate its completion now by performing 2 tasks: // 1) update the config map manually (which doesn't trigger a watch event, see https://github.com/kubernetes/kubernetes/issues/54075#issuecomment-337298950) - status = &OrchestrationStatus{Status: OrchestrationStatusCompleted} + status = &OrchestrationStatus{ + OSDs: []OSDInfo{ + { + ID: 1, + DataPath: "/tmp", + Config: "/foo/bar/ceph.conf", + Cluster: "rook", + KeyringPath: "/foo/bar/key", + }, + }, + Status: OrchestrationStatusCompleted, + } UpdateOrchestrationStatusMap(c.context.Clientset, c.Namespace, nodeName, *status) // 2) call modify on the fake watcher so a watch event will get triggered @@ -245,7 +287,7 @@ func mockNodeOrchestrationCompletion(c *Cluster, nodeName string, statusMapWatch statusMapWatcher.Modify(cm) break } else { - logger.Infof("waiting for node %s orchestration to start. status: %+v", nodeName, *status) + logger.Debugf("waiting for node %s orchestration to start. status: %+v", nodeName, *status) } } else { logger.Warningf("failed to get node %s orchestration status, will try again: %+v", nodeName, err) @@ -263,7 +305,7 @@ func waitForOrchestrationCompletion(c *Cluster, nodeName string, startCompleted if err == nil { status := parseOrchestrationStatus(cm.Data, nodeName) if status != nil { - logger.Infof("start has not completed, status is %+v", status) + logger.Debugf("start has not completed, status is %+v", status) } } <-time.After(50 * time.Millisecond) diff --git a/pkg/operator/ceph/cluster/osd/pod.go b/pkg/operator/ceph/cluster/osd/pod.go index 41212d687..6d1ab4b08 100644 --- a/pkg/operator/ceph/cluster/osd/pod.go +++ b/pkg/operator/ceph/cluster/osd/pod.go @@ -27,8 +27,11 @@ import ( opmon "github.com/rook/rook/pkg/operator/ceph/cluster/mon" "github.com/rook/rook/pkg/operator/ceph/cluster/osd/config" "github.com/rook/rook/pkg/operator/k8sutil" + + batch "k8s.io/api/batch/v1" "k8s.io/api/core/v1" extensions "k8s.io/api/extensions/v1beta1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/kubernetes/pkg/kubelet/apis" ) @@ -40,67 +43,154 @@ const ( osdWalSizeEnvVarName = "ROOK_OSD_WAL_SIZE" osdJournalSizeEnvVarName = "ROOK_OSD_JOURNAL_SIZE" osdMetadataDeviceEnvVarName = "ROOK_METADATA_DEVICE" + osdPrepareOnlyEnvVarName = "ROOK_OSD_PREPARE_ONLY" ) -func (c *Cluster) makeDaemonSet(selection rookalpha.Selection, storeConfig config.StoreConfig, metadataDevice, location string) *extensions.DaemonSet { - podSpec := c.podTemplateSpec(nil, selection, c.resources, storeConfig, metadataDevice, location) - ds := &extensions.DaemonSet{ +func (c *Cluster) makeJob(nodeName string, devices []rookalpha.Device, + selection rookalpha.Selection, resources v1.ResourceRequirements, storeConfig config.StoreConfig, metadataDevice, location string) (*batch.Job, error) { + + podSpec, err := c.podTemplateSpec(devices, selection, resources, storeConfig, metadataDevice, location, true /* prepare-only */, v1.RestartPolicyOnFailure) + if err != nil { + return nil, err + } + podSpec.Spec.NodeSelector = map[string]string{apis.LabelHostname: nodeName} + + job := &batch.Job{ ObjectMeta: metav1.ObjectMeta{ - Name: appName, - Namespace: c.Namespace, + Name: fmt.Sprintf(prepareAppNameFmt, nodeName), + Namespace: c.Namespace, + OwnerReferences: []metav1.OwnerReference{c.ownerRef}, Labels: map[string]string{ - k8sutil.AppAttr: appName, + k8sutil.AppAttr: prepareAppName, k8sutil.ClusterAttr: c.Namespace, }, }, - Spec: extensions.DaemonSetSpec{ - UpdateStrategy: extensions.DaemonSetUpdateStrategy{ - Type: extensions.RollingUpdateDaemonSetStrategyType, - }, - Template: podSpec, + Spec: batch.JobSpec{ + Template: *podSpec, }, } - k8sutil.SetOwnerRef(c.context.Clientset, c.Namespace, &ds.ObjectMeta, &c.ownerRef) - return ds + k8sutil.SetOwnerRef(c.context.Clientset, c.Namespace, &job.ObjectMeta, &c.ownerRef) + return job, nil } -func (c *Cluster) makeReplicaSet(nodeName string, devices []rookalpha.Device, selection rookalpha.Selection, resources v1.ResourceRequirements, - storeConfig config.StoreConfig, metadataDevice, location string) *extensions.ReplicaSet { - - podSpec := c.podTemplateSpec(devices, selection, resources, storeConfig, metadataDevice, location) - podSpec.Spec.NodeSelector = map[string]string{apis.LabelHostname: nodeName} +func (c *Cluster) makeDeployment(nodeName string, devices []rookalpha.Device, selection rookalpha.Selection, resources v1.ResourceRequirements, + storeConfig config.StoreConfig, metadataDevice, location string, osd OSDInfo) (*extensions.Deployment, error) { replicaCount := int32(1) + volumeMounts := []v1.VolumeMount{ + {Name: k8sutil.DataDirVolume, MountPath: k8sutil.DataDir}, + k8sutil.ConfigOverrideMount(), + } + volumes := []v1.Volume{k8sutil.ConfigOverrideVolume()} + if c.dataDirHostPath != "" { + // the user has specified a host path to use for the data dir, use that instead + dataDirSource := v1.VolumeSource{HostPath: &v1.HostPathVolumeSource{Path: c.dataDirHostPath}} + volumes = append(volumes, v1.Volume{Name: k8sutil.DataDirVolume, VolumeSource: dataDirSource}) + } else { + logger.Infof("no data dir provided") + } - rs := &extensions.ReplicaSet{ + // by default, don't define any volume config unless it is required + if len(devices) > 0 || 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) + devMount := v1.VolumeMount{Name: "devices", MountPath: "/dev"} + volumeMounts = append(volumeMounts, devMount) + } + + if len(volumes) == 0 { + return nil, fmt.Errorf("empty volumes") + } + + envVars := []v1.EnvVar{ + nodeNameEnvVar(), + k8sutil.PodIPEnvVar(k8sutil.PrivateIPEnvVar), + k8sutil.PodIPEnvVar(k8sutil.PublicIPEnvVar), + } + + privileged := true + runAsUser := int64(0) + readOnlyRootFilesystem := false + DNSPolicy := v1.DNSClusterFirst + if c.HostNetwork { + DNSPolicy = v1.DNSClusterFirstWithHostNet + } + journalStr := "" + if osd.IsFileStore { + journalStr = fmt.Sprintf("--osd-journal=%s", osd.Journal) + } + deployment := &extensions.Deployment{ ObjectMeta: metav1.ObjectMeta{ - Name: fmt.Sprintf(appNameFmt, nodeName), + Name: fmt.Sprintf(osdAppNameFmt, osd.ID), Namespace: c.Namespace, Labels: map[string]string{ k8sutil.AppAttr: appName, k8sutil.ClusterAttr: c.Namespace, + osdLabelKey: fmt.Sprintf("%d", osd.ID), }, }, - Spec: extensions.ReplicaSetSpec{ - Template: podSpec, + Spec: extensions.DeploymentSpec{ + Template: v1.PodTemplateSpec{ + ObjectMeta: metav1.ObjectMeta{ + Name: appName, + Labels: map[string]string{ + k8sutil.AppAttr: appName, + k8sutil.ClusterAttr: c.Namespace, + osdLabelKey: fmt.Sprintf("%d", osd.ID), + }, + Annotations: map[string]string{}, + }, + Spec: v1.PodSpec{ + NodeSelector: map[string]string{apis.LabelHostname: nodeName}, + //FIXME: use a diff SA for osd daemon + // ServiceAccountName: appName, + RestartPolicy: v1.RestartPolicyAlways, + HostNetwork: c.HostNetwork, + DNSPolicy: DNSPolicy, + Containers: []v1.Container{ + { + Command: []string{"sh", "-c", + fmt.Sprintf("ceph-osd --foreground --id %d --conf %s --osd-data %s --keyring %s --cluster %s --osd-uuid %s %s %s %s", + osd.ID, + osd.Config, + osd.DataPath, + osd.KeyringPath, + osd.Cluster, + osd.UUID, + fmt.Sprintf("--public-addr=${%s}", k8sutil.PublicIPEnvVar), + fmt.Sprintf("--cluster-addr=${%s}", k8sutil.PrivateIPEnvVar), + journalStr), + }, + Name: appName, + Image: k8sutil.MakeRookImage(c.Version), + VolumeMounts: volumeMounts, + Env: envVars, + Resources: resources, + SecurityContext: &v1.SecurityContext{ + Privileged: &privileged, + RunAsUser: &runAsUser, + ReadOnlyRootFilesystem: &readOnlyRootFilesystem, + }, + }, + }, + Volumes: volumes, + }, + }, Replicas: &replicaCount, }, } - k8sutil.SetOwnerRef(c.context.Clientset, c.Namespace, &rs.ObjectMeta, &c.ownerRef) - return rs + k8sutil.SetOwnerRef(c.context.Clientset, c.Namespace, &deployment.ObjectMeta, &c.ownerRef) + return deployment, nil } func (c *Cluster) podTemplateSpec(devices []rookalpha.Device, selection rookalpha.Selection, resources v1.ResourceRequirements, - storeConfig config.StoreConfig, metadataDevice, location string) v1.PodTemplateSpec { - // by default, the data/config dir will be an empty volume - dataDirSource := v1.VolumeSource{EmptyDir: &v1.EmptyDirVolumeSource{}} + storeConfig config.StoreConfig, metadataDevice, location string, prepareOnly bool, restart v1.RestartPolicy) (*v1.PodTemplateSpec, error) { + volumes := []v1.Volume{k8sutil.ConfigOverrideVolume()} + if c.dataDirHostPath != "" { // the user has specified a host path to use for the data dir, use that instead - dataDirSource = v1.VolumeSource{HostPath: &v1.HostPathVolumeSource{Path: c.dataDirHostPath}} - } - - volumes := []v1.Volume{ - {Name: k8sutil.DataDirVolume, VolumeSource: dataDirSource}, - k8sutil.ConfigOverrideVolume(), + dataDirSource := v1.VolumeSource{HostPath: &v1.HostPathVolumeSource{Path: c.dataDirHostPath}} + volumes = append(volumes, v1.Volume{Name: k8sutil.DataDirVolume, VolumeSource: dataDirSource}) } // by default, don't define any volume config unless it is required @@ -121,10 +211,14 @@ func (c *Cluster) podTemplateSpec(devices []rookalpha.Device, selection rookalph volumes = append(volumes, dirVolume) } + if len(volumes) == 0 { + return nil, fmt.Errorf("empty volumes") + } + podSpec := v1.PodSpec{ ServiceAccountName: c.serviceAccount, - Containers: []v1.Container{c.osdContainer(devices, selection, resources, storeConfig, metadataDevice, location)}, - RestartPolicy: v1.RestartPolicyAlways, + Containers: []v1.Container{c.osdContainer(devices, selection, resources, storeConfig, metadataDevice, location, prepareOnly)}, + RestartPolicy: restart, Volumes: volumes, HostNetwork: c.HostNetwork, } @@ -133,21 +227,21 @@ func (c *Cluster) podTemplateSpec(devices []rookalpha.Device, selection rookalph } c.placement.ApplyToPodSpec(&podSpec) - return v1.PodTemplateSpec{ + return &v1.PodTemplateSpec{ ObjectMeta: metav1.ObjectMeta{ Name: appName, Labels: map[string]string{ - k8sutil.AppAttr: appName, + k8sutil.AppAttr: prepareAppName, k8sutil.ClusterAttr: c.Namespace, }, Annotations: map[string]string{}, }, Spec: podSpec, - } + }, nil } func (c *Cluster) osdContainer(devices []rookalpha.Device, selection rookalpha.Selection, resources v1.ResourceRequirements, - storeConfig config.StoreConfig, metadataDevice, location string) v1.Container { + storeConfig config.StoreConfig, metadataDevice, location string, prepareOnly bool) v1.Container { envVars := []v1.EnvVar{ nodeNameEnvVar(), @@ -160,9 +254,11 @@ func (c *Cluster) osdContainer(devices []rookalpha.Device, selection rookalpha.S opmon.AdminSecretEnvVar(), k8sutil.ConfigDirEnvVar(), k8sutil.ConfigOverrideEnvVar(), + {Name: osdPrepareOnlyEnvVarName, Value: strconv.FormatBool(prepareOnly)}, } 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 { @@ -185,8 +281,14 @@ func (c *Cluster) osdContainer(devices []rookalpha.Device, selection rookalpha.S devMountNeeded = true } + dataVolume := v1.VolumeMount{Name: k8sutil.DataDirVolume, MountPath: k8sutil.DataDir} + if storeConfig.StoreType == config.Filestore { + prop := v1.MountPropagationBidirectional + dataVolume.MountPropagation = &prop + privileged = true + } volumeMounts := []v1.VolumeMount{ - {Name: k8sutil.DataDirVolume, MountPath: k8sutil.DataDir}, + dataVolume, k8sutil.ConfigOverrideMount(), } if devMountNeeded { @@ -230,7 +332,6 @@ func (c *Cluster) osdContainer(devices []rookalpha.Device, selection rookalpha.S envVars = append(envVars, rookalpha.LocationEnvVar(location)) } - privileged := false // 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" { privileged = true diff --git a/pkg/operator/ceph/cluster/osd/pod_test.go b/pkg/operator/ceph/cluster/osd/pod_test.go index ab86f7887..543eef15c 100644 --- a/pkg/operator/ceph/cluster/osd/pod_test.go +++ b/pkg/operator/ceph/cluster/osd/pod_test.go @@ -34,8 +34,9 @@ import ( func TestPodContainer(t *testing.T) { cluster := &Cluster{Namespace: "myosd", Version: "23"} - c := cluster.podTemplateSpec([]rookalpha.Device{}, rookalpha.Selection{}, v1.ResourceRequirements{}, config.StoreConfig{}, "", "") + c, err := cluster.podTemplateSpec([]rookalpha.Device{}, rookalpha.Selection{}, v1.ResourceRequirements{}, config.StoreConfig{}, "", "", false /* prepareOnly */, v1.RestartPolicyAlways) assert.NotNil(t, c) + assert.Nil(t, err) assert.Equal(t, 1, len(c.Spec.Containers)) container := c.Spec.Containers[0] assert.Equal(t, "ceph", container.Args[0]) @@ -49,74 +50,64 @@ func TestDaemonset(t *testing.T) { testPodDevices(t, "", "", false) } -func testPodDevices(t *testing.T, dataDir, deviceFilter string, allDevices bool) { +func testPodDevices(t *testing.T, dataDir, deviceName string, allDevices bool) { storageSpec := rookalpha.StorageScopeSpec{ - Selection: rookalpha.Selection{UseAllDevices: &allDevices, DeviceFilter: deviceFilter}, + Selection: rookalpha.Selection{UseAllDevices: &allDevices, DeviceFilter: deviceName}, Nodes: []rookalpha.Node{{Name: "node1"}}, } + devices := []rookalpha.Device{ + {Name: deviceName}, + } clientset := fake.NewSimpleClientset() c := New(&clusterd.Context{Clientset: clientset, ConfigDir: "/var/lib/rook", Executor: &exectest.MockExecutor{}}, "ns", "rook/rook:myversion", "mysa", storageSpec, dataDir, rookalpha.Placement{}, false, v1.ResourceRequirements{}, metav1.OwnerReference{}) - devMountNeeded := deviceFilter != "" || allDevices + devMountNeeded := deviceName != "" || allDevices - n := c.resolveNode(storageSpec.Nodes[0]) - replicaSet := c.makeReplicaSet(n.Name, n.Devices, n.Selection, v1.ResourceRequirements{}, config.StoreConfig{}, "", n.Location) - assert.NotNil(t, replicaSet) - assert.Equal(t, "rook-ceph-osd-node1", replicaSet.Name) - assert.Equal(t, c.Namespace, replicaSet.Namespace) + n := c.Storage.ResolveNode(storageSpec.Nodes[0].Name) + if len(devices) == 0 && len(dataDir) == 0 { + return + } + osd := OSDInfo{ + ID: 0, + } + + deployment, err := c.makeDeployment(n.Name, devices, n.Selection, v1.ResourceRequirements{}, config.StoreConfig{}, "", n.Location, osd) + assert.Nil(t, err) + assert.NotNil(t, deployment) + assert.Equal(t, "rook-ceph-osd-id-0", deployment.Name) + assert.Equal(t, c.Namespace, deployment.Namespace) assert.Equal(t, c.serviceAccount, replicaSet.Spec.Template.Spec.ServiceAccountName) - assert.Equal(t, int32(1), *(replicaSet.Spec.Replicas)) - assert.Equal(t, "node1", replicaSet.Spec.Template.Spec.NodeSelector[apis.LabelHostname]) - assert.Equal(t, v1.RestartPolicyAlways, replicaSet.Spec.Template.Spec.RestartPolicy) - if devMountNeeded { - assert.Equal(t, 4, len(replicaSet.Spec.Template.Spec.Volumes)) - } else { - assert.Equal(t, 2, len(replicaSet.Spec.Template.Spec.Volumes)) + assert.Equal(t, int32(1), *(deployment.Spec.Replicas)) + assert.Equal(t, "node1", deployment.Spec.Template.Spec.NodeSelector[apis.LabelHostname]) + assert.Equal(t, v1.RestartPolicyAlways, deployment.Spec.Template.Spec.RestartPolicy) + if devMountNeeded && len(dataDir) > 0 { + assert.Equal(t, 3, len(deployment.Spec.Template.Spec.Volumes)) } - assert.Equal(t, "rook-data", replicaSet.Spec.Template.Spec.Volumes[0].Name) - assert.Equal(t, "rook-config-override", replicaSet.Spec.Template.Spec.Volumes[1].Name) - if devMountNeeded { - assert.Equal(t, "devices", replicaSet.Spec.Template.Spec.Volumes[2].Name) + if devMountNeeded && len(dataDir) == 0 { + assert.Equal(t, 2, len(deployment.Spec.Template.Spec.Volumes)) } - if dataDir == "" { - assert.NotNil(t, replicaSet.Spec.Template.Spec.Volumes[0].EmptyDir) - assert.Nil(t, replicaSet.Spec.Template.Spec.Volumes[0].HostPath) - } else { - assert.Nil(t, replicaSet.Spec.Template.Spec.Volumes[0].EmptyDir) - assert.Equal(t, dataDir, replicaSet.Spec.Template.Spec.Volumes[0].HostPath.Path) + if !devMountNeeded && len(dataDir) > 0 { + assert.Equal(t, 2, len(deployment.Spec.Template.Spec.Volumes)) } - assert.Equal(t, appName, replicaSet.Spec.Template.ObjectMeta.Name) - assert.Equal(t, appName, replicaSet.Spec.Template.ObjectMeta.Labels["app"]) - assert.Equal(t, c.Namespace, replicaSet.Spec.Template.ObjectMeta.Labels["rook_cluster"]) - assert.Equal(t, 0, len(replicaSet.Spec.Template.ObjectMeta.Annotations)) + //assert.Equal(t, "rook-data", deployment.Spec.Template.Spec.Volumes[0].Name) + assert.Equal(t, "rook-config-override", deployment.Spec.Template.Spec.Volumes[0].Name) - cont := replicaSet.Spec.Template.Spec.Containers[0] + assert.Equal(t, appName, deployment.Spec.Template.ObjectMeta.Name) + assert.Equal(t, appName, deployment.Spec.Template.ObjectMeta.Labels["app"]) + assert.Equal(t, c.Namespace, deployment.Spec.Template.ObjectMeta.Labels["rook_cluster"]) + assert.Equal(t, 0, len(deployment.Spec.Template.ObjectMeta.Annotations)) + + cont := deployment.Spec.Template.Spec.Containers[0] assert.Equal(t, "rook/rook:myversion", cont.Image) if devMountNeeded { - assert.Equal(t, 4, len(cont.VolumeMounts)) + assert.Equal(t, 3, len(cont.VolumeMounts)) } else { - assert.Equal(t, 2, len(cont.VolumeMounts)) - } - assert.Equal(t, "ceph", cont.Args[0]) - assert.Equal(t, "osd", cont.Args[1]) - - // verify the config dir env var - verifyEnvVar(t, cont.Env, "ROOK_CONFIG_DIR", "/var/lib/rook", true) - - // verify the osd store type env var uses the default - verifyEnvVar(t, cont.Env, "ROOK_OSD_STORE", "", false) - - // verify the device filter env var - if deviceFilter != "" { - verifyEnvVar(t, cont.Env, "ROOK_DATA_DEVICE_FILTER", deviceFilter, true) - } else if allDevices { - verifyEnvVar(t, cont.Env, "ROOK_DATA_DEVICE_FILTER", "all", true) - } else { - verifyEnvVar(t, cont.Env, "ROOK_DATA_DEVICE_FILTER", "", false) + assert.Equal(t, 3, len(cont.VolumeMounts)) } + assert.Equal(t, "sh", cont.Command[0]) } func verifyEnvVar(t *testing.T, envVars []v1.EnvVar, expectedName, expectedValue string, expectedFound bool) { @@ -152,24 +143,16 @@ func TestStorageSpecDevicesAndDirectories(t *testing.T) { c := New(&clusterd.Context{Clientset: clientset, ConfigDir: "/var/lib/rook", Executor: &exectest.MockExecutor{}}, "ns", "rook/rook:myversion", "", storageSpec, "", rookalpha.Placement{}, false, v1.ResourceRequirements{}, metav1.OwnerReference{}) - n := c.resolveNode(storageSpec.Nodes[0]) - replicaSet := c.makeReplicaSet(n.Name, n.Devices, n.Selection, v1.ResourceRequirements{}, config.StoreConfig{}, "", n.Location) - assert.NotNil(t, replicaSet) - + n := c.Storage.ResolveNode(storageSpec.Nodes[0].Name) + osd := OSDInfo{ + ID: 0, + } + deployment, err := c.makeDeployment(n.Name, n.Devices, n.Selection, v1.ResourceRequirements{}, config.StoreConfig{}, "", n.Location, osd) + assert.NotNil(t, deployment) + assert.Nil(t, err) // pod spec should have a volume for the given dir - podSpec := replicaSet.Spec.Template.Spec - assert.Equal(t, 5, len(podSpec.Volumes)) - assert.Equal(t, "rook-dir1", podSpec.Volumes[4].Name) - assert.Equal(t, "/rook/dir1", podSpec.Volumes[4].VolumeSource.HostPath.Path) - - // container should have a volume mount for the given dir - container := podSpec.Containers[0] - assert.Equal(t, "rook-dir1", container.VolumeMounts[4].Name) - assert.Equal(t, "/rook/dir1", container.VolumeMounts[4].MountPath) - - // container command should have the given dir and device - verifyEnvVar(t, container.Env, "ROOK_DATA_DIRECTORIES", "/rook/dir1", true) - verifyEnvVar(t, container.Env, "ROOK_DATA_DEVICES", "sda", true) + podSpec := deployment.Spec.Template.Spec + assert.Equal(t, 2, len(podSpec.Volumes)) } func TestStorageSpecConfig(t *testing.T) { @@ -204,13 +187,14 @@ func TestStorageSpecConfig(t *testing.T) { c := New(&clusterd.Context{Clientset: clientset, ConfigDir: "/var/lib/rook", Executor: &exectest.MockExecutor{}}, "ns", "rook/rook:myversion", "", storageSpec, "", rookalpha.Placement{}, false, v1.ResourceRequirements{}, metav1.OwnerReference{}) - n := c.resolveNode(storageSpec.Nodes[0]) + n := c.Storage.ResolveNode(storageSpec.Nodes[0].Name) storeConfig := config.ToStoreConfig(storageSpec.Nodes[0].Config) metadataDevice := config.MetadataDevice(storageSpec.Nodes[0].Config) - replicaSet := c.makeReplicaSet(n.Name, n.Devices, n.Selection, c.Storage.Nodes[0].Resources, storeConfig, metadataDevice, n.Location) - assert.NotNil(t, replicaSet) - container := replicaSet.Spec.Template.Spec.Containers[0] + job, err := c.makeJob(n.Name, n.Devices, n.Selection, c.Storage.Nodes[0].Resources, storeConfig, metadataDevice, n.Location) + assert.NotNil(t, job) + assert.Nil(t, err) + container := job.Spec.Template.Spec.Containers[0] assert.NotNil(t, container) verifyEnvVar(t, container.Env, "ROOK_OSD_STORE", "bluestore", true) verifyEnvVar(t, container.Env, "ROOK_OSD_DATABASE_SIZE", "10", true) @@ -249,9 +233,13 @@ func TestHostNetwork(t *testing.T) { c := New(&clusterd.Context{Clientset: clientset, ConfigDir: "/var/lib/rook", Executor: &exectest.MockExecutor{}}, "ns", "myversion", "", storageSpec, "", rookalpha.Placement{}, true, v1.ResourceRequirements{}, metav1.OwnerReference{}) - n := c.resolveNode(storageSpec.Nodes[0]) - r := c.makeReplicaSet(n.Name, n.Devices, n.Selection, v1.ResourceRequirements{}, config.StoreConfig{}, "", n.Location) + n := c.Storage.ResolveNode(storageSpec.Nodes[0].Name) + osd := OSDInfo{ + ID: 0, + } + r, err := c.makeDeployment(n.Name, n.Devices, n.Selection, v1.ResourceRequirements{}, config.StoreConfig{}, "", n.Location, osd) assert.NotNil(t, r) + assert.Nil(t, err) assert.Equal(t, true, r.Spec.Template.Spec.HostNetwork) assert.Equal(t, v1.DNSClusterFirstWithHostNet, r.Spec.Template.Spec.DNSPolicy) diff --git a/pkg/operator/discover/discover.go b/pkg/operator/discover/discover.go index 5c410c20f..c26cc9cdc 100644 --- a/pkg/operator/discover/discover.go +++ b/pkg/operator/discover/discover.go @@ -22,6 +22,7 @@ import ( "fmt" "os" "regexp" + "time" "github.com/coreos/pkg/capnslog" rookalpha "github.com/rook/rook/pkg/apis/rook.io/v1alpha2" @@ -42,6 +43,9 @@ const ( discoverDaemonsetName = "rook-discover" discoverDaemonsetTolerationEnv = "DISCOVER_TOLERATION" discoverDaemonsetTolerationKeyEnv = "DISCOVER_TOLERATION_KEY" + deviceInUseCMName = "local-device-in-use-cluster-%s-node-%s" + deviceInUseAppName = "rook-claimed-devices" + deviceInUseClusterAttr = "rook.io/cluster" ) var logger = capnslog.NewPackageLogger("github.com/rook/rook", "op-discover") @@ -106,7 +110,8 @@ func (d *Discover) createDiscoverDaemonSet(namespace, discoverImage, securityAcc { Name: "dev", MountPath: "/dev", - ReadOnly: true, + // discovery pod could fail to start if /dev is mounted ro + ReadOnly: false, }, { Name: "sys", @@ -186,21 +191,81 @@ func (d *Discover) createDiscoverDaemonSet(namespace, discoverImage, securityAcc } +// ListDevices lists all devices discovered on all nodes or specific node if node name is provided. func ListDevices(context *clusterd.Context, namespace, nodeName string) (map[string][]sys.LocalDisk, error) { var devices map[string][]sys.LocalDisk listOpts := metav1.ListOptions{LabelSelector: fmt.Sprintf("%s=%s", k8sutil.AppAttr, discoverDaemon.AppName)} + // wait for device discovery configmaps + retryCount := 0 + retryMax := 30 + sleepTime := 5 + for { + retryCount++ + if retryCount > retryMax { + return devices, fmt.Errorf("exceeded max retry count waiting for device configmap to appear") + } + + if retryCount > 1 { + // only sleep after the first time + <-time.After(time.Duration(sleepTime) * time.Second) + } + + cms, err := context.Clientset.CoreV1().ConfigMaps(namespace).List(listOpts) + if err != nil { + logger.Warningf("failed to list device configmaps: %v", err) + return devices, fmt.Errorf("failed to list device configmaps: %+v", err) + } + if len(cms.Items) == 0 { + logger.Infof("no configmap match, retry #%d", retryCount) + continue + } + devices = make(map[string][]sys.LocalDisk, len(cms.Items)) + for _, cm := range cms.Items { + node := cm.ObjectMeta.Labels[discoverDaemon.NodeAttr] + if len(nodeName) > 0 && node != nodeName { + continue + } + deviceJson := cm.Data[discoverDaemon.LocalDiskCMData] + logger.Debugf("node %s, device %s", node, deviceJson) + + if len(node) == 0 || len(deviceJson) == 0 { + continue + } + var d []sys.LocalDisk + err = json.Unmarshal([]byte(deviceJson), &d) + if err != nil { + logger.Warningf("failed to unmarshal %s", deviceJson) + continue + } + devices[node] = d + } + break + } + logger.Debugf("discovery found the following devices %+v", devices) + return devices, nil +} + +// ListDevicesInUse lists all devices on a node that are already used by existing clusters. +func ListDevicesInUse(context *clusterd.Context, namespace, nodeName string) ([]sys.LocalDisk, error) { + var devices []sys.LocalDisk + + if len(nodeName) == 0 { + return devices, fmt.Errorf("empty node name") + } + + listOpts := metav1.ListOptions{LabelSelector: fmt.Sprintf("%s=%s", k8sutil.AppAttr, deviceInUseAppName)} cms, err := context.Clientset.CoreV1().ConfigMaps(namespace).List(listOpts) if err != nil { - return devices, fmt.Errorf("failed to list device configmaps: %+v", err) + return devices, fmt.Errorf("failed to list device in use configmaps: %+v", err) } - devices = make(map[string][]sys.LocalDisk, len(cms.Items)) + for _, cm := range cms.Items { node := cm.ObjectMeta.Labels[discoverDaemon.NodeAttr] - if len(nodeName) > 0 && node != nodeName { + if node != nodeName { continue } deviceJson := cm.Data[discoverDaemon.LocalDiskCMData] - logger.Debugf("node %s, device %s", node, deviceJson) + logger.Debugf("node %s, device in use %s", node, deviceJson) if len(node) == 0 || len(deviceJson) == 0 { continue @@ -211,31 +276,97 @@ func ListDevices(context *clusterd.Context, namespace, nodeName string) (map[str logger.Warningf("failed to unmarshal %s", deviceJson) continue } - devices[node] = d + for i := range d { + devices = append(devices, d[i]) + } } - logger.Debugf("devices %+v", devices) + logger.Debugf("devices in use %+v", devices) return devices, nil } +// FreeDevices frees up devices used by a cluster on a node. +func FreeDevices(context *clusterd.Context, nodeName, clusterName string) error { + if len(nodeName) == 0 || len(clusterName) == 0 { + return nil + } + namespace := os.Getenv(k8sutil.PodNamespaceEnvVar) + cmName := fmt.Sprintf(deviceInUseCMName, clusterName, nodeName) + // delete configmap + err := context.Clientset.CoreV1().ConfigMaps(namespace).Delete(cmName, &metav1.DeleteOptions{}) + if err != nil && !kserrors.IsNotFound(err) { + return fmt.Errorf("failed to delete configmap %s/%s. %+v", namespace, cmName, err) + } + + return nil +} + +// FreeDevicesByCluster frees devices on all nodes that are used by the cluster +func FreeDevicesByCluster(context *clusterd.Context, clusterName string) error { + logger.Infof("freeing devices used by cluster %s", clusterName) + namespace := os.Getenv(k8sutil.PodNamespaceEnvVar) + listOpts := metav1.ListOptions{LabelSelector: fmt.Sprintf("%s=%s", deviceInUseClusterAttr, clusterName)} + cms, err := context.Clientset.CoreV1().ConfigMaps(namespace).List(listOpts) + if err != nil { + return fmt.Errorf("failed to list device in use configmaps for cluster %s: %+v", clusterName, err) + } + + for _, cm := range cms.Items { + // delete configmap + cmName := cm.Name + err := context.Clientset.CoreV1().ConfigMaps(namespace).Delete(cmName, &metav1.DeleteOptions{}) + logger.Infof("deleting configmap %s: %+v", cmName, err) + if err != nil { + return err + } + } + return nil +} + +// GetAvailableDevices conducts outter join using input filters with free devices that a node has. It marks the devices from join result as in-use. func GetAvailableDevices(context *clusterd.Context, nodeName, clusterName string, devices []rookalpha.Device, filter string, useAllDevices bool) ([]rookalpha.Device, error) { results := []rookalpha.Device{} if len(devices) == 0 && len(filter) == 0 && !useAllDevices { return results, nil } namespace := os.Getenv(k8sutil.PodNamespaceEnvVar) + // find all devices allDevices, err := ListDevices(context, namespace, nodeName) if err != nil { return results, err } - nodeDevices, ok := allDevices[nodeName] + // find those on the node + nodeAllDevices, ok := allDevices[nodeName] if !ok { return results, fmt.Errorf("node %s has no devices", nodeName) } + // find those in use on the node + devicesInUse, err := ListDevicesInUse(context, namespace, nodeName) + if err != nil { + return results, err + } + + // filter those in use + nodeDevices := []sys.LocalDisk{} + for i := range nodeAllDevices { + isInUse := false + for j := range devicesInUse { + if nodeAllDevices[i].Name == devicesInUse[j].Name { + isInUse = true + break + } + } + if !isInUse { + nodeDevices = append(nodeDevices, nodeAllDevices[i]) + } + } + claimedDevices := []sys.LocalDisk{} + // now those left are free to use if len(devices) > 0 { for i := range devices { for j := range nodeDevices { if devices[i].Name == nodeDevices[j].Name { results = append(results, devices[i]) + claimedDevices = append(claimedDevices, nodeDevices[j]) } } } @@ -247,6 +378,7 @@ func GetAvailableDevices(context *clusterd.Context, nodeName, clusterName string d := rookalpha.Device{ Name: nodeDevices[i].Name, } + claimedDevices = append(claimedDevices, nodeDevices[i]) results = append(results, d) } } @@ -256,8 +388,37 @@ func GetAvailableDevices(context *clusterd.Context, nodeName, clusterName string Name: nodeDevices[i].Name, } results = append(results, d) + claimedDevices = append(claimedDevices, nodeDevices[i]) } } + // mark these devices in use + if len(claimedDevices) > 0 { + deviceJson, err := json.Marshal(claimedDevices) + if err != nil { + logger.Infof("failed to marshal: %v", err) + return results, err + } + data := make(map[string]string, 1) + data[discoverDaemon.LocalDiskCMData] = string(deviceJson) + cm := &v1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{ + Name: fmt.Sprintf(deviceInUseCMName, clusterName, nodeName), + Namespace: namespace, + Labels: map[string]string{ + k8sutil.AppAttr: deviceInUseAppName, + discoverDaemon.NodeAttr: nodeName, + deviceInUseClusterAttr: clusterName, + }, + }, + Data: data, + } + cm, err = context.Clientset.CoreV1().ConfigMaps(namespace).Create(cm) + + if err != nil { + logger.Warningf("failed to update device in use for cluster %s node %s: %v", clusterName, nodeName, err) + } + return results, err + } return results, nil } diff --git a/pkg/operator/discover/discover_test.go b/pkg/operator/discover/discover_test.go index 41eb8a8a1..74a0057e1 100644 --- a/pkg/operator/discover/discover_test.go +++ b/pkg/operator/discover/discover_test.go @@ -112,10 +112,33 @@ func TestGetAvailableDevices(t *testing.T) { devices, err := GetAvailableDevices(context, nodeName, ns, d, "^sd.", false) assert.Nil(t, err) assert.Equal(t, 1, len(devices)) + // devices should be in use now, 2nd try gets empty devices + devices, err = GetAvailableDevices(context, nodeName, ns, d, "^sd.", false) + assert.Nil(t, err) + assert.Equal(t, 0, len(devices)) + + err = FreeDevices(context, nodeName, ns) + assert.Nil(t, err) + // all devices freed devices, err = GetAvailableDevices(context, nodeName, ns, nil, "^sd.", false) assert.Nil(t, err) assert.Equal(t, 4, len(devices)) + // devices should be in use now, 2nd try gets empty devices + devices, err = GetAvailableDevices(context, nodeName, ns, nil, "^sd.", false) + assert.Nil(t, err) + assert.Equal(t, 0, len(devices)) + + err = FreeDevices(context, nodeName, ns) + assert.Nil(t, err) + devices, err = GetAvailableDevices(context, nodeName, ns, nil, "", true) assert.Nil(t, err) assert.Equal(t, 5, len(devices)) + // devices should be in use now, 2nd try gets empty devices + devices, err = GetAvailableDevices(context, nodeName, ns, nil, "", true) + assert.Nil(t, err) + assert.Equal(t, 0, len(devices)) + + err = FreeDevices(context, nodeName, ns) + assert.Nil(t, err) } diff --git a/pkg/util/sys/device.go b/pkg/util/sys/device.go index c8c1608e8..2358a5658 100644 --- a/pkg/util/sys/device.go +++ b/pkg/util/sys/device.go @@ -30,6 +30,7 @@ const ( SSDType = "ssd" PartType = "part" CryptType = "crypt" + LVMType = "lvm" sgdisk = "sgdisk" mountCmd = "mount" ) @@ -118,17 +119,16 @@ func GetDevicePartitions(device string, executor exec.Executor) (partitions []Pa } totalPartitionSize += p.Size - label, err := GetPartitionLabel(name, executor) + info, err := GetUdevInfo(name, executor) if err != nil { return nil, 0, err } - p.Label = label - - fs, err := GetDeviceFilesystems(name, executor) - if err != nil { - return nil, 0, err + if v, ok := info["ID_PART_ENTRY_NAME"]; ok { + p.Label = v + } + if v, ok := info["ID_FS_TYPE"]; ok { + p.Filesystem = v } - p.Filesystem = fs partitions = append(partitions, p) } diff --git a/pkg/util/sys/device_test.go b/pkg/util/sys/device_test.go index bae6197df..9c1102a2e 100644 --- a/pkg/util/sys/device_test.go +++ b/pkg/util/sys/device_test.go @@ -155,14 +155,11 @@ NAME="sdb3" SIZE="20" TYPE="part" PKNAME="sdb" NAME="sdb1" SIZE="30" TYPE="part" PKNAME="sdb"`, nil case run == 3: return fmt.Sprintf(udevPartOutput, "ROOK-OSD0-DB"), nil - case run == 4 || run == 6 || run == 8: - // get filesystem - return "", nil - case run == 5: + case run == 4: return fmt.Sprintf(udevPartOutput, "ROOK-OSD0-BLOCK"), nil - case run == 7: + case run == 5: return fmt.Sprintf(udevPartOutput, "ROOK-OSD0-WAL"), nil - case run == 9: + case run == 6: return `NAME="sda" SIZE="19818086400" TYPE="disk" PKNAME="" NAME="sda4" SIZE="1073741824" TYPE="part" PKNAME="sda" NAME="sda2" SIZE="2097152" TYPE="part" PKNAME="sda" @@ -172,14 +169,6 @@ NAME="sda3" SIZE="1073741824" TYPE="part" PKNAME="sda" NAME="usr" SIZE="1065345024" TYPE="crypt" PKNAME="sda3" NAME="sda1" SIZE="134217728" TYPE="part" PKNAME="sda" NAME="sda6" SIZE="134217728" TYPE="part" PKNAME="sda"`, nil - case run == 9: - return fmt.Sprintf(udevPartOutput, "ROOT"), nil - case run == 10: - return fmt.Sprintf(udevPartOutput, "OEM-CONFIG"), nil - case run == 11: - return fmt.Sprintf(udevPartOutput, "USR-A"), nil - case run == 12: - return fmt.Sprintf(udevPartOutput, "EFI-SYSTEM"), nil } return "", nil }, diff --git a/tests/framework/installer/install_data.go b/tests/framework/installer/install_data.go index 0b96cc8b2..d29a74fe7 100644 --- a/tests/framework/installer/install_data.go +++ b/tests/framework/installer/install_data.go @@ -17,6 +17,7 @@ limitations under the License. package installer import ( + "github.com/google/uuid" "strconv" ) @@ -220,6 +221,17 @@ rules: - get - list - watch +- apiGroups: + - batch + resources: + - jobs + verbs: + - get + - list + - watch + - create + - update + - delete - apiGroups: - ceph.rook.io resources: @@ -443,3 +455,31 @@ spec: - key: data path: mon-endpoints` } + +//GetCleanupPod gets a cleanup Pod manifest +func (i *InstallData) GetCleanupPod(removalDir string) string { + return `apiVersion: batch/v1 +kind: Job +metadata: + name: rook-cleanup-` + uuid.Must(uuid.NewRandom()).String() + ` +spec: + template: + spec: + restartPolicy: Never + containers: + - name: rook-cleaner + image: rook/rook:master + securityContext: + privileged: true + volumeMounts: + - name: cleaner + mountPath: /scrub + command: + - "sh" + - "-c" + - "rm -rf /scrub/*" + volumes: + - name: cleaner + hostPath: + path: ` + removalDir +} diff --git a/tests/framework/installer/install_helper.go b/tests/framework/installer/install_helper.go index 1066205aa..9a0f61ee2 100644 --- a/tests/framework/installer/install_helper.go +++ b/tests/framework/installer/install_helper.go @@ -18,6 +18,9 @@ package installer import ( "fmt" + "io/ioutil" + "os" + "path" "strings" "testing" "time" @@ -62,6 +65,7 @@ var ( type InstallHelper struct { k8shelper *utils.K8sHelper installData *InstallData + dataDir string helmHelper *utils.HelmHelper Env objects.EnvironmentManifest k8sVersion string @@ -158,10 +162,20 @@ func (h *InstallHelper) CreateK8sRookCluster(namespace, systemNamespace string, //CreateK8sRookCluster creates rook cluster via kubectl func (h *InstallHelper) CreateK8sRookClusterWithHostPathAndDevices(namespace, systemNamespace, storeType, dataDirHostPath string, useAllDevices bool, mon cephv1alpha1.MonSpec, startWithAllNodes bool) error { + if err := os.MkdirAll(DefaultDataDirHostPath(namespace), 0777); err != nil { + return err + } + + dataDir, err := ioutil.TempDir(DefaultDataDirHostPath(namespace), "test-") + if err != nil { + return err + } + dataDirHostPath = dataDir + h.dataDir = dataDir logger.Infof("Creating namespace %s", namespace) ns := &v1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: namespace}} - _, err := h.k8shelper.Clientset.CoreV1().Namespaces().Create(ns) + _, err = h.k8shelper.Clientset.CoreV1().Namespaces().Create(ns) if err != nil && !errors.IsAlreadyExists(err) { return fmt.Errorf("failed to create namespace %s. %+v", namespace, err) } @@ -380,6 +394,18 @@ func (h *InstallHelper) UninstallRookFromMultipleNS(helmInstalled bool, systemNa h.k8shelper.Clientset.RbacV1beta1().Roles(systemNamespace).Delete("rook-ceph-system", nil) logger.Infof("done removing the operator from namespace %s", systemNamespace) + logger.Infof("removing host data dir %s", h.dataDir) + // removing data dir if exists + if len(h.dataDir) > 0 { + err = h.cleanupDir(h.dataDir) + logger.Infof("removing %s, err %v", h.dataDir, err) + } +} + +func (h *InstallHelper) cleanupDir(dir string) error { + resources := h.installData.GetCleanupPod(dir) + _, err := h.k8shelper.KubectlWithStdin(resources, createArgs...) + return err } func (h *InstallHelper) checkError(err error, message string) { @@ -420,18 +446,21 @@ func (h *InstallHelper) CleanupCluster(clusterName string) { logger.Infof("Uninstalling All Rook Clusters - %s", clusterName) _, err := h.k8shelper.DeleteResource("-n", clusterName, "cluster", clusterName) if err != nil { - logger.Errorf("Rook Cluster %s cannot be deleted,err -> %v", clusterName, err) + logger.Warningf("Rook Cluster %s cannot be deleted,err -> %v", clusterName, err) } _, err = h.k8shelper.DeleteResource("-n", clusterName, "serviceaccount", "rook-ceph-osd") if err != nil { - logger.Errorf("rook-ceph-osd service account in namespace %s cannot be deleted,err -> %v", clusterName, err) - panic(err) + logger.Warningf("rook-ceph-osd service account in namespace %s cannot be deleted,err -> %v", clusterName, err) } _, err = h.k8shelper.DeleteResource("namespace", clusterName) if err != nil { - logger.Errorf("namespace %s cannot be deleted,err -> %v", clusterName, err) + logger.Warningf("namespace %s cannot be deleted,err -> %v", clusterName, err) + } + if len(h.dataDir) > 0 { + err = os.RemoveAll(h.dataDir) + logger.Infof("removing %s, err %v", h.dataDir, err) } } @@ -496,3 +525,9 @@ func IsAdditionalDeviceAvailableOnCluster() bool { logger.Info("No additional disks found on cluster") return false } + +func DefaultDataDirHostPath(namespace string) string { + cwd, _ := os.Getwd() + testDir := path.Join(cwd, "rook-test", namespace) + return testDir +} diff --git a/tests/integration/base_deploy_test.go b/tests/integration/base_deploy_test.go index 19c64431e..61736a738 100644 --- a/tests/integration/base_deploy_test.go +++ b/tests/integration/base_deploy_test.go @@ -18,6 +18,8 @@ package integration import ( "fmt" + "io/ioutil" + "os" "time" "testing" @@ -125,6 +127,16 @@ func StartBaseTestOperations(t func() *testing.T, namespace, storeType, dataDirH //SetUpRook is a wrapper for setting up rook func (op BaseTestOperations) SetUp() { + if err := os.MkdirAll(op.dataDirHostPath, 0755); err != nil { + logger.Warningf("failed to mkdir %s:%v", op.dataDirHostPath, err) + return + } + dataDir, err := ioutil.TempDir(op.dataDirHostPath, "test-") + if err != nil { + logger.Warningf("failed to make temp dir %s: %v", op.dataDirHostPath, err) + return + } + op.dataDirHostPath = dataDir isRookInstalled, err := op.installer.InstallRookOnK8sWithHostPathAndDevices(op.namespace, op.storeType, op.dataDirHostPath, op.helmInstalled, op.useDevices, cephv1alpha1.MonSpec{Count: op.mons, AllowMultiplePerNode: true}, false /* startWithAllNodes */) diff --git a/tests/integration/block_createBlock_k8s_test.go b/tests/integration/block_createBlock_k8s_test.go index 5cd043811..496504904 100644 --- a/tests/integration/block_createBlock_k8s_test.go +++ b/tests/integration/block_createBlock_k8s_test.go @@ -56,7 +56,7 @@ func (s *BlockCreateSuite) SetupSuite() { var err error s.namespace = "block-k8s-ns" - s.op, s.kh = StartBaseTestOperations(s.T, s.namespace, "bluestore", "", false, false, 1) + s.op, s.kh = StartBaseTestOperations(s.T, s.namespace, "bluestore", installer.DefaultDataDirHostPath(s.namespace), false, false, 1) s.testClient = GetTestClient(s.kh, s.namespace, s.op, s.T) initialBlocks, err := s.testClient.BlockClient.List(s.namespace) assert.Nil(s.T(), err) diff --git a/tests/integration/helm_test.go b/tests/integration/helm_test.go index 79dd7323a..720aac069 100644 --- a/tests/integration/helm_test.go +++ b/tests/integration/helm_test.go @@ -18,6 +18,8 @@ package integration import ( "testing" + "github.com/rook/rook/tests/framework/installer" + "github.com/coreos/pkg/capnslog" "github.com/rook/rook/tests/framework/clients" "github.com/rook/rook/tests/framework/contracts" @@ -63,7 +65,7 @@ type HelmSuite struct { func (hs *HelmSuite) SetupSuite() { hs.namespace = "helm-ns" - hs.op, hs.kh = StartBaseTestOperations(hs.T, hs.namespace, "bluestore", "", true, false, 1) + hs.op, hs.kh = StartBaseTestOperations(hs.T, hs.namespace, "bluestore", installer.DefaultDataDirHostPath(hs.namespace), true, false, 1) hs.helper = GetTestClient(hs.kh, hs.namespace, hs.op, hs.T) } diff --git a/tests/integration/smoke_test.go b/tests/integration/smoke_test.go index 91d9c0b8a..d41e4a9b4 100644 --- a/tests/integration/smoke_test.go +++ b/tests/integration/smoke_test.go @@ -75,7 +75,7 @@ type SmokeSuite struct { func (suite *SmokeSuite) SetupSuite() { suite.namespace = "smoke-ns" - suite.op, suite.k8sh = StartBaseTestOperations(suite.T, suite.namespace, "bluestore", "", false, false, 3) + suite.op, suite.k8sh = StartBaseTestOperations(suite.T, suite.namespace, "bluestore", installer.DefaultDataDirHostPath(suite.namespace), false, false, 3) suite.helper = GetTestClient(suite.k8sh, suite.namespace, suite.op, suite.T) } diff --git a/tests/integration/zblock_mount_unmount_test.go b/tests/integration/zblock_mount_unmount_test.go index e45ffd0c3..9de084ac0 100644 --- a/tests/integration/zblock_mount_unmount_test.go +++ b/tests/integration/zblock_mount_unmount_test.go @@ -85,7 +85,8 @@ func (s *BlockMountUnMountSuite) SetupSuite() { s.namespace = "block-test-ns" s.pvcNameRWO = "block-persistent-rwo" s.pvcNameRWX = "block-persistent-rwx" - s.op, s.kh = StartBaseTestOperations(s.T, s.namespace, "bluestore", "", false, false, 1) + + s.op, s.kh = StartBaseTestOperations(s.T, s.namespace, "bluestore", installer.DefaultDataDirHostPath(s.namespace), false, false, 1) s.testClient = GetTestClient(s.kh, s.namespace, s.op, s.T) s.bc = s.testClient.BlockClient } diff --git a/tests/scripts/minikube.sh b/tests/scripts/minikube.sh index ed2f8f2fa..fe69ef81c 100755 --- a/tests/scripts/minikube.sh +++ b/tests/scripts/minikube.sh @@ -115,6 +115,7 @@ case "${1:-}" in wait_for_ssh fi # create a link so the default dataDirHostPath will work for this environment + minikube ssh "sudo mkdir -p /mnt/sda1/${PWD}; sudo mkdir -p $(dirname $PWD); sudo ln -s /mnt/sda1/${PWD} $(dirname $PWD)/" minikube ssh "sudo mkdir /mnt/sda1/var/lib/rook;sudo ln -s /mnt/sda1/var/lib/rook /var/lib/rook" copy_image_to_cluster "${BUILD_REGISTRY}/ceph-amd64" rook/ceph:master copy_image_to_cluster "${BUILD_REGISTRY}/ceph-toolbox-amd64" rook/ceph-toolbox:master