feedc0de/my-rook-config
1
0 Fork 0
forked from rook/rook
Code Pull Requests Activity

prepare osd in a job per node. Once all osds are prepared, store osd info in orchestration configmap.

Browse Source 
Operator watches the configmap, starts one osd replica set per osd.

Signed-off-by: Huamin Chen <hchen@redhat.com>
This commit is contained in:
Huamin Chen authored and travisn committed 2018-07-05 16:02:27 -06:00
1 parent c568dd6aa3
commit 82425aafcc
31 files changed
+949 -356
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

.gitignore
+2
View File
@@ -6,3 +6,5 @@
/_output
/vendor/
/vendor.orig/
/tests/integration/rook-test/
Jenkinsfile
Vendored
+2
View File
@@ -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()
}
}
cluster/charts/rook-ceph/templates/clusterrole.yaml
+11
View File
@@ -81,6 +81,17 @@ rules:
- get
- list
- watch
- apiGroups:
- batch
resources:
- jobs
verbs:
- get
- list
- watch
- create
- update
- delete
- apiGroups:
- ceph.rook.io
resources:
cluster/examples/kubernetes/ceph/operator.yaml
+11
View File
@@ -200,6 +200,17 @@ rules:
- get
- list
- watch
- apiGroups:
- batch
resources:
- jobs
verbs:
- get
- list
- watch
- create
- update
- delete
- apiGroups:
- ceph.rook.io
resources:
cmd/rook/ceph/osd.go
+10 -2
View File
@@ -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{
pkg/clusterd/disk.go
+3 -5
View File
@@ -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
pkg/daemon/ceph/osd/agent.go
+84 -48
View File
@@ -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
pkg/daemon/ceph/osd/agent_test.go
+3 -3
View File
@@ -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
}
pkg/daemon/ceph/osd/daemon.go
+34 -14
View File
@@ -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
}
pkg/daemon/ceph/osd/daemon_test.go
+6 -4
View File
@@ -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)
}
pkg/daemon/ceph/osd/device.go
+9 -4
View File
@@ -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)
}
pkg/daemon/ceph/osd/device_test.go
+13 -16
View File
@@ -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) {
pkg/daemon/ceph/osd/health_test.go
+1 -1
View File
@@ -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
pkg/daemon/discover/discover.go
+10 -6
View File
@@ -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)
}
pkg/operator/ceph/cluster/controller.go
+3
View File
@@ -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 {
pkg/operator/ceph/cluster/osd/osd.go
+191 -92
View File
@@ -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{})
pkg/operator/ceph/cluster/osd/osd_test.go
+51 -9
View File
@@ -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)
pkg/operator/ceph/cluster/osd/pod.go
+142 -41
View File
@@ -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
pkg/operator/ceph/cluster/osd/pod_test.go
+61 -73
View File
@@ -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)
pkg/operator/discover/discover.go
+169 -8
View File
@@ -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
}
pkg/operator/discover/discover_test.go
+23
View File
@@ -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)
}
pkg/util/sys/device.go
+7 -7
View File
@@ -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)
}
pkg/util/sys/device_test.go
+3 -14
View File
@@ -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
},
tests/framework/installer/install_data.go
+40
View File
@@ -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
}
tests/framework/installer/install_helper.go
+40 -5
View File
@@ -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
}
tests/integration/base_deploy_test.go
+12
View File
@@ -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 */)
tests/integration/block_createBlock_k8s_test.go
+1 -1
View File
@@ -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)
tests/integration/helm_test.go
+3 -1
View File
@@ -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)
}
tests/integration/smoke_test.go
+1 -1
View File
@@ -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)
}
tests/integration/zblock_mount_unmount_test.go
+2 -1
View File
@@ -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
}
tests/scripts/minikube.sh
+1
View File
@@ -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