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

Continued refactoring for clustered orchestration of ceph mons

Browse Source
This commit is contained in:
Travis Nielsen
2016-08-18 10:56:44 -07:00
parent ed074c5e58
commit dc16da55b8
7 changed files with 241 additions and 262 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/castled/ceph.go
+11 -23
View File
@@ -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)
}
pkg/castled/cmd/castled.go
+50 -42
View File
@@ -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
}
pkg/castled/config.go
+4 -32
View File
@@ -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
}
pkg/castled/mon.go
+7 -99
View File
@@ -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)),
pkg/castled/monagent.go
+117 -9
View File
@@ -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
}
pkg/castled/monleader.go
+51 -56
View File
@@ -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)
pkg/castled/osd.go
+1 -1
View File
@@ -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