diff --git a/pkg/cephmgr/mon.go b/pkg/cephmgr/mon.go index 2b39bed09..e2c884be1 100644 --- a/pkg/cephmgr/mon.go +++ b/pkg/cephmgr/mon.go @@ -79,7 +79,7 @@ type MonMapEntry struct { // calls mon_status mon_command func GetMonStatus(adminConn client.Connection) (MonStatusResponse, error) { cmd := map[string]interface{}{"prefix": "mon_status"} - buf, err := executeMonCommand(adminConn, cmd, fmt.Sprintf("retrieving mon_status")) + buf, err := ExecuteMonCommand(adminConn, cmd, fmt.Sprintf("retrieving mon_status")) var resp MonStatusResponse err = json.Unmarshal(buf, &resp) diff --git a/pkg/cephmgr/monagent.go b/pkg/cephmgr/monagent.go index d5a1024f5..19cdda95a 100644 --- a/pkg/cephmgr/monagent.go +++ b/pkg/cephmgr/monagent.go @@ -5,7 +5,6 @@ import ( "io/ioutil" "log" "os" - "os/exec" "path" "path/filepath" "regexp" @@ -22,7 +21,7 @@ const ( type monAgent struct { factory client.ConnectionFactory - monCmd *exec.Cmd + monProc *proc.MonitoredProc } func (a *monAgent) Name() string { @@ -74,18 +73,18 @@ func (a *monAgent) ConfigureLocalService(context *clusterd.Context) error { // stops and removes the monitor from this node func (a *monAgent) DestroyLocalService(context *clusterd.Context) error { - if a.monCmd == nil { + if a.monProc == nil { log.Printf("no need to stop a monitor that is not running") return nil } - if err := context.ProcMan.Stop(a.monCmd); err != nil { + if err := a.monProc.Stop(); err != nil { log.Printf("failed to stop mon. %v", err) return err } log.Printf("stopped ceph monitor") - a.monCmd = nil + a.monProc = nil // TODO: Clean up the monitor folder @@ -136,7 +135,7 @@ func (a *monAgent) runMonitor(context *clusterd.Context, cluster *ClusterInfo, m // start the monitor daemon in the foreground with the given config log.Printf("starting monitor %s", monitor.Name) monNameArg := fmt.Sprintf("--name=mon.%s", monitor.Name) - monCmd, err := context.ProcMan.Start( + monProc, err := context.ProcMan.Start( "mon", regexp.QuoteMeta(monNameArg), proc.ReuseExisting, @@ -150,9 +149,9 @@ func (a *monAgent) runMonitor(context *clusterd.Context, cluster *ClusterInfo, m return fmt.Errorf("failed to start monitor %s: %v", monitor.Name, err) } - if monCmd != nil { + if monProc != nil { // if the process was already running Start will return nil in which case we don't want to overwrite it - a.monCmd = monCmd + a.monProc = monProc } return nil diff --git a/pkg/cephmgr/monagent_test.go b/pkg/cephmgr/monagent_test.go index 6b18ed00b..d12606cb7 100644 --- a/pkg/cephmgr/monagent_test.go +++ b/pkg/cephmgr/monagent_test.go @@ -2,6 +2,7 @@ package cephmgr import ( "fmt" + "log" "os/exec" "path" "testing" @@ -17,31 +18,28 @@ func TestMonAgent(t *testing.T) { commands := 0 procTrap := func(action string, c *exec.Cmd) error { + log.Printf("PROC TRAP %d for %s. %+v", commands, action, c) + assert.Equal(t, "daemon", c.Args[1]) + assert.Equal(t, "--type=mon", c.Args[2]) + assert.Equal(t, "--", c.Args[3]) + assert.Equal(t, "--cluster=castlecluster", c.Args[5]) + assert.Equal(t, "--name=mon.mon0", c.Args[6]) + assert.Equal(t, "--mon-data=/tmp/mon0/mon.mon0", c.Args[7]) + assert.Equal(t, "--conf=/tmp/mon0/castlecluster.config", c.Args[8]) switch { case commands == 0: assert.Equal(t, proc.RunAction, action) - assert.Equal(t, "daemon", c.Args[1]) - assert.Equal(t, "--type=mon", c.Args[2]) - assert.Equal(t, "--", c.Args[3]) assert.Equal(t, "--mkfs", c.Args[4]) - assert.Equal(t, "--cluster=castlecluster", c.Args[5]) - assert.Equal(t, "--name=mon.mon0", c.Args[6]) - assert.Equal(t, "--mon-data=/tmp/mon0/mon.mon0", c.Args[7]) - assert.Equal(t, "--conf=/tmp/mon0/castlecluster.config", c.Args[8]) assert.Equal(t, "--keyring=/tmp/mon0/keyring", c.Args[9]) case commands == 1: assert.Equal(t, proc.StartAction, action) - assert.Equal(t, "daemon", c.Args[1]) - assert.Equal(t, "--type=mon", c.Args[2]) - assert.Equal(t, "--", c.Args[3]) assert.Equal(t, "--foreground", c.Args[4]) - assert.Equal(t, "--cluster=castlecluster", c.Args[5]) - assert.Equal(t, "--name=mon.mon0", c.Args[6]) - assert.Equal(t, "--mon-data=/tmp/mon0/mon.mon0", c.Args[7]) - assert.Equal(t, "--conf=/tmp/mon0/castlecluster.config", c.Args[8]) assert.Equal(t, "--public-addr=1.2.3.4:2345", c.Args[9]) + case commands == 2: + assert.Equal(t, proc.StopAction, action) default: return fmt.Errorf("unexpected case %d", commands) + assert.Equal(t, "--foreground", c.Args[4]) } commands++ return nil @@ -63,7 +61,7 @@ func TestMonAgent(t *testing.T) { err = agent.ConfigureLocalService(context) assert.Nil(t, err) assert.Equal(t, 0, commands) - assert.Nil(t, agent.monCmd) + assert.Nil(t, agent.monProc) // set the agent in the desired state key := path.Join(cephKey, monitorAgentName, desiredKey, context.NodeID) @@ -75,12 +73,12 @@ func TestMonAgent(t *testing.T) { err = agent.ConfigureLocalService(context) assert.Nil(t, err) assert.Equal(t, 2, commands) - assert.NotNil(t, agent.monCmd) + assert.NotNil(t, agent.monProc) // when the mon is not in desired state, it will be removed etcdClient.DeleteDir(key) err = agent.ConfigureLocalService(context) assert.Nil(t, err) - assert.Equal(t, 2, commands) - assert.Nil(t, agent.monCmd) + assert.Equal(t, 3, commands) + assert.Nil(t, agent.monProc) } diff --git a/pkg/cephmgr/osd.go b/pkg/cephmgr/osd.go index 8e46c3357..f29ac2c35 100644 --- a/pkg/cephmgr/osd.go +++ b/pkg/cephmgr/osd.go @@ -130,7 +130,7 @@ func createOSDBootstrapKeyring(conn client.Connection, clusterName string) error "entity": "client.bootstrap-osd", "caps": []string{"mon", "allow profile bootstrap-osd"}, } - buf, err := executeMonCommand(conn, cmd, "create osd bootstrap key") + buf, err := ExecuteMonCommand(conn, cmd, "create osd bootstrap key") if err != nil { return fmt.Errorf("failed to create osd bootstrap key: %+v", err) } @@ -406,7 +406,7 @@ func createOSD(bootstrapConn client.Connection, osdUUID uuid.UUID) (int, error) "entity": "client.bootstrap-osd", "uuid": osdUUID.String(), } - buf, err := executeMonCommand(bootstrapConn, cmd, fmt.Sprintf("create osd %s", osdUUID)) + buf, err := ExecuteMonCommand(bootstrapConn, cmd, fmt.Sprintf("create osd %s", osdUUID)) if err != nil { return 0, fmt.Errorf("failed to create osd %s: %+v", osdUUID, err) } @@ -427,7 +427,7 @@ func getMonMap(bootstrapConn client.Connection) ([]byte, error) { "entity": "client.bootstrap-osd", } - buf, err := executeMonCommand(bootstrapConn, cmd, "get mon map") + buf, err := ExecuteMonCommand(bootstrapConn, cmd, "get mon map") if err != nil { return nil, fmt.Errorf("failed to get mon map: %+v", cmd, err) } @@ -526,7 +526,7 @@ func addOSDToCrushMap(osdConn client.Connection, context *clusterd.Context, osdI "weight": weight, "args": locArgs, } - _, err = executeMonCommand(osdConn, cmd, fmt.Sprintf("adding %s to crush map", osdEntity)) + _, err = ExecuteMonCommand(osdConn, cmd, fmt.Sprintf("adding %s to crush map", osdEntity)) if err != nil { return fmt.Errorf("failed adding %s to crush map: %+v", osdEntity, err) } @@ -538,7 +538,7 @@ func addOSDToCrushMap(osdConn client.Connection, context *clusterd.Context, osdI return nil } -func executeMonCommand(connection client.Connection, cmd map[string]interface{}, message string) ([]byte, error) { +func ExecuteMonCommand(connection client.Connection, cmd map[string]interface{}, message string) ([]byte, error) { // ensure the json attribute is included in the request cmd["format"] = "json" @@ -568,7 +568,7 @@ func markOSDOut(connection client.Connection, id int) error { "prefix": "osd out", "ids": []int{id}, } - _, err := executeMonCommand(connection, command, fmt.Sprintf("mark osd %d out", id)) + _, err := ExecuteMonCommand(connection, command, fmt.Sprintf("mark osd %d out", id)) return err } @@ -578,7 +578,7 @@ func purgeOSD(connection client.Connection, name string, id int) error { "prefix": "osd crush remove", "name": fmt.Sprintf("osd.%d", id), } - _, err := executeMonCommand(connection, command, fmt.Sprintf("remove osd %s from crush map", name)) + _, err := ExecuteMonCommand(connection, command, fmt.Sprintf("remove osd %s from crush map", name)) if err != nil { return fmt.Errorf("failed to remove osd %s from crush map. %v", name, err) } @@ -588,7 +588,7 @@ func purgeOSD(connection client.Connection, name string, id int) error { "prefix": "auth del", "entity": fmt.Sprintf("osd.%d", id), } - _, err = executeMonCommand(connection, command, fmt.Sprintf("delete auth for osd %s", name)) + _, err = ExecuteMonCommand(connection, command, fmt.Sprintf("delete auth for osd %s", name)) if err != nil { return fmt.Errorf("failed to delete auth for osd %s. %v", name, err) } @@ -598,7 +598,7 @@ func purgeOSD(connection client.Connection, name string, id int) error { "prefix": "osd rm", "ids": []int{id}, } - _, err = executeMonCommand(connection, command, fmt.Sprintf("rm osds %v", name)) + _, err = ExecuteMonCommand(connection, command, fmt.Sprintf("rm osds %v", name)) if err != nil { return fmt.Errorf("failed to rm osd %s. %v", name, err) } @@ -609,7 +609,7 @@ func purgeOSD(connection client.Connection, name string, id int) error { // calls osd getcrushmap func GetCrushMap(adminConn client.Connection) (string, error) { cmd := map[string]interface{}{"prefix": "osd crush dump"} - buf, err := executeMonCommand(adminConn, cmd, fmt.Sprintf("retrieving crush map")) + buf, err := ExecuteMonCommand(adminConn, cmd, fmt.Sprintf("retrieving crush map")) if err != nil { return "", fmt.Errorf("failed to get crush map. %v", err) } diff --git a/pkg/cephmgr/osdagent.go b/pkg/cephmgr/osdagent.go index 37f793d7d..e0498199b 100644 --- a/pkg/cephmgr/osdagent.go +++ b/pkg/cephmgr/osdagent.go @@ -4,7 +4,6 @@ import ( "fmt" "log" "os" - "os/exec" "path" "regexp" "strconv" @@ -33,7 +32,7 @@ type osdAgent struct { forceFormat bool location string factory client.ConnectionFactory - osdCmd map[string]*exec.Cmd + osdProc map[string]*proc.MonitoredProc devices string } @@ -172,15 +171,15 @@ func (a *osdAgent) removeOSD(context *clusterd.Context, connection client.Connec } // stop the osd process if running - cmd, ok := a.osdCmd[name] + proc, ok := a.osdProc[name] if ok { - err := context.ProcMan.Stop(cmd) + err := proc.Stop() if err != nil { log.Printf("failed to stop osd for device %s. %v", name, err) return err } - delete(a.osdCmd, name) + delete(a.osdProc, name) } err = purgeOSD(connection, name, osdID) @@ -203,13 +202,13 @@ func (a *osdAgent) removeOSD(context *clusterd.Context, connection client.Connec func (a *osdAgent) DestroyLocalService(context *clusterd.Context) error { // stop the OSD processes - for device, cmd := range a.osdCmd { + for device, proc := range a.osdProc { log.Printf("stopping osd on device %s", device) - context.ProcMan.Stop(cmd) + proc.Stop() } // clear out the osd procs - a.osdCmd = map[string]*exec.Cmd{} + a.osdProc = map[string]*proc.MonitoredProc{} return nil } @@ -354,7 +353,7 @@ func (a *osdAgent) runOSD(context *clusterd.Context, clusterName string, osdID i // start the OSD daemon in the foreground with the given config log.Printf("starting osd %d at %s", osdID, osdDataPath) osdUUIDArg := fmt.Sprintf("--osd-uuid=%s", osdUUID.String()) - cmd, err := context.ProcMan.Start( + process, err := context.ProcMan.Start( "osd", regexp.QuoteMeta(osdUUIDArg), proc.ReuseExisting, @@ -370,13 +369,13 @@ func (a *osdAgent) runOSD(context *clusterd.Context, clusterName string, osdID i return fmt.Errorf("failed to start osd %d: %+v", osdID, err) } - if a.osdCmd == nil { + if a.osdProc == nil { // initialize the osd map - a.osdCmd = map[string]*exec.Cmd{} + a.osdProc = make(map[string]*proc.MonitoredProc) } - if cmd != nil { + if process != nil { // if the process was already running Start will return nil in which case we don't want to overwrite it - a.osdCmd[device] = cmd + a.osdProc[device] = process } return nil diff --git a/pkg/cephmgr/osdagent_test.go b/pkg/cephmgr/osdagent_test.go index 2c69dfff7..b0a303159 100644 --- a/pkg/cephmgr/osdagent_test.go +++ b/pkg/cephmgr/osdagent_test.go @@ -93,7 +93,7 @@ func TestOSDAgentWithDevices(t *testing.T) { EtcdClient: etcdClient, Executor: executor, NodeID: "abc", - ProcMan: &proc.ProcManager{Trap: createOSDAgentProcTrap(t, &procCommands)}, + ProcMan: &proc.ProcManager{Trap: createOSDAgentProcTrap(t, &procCommands, map[int]string{1: proc.StartAction, 3: proc.StartAction, 4: proc.StopAction})}, } // prep the OSD agent and related orcehstration data @@ -109,11 +109,11 @@ func TestOSDAgentWithDevices(t *testing.T) { assert.Equal(t, 6, execCount) assert.Equal(t, 2, outputExecCount) assert.Equal(t, 4, procCommands) - assert.Equal(t, 2, len(agent.osdCmd)) + assert.Equal(t, 2, len(agent.osdProc)) err = agent.DestroyLocalService(context) assert.Nil(t, err) - assert.Equal(t, 0, len(agent.osdCmd)) + assert.Equal(t, 0, len(agent.osdProc)) } func TestOSDAgentNoDevices(t *testing.T) { @@ -148,7 +148,7 @@ func TestOSDAgentNoDevices(t *testing.T) { EtcdClient: etcdClient, Executor: executor, NodeID: "abc", - ProcMan: &proc.ProcManager{Trap: createOSDAgentProcTrap(t, &procCommands)}, + ProcMan: &proc.ProcManager{Trap: createOSDAgentProcTrap(t, &procCommands, map[int]string{1: proc.StartAction, 2: proc.StopAction})}, } // prep the OSD agent and related orcehstration data @@ -160,7 +160,7 @@ func TestOSDAgentNoDevices(t *testing.T) { assert.Equal(t, 0, execCount) assert.Equal(t, 0, outputExecCount) assert.Equal(t, 2, procCommands) - assert.Equal(t, 1, len(agent.osdCmd)) + assert.Equal(t, 1, len(agent.osdProc)) // the local device should be marked as an applied OSD now osds, err := GetAppliedOSDs(context.NodeID, etcdClient) @@ -171,7 +171,7 @@ func TestOSDAgentNoDevices(t *testing.T) { // destroy the OSD and verify the results err = agent.DestroyLocalService(context) assert.Nil(t, err) - assert.Equal(t, 0, len(agent.osdCmd)) + assert.Equal(t, 0, len(agent.osdProc)) } func TestAppliedDevices(t *testing.T) { @@ -266,12 +266,19 @@ func prepAgentOrchestrationData(t *testing.T, agent *osdAgent, etcdClient *util. etcdClient.SetValue(path.Join(monKey, "port"), "8743") } -func createOSDAgentProcTrap(t *testing.T, commands *int) func(action string, c *exec.Cmd) error { +func createOSDAgentProcTrap(t *testing.T, commands *int, actions map[int]string) func(action string, c *exec.Cmd) error { return func(action string, c *exec.Cmd) error { - log.Printf("PROC TRAP %d for %s. %+v", commands, action, c) + log.Printf("PROC TRAP %d for %s. %+v", *commands, action, c) assert.Equal(t, "daemon", c.Args[1]) assert.Equal(t, "--type=osd", c.Args[2]) assert.Equal(t, "--", c.Args[3]) + if a, ok := actions[*commands]; ok { + assert.Equal(t, a, action) + } else { + // the default action expected is Run + assert.Equal(t, proc.RunAction, action, fmt.Sprintf("command=%d, action=%s", *commands, action)) + } + switch { case *commands == 0: err := ioutil.WriteFile("/tmp/osd3/mycluster-3/keyring", []byte("mykeyring"), 0644) @@ -281,7 +288,11 @@ func createOSDAgentProcTrap(t *testing.T, commands *int) func(action string, c * case *commands == 1: assert.Equal(t, "--foreground", c.Args[4]) case *commands == 2: - assert.Equal(t, "--mkfs", c.Args[4]) + if action == proc.StopAction { + assert.Equal(t, "--foreground", c.Args[4]) + } else { + assert.Equal(t, "--mkfs", c.Args[4]) + } case *commands == 3: assert.Equal(t, "--foreground", c.Args[4]) default: diff --git a/pkg/proc/monitoredproc.go b/pkg/proc/monitoredproc.go new file mode 100644 index 000000000..7dd131106 --- /dev/null +++ b/pkg/proc/monitoredproc.go @@ -0,0 +1,103 @@ +package proc + +import ( + "log" + "math" + "os/exec" + "syscall" + "time" +) + +const ( + maxDelaySeconds = 30 +) + +type MonitoredProc struct { + parent *ProcManager + cmd *exec.Cmd + monitor bool + retries int + totalRetries int + retrySecondsExponentBase float64 + waitForExit func() +} + +func newMonitoredProc(p *ProcManager, cmd *exec.Cmd) *MonitoredProc { + m := &MonitoredProc{parent: p, cmd: cmd, retrySecondsExponentBase: 2} + m.waitForExit = m.waitForProcessExit + return m +} + +func (p *MonitoredProc) Monitor() { + p.monitor = true + var err error + + for { + // wait for the given process to complete, unless the last retry had failed immediately + if err == nil { + p.waitForExit() + } + + if !p.monitor { + log.Printf("done monitoring process %v", p.cmd.Args) + break + } + + // calculate the delay + delaySeconds := calcRetryDelay(p.retrySecondsExponentBase, p.retries) + + log.Printf("starting process %v again after %.1f seconds", p.cmd.Args, delaySeconds) + <-time.After(time.Second * time.Duration(delaySeconds)) + + // initialize a new cmd object with the same arguments + p.cmd = exec.Command(p.cmd.Args[0], p.cmd.Args[1:]...) + + // start the process + err = p.parent.startChildProcessCmd(p.cmd) + if err != nil { + log.Printf("retry %d (total %d): process %v failed to restart. %v", p.retries, p.totalRetries, p.cmd.Args, err) + p.retries++ + } else { + log.Printf("retry (total %d). started process %v", p.totalRetries, p.cmd.Args) + p.retries = 0 + } + + p.totalRetries++ + } +} + +func (p *MonitoredProc) waitForProcessExit() { + state, err := p.cmd.Process.Wait() + if err != nil { + log.Printf("waiting for process %d had an error: %+v", p.cmd.Process.Pid, err) + return + } + + // check the wait status of the process which has all the exit information + waitStatus, ok := state.Sys().(syscall.WaitStatus) + if !ok { + log.Printf("unknown waitStatus for process %d: %+v", p.cmd.Process.Pid, state.Sys()) + return + } + + log.Printf("process %d completed. Exited: %t, ExitStatus: %d, Signaled: %t, Signal: %d", + p.cmd.Process.Pid, waitStatus.Exited(), waitStatus.ExitStatus(), waitStatus.Signaled(), waitStatus.Signal()) +} + +func (p *MonitoredProc) Stop() error { + p.monitor = false + return p.parent.stopChildProcess(p.cmd) +} + +func calcRetryDelay(base float64, power int) float64 { + if base == 0 { + return 0 + } + + delay := math.Pow(base, float64(power)) + if delay > maxDelaySeconds { + return maxDelaySeconds + } + + return delay +} diff --git a/pkg/proc/procmanager.go b/pkg/proc/procmanager.go index 88a9e5427..452fd61fd 100644 --- a/pkg/proc/procmanager.go +++ b/pkg/proc/procmanager.go @@ -12,16 +12,17 @@ import ( type ProcStartPolicy int const ( - StartAction = "start" - RunAction = "run" - StopAction = "stop" + StartAction = "start" + RunAction = "run" + StopAction = "stop" + RestartExisting ProcStartPolicy = iota ReuseExisting ) type ProcManager struct { sync.RWMutex - procs []*exec.Cmd + procs []*MonitoredProc // test override method to handle starting processes Trap func(action string, cmd *exec.Cmd) error @@ -41,7 +42,7 @@ func (p *ProcManager) Run(daemon string, args ...string) error { // with the given ProcStartPolicy. The search pattern will be used to search through the cmdline args of existing // processes to find any matching existing process. Therefore, it should be a regex pattern that can uniquely // identify the process (e.g., --id=1) -func (p *ProcManager) Start(daemon, procSearchPattern string, policy ProcStartPolicy, args ...string) (*exec.Cmd, error) { +func (p *ProcManager) Start(daemon, procSearchPattern string, policy ProcStartPolicy, args ...string) (*MonitoredProc, error) { // look for an existing process first shouldStart, err := p.checkProcessExists(os.Args[0], procSearchPattern, policy) if err != nil { @@ -59,37 +60,43 @@ func (p *ProcManager) Start(daemon, procSearchPattern string, policy ProcStartPo return nil, fmt.Errorf("failed to start daemon %s: %+v", daemon, err) } - // in a goroutine, wait for and monitor the process - go p.monitorProcess(process.Process) + proc := newMonitoredProc(p, process) + + // monitor the process if it was not mocked + if process != nil && process.Process != nil { + go proc.Monitor() + } p.Lock() - p.procs = append(p.procs, process) + p.procs = append(p.procs, proc) p.Unlock() - return process, nil + return proc, nil } -func (p *ProcManager) Stop(cmd *exec.Cmd) error { - if cmd == nil || cmd.Process == nil { - return nil +func (p *ProcManager) stopChildProcess(cmd *exec.Cmd) error { + + if p.Trap != nil { + // mock stopping the process + return p.Trap(StopAction, cmd) } pid := cmd.Process.Pid fmt.Printf("stopping child process %d\n", pid) - if err := p.stopChildProcess(cmd); err != nil { + if err := cmd.Process.Kill(); err != nil { fmt.Printf("failed to stop child process %d: %+v\n", pid, err) return err - } else { - fmt.Printf("child process %d stopped successfully\n", pid) } + fmt.Printf("child process %d stopped successfully\n", pid) return nil } func (p *ProcManager) Shutdown() { p.RLock() for _, proc := range p.procs { - p.Stop(proc) + proc.Stop() } + p.procs = nil p.RUnlock() } @@ -115,8 +122,9 @@ func (p *ProcManager) checkProcessExists(binary, procSearchPattern string, polic stopped := false p.RLock() for i := range p.procs { - if p.procs[i] != nil && p.procs[i].Process != nil && p.procs[i].Process.Pid == existingProc.pid { - if err := p.stopChildProcess(p.procs[i]); err != nil { + proc := p.procs[i].cmd + if proc != nil && proc.Process != nil && proc.Process.Pid == existingProc.pid { + if err := p.procs[i].Stop(); err != nil { log.Printf("failed to stop child process %d: %v", existingProc.pid, err) break } @@ -144,42 +152,13 @@ func (p *ProcManager) checkProcessExists(binary, procSearchPattern string, polic return true, nil } -func (p *ProcManager) monitorProcess(process *os.Process) { - if process == nil { - // the process was not started, the process was mocked - return - } - - // wait for the given process to complete - state, err := process.Wait() - if err != nil { - log.Printf("waiting for process %d had an error: %+v", process.Pid, err) - return - } - - // check the wait status of the process which has all the exit information - waitStatus, ok := state.Sys().(syscall.WaitStatus) - if !ok { - log.Printf("unknown waitStatus for process %d: %+v", process.Pid, state.Sys()) - return - } - - // TODO: improve the process monitoring here. We know the process has exited, if it was because - // of a crash then we should restart the process with the same arguments. - - log.Printf("process %d completed. Exited: %t, ExitStatus: %d, Signaled: %t, Signal: %d", - process.Pid, waitStatus.Exited(), waitStatus.ExitStatus(), waitStatus.Signaled(), waitStatus.Signal()) - - // find the managed process and remove it if it exists in our managed set - p.removeManagedProc(process.Pid) -} - func (p *ProcManager) removeManagedProc(pid int) { procNum := -1 p.Lock() for i := range p.procs { - if p.procs[i] != nil && p.procs[i].Process != nil && p.procs[i].Process.Pid == pid { + proc := p.procs[i].cmd + if proc != nil && proc.Process != nil && proc.Process.Pid == pid { procNum = i break } @@ -203,28 +182,19 @@ func (p *ProcManager) runChildProcess(daemon string, args ...string) error { return cmd.Run() } -func (p *ProcManager) startChildProcess(daemon string, args ...string) (cmd *exec.Cmd, err error) { - cmd = createCmd(daemon, args...) - if p.Trap != nil { - // mock starting the process - err := p.Trap(StartAction, cmd) - return cmd, err - } - - err = cmd.Start() - if err != nil { - return nil, err - } - return cmd, nil +func (p *ProcManager) startChildProcess(daemon string, args ...string) (*exec.Cmd, error) { + cmd := createCmd(daemon, args...) + err := p.startChildProcessCmd(cmd) + return cmd, err } -func (p *ProcManager) stopChildProcess(cmd *exec.Cmd) error { +func (p *ProcManager) startChildProcessCmd(cmd *exec.Cmd) (err error) { if p.Trap != nil { - // mock stopping the process - return p.Trap(StopAction, cmd) + // mock starting the process + return p.Trap(StartAction, cmd) } - return cmd.Process.Kill() + return cmd.Start() } func createCmd(daemon string, args ...string) (cmd *exec.Cmd) { diff --git a/pkg/util/proc/monitoredproc_test.go b/pkg/util/proc/monitoredproc_test.go new file mode 100644 index 000000000..36520a04e --- /dev/null +++ b/pkg/util/proc/monitoredproc_test.go @@ -0,0 +1,60 @@ +package proc + +import ( + "errors" + "log" + "os/exec" + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestRestartDelay(t *testing.T) { + assert.Equal(t, 0, calcRetryDelay(0, 0)) + assert.Equal(t, 0, calcRetryDelay(0, 2)) + assert.Equal(t, 1, calcRetryDelay(2, 0)) + assert.Equal(t, 2, calcRetryDelay(2, 1)) + assert.Equal(t, 4, calcRetryDelay(2, 2)) + assert.Equal(t, 30, calcRetryDelay(2, 5)) + assert.Equal(t, 30, calcRetryDelay(2, 100)) +} + +func TestMonitoredRestart(t *testing.T) { + procMgr := &ProcManager{} + cmd := &exec.Cmd{Args: []string{"/my/path", "1", "2", "3"}} + proc := newMonitoredProc(procMgr, cmd) + proc.retrySecondsExponentBase = 0.0 + + traps := 0 + procMgr.Trap = func(action string, cmd *exec.Cmd) error { + traps++ + switch { + case traps == 1: + return errors.New("test failure") + case traps == 2: + assert.Equal(t, 1, proc.retries) + break + } + return nil + } + + iter := 0 + proc.waitForExit = func() { + log.Printf("waited for process") + assert.True(t, proc.monitor) + switch { + case iter == 0: + case iter == 1: + assert.True(t, proc.monitor) + log.Printf("stop monitoring") + proc.monitor = false + break + } + iter++ + } + + proc.Monitor() + assert.False(t, proc.monitor) + assert.Equal(t, proc.retries, 0) + assert.Equal(t, proc.totalRetries, 2) +}