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

CAS-1262 start monitor after bootstrapping

Browse Source
This commit is contained in:
Jared Watts
2016-08-08 14:28:01 -07:00
parent f58ee270a1
commit 07262efab3
3 changed files with 194 additions and 116 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/castled/main.go
+90 -39
View File
@@ -4,6 +4,7 @@ import (
"fmt"
"log"
"os"
"os/exec"
"os/signal"
"github.com/namsral/flag"
@@ -14,27 +15,29 @@ import (
func main() {
// supported top level commands
versionCommand := flag.NewFlagSet("version", flag.ExitOnError)
bootstrapCommand := flag.NewFlagSet("bootstrap", flag.ExitOnError)
daemonCommand := flag.NewFlagSet("daemon", flag.ExitOnError)
daemonCommand := flag.NewFlagSet("daemon", flag.ContinueOnError)
// bootstrap command flags/options
clusterNamePtr := bootstrapCommand.String("cluster-name", "defaultCluster", "name of ceph cluster")
etcdURLsPtr := bootstrapCommand.String("etcd-urls", "http://127.0.0.1:4001", "comma separated list of etcd listen URLs")
privateIPv4Ptr := bootstrapCommand.String("private-ipv4", "", "private IPv4 address for this machine (required)")
monNamePtr := bootstrapCommand.String("mon-name", "mon1", "monitor name")
initMonSetPtr := bootstrapCommand.String("initial-monitors", "mon1", "comma separated list of initial monitor names")
daemonTypePointer := daemonCommand.String("type", "", "type of daemon [mon|osd]")
// daemon command flags/options
daemonTypePtr := daemonCommand.String("type", "", "type of daemon [mon|osd]")
allFlagSets := map[string]*flag.FlagSet{
"version": versionCommand,
"bootstrap": bootstrapCommand,
"daemon": daemonCommand,
}
if len(os.Args) < 2 {
fmt.Println("version")
versionCommand.PrintDefaults()
fmt.Println("bootstrap")
bootstrapCommand.PrintDefaults()
fmt.Println("daemon")
daemonCommand.PrintDefaults()
os.Exit(1)
usage(allFlagSets)
}
switch os.Args[1] {
@@ -45,34 +48,15 @@ func main() {
case "daemon":
daemonCommand.Parse(os.Args[2:])
default:
handleFlagError()
fmt.Printf("unknown command: %s", os.Args[1])
usage(allFlagSets)
}
if bootstrapCommand.Parsed() {
verifyRequiredFlags([]*string{clusterNamePtr, etcdURLsPtr, privateIPv4Ptr})
// 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(*clusterNamePtr, *etcdURLsPtr, *privateIPv4Ptr, *monNamePtr, *initMonSetPtr)
go func(cfg castled.Config) {
if err := castled.Bootstrap(cfg); err != nil {
log.Fatalf("failed to bootstrap castled: %+v", err)
}
log.Printf("castled bootstrapped successfully!")
}(cfg)
handleBootstrapCmd(bootstrapCommand, clusterNamePtr, etcdURLsPtr, privateIPv4Ptr, monNamePtr, initMonSetPtr)
} else if daemonCommand.Parsed() {
if *daemonTypePointer == "" {
handleFlagError()
}
cephd.RunDaemon(*daemonTypePointer, os.Args[3:]...)
handleDaemonCmd(daemonCommand, daemonTypePtr)
}
// wait for user to interrupt/terminate the process
c := make(chan os.Signal, 1)
signal.Notify(c, os.Interrupt)
log.Printf("waiting for ctrl-c interrupt...")
<-c
log.Printf("terminating due to ctrl-c interrupt...")
}
func handleVersionCmd() {
@@ -82,15 +66,82 @@ func handleVersionCmd() {
os.Exit(0)
}
func verifyRequiredFlags(requiredFlags []*string) {
for i := range requiredFlags {
if *(requiredFlags[i]) == "" {
handleFlagError()
func handleBootstrapCmd(bootstrapCommand *flag.FlagSet,
clusterNamePtr, etcdURLsPtr, privateIPv4Ptr, monNamePtr, initMonSetPtr *string) {
verifyRequiredFlags(bootstrapCommand, []*string{clusterNamePtr, etcdURLsPtr, privateIPv4Ptr})
// 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(*clusterNamePtr, *etcdURLsPtr, *privateIPv4Ptr, *monNamePtr, *initMonSetPtr)
var proc *exec.Cmd
go func(cfg castled.Config) {
var err error
proc, err = castled.Bootstrap(cfg)
if err != nil {
log.Fatalf("failed to bootstrap castled: %+v", err)
}
fmt.Println("castled bootstrapped successfully!")
}(cfg)
// wait for user to interrupt/terminate the process
c := make(chan os.Signal, 1)
signal.Notify(c, os.Interrupt)
fmt.Println("waiting for ctrl-c interrupt...")
<-c
fmt.Println("terminating due to ctrl-c interrupt...")
if proc != nil {
fmt.Println("stopping child process...")
if err := castled.StopChildProcess(proc); err != nil {
fmt.Printf("failed to stop child process: %+v", err)
} else {
fmt.Println("child process stopped successfully")
}
}
}
func handleFlagError() {
flag.PrintDefaults()
func handleDaemonCmd(daemonCommand *flag.FlagSet, daemonTypePtr *string) {
if *daemonTypePtr == "" {
handleFlagError(daemonCommand)
}
if *daemonTypePtr != "mon" && *daemonTypePtr != "osd" {
fmt.Printf("unknown daemon type: %s\n", *daemonTypePtr)
handleFlagError(daemonCommand)
}
// daemon command passes through args to the child daemon process. Look for the
// terminator arg, and pass through all args after that (without a terminator arg,
// FlagSet.Parse prints errors for args it doesn't recognize)
passthruIndex := 3
for i := range os.Args {
if os.Args[i] == "--" {
passthruIndex = i + 1
break
}
}
// run the specified daemon
cephd.RunDaemon(*daemonTypePtr, os.Args[passthruIndex:]...)
}
func verifyRequiredFlags(flagSet *flag.FlagSet, requiredFlags []*string) {
for i := range requiredFlags {
if *(requiredFlags[i]) == "" {
handleFlagError(flagSet)
}
}
}
func handleFlagError(flagSet *flag.FlagSet) {
flagSet.PrintDefaults()
os.Exit(1)
}
func usage(allFlagSets map[string]*flag.FlagSet) {
for name, cmd := range allFlagSets {
fmt.Println(name)
cmd.PrintDefaults()
}
os.Exit(1)
}
pkg/castled/ceph.go
+68 -77
View File
@@ -46,67 +46,38 @@ const (
`
)
// BUGUBG: we could use a better process manager here with support for rediecting
// stdout, stderr, and signals. And can monitor the child.
func createCmd(daemon string, args ...string) (cmd *exec.Cmd) {
cmd = exec.Command(os.Args[0], append([]string{"daemon", daemon}, args...)...)
cmd.Stdout = os.Stdout
cmd.Stderr = os.Stderr
return cmd
}
func runChildProcess(daemon string, args ...string) error {
cmd := createCmd(daemon, args...)
return cmd.Run()
}
func startChildProcess(daemon string, args ...string) (cmd *exec.Cmd, err error) {
cmd = createCmd(daemon, args...)
err = cmd.Start()
if err != nil {
return nil, err
}
return cmd, nil
}
func stopChildProcess(cmd *exec.Cmd) error {
return cmd.Process.Kill()
}
type clusterInfo struct {
FSID string
MonitorSecret string
AdminSecret string
}
func Bootstrap(cfg Config) error {
func Bootstrap(cfg Config) (*exec.Cmd, error) {
// TODO: some of these operations should be done by only one member of the cluster, (e.g. leader election)
// get an etcd client to coordinate with the rest of the cluster and load/save config
etcdClient, err := kvstore.GetEtcdClient(cfg.EtcdURLs)
if err != nil {
return fmt.Errorf("failed to get etcdClient: %+v", err)
return nil, fmt.Errorf("failed to get etcdClient: %+v", err)
}
// load any existing cluster info that may have previously been created
cluster, err := LoadClusterInfo(etcdClient)
cluster, err := loadClusterInfo(etcdClient)
if err != nil {
return fmt.Errorf("failed to load cluster info: %+v", err)
return nil, fmt.Errorf("failed to load cluster info: %+v", err)
}
if !isClusterInfoSet(cluster) {
// the cluster info is not yet set, go ahead and set it now
cluster, err := createClusterInfo()
cluster, err = createClusterInfo()
if err != nil {
return fmt.Errorf("failed to create cluster info: %+v", err)
return nil, fmt.Errorf("failed to create cluster info: %+v", err)
}
log.Printf("Created new cluster info: %+v", cluster)
err = SaveClusterInfo(cluster, etcdClient)
err = saveClusterInfo(cluster, etcdClient)
if err != nil {
return fmt.Errorf("failed to save new cluster info: %+v", err)
return nil, fmt.Errorf("failed to save new cluster info: %+v", err)
}
} else {
// the cluster has already been created
@@ -114,28 +85,29 @@ func Bootstrap(cfg Config) error {
}
if err := registerMonitor(cfg, etcdClient); err != nil {
return fmt.Errorf("failed to register monitor: %+v", err)
return nil, fmt.Errorf("failed to register monitor: %+v", err)
}
// wait for monitor registration to complete for all expected initial monitors
if err := waitForMonitorRegistration(cfg, etcdClient); err != nil {
return fmt.Errorf("failed to wait for monitors to register: %+v", err)
return nil, fmt.Errorf("failed to wait for monitors to register: %+v", err)
}
if err := makeMonitorFileSystem(cfg, cluster); err != nil {
return fmt.Errorf("failed to make monitor filesystem: %+v", err)
return nil, fmt.Errorf("failed to make monitor filesystem: %+v", err)
}
if err := runMonitor(cfg); err != nil {
return fmt.Errorf("failed to run monitor: %+v", err)
proc, err := runMonitor(cfg)
if err != nil {
return nil, fmt.Errorf("failed to run monitor: %+v", err)
}
log.Printf("successfully started monitor %s", cfg.MonName)
return nil
return proc, nil
}
func LoadClusterInfo(etcdClient etcd.KeysAPI) (clusterInfo, error) {
func loadClusterInfo(etcdClient etcd.KeysAPI) (clusterInfo, error) {
resp, err := etcdClient.Get(context.Background(), path.Join(cephKey, "fsid"), nil)
if err != nil {
return handleLoadClusterInfoErr(err)
@@ -197,7 +169,7 @@ func createClusterInfo() (clusterInfo, error) {
}, nil
}
func SaveClusterInfo(c clusterInfo, etcdClient etcd.KeysAPI) error {
func saveClusterInfo(c clusterInfo, etcdClient etcd.KeysAPI) error {
_, err := etcdClient.Set(context.Background(), path.Join(cephKey, "fsid"), c.FSID, nil)
if err != nil {
return err
@@ -268,15 +240,53 @@ func waitForMonitorRegistration(cfg Config, etcdClient etcd.KeysAPI) error {
func makeMonitorFileSystem(cfg Config, c clusterInfo) error {
// write the keyring to disk
keyringPath, err := writeMonitorKeyring(c)
if err != nil {
return err
}
// write the config file to disk
monConfFilePath, err := writeMonitorConfigFile(cfg, c)
if err != nil {
return err
}
// create monitor data dir
monDataDir := fmt.Sprintf("/tmp/mon/mon.%s", cfg.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 = runChildProcess(
"mon",
"--mkfs",
fmt.Sprintf("--cluster=%s", cfg.ClusterName),
fmt.Sprintf("--name=mon.%s", cfg.MonName),
fmt.Sprintf("--mon-data=%s", monDataDir),
fmt.Sprintf("--conf=%s", monConfFilePath),
fmt.Sprintf("--keyring=%s", keyringPath))
if err != nil {
return fmt.Errorf("failed mon --mkfs: %+v", err)
}
return nil
}
func writeMonitorKeyring(c clusterInfo) (string, error) {
keyring := fmt.Sprintf(monitorKeyringTemplate, c.MonitorSecret, c.AdminSecret)
keyringPath := "/tmp/mon/tmp_keyring"
if err := os.MkdirAll(filepath.Dir(keyringPath), 0744); err != nil {
fmt.Printf("failed to create keyring directory for %s: %+v", keyringPath, err)
return "", fmt.Errorf("failed to create keyring directory for %s: %+v", keyringPath, err)
}
if err := ioutil.WriteFile(keyringPath, []byte(keyring), 0644); err != nil {
return fmt.Errorf("failed to write monitor keyring to %s: %+v", keyringPath, err)
return "", fmt.Errorf("failed to write monitor keyring to %s: %+v", keyringPath, err)
}
return keyringPath, nil
}
func writeMonitorConfigFile(cfg Config, c clusterInfo) (string, error) {
// extract a list of just the monitor names, which will populate the "mon initial members"
// global config field
initialMonMembers := make([]string, len(cfg.InitialMonitors))
@@ -291,7 +301,7 @@ func makeMonitorFileSystem(cfg Config, c clusterInfo) error {
c.FSID,
strings.Join(initialMonMembers, " ")))
if err != nil {
return fmt.Errorf("failed to write global config section, %+v", err)
return "", fmt.Errorf("failed to write global config section, %+v", err)
}
// write the config for each individual monitor member of the cluster to the content buffer
@@ -299,7 +309,7 @@ func makeMonitorFileSystem(cfg Config, c clusterInfo) error {
mon := cfg.InitialMonitors[i]
_, err := contentBuffer.WriteString(fmt.Sprintf(monitorConfigTemplate, mon.Name, mon.Name, mon.Endpoint))
if err != nil {
return fmt.Errorf("failed to write monitor config section for mon %s, %+v", mon.Name, err)
return "", fmt.Errorf("failed to write monitor config section for mon %s, %+v", mon.Name, err)
}
}
@@ -309,32 +319,13 @@ func makeMonitorFileSystem(cfg Config, c clusterInfo) error {
fmt.Printf("failed to create monitor config file directory for %s: %+v", monConfFilePath, err)
}
if err := ioutil.WriteFile(monConfFilePath, contentBuffer.Bytes(), 0644); err != nil {
return fmt.Errorf("failed to write monitor config file to %s: %+v", monConfFilePath, err)
return "", fmt.Errorf("failed to write monitor config file to %s: %+v", monConfFilePath, err)
}
// create monitor data dir
monDataDir := fmt.Sprintf("/tmp/mon/mon.%s", cfg.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 ceph-mon --mkfs
err = runChildProcess(
"mon",
"--mkfs",
fmt.Sprintf("--cluster=%s", cfg.ClusterName),
fmt.Sprintf("--name=%s", cfg.MonName),
fmt.Sprintf("--mon-data=%s", monDataDir),
fmt.Sprintf("--conf=%s", monConfFilePath),
fmt.Sprintf("--keyring=%s", keyringPath))
if err != nil {
return fmt.Errorf("failed ceph-mon --mkfs: %+v", err)
}
return nil
return monConfFilePath, nil
}
func runMonitor(cfg Config) error {
func runMonitor(cfg Config) (*exec.Cmd, error) {
monConfFilePath := "/tmp/mon/tmp_config"
monDataDir := fmt.Sprintf("/tmp/mon/mon.%s", cfg.MonName)
@@ -345,22 +336,22 @@ func runMonitor(cfg Config) error {
}
}
if monEndpoint == "" {
return fmt.Errorf("failed to find endpoint for mon %s", cfg.MonName)
return nil, fmt.Errorf("failed to find endpoint for mon %s", cfg.MonName)
}
// now run the mon
// start the monitor daemon in the foreground with the given config
log.Printf("starting monitor %s", cfg.MonName)
_, err := startChildProcess(
cmd, err := startChildProcess(
"mon",
"--foreground",
fmt.Sprintf("--cluster=%v", cfg.ClusterName),
fmt.Sprintf("--name=%v", cfg.MonName),
fmt.Sprintf("--name=mon.%v", cfg.MonName),
fmt.Sprintf("--mon-data=%s", monDataDir),
fmt.Sprintf("--conf=%s", monConfFilePath),
fmt.Sprintf("--public-addr=%v", monEndpoint))
if err != nil {
return fmt.Errorf("failed to start monitor %s: %+v", cfg.MonName, err)
return nil, fmt.Errorf("failed to start monitor %s: %+v", cfg.MonName, err)
}
return nil
return cmd, nil
}
pkg/castled/proc.go
+36
View File
@@ -0,0 +1,36 @@
package castled
import (
"fmt"
"os"
"os/exec"
)
// BUGUBG: we could use a better process manager here with support for rediecting
// stdout, stderr, and signals. And can monitor the child.
func createCmd(daemon string, args ...string) (cmd *exec.Cmd) {
cmd = exec.Command(os.Args[0], append([]string{"daemon", fmt.Sprintf("--type=%s", daemon), "--"}, args...)...)
cmd.Stdout = os.Stdout
cmd.Stderr = os.Stderr
return cmd
}
func runChildProcess(daemon string, args ...string) error {
cmd := createCmd(daemon, args...)
return cmd.Run()
}
func startChildProcess(daemon string, args ...string) (cmd *exec.Cmd, err error) {
cmd = createCmd(daemon, args...)
err = cmd.Start()
if err != nil {
return nil, err
}
return cmd, nil
}
func StopChildProcess(cmd *exec.Cmd) error {
return cmd.Process.Kill()
}