diff --git a/pkg/castled/ceph.go b/pkg/castled/ceph.go index ccc433140..19d289d8f 100644 --- a/pkg/castled/ceph.go +++ b/pkg/castled/ceph.go @@ -26,38 +26,26 @@ type clusterInfo struct { FSID string MonitorSecret string AdminSecret string + Name string } -func Bootstrap(cfg Config, executor proc.Executor) ([]*exec.Cmd, error) { - - // Start the monitors - cluster, procs, err := startMonitors(cfg) - if err != nil { - return nil, err - } - - // Start the OSDs - osdProcs, err := startOSDs(cfg, cluster, executor) - if err != nil { - return nil, err - } - - procs = append(procs, osdProcs...) - - return procs, nil -} - -func startOSDs(cfg Config, cluster *clusterInfo, executor proc.Executor) ([]*exec.Cmd, error) { +func startOSDs(cluster *clusterInfo, executor proc.Executor) ([]*exec.Cmd, error) { user := "client.admin" - adminConn, err := connectToCluster(cfg.ClusterName, user, getMonConfFilePath(cfg.MonNames[0])) + config, err := getCephConnectionConfig(cluster) + if err != nil { + return nil, err + } + + adminConn, err := connectToCluster(cluster.Name, user, config) if err != nil { return nil, err } defer adminConn.Shutdown() // create/start an OSD for each of the specified devices - if len(cfg.Devices) > 0 { - osdProcs, err := createOSDs(adminConn, cfg, cluster, executor) + devices := []string{} + if len(devices) > 0 { + osdProcs, err := createOSDs(adminConn, cluster, executor) if err != nil { return nil, fmt.Errorf("failed to create OSDs: %+v", err) } diff --git a/pkg/castled/cmd/castled.go b/pkg/castled/cmd/castled.go index 055211811..78f784628 100644 --- a/pkg/castled/cmd/castled.go +++ b/pkg/castled/cmd/castled.go @@ -2,26 +2,26 @@ package cmd import ( "fmt" + "log" "os" - "os/exec" "os/signal" "strings" "github.com/spf13/cobra" "github.com/quantum/castle/pkg/castled" - "github.com/quantum/castle/pkg/kvstore" - "github.com/quantum/castle/pkg/proc" + "github.com/quantum/clusterd/pkg/orchestrator" + "github.com/quantum/clusterd/pkg/store" ) var ( - clusterName string - etcdURLs string - privateIPv4 string - monNames string - initMonSet string - devices string - forceFormat bool + clusterName string + discoveryURL string + etcdURLs string + nodeID string + privateIPv4 string + devices string + forceFormat bool ) var rootCmd = &cobra.Command{ @@ -31,23 +31,23 @@ var rootCmd = &cobra.Command{ } func init() { + rootCmd.Flags().StringVar(&nodeID, "node-id", "12345", "unique ID for the node (required)") rootCmd.Flags().StringVar(&clusterName, "cluster-name", "defaultCluster", "name of ceph cluster (required)") + rootCmd.Flags().StringVar(&discoveryURL, "discovery-url", "http://discovery.castle.com/26bd83c92e7145e6b103f623263f61df", + "etcd discovery URL") rootCmd.Flags().StringVar(&etcdURLs, "etcd-urls", "http://127.0.0.1:4001", - "comma separated list of etcd listen URLs (required)") + "comma separated list of etcd listen URLs (required), ignored if the discovery URL is specified") rootCmd.Flags().StringVar(&privateIPv4, "private-ipv4", "", "private IPv4 address for this machine (required)") - rootCmd.Flags().StringVar(&monNames, "mon-names", "mon1", - "comma separated list of monitor names to run on this machine (only 1 recommended)") - rootCmd.Flags().StringVar(&initMonSet, "initial-monitors", "mon1", - "comma separated list of initial monitor names in the cluster") rootCmd.Flags().StringVar(&devices, "devices", "", "comma separated list of devices to use") rootCmd.Flags().BoolVar(&forceFormat, "force-format", false, "true to force the format of any specified devices, even if they already have a filesystem. BE CAREFUL!") rootCmd.MarkFlagRequired("cluster-name") + rootCmd.MarkFlagRequired("node-id") rootCmd.MarkFlagRequired("etcd-urls") rootCmd.MarkFlagRequired("private-ipv4") - rootCmd.RunE = bootstrap + rootCmd.RunE = joinCluster } func Execute() error { @@ -60,30 +60,48 @@ func addCommands() { rootCmd.AddCommand(daemonCmd) } -func bootstrap(cmd *cobra.Command, args []string) error { - if err := verifyRequiredFlags(cmd, []string{"cluster-name", "etcd-urls", "private-ipv4"}); err != nil { +func joinCluster(cmd *cobra.Command, args []string) error { + if err := verifyRequiredFlags(cmd, []string{"node-id", "cluster-name", "etcd-urls", "private-ipv4"}); err != nil { return err } + // TODO: Get the etcd client with the discovery token rather than the etcd endpoints // get an etcd client to coordinate with the rest of the cluster and load/save config - etcdClient, err := kvstore.GetEtcdClient(strings.Split(etcdURLs, ",")) + etcdClient, err := store.GetEtcdClient(strings.Split(etcdURLs, ",")) if err != nil { return err } - // TODO: add the discovery URL to the command line/env var config, - // then we could just ask the discovery service where to find things like etcd - cfg := castled.NewConfig(etcdClient, clusterName, privateIPv4, monNames, initMonSet, devices, forceFormat) - var procs []*exec.Cmd - go func(cfg castled.Config) { - var err error - procs, err = castled.Bootstrap(cfg, &proc.CommandExecutor{}) - if err != nil { - fmt.Printf("failed to bootstrap castled: %+v", err) - } else { - fmt.Println("castled bootstrapped successfully!") - } - }(cfg) + // TODO: Write desired state for this machine to etcd + + // initialize a leadership lease manager + leaseManager, err := orchestrator.InitLeaseManager(etcdClient) + if err != nil { + log.Fatalf("failed to initialize lease manager: %s", err.Error()) + return err + } + + context := &orchestrator.ClusterContext{ + EtcdClient: etcdClient, + Executor: &orchestrator.CommandExecutor{}, + NodeID: nodeID, + Services: []*orchestrator.ClusterService{castled.NewMonitorService()}, + } + clusterLeader := &orchestrator.SimpleLeader{LeaseName: orchestrator.LeaderElectionKey} + clusterMember := orchestrator.NewClusterMember(context, leaseManager, clusterLeader) + + err = clusterMember.Initialize() + if err != nil { + log.Fatalf("failed to initialize local cluster: %v", err) + return err + } + + p := &orchestrator.ProcessManager{} + defer p.Shutdown() + go func() { + // Watch for commands from the leader + orchestrator.WatchForAgentServiceConfig(context, p) + }() // wait for user to interrupt/terminate the process ch := make(chan os.Signal, 1) @@ -91,16 +109,6 @@ func bootstrap(cmd *cobra.Command, args []string) error { fmt.Println("waiting for ctrl-c interrupt...") <-ch fmt.Println("terminating due to ctrl-c interrupt...") - for i := range procs { - if procs[i] != nil { - fmt.Printf("stopping child process %d\n", i) - if err := proc.StopChildProcess(procs[i]); err != nil { - fmt.Printf("failed to stop child process %d: %+v\n", i, err) - } else { - fmt.Printf("child process %d stopped successfully\n", i) - } - } - } return nil } diff --git a/pkg/castled/config.go b/pkg/castled/config.go index fb0dfdb77..107a60b24 100644 --- a/pkg/castled/config.go +++ b/pkg/castled/config.go @@ -7,45 +7,13 @@ import ( "os" "path/filepath" "strings" - - etcd "github.com/coreos/etcd/client" ) -type Config struct { - ClusterName string - EtcdClient etcd.KeysAPI - PrivateIPv4 string - MonNames []string - InitialMonitors []CephMonitorConfig - Devices []string - ForceFormat bool -} - type CephMonitorConfig struct { Name string Endpoint string } -func NewConfig(etcdClient etcd.KeysAPI, clusterName, privateIPv4, monNames, initMonitorNames, devices string, forceFormat bool) Config { - // caller should have provided a comma separated list of monitor names, split those into a - // list/slice, then create a slice of CephMonitorConfig structs based off those names - initMonNameSet := splitList(initMonitorNames) - initMonSet := make([]CephMonitorConfig, len(initMonNameSet)) - for i := range initMonNameSet { - initMonSet[i] = CephMonitorConfig{Name: initMonNameSet[i]} - } - - return Config{ - ClusterName: clusterName, - EtcdClient: etcdClient, - PrivateIPv4: privateIPv4, - MonNames: splitList(monNames), - InitialMonitors: initMonSet, - Devices: splitList(devices), - ForceFormat: forceFormat, - } -} - func splitList(list string) []string { if list == "" { return nil @@ -95,3 +63,7 @@ func writeInitialMonitorsConfigFileSections(contentBuffer *bytes.Buffer, cfg Con return nil } + +func getCephConnectionConfig(cluster *clusterInfo) (string, err) { + return "config", nil +} diff --git a/pkg/castled/mon.go b/pkg/castled/mon.go index b8850817c..07d2bd78b 100644 --- a/pkg/castled/mon.go +++ b/pkg/castled/mon.go @@ -12,11 +12,7 @@ import ( "path/filepath" "time" - etcd "github.com/coreos/etcd/client" - "golang.org/x/net/context" - "github.com/quantum/castle/pkg/cephd" - "github.com/quantum/castle/pkg/kvstore" "github.com/quantum/castle/pkg/proc" "github.com/quantum/clusterd/pkg/orchestrator" ) @@ -40,7 +36,7 @@ const ( ` ) -func NewService() *orchestrator.ClusterService { +func NewMonitorService() *orchestrator.ClusterService { service := &orchestrator.ClusterService{Name: "ceph-mon"} service.Leader = &monLeader{} @@ -74,58 +70,6 @@ func getMonDataDirPath(monName string) string { return filepath.Join(getMonRunDirPath(monName), fmt.Sprintf("mon.%s", monName)) } -// register the names and endpoints of all monitors on this machine -func registerMonitors(cfg Config, etcdClient etcd.KeysAPI) error { - port := 6790 - for i, monName := range cfg.MonNames { - key := getMonitorEndpointKey(monName) - val := fmt.Sprintf("%s:%d", cfg.PrivateIPv4, port+i) - - _, err := etcdClient.Set(context.Background(), key, val, nil) - if err == nil || kvstore.IsEtcdNodeExist(err) { - log.Printf("registered monitor %s endpoint of %s", monName, val) - } else { - return fmt.Errorf("failed to register mon %s endpoint: %+v", monName, err) - } - } - - return nil -} - -// wait for all expected initial monitors to register (report their names/endpoints) -func waitForMonitorRegistration(cfg Config, etcdClient etcd.KeysAPI) error { - for i := range cfg.InitialMonitors { - monName := cfg.InitialMonitors[i].Name - key := getMonitorEndpointKey(monName) - - registered := false - retryCount := 0 - retryMax := 40 - sleepTime := 5 - for { - resp, err := etcdClient.Get(context.Background(), key, nil) - if err == nil && resp != nil && resp.Node != nil && resp.Node.Value != "" { - cfg.InitialMonitors[i].Endpoint = resp.Node.Value - registered = true - } - - if registered { - log.Printf("monitor %s registered at %s", monName, cfg.InitialMonitors[i].Endpoint) - break - } - - retryCount++ - if retryCount > retryMax { - return fmt.Errorf("exceeded max retry count waiting for monitor %s to register", monName) - } - - <-time.After(time.Duration(sleepTime) * time.Second) - } - } - - return nil -} - // represents the response from a mon_status mon_command (subset of all available fields, only // marshal ones we care about) type MonStatusResponse struct { @@ -143,7 +87,7 @@ type MonMapEntry struct { } // waits for all expected initial monitors to establish and join quorum -func waitForMonitorQuorum(adminConn *cephd.Conn, cfg Config) error { +func (m *monLeader) waitForMonitorQuorum(adminConn *cephd.Conn) error { retryCount := 0 retryMax := 20 sleepTime := 5 @@ -168,7 +112,7 @@ func waitForMonitorQuorum(adminConn *cephd.Conn, cfg Config) error { // check if each of the initial monitors is in quorum allInQuorum := true - for _, im := range cfg.InitialMonitors { + for _, im := range m.monNames { // first get the initial monitors corresponding mon map entry var monMapEntry *MonMapEntry for i := range monStatusResp.MonMap.Mons { @@ -232,42 +176,6 @@ func getMonStatus(adminConn *cephd.Conn) (MonStatusResponse, error) { return resp, nil } -// creates and initializes the given monitors file systems -func makeMonitorFileSystems(cfg Config, c *clusterInfo) error { - for _, monName := range cfg.MonNames { - // write the keyring to disk - if err := writeMonitorKeyring(monName, c); err != nil { - return err - } - - // write the config file to disk - if err := writeMonitorConfigFile(monName, cfg, c, getMonKeyringPath(monName)); err != nil { - return err - } - - // create monitor data dir - monDataDir := getMonDataDirPath(monName) - if err := os.MkdirAll(filepath.Dir(monDataDir), 0744); err != nil { - fmt.Printf("failed to create monitor data directory at %s: %+v", monDataDir, err) - } - - // call mon --mkfs in a child process - err := proc.RunChildProcess( - "mon", - "--mkfs", - fmt.Sprintf("--cluster=%s", cfg.ClusterName), - fmt.Sprintf("--name=mon.%s", monName), - fmt.Sprintf("--mon-data=%s", monDataDir), - fmt.Sprintf("--conf=%s", getMonConfFilePath(monName)), - fmt.Sprintf("--keyring=%s", getMonKeyringPath(monName))) - if err != nil { - return fmt.Errorf("failed mon %s --mkfs: %+v", monName, err) - } - } - - return nil -} - // writes the monitor keyring to disk func writeMonitorKeyring(monName string, c *clusterInfo) error { keyring := fmt.Sprintf(monitorKeyringTemplate, c.MonitorSecret, c.AdminSecret) @@ -283,10 +191,10 @@ func writeMonitorKeyring(monName string, c *clusterInfo) error { } // generates and writes the monitor config file to disk -func writeMonitorConfigFile(monName string, cfg Config, c *clusterInfo, adminKeyringPath string) error { +func (m *monAgent) writeMonitorConfigFile(monName string, adminKeyringPath string) error { var contentBuffer bytes.Buffer - if err := writeGlobalConfigFileSection(&contentBuffer, cfg, c, getMonRunDirPath(monName)); err != nil { + if err := writeGlobalConfigFileSection(&contentBuffer, getMonRunDirPath(monName)); err != nil { return fmt.Errorf("failed to write mon %s global config section, %+v", monName, err) } @@ -308,7 +216,7 @@ func writeMonitorConfigFile(monName string, cfg Config, c *clusterInfo, adminKey } // runs all the given monitors in child processes -func runMonitors(cfg Config) ([]*exec.Cmd, error) { +func (m *monAgent) runMonitor() ([]*exec.Cmd, error) { procs := make([]*exec.Cmd, len(cfg.MonNames)) for i, monName := range cfg.MonNames { @@ -329,7 +237,7 @@ func runMonitors(cfg Config) ([]*exec.Cmd, error) { cmd, err := proc.StartChildProcess( "mon", "--foreground", - fmt.Sprintf("--cluster=%v", cfg.ClusterName), + fmt.Sprintf("--cluster=%v", m.cluster.Name), fmt.Sprintf("--name=mon.%v", monName), fmt.Sprintf("--mon-data=%s", getMonDataDirPath(monName)), fmt.Sprintf("--conf=%s", getMonConfFilePath(monName)), diff --git a/pkg/castled/monagent.go b/pkg/castled/monagent.go index 6f7c2a6d8..9f94a8e61 100644 --- a/pkg/castled/monagent.go +++ b/pkg/castled/monagent.go @@ -1,18 +1,50 @@ package castled -import "github.com/quantum/clusterd/pkg/orchestrator" +import ( + "fmt" + "log" + "os" + "path/filepath" + "time" + + etcd "github.com/coreos/etcd/client" + "golang.org/x/net/context" + + "github.com/quantum/clusterd/pkg/orchestrator" + "github.com/quantum/clusterd/pkg/store" +) type monAgent struct { + cluster clusterInfo + procMan *orchestrator.ProcessManager + privateIPv4 string + etcdClient etcd.KeysAPI } -func (m *monAgent) ConfigureAgent(context *orchestrator.ClusterContext, changeList []orchestrator.ChangeElement) error { - monNames := "foo:1235" - initMonitorNames := "mon1" - devices := "sda,sdb" - forceFormat := true - privateIP4 := "1.2.3.4" - config := NewConfig(context.EtcdClient, "default", privateIP4, monNames, initMonitorNames, devices, forceFormat) - _, err := Bootstrap(config, context.Executor) +func (a *monAgent) ConfigureAgent(context *orchestrator.ClusterContext, changeList []orchestrator.ChangeElement) error { + monName := "mon1" // FIX + port := 6790 + if err := a.registerMonitor(monName, port); err != nil { + return cluster, nil, fmt.Errorf("failed to register monitors: %+v", err) + } + + // wait for monitor registration to complete for all expected initial monitors + if err := a.waitForMonitorRegistration(); err != nil { + return cluster, nil, fmt.Errorf("failed to wait for monitors to register: %+v", err) + } + + // initialze the file systems for the monitors + if err := a.makeMonitorFileSystem(monName); err != nil { + return cluster, nil, fmt.Errorf("failed to make monitor filesystems: %+v", err) + } + + // run the monitors + procs, err := a.runMonitor() + if err != nil { + return cluster, nil, fmt.Errorf("failed to run monitors: %+v", err) + } + + log.Printf("successfully started monitor %s", monName) return err } @@ -20,3 +52,79 @@ func (m *monAgent) ConfigureAgent(context *orchestrator.ClusterContext, changeLi func (m *monAgent) DestroyAgent(context *orchestrator.ClusterContext) error { return nil } + +// register the names and endpoints of all monitors on this machine +func (a *monAgent) registerMonitor(monName string, port int) error { + + key := getMonitorEndpointKey(monName) + val := fmt.Sprintf("%s:%d", a.privateIPv4, port) + + _, err := a.etcdClient.Set(context.Background(), key, val, nil) + if err == nil || store.IsEtcdNodeExist(err) { + log.Printf("registered monitor %s endpoint of %s", monName, val) + } else { + return fmt.Errorf("failed to register mon %s endpoint: %+v", monName, err) + } + + return nil +} + +// wait for all expected initial monitors to register (report their names/endpoints) +func (a *monAgent) waitForMonitorRegistration(monName string) error { + key := getMonitorEndpointKey(monName) + + registered := false + retryCount := 0 + retryMax := 40 + sleepTime := 5 + for { + resp, err := a.etcdClient.Get(context.Background(), key, nil) + if err == nil && resp != nil && resp.Node != nil && resp.Node.Value != "" { + log.Printf("monitor %s registered at %s", monName, resp.Node.Value) + break + } + + retryCount++ + if retryCount > retryMax { + return fmt.Errorf("exceeded max retry count waiting for monitor %s to register", monName) + } + + <-time.After(time.Duration(sleepTime) * time.Second) + } + + return nil +} + +// creates and initializes the given monitors file systems +func (a *monAgent) makeMonitorFileSystem(monName string) error { + // write the keyring to disk + if err := a.writeMonitorKeyring(monName); err != nil { + return err + } + + // write the config file to disk + if err := a.writeMonitorConfigFile(monName, getMonKeyringPath(monName)); err != nil { + return err + } + + // create monitor data dir + monDataDir := getMonDataDirPath(monName) + if err := os.MkdirAll(filepath.Dir(monDataDir), 0744); err != nil { + fmt.Printf("failed to create monitor data directory at %s: %+v", monDataDir, err) + } + + // call mon --mkfs in a child process + err := a.procMan.Start( + "mon", + "--mkfs", + fmt.Sprintf("--cluster=%s", a.clusterName), + fmt.Sprintf("--name=mon.%s", monName), + fmt.Sprintf("--mon-data=%s", monDataDir), + fmt.Sprintf("--conf=%s", getMonConfFilePath(monName)), + fmt.Sprintf("--keyring=%s", getMonKeyringPath(monName))) + if err != nil { + return fmt.Errorf("failed mon %s --mkfs: %+v", monName, err) + } + + return nil +} diff --git a/pkg/castled/monleader.go b/pkg/castled/monleader.go index c2e1cf674..aa698eed1 100644 --- a/pkg/castled/monleader.go +++ b/pkg/castled/monleader.go @@ -3,30 +3,51 @@ package castled import ( "fmt" "log" - "os/exec" "path" "golang.org/x/net/context" etcd "github.com/coreos/etcd/client" "github.com/quantum/castle/pkg/cephd" - "github.com/quantum/castle/pkg/kvstore" "github.com/quantum/clusterd/pkg/orchestrator" ) // Interface implemented by a service that has been elected leader type monLeader struct { + cluster clusterInfo + privateIPv4 string + devices []string + forceFormat bool + etcdClient etcd.KeysAPI + monNames []string } // Load the state of the service from etcd. Typically a service will populate the desired/discovered state and the applied state // from etcd, then compute the difference and cache it. // Returns whether the service has updates to be applied. -func (m *monLeader) LoadState(ctx *orchestrator.ClusterContext) (bool, error) { +func (m *monLeader) LoadState(context *orchestrator.ClusterContext) (bool, error) { return true, nil } // Apply the desired state to the cluster. The context provides all the information needed to make changes to the service. -func (m *monLeader) ApplyState(ctx *orchestrator.ClusterContext) error { +func (m *monLeader) ApplyState(context *orchestrator.ClusterContext) error { + + // Create or get the basic cluster info + var err error + m.cluster, err = createOrGetClusterInfo(m.etcdClient) + if err != nil { + return nil, nil, err + } + + // TODO: Get existing monitors + // TODO: Select new monitors + // TODO: Send instructions to monitors + + err = m.waitForQuorum() + if err != nil { + return nil, err + } + log.Printf("Applied state for the ceph monitor") return nil } @@ -43,7 +64,7 @@ func createOrGetClusterInfo(etcdClient etcd.KeysAPI) (*clusterInfo, error) { return nil, fmt.Errorf("failed to load cluster info: %+v", err) } - if !isClusterInfoSet(cluster) { + if cluster == nil { // the cluster info is not yet set, go ahead and set it now cluster, err = createClusterInfo() if err != nil { @@ -63,71 +84,50 @@ func createOrGetClusterInfo(etcdClient etcd.KeysAPI) (*clusterInfo, error) { return cluster, nil } -func startMonitors(cfg Config) (*clusterInfo, []*exec.Cmd, error) { - - cluster, err := createOrGetClusterInfo(cfg.EtcdClient) - if err != nil { - return nil, nil, err - } - - if err := registerMonitors(cfg, cfg.EtcdClient); err != nil { - return cluster, nil, fmt.Errorf("failed to register monitors: %+v", err) - } - - // wait for monitor registration to complete for all expected initial monitors - if err := waitForMonitorRegistration(cfg, cfg.EtcdClient); err != nil { - return cluster, nil, fmt.Errorf("failed to wait for monitors to register: %+v", err) - } - - // initialze the file systems for the monitors - if err := makeMonitorFileSystems(cfg, cluster); err != nil { - return cluster, nil, fmt.Errorf("failed to make monitor filesystems: %+v", err) - } - - // run the monitors - procs, err := runMonitors(cfg) - if err != nil { - return cluster, nil, fmt.Errorf("failed to run monitors: %+v", err) - } - - log.Printf("successfully started monitors") +func (m *monLeader) waitForQuorum() error { // open an admin connection to the cluster user := "client.admin" - adminConn, err := connectToCluster(cfg.ClusterName, user, getMonConfFilePath(cfg.MonNames[0])) + adminConn, err := connectToCluster(cluster.Name, user, getCephConnectionConfig(m.cluster)) if err != nil { - return cluster, procs, err + return err } defer adminConn.Shutdown() // wait for monitors to establish quorum - if err := waitForMonitorQuorum(adminConn, cfg); err != nil { - return cluster, procs, fmt.Errorf("failed to wait for monitors to establish quorum: %+v", err) + if err := m.waitForMonitorQuorum(adminConn); err != nil { + return fmt.Errorf("failed to wait for monitors to establish quorum: %+v", err) } log.Printf("monitors formed quorum") - return cluster, procs, nil + return nil } // attempt to load any previously created and saved cluster info func loadClusterInfo(etcdClient etcd.KeysAPI) (*clusterInfo, error) { resp, err := etcdClient.Get(context.Background(), path.Join(cephKey, "fsid"), nil) if err != nil { - return handleLoadClusterInfoErr(err) + return nil, err } fsid := resp.Node.Value + resp, err = etcdClient.Get(context.Background(), path.Join(cephKey, "name"), nil) + if err != nil { + return nil, err + } + name := resp.Node.Value + secretsKey := path.Join(cephKey, "_secrets") resp, err = etcdClient.Get(context.Background(), path.Join(secretsKey, "monitor"), nil) if err != nil { - return handleLoadClusterInfoErr(err) + return nil, err } monSecret := resp.Node.Value resp, err = etcdClient.Get(context.Background(), path.Join(secretsKey, "admin"), nil) if err != nil { - return handleLoadClusterInfoErr(err) + return nil, err } adminSecret := resp.Node.Value @@ -135,42 +135,32 @@ func loadClusterInfo(etcdClient etcd.KeysAPI) (*clusterInfo, error) { FSID: fsid, MonitorSecret: monSecret, AdminSecret: adminSecret, + Name: name, }, nil } -func handleLoadClusterInfoErr(err error) (*clusterInfo, error) { - if kvstore.IsEtcdKeyNotFound(err) { - return &clusterInfo{}, nil - } - - return nil, err -} - -func isClusterInfoSet(c *clusterInfo) bool { - return c.FSID != "" && c.MonitorSecret != "" && c.AdminSecret != "" -} - // create new cluster info (FSID, shared keys) func createClusterInfo() (*clusterInfo, error) { fsid, err := cephd.NewFsid() if err != nil { - return handleLoadClusterInfoErr(err) + return nil, err } monSecret, err := cephd.NewSecretKey() if err != nil { - return handleLoadClusterInfoErr(err) + return nil, err } adminSecret, err := cephd.NewSecretKey() if err != nil { - return handleLoadClusterInfoErr(err) + return nil, err } return &clusterInfo{ FSID: fsid, MonitorSecret: monSecret, AdminSecret: adminSecret, + Name: "castlecluster", }, nil } @@ -181,6 +171,11 @@ func saveClusterInfo(c *clusterInfo, etcdClient etcd.KeysAPI) error { return err } + _, err = etcdClient.Set(context.Background(), path.Join(cephKey, "name"), c.Name, nil) + if err != nil { + return err + } + secretsKey := path.Join(cephKey, "_secrets") _, err = etcdClient.Set(context.Background(), path.Join(secretsKey, "monitor"), c.MonitorSecret, nil) diff --git a/pkg/castled/osd.go b/pkg/castled/osd.go index 8400c22d9..eba4653a0 100644 --- a/pkg/castled/osd.go +++ b/pkg/castled/osd.go @@ -72,7 +72,7 @@ func getOSDTempMonMapPath(osdDataPath string) string { } // create and initalize OSDs for all the devices specified in the given config -func createOSDs(adminConn *cephd.Conn, cfg Config, c *clusterInfo, executor proc.Executor) ([]*exec.Cmd, error) { +func createOSDs(adminConn *cephd.Conn, c *clusterInfo, executor proc.Executor) ([]*exec.Cmd, error) { // generate and write the OSD bootstrap keyring if err := createOSDBootstrapKeyring(adminConn, cfg.ClusterName, executor); err != nil { return nil, err