From eea62192db23775b7fd2edcd9282d37b3a14010f Mon Sep 17 00:00:00 2001 From: Travis Nielsen Date: Wed, 15 Feb 2017 10:27:41 -0800 Subject: [PATCH 1/2] create osds in directories instead of devices by default --- pkg/cephmgr/osd/agent.go | 9 +++-- pkg/cephmgr/osd/daemon.go | 74 ++++++++++++++++++++++++++++++---- pkg/cephmgr/osd/daemon_test.go | 61 ++++++++++++++++++++++++++++ 3 files changed, 133 insertions(+), 11 deletions(-) create mode 100644 pkg/cephmgr/osd/daemon_test.go diff --git a/pkg/cephmgr/osd/agent.go b/pkg/cephmgr/osd/agent.go index 1be60977e..2d30e554e 100644 --- a/pkg/cephmgr/osd/agent.go +++ b/pkg/cephmgr/osd/agent.go @@ -329,13 +329,16 @@ func (a *OsdAgent) configureDirs(context *clusterd.Context, dirs map[string]int) return err } + dirs[dirPath] = *osdID config.id = *osdID config.uuid = *osdUUID // set the desired state of the dir with the osd id - err = associateOsdIDWithDevice(context.EtcdClient, context.NodeID, dirPath, config.id, config.dir) - if err != nil { - return fmt.Errorf("failed to associate osd id %d with the data dir", config.id) + if context.EtcdClient != nil { + err = associateOsdIDWithDevice(context.EtcdClient, context.NodeID, dirPath, config.id, config.dir) + if err != nil { + return fmt.Errorf("failed to associate osd id %d with the data dir", config.id) + } } } diff --git a/pkg/cephmgr/osd/daemon.go b/pkg/cephmgr/osd/daemon.go index 7a04372d7..7d9468ddd 100644 --- a/pkg/cephmgr/osd/daemon.go +++ b/pkg/cephmgr/osd/daemon.go @@ -16,8 +16,12 @@ limitations under the License. package osd import ( + "encoding/json" "fmt" + "io/ioutil" "log" + "os" + "path" "time" "github.com/rook/rook/pkg/cephmgr/mon" @@ -26,6 +30,10 @@ import ( "github.com/rook/rook/pkg/util/sys" ) +const ( + dirOSDConfigFilename = "osd-dirs" +) + func Run(dcontext *clusterd.DaemonContext, agent *OsdAgent) error { logger.Infof("discovering hardware") @@ -64,13 +72,19 @@ func Run(dcontext *clusterd.DaemonContext, agent *OsdAgent) error { return fmt.Errorf("failed to configure devices. %+v", err) } - if len(devices.Entries) == 0 { - dirs := map[string]int{context.ConfigDir: -1} - logger.Infof("configuring osd dir: %+v", dirs) - err = agent.configureDirs(context, dirs) - if err != nil { - return fmt.Errorf("failed to configure dir %s. %+v", context.ConfigDir, err) - } + // initialize the data directories + dirs, err := getDataDirs(dcontext, true) + if err != nil { + return fmt.Errorf("failed to get data dirs. %+v", err) + } + logger.Infof("configuring osd dirs: %+v", dirs) + err = agent.configureDirs(context, dirs) + if err != nil { + return fmt.Errorf("failed to configure dirs %v. %+v", dirs, err) + } + err = saveDirConfig(dcontext, dirs) + if err != nil { + return fmt.Errorf("failed to save osd dir config. %+v", err) } // FIX @@ -90,9 +104,53 @@ func getAvailableDevices(context *clusterd.DaemonContext, devices []*inventory.L return nil, fmt.Errorf("failed to get device %s info. %+v", device.Name, err) } if fs == "" && ownPartitions { - available.Entries[device.Name] = &DeviceOsdIDEntry{Data: unassignedOSDID} + logger.Infof("skipping device %s until the admin specifies it can be used by an osd", device.Name) + //available.Entries[device.Name] = &DeviceOsdIDEntry{Data: (for existing osds get their ids) unassignedOSDID} } } return available, nil } + +func getDataDirs(context *clusterd.DaemonContext, useDataDir bool) (map[string]int, error) { + filePath := path.Join(context.ConfigDir, dirOSDConfigFilename) + if _, err := os.Stat(filePath); os.IsNotExist(err) { + // the config file doesn't exist yet, just return the empty map unless the data dir is desired + if useDataDir { + return map[string]int{context.ConfigDir: unassignedOSDID}, nil + } + + return map[string]int{}, nil + } + + b, err := ioutil.ReadFile(filePath) + if err != nil { + return nil, err + } + + var dirs map[string]int + err = json.Unmarshal(b, &dirs) + if err != nil { + return nil, err + } + return dirs, nil +} + +func saveDirConfig(context *clusterd.DaemonContext, config map[string]int) error { + if len(config) == 0 { + return nil + } + + b, err := json.Marshal(config) + if err != nil { + return err + } + filePath := path.Join(context.ConfigDir, dirOSDConfigFilename) + err = ioutil.WriteFile(filePath, b, 0644) + if err != nil { + return err + } + + logger.Debugf("saved osd dir config to %s", filePath) + return nil +} diff --git a/pkg/cephmgr/osd/daemon_test.go b/pkg/cephmgr/osd/daemon_test.go new file mode 100644 index 000000000..93083e454 --- /dev/null +++ b/pkg/cephmgr/osd/daemon_test.go @@ -0,0 +1,61 @@ +/* +Copyright 2016 The Rook Authors. All rights reserved. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ +package osd + +import ( + "testing" + + "os" + + "github.com/coreos/pkg/capnslog" + "github.com/rook/rook/pkg/clusterd" + "github.com/stretchr/testify/assert" +) + +func TestStoreOSDDirMap(t *testing.T) { + context := clusterd.NewDaemonContext("/tmp/testdir", "", capnslog.INFO) + defer os.RemoveAll(context.ConfigDir) + os.MkdirAll(context.ConfigDir, 0755) + + dirMap, err := getDataDirs(context, false) + assert.Nil(t, err) + assert.Equal(t, 0, len(dirMap)) + + dirMap, err = getDataDirs(context, true) + assert.Nil(t, err) + assert.Equal(t, 1, len(dirMap)) + assert.Equal(t, unassignedOSDID, dirMap[context.ConfigDir]) + dirMap[context.ConfigDir] = 0 + + err = saveDirConfig(context, dirMap) + assert.Nil(t, err) + + dirMap, err = getDataDirs(context, false) + assert.Nil(t, err) + assert.Equal(t, 1, len(dirMap)) + assert.Equal(t, 0, dirMap[context.ConfigDir]) + + // add another directory to the map + dirMap["/tmp/mydir"] = 23 + err = saveDirConfig(context, dirMap) + assert.Nil(t, err) + + dirMap, err = getDataDirs(context, false) + assert.Nil(t, err) + assert.Equal(t, 2, len(dirMap)) + assert.Equal(t, 0, dirMap[context.ConfigDir]) + assert.Equal(t, 23, dirMap["/tmp/mydir"]) +} From a07afb2e10f3d9d0f478548d01ecb16674df7acf Mon Sep 17 00:00:00 2001 From: Travis Nielsen Date: Wed, 15 Feb 2017 14:13:12 -0800 Subject: [PATCH 2/2] flag to allow osds to use all devices --- cmd/rook-operator/main.go | 5 ++++- cmd/rookd/osd.go | 9 ++++++--- pkg/cephmgr/osd/daemon.go | 11 +++++++---- pkg/operator/operator.go | 6 ++++-- pkg/operator/operator_test.go | 2 +- pkg/operator/osd/osd.go | 18 ++++++++++++------ 6 files changed, 34 insertions(+), 17 deletions(-) diff --git a/cmd/rook-operator/main.go b/cmd/rook-operator/main.go index 9a91e2831..dd94b264b 100644 --- a/cmd/rook-operator/main.go +++ b/cmd/rook-operator/main.go @@ -47,6 +47,7 @@ type config struct { cephConfigOverride string clusterInfo mon.ClusterInfo monEndpoints string + useAllDevices bool } var logLevelRaw string @@ -74,6 +75,8 @@ func init() { rootCmd.PersistentFlags().StringVar(&cfg.dataDir, "data-dir", "/var/lib/rook", "directory for storing configuration") rootCmd.PersistentFlags().StringVar(&logLevelRaw, "log-level", "INFO", "logging level for logging/tracing output (valid values: CRITICAL,ERROR,WARNING,NOTICE,INFO,DEBUG,TRACE)") + rootCmd.Flags().BoolVar(&cfg.useAllDevices, "use-all-devices", false, "true to use all storage devices, false to require local device selection") + // load the environment variables flags.SetFlagsFromEnv(rootCmd.Flags(), "ROOK_OPERATOR") flags.SetFlagsFromEnv(rootCmd.PersistentFlags(), "ROOK_OPERATOR") @@ -96,7 +99,7 @@ func startOperator(cmd *cobra.Command, args []string) error { } logger.Infof("starting operator. containerVersion=%s", cfg.containerVersion) - op := operator.New(k8sutil.Namespace, cephd.New(), clientset, cfg.containerVersion) + op := operator.New(k8sutil.Namespace, cephd.New(), clientset, cfg.containerVersion, cfg.useAllDevices) err = op.Run() if err != nil { fmt.Printf("failed to run operator. %+v\n", err) diff --git a/cmd/rookd/osd.go b/cmd/rookd/osd.go index 709d95fe0..af2b10cb6 100644 --- a/cmd/rookd/osd.go +++ b/cmd/rookd/osd.go @@ -33,10 +33,14 @@ var osdCmd = &cobra.Command{ Hidden: true, } var ( - osdCluster mon.ClusterInfo + osdCluster mon.ClusterInfo + osdDataDevices string ) func init() { + osdCmd.Flags().StringVar(&osdDataDevices, "data-devices", "", "the device names to use, or \"all\"") + flags.SetFlagsFromEnv(monCmd.Flags(), "ROOKD") + osdCmd.RunE = startOSD } @@ -47,13 +51,12 @@ func startOSD(cmd *cobra.Command, args []string) error { setLogLevel() - devices := "" metadataDevice := "" forceFormat := false location := "" bluestoreConfig := partition.BluestoreConfig{DatabaseSizeMB: 512} // FIX clusterInfo.Monitors = mon.ParseMonEndpoints(cfg.monEndpoints) - agent := osd.NewAgent(cephd.New(), devices, metadataDevice, forceFormat, location, bluestoreConfig, &clusterInfo) + agent := osd.NewAgent(cephd.New(), osdDataDevices, metadataDevice, forceFormat, location, bluestoreConfig, &clusterInfo) context := clusterd.NewDaemonContext(cfg.dataDir, cfg.cephConfigOverride, cfg.logLevel) return osd.Run(context, agent) diff --git a/pkg/cephmgr/osd/daemon.go b/pkg/cephmgr/osd/daemon.go index 7d9468ddd..525856082 100644 --- a/pkg/cephmgr/osd/daemon.go +++ b/pkg/cephmgr/osd/daemon.go @@ -61,7 +61,7 @@ func Run(dcontext *clusterd.DaemonContext, agent *OsdAgent) error { } // initialize the desired osds - devices, err := getAvailableDevices(dcontext, hardware.Disks) + devices, err := getAvailableDevices(dcontext, hardware.Disks, agent.devices) if err != nil { return fmt.Errorf("failed to get available devices. %+v", err) } @@ -93,7 +93,7 @@ func Run(dcontext *clusterd.DaemonContext, agent *OsdAgent) error { return nil } -func getAvailableDevices(context *clusterd.DaemonContext, devices []*inventory.LocalDisk) (*DeviceOsdMapping, error) { +func getAvailableDevices(context *clusterd.DaemonContext, devices []*inventory.LocalDisk, desiredDevices string) (*DeviceOsdMapping, error) { available := &DeviceOsdMapping{Entries: map[string]*DeviceOsdIDEntry{}} for _, device := range devices { if device.Type == sys.PartType { @@ -104,8 +104,11 @@ func getAvailableDevices(context *clusterd.DaemonContext, devices []*inventory.L return nil, fmt.Errorf("failed to get device %s info. %+v", device.Name, err) } if fs == "" && ownPartitions { - logger.Infof("skipping device %s until the admin specifies it can be used by an osd", device.Name) - //available.Entries[device.Name] = &DeviceOsdIDEntry{Data: (for existing osds get their ids) unassignedOSDID} + if desiredDevices == "all" { + available.Entries[device.Name] = &DeviceOsdIDEntry{Data: unassignedOSDID} + } else { + logger.Infof("skipping device %s until the admin specifies it can be used by an osd", device.Name) + } } } diff --git a/pkg/operator/operator.go b/pkg/operator/operator.go index 8d1b47c19..1fd67cad2 100644 --- a/pkg/operator/operator.go +++ b/pkg/operator/operator.go @@ -36,14 +36,16 @@ type Operator struct { clientset *kubernetes.Clientset waitCluster sync.WaitGroup factory client.ConnectionFactory + useAllDevices bool } -func New(namespace string, factory client.ConnectionFactory, clientset *kubernetes.Clientset, containerVersion string) *Operator { +func New(namespace string, factory client.ConnectionFactory, clientset *kubernetes.Clientset, containerVersion string, useAllDevices bool) *Operator { return &Operator{ Namespace: namespace, factory: factory, clientset: clientset, containerVersion: containerVersion, + useAllDevices: useAllDevices, } } @@ -63,7 +65,7 @@ func (o *Operator) Run() error { } // Start the OSDs - osds := osd.New(o.Namespace, o.containerVersion) + osds := osd.New(o.Namespace, o.containerVersion, o.useAllDevices) err = osds.Start(o.clientset, cluster) if err != nil { return fmt.Errorf("failed to start the osds. %+v", err) diff --git a/pkg/operator/operator_test.go b/pkg/operator/operator_test.go index add518320..4773f9685 100644 --- a/pkg/operator/operator_test.go +++ b/pkg/operator/operator_test.go @@ -18,5 +18,5 @@ package operator import "testing" func TestOperator(t *testing.T) { - New("foo", nil, nil, "version") + New("foo", nil, nil, "version", true) } diff --git a/pkg/operator/osd/osd.go b/pkg/operator/osd/osd.go index 7cd5ce1db..e262ec988 100644 --- a/pkg/operator/osd/osd.go +++ b/pkg/operator/osd/osd.go @@ -31,15 +31,17 @@ const ( ) type Cluster struct { - Namespace string - Keyring string - Version string + Namespace string + Keyring string + Version string + useAllDevices bool } -func New(namespace, version string) *Cluster { +func New(namespace, version string, useAllDevices bool) *Cluster { return &Cluster{ - Namespace: namespace, - Version: version, + Namespace: namespace, + Version: version, + useAllDevices: useAllDevices, } } @@ -97,6 +99,10 @@ func (c *Cluster) osdContainer(cluster *mon.ClusterInfo) v1.Container { command := fmt.Sprintf("/usr/bin/rookd osd --data-dir=%s --mon-endpoints=%s --cluster-name=%s ", k8sutil.DataDir, mon.FlattenMonEndpoints(cluster.Monitors), cluster.Name) + if c.useAllDevices { + command += fmt.Sprintf("--data-devices=all ") + } + privileged := true return v1.Container{ // TODO: fix "sleep 5".