diff --git a/cmd/rookd/main.go b/cmd/rookd/main.go index 209a7ebca..2e6b75836 100644 --- a/cmd/rookd/main.go +++ b/cmd/rookd/main.go @@ -21,7 +21,6 @@ import ( "github.com/rook/rook/pkg/util" "github.com/rook/rook/pkg/util/flags" - "github.com/rook/rook/pkg/util/proc" ) var rootCmd = &cobra.Command{ @@ -123,11 +122,6 @@ func joinCluster() error { cephmgr.NewCephService(cephd.New(), cfg.devices, cfg.forceFormat, cfg.location), //etcdmgr.NewEtcdMgrService(cfg.discoveryURL), } - procMan := &proc.ProcManager{} - defer func() { - procMan.Shutdown() - <-time.After(time.Duration(1) * time.Second) - }() if cfg.nodeID == "" { // read /etc/machine-id @@ -139,11 +133,15 @@ func joinCluster() error { } // start the cluster orchestration services - context, err := clusterd.StartJoinCluster(services, procMan, cfg.dataDir, cfg.nodeID, cfg.discoveryURL, + context, err := clusterd.StartJoinCluster(services, cfg.dataDir, cfg.nodeID, cfg.discoveryURL, cfg.etcdMembers, cfg.publicIPv4, cfg.privateIPv4, cfg.debug) if err != nil { return err } + defer func() { + context.ProcMan.Shutdown() + <-time.After(time.Duration(1) * time.Second) + }() go func() { // set up routes and start HTTP server for REST API diff --git a/pkg/cephmgr/monagent_test.go b/pkg/cephmgr/monagent_test.go index 65f849755..3f094b60e 100644 --- a/pkg/cephmgr/monagent_test.go +++ b/pkg/cephmgr/monagent_test.go @@ -10,44 +10,55 @@ import ( testceph "github.com/rook/rook/pkg/cephmgr/client/test" "github.com/rook/rook/pkg/clusterd" "github.com/rook/rook/pkg/util" + "github.com/rook/rook/pkg/util/exec/test" "github.com/rook/rook/pkg/util/proc" "github.com/stretchr/testify/assert" ) func TestMonAgent(t *testing.T) { + executor := &test.MockExecutor{} - 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=rookcluster", 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/rookcluster.config", c.Args[8]) + runCommands := 0 + executor.MockExecuteCommand = func(name string, command string, args ...string) error { + log.Printf("RUN %d. %s %+v", runCommands, command, args) switch { - case commands == 0: - assert.Equal(t, proc.RunAction, action) - assert.Equal(t, "--mkfs", c.Args[4]) - assert.Equal(t, "--keyring=/tmp/mon0/keyring", c.Args[9]) - case commands == 1: - assert.Equal(t, proc.StartAction, action) - assert.Equal(t, "--foreground", c.Args[4]) - assert.Equal(t, "--public-addr=1.2.3.4:2345", c.Args[9]) - case commands == 2: - assert.Equal(t, proc.StopAction, action) + case runCommands == 0: + assert.Equal(t, "--mkfs", args[3]) default: - return fmt.Errorf("unexpected case %d", commands) + assert.Fail(t, fmt.Sprintf("unexpected case %d", runCommands)) } - commands++ + + runCommands++ return nil } + + startCommands := 0 + executor.MockStartExecuteCommand = func(name string, command string, args ...string) (*exec.Cmd, error) { + log.Printf("START %d. %s %+v", startCommands, command, args) + cmd := &exec.Cmd{Args: append([]string{command}, args...)} + assert.Equal(t, "daemon", args[0]) + assert.Equal(t, "--type=mon", args[1]) + assert.Equal(t, "--", args[2]) + assert.Equal(t, "--cluster=rookcluster", args[4]) + assert.Equal(t, "--name=mon.mon0", args[5]) + assert.Equal(t, "--mon-data=/tmp/mon0/mon.mon0", args[6]) + assert.Equal(t, "--conf=/tmp/mon0/rookcluster.config", args[7]) + switch { + case startCommands == 0: + assert.Equal(t, "--foreground", args[3]) + assert.Equal(t, "--public-addr=1.2.3.4:2345", args[8]) + default: + return cmd, fmt.Errorf("unexpected case %d", startCommands) + } + startCommands++ + return cmd, nil + } + etcdClient := util.NewMockEtcdClient() context := &clusterd.Context{ EtcdClient: etcdClient, NodeID: "a", - ProcMan: &proc.ProcManager{Trap: procTrap}, + ProcMan: proc.New(executor), ConfigDir: "/tmp", } @@ -60,7 +71,8 @@ func TestMonAgent(t *testing.T) { agent := &monAgent{} err = agent.ConfigureLocalService(context) assert.Nil(t, err) - assert.Equal(t, 0, commands) + assert.Equal(t, 0, runCommands) + assert.Equal(t, 0, startCommands) assert.Nil(t, agent.monProc) // set the agent in the desired state @@ -72,13 +84,15 @@ func TestMonAgent(t *testing.T) { // now the monitor will be configured err = agent.ConfigureLocalService(context) assert.Nil(t, err) - assert.Equal(t, 2, commands) + assert.Equal(t, 1, runCommands) + assert.Equal(t, 1, startCommands) 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, 3, commands) + assert.Equal(t, 1, runCommands) + assert.Equal(t, 1, startCommands) assert.Nil(t, agent.monProc) } diff --git a/pkg/cephmgr/osdagent_test.go b/pkg/cephmgr/osdagent_test.go index 33605c571..ea21838cd 100644 --- a/pkg/cephmgr/osdagent_test.go +++ b/pkg/cephmgr/osdagent_test.go @@ -28,20 +28,37 @@ func TestOSDAgentWithDevices(t *testing.T) { etcdClient, agent, _ := createTestAgent(t, nodeID, "sdx,sdy") - execCount := 0 + startCount := 0 executor := &exectest.MockExecutor{} + executor.MockStartExecuteCommand = func(name string, command string, args ...string) (*exec.Cmd, error) { + log.Printf("START %d for %s. %s %+v", startCount, name, command, args) + cmd := &exec.Cmd{Args: append([]string{command}, args...)} + + switch { + case startCount < 2: + assert.Equal(t, "--type=osd", args[1]) + default: + assert.Fail(t, fmt.Sprintf("unexpected case %d", startCount)) + } + startCount++ + return cmd, nil + } + + execCount := 0 executor.MockExecuteCommand = func(name string, command string, args ...string) error { - log.Printf("EXECUTE %d for %s. %s %+v", execCount, name, command, args) + log.Printf("RUN %d for %s. %s %+v", execCount, name, command, args) parts := strings.Split(name, " ") nameSuffix := parts[len(parts)-1] - assert.Equal(t, "sgdisk", command) switch { - case execCount == 0: - case execCount%2 == 0: + case execCount%3 == 0: + assert.Equal(t, "sgdisk", command) assert.Equal(t, "--zap-all", args[0]) assert.Equal(t, "/dev/"+nameSuffix, args[1]) - case execCount%2 == 1: + case execCount%3 == 1: assert.Equal(t, "/dev/"+nameSuffix, args[10]) + case execCount%3 == 2: + assert.Equal(t, "--mkfs", args[3]) + createTestKeyring(t, args) default: assert.Fail(t, fmt.Sprintf("unexpected case %d", execCount)) } @@ -50,18 +67,17 @@ func TestOSDAgentWithDevices(t *testing.T) { } outputExecCount := 0 executor.MockExecuteCommandWithOutput = func(name string, command string, args ...string) (string, error) { - log.Printf("OUTPUT EXECUTE %d for %s. %s %+v", outputExecCount, name, command, args) + log.Printf("OUTPUT %d for %s. %s %+v", outputExecCount, name, command, args) outputExecCount++ return "", nil } - procCommands := 0 context := &clusterd.Context{ EtcdClient: etcdClient, Executor: executor, NodeID: nodeID, ConfigDir: "/tmp", - ProcMan: &proc.ProcManager{Trap: createOSDAgentProcTrap(t, &procCommands, map[int]string{1: proc.StartAction, 3: proc.StartAction, 4: proc.StopAction, 5: proc.StopAction})}, + ProcMan: proc.New(executor), } // prep the etcd keys that would have been discovered by inventory @@ -80,9 +96,9 @@ func TestOSDAgentWithDevices(t *testing.T) { err := agent.ConfigureLocalService(context) assert.Nil(t, err) - assert.Equal(t, 4, execCount) + assert.Equal(t, 6, execCount) assert.Equal(t, 2, outputExecCount) - assert.Equal(t, 4, procCommands) + assert.Equal(t, 2, startCount) assert.Equal(t, 2, len(agent.osdProc), fmt.Sprintf("procs=%+v", agent.osdProc)) err = agent.DestroyLocalService(context) @@ -101,11 +117,19 @@ func TestOSDAgentNoDevices(t *testing.T) { nodeID := "abc" etcdClient, agent, _ := createTestAgent(t, nodeID, "") - // should be no executeCommand calls - execCount := 0 + startCount := 0 executor := &exectest.MockExecutor{} + executor.MockStartExecuteCommand = func(name string, command string, args ...string) (*exec.Cmd, error) { + startCount++ + cmd := &exec.Cmd{Args: append([]string{command}, args...)} + return cmd, nil + } + + // should be no executeCommand calls + runCount := 0 executor.MockExecuteCommand = func(name string, command string, args ...string) error { - execCount++ + runCount++ + createTestKeyring(t, args) return nil } @@ -118,12 +142,11 @@ func TestOSDAgentNoDevices(t *testing.T) { } // set up expected ProcManager commands - procCommands := 0 context := &clusterd.Context{ EtcdClient: etcdClient, Executor: executor, NodeID: nodeID, - ProcMan: &proc.ProcManager{Trap: createOSDAgentProcTrap(t, &procCommands, map[int]string{1: proc.StartAction, 2: proc.StopAction})}, + ProcMan: proc.New(executor), ConfigDir: "/tmp", } @@ -137,9 +160,9 @@ func TestOSDAgentNoDevices(t *testing.T) { err = agent.ConfigureLocalService(context) assert.Nil(t, err) - assert.Equal(t, 0, execCount) + assert.Equal(t, 1, runCount) + assert.Equal(t, 1, startCount) assert.Equal(t, 0, outputExecCount) - assert.Equal(t, 2, procCommands) assert.Equal(t, 1, len(agent.osdProc)) // the local device should be marked as an applied OSD now @@ -184,11 +207,8 @@ func TestRemoveDevice(t *testing.T) { nodeID := "a" etcdClient, agent, conn := createTestAgent(t, nodeID, "") executor := &exectest.MockExecutor{} - procTrap := func(action string, c *exec.Cmd) error { - return nil - } - context := &clusterd.Context{EtcdClient: etcdClient, NodeID: nodeID, Executor: executor, ProcMan: &proc.ProcManager{Trap: procTrap}} + context := &clusterd.Context{EtcdClient: etcdClient, NodeID: nodeID, Executor: executor, ProcMan: proc.New(executor)} etcdClient.SetValue("/rook/services/ceph/osd/desired/a/device/sda/osd-id", "23") // create two applied osds, one of which is desired @@ -243,48 +263,13 @@ func prepAgentOrchestrationData(t *testing.T, agent *osdAgent, etcdClient *util. etcdClient.SetValue(path.Join(monKey, "port"), "8743") } -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) - command := fmt.Sprintf("[%d] %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, command) - } else { - // the default action expected is Run - assert.Equal(t, proc.RunAction, action, fmt.Sprintf("command=%d, action=%s", *commands, action)) - } - - var configDir string - if len(c.Args) > 6 && strings.HasPrefix(c.Args[6], "--id") { - configDir = "/tmp/osd" + c.Args[6][5:] - err := os.MkdirAll(configDir, 0744) - assert.Nil(t, err) - err = ioutil.WriteFile(path.Join(configDir, "keyring"), []byte("mykeyring"), 0644) - assert.Nil(t, err) - } - - switch { - case *commands == 0: - assert.Equal(t, "--mkfs", c.Args[4], command) - assert.Equal(t, "--mkkey", c.Args[5], command) - case *commands == 1: - assert.Equal(t, "--foreground", c.Args[4], command) - case *commands == 2: - if action == proc.StopAction { - assert.Equal(t, "--foreground", c.Args[4], command) - } else { - assert.Equal(t, "--mkfs", c.Args[4], command) - } - case *commands >= 3 && *commands <= 5: - assert.Equal(t, "--foreground", c.Args[4], command) - default: - assert.Fail(t, "unexpected case %d. %s", *commands, command) - } - *commands++ - return nil +func createTestKeyring(t *testing.T, args []string) { + var configDir string + if len(args) > 5 && strings.HasPrefix(args[5], "--id") { + configDir = "/tmp/osd" + args[5][5:] + err := os.MkdirAll(configDir, 0744) + assert.Nil(t, err) + err = ioutil.WriteFile(path.Join(configDir, "keyring"), []byte("mykeyring"), 0644) + assert.Nil(t, err) } } diff --git a/pkg/clusterd/join.go b/pkg/clusterd/join.go index 413690f1f..932080f19 100644 --- a/pkg/clusterd/join.go +++ b/pkg/clusterd/join.go @@ -13,7 +13,7 @@ import ( ) // StartJoinCluster initializes the cluster services to enable joining the cluster and listening for orchestration. -func StartJoinCluster(services []*ClusterService, procMan *proc.ProcManager, configDir, nodeID, discoveryURL, +func StartJoinCluster(services []*ClusterService, configDir, nodeID, discoveryURL, etcdMembers, publicIPv4, privateIPv4 string, debug bool) (*Context, error) { log.Printf("Starting cluster. configDir=%s, nodeID=%s, url=%s, members=%s, publicIPv4=%s, privateIPv4=%s, debug=%t", @@ -53,12 +53,13 @@ func StartJoinCluster(services []*ClusterService, procMan *proc.ProcManager, con return nil, fmt.Errorf("failed to initialize lease manager: %+v", err) } + executor := &exec.CommandExecutor{} context := &Context{ EtcdClient: etcdClient, - Executor: &exec.CommandExecutor{}, + Executor: executor, NodeID: nodeID, Services: services, - ProcMan: procMan, + ProcMan: proc.New(executor), ConfigDir: configDir, Debug: debug, } diff --git a/pkg/util/exec/error.go b/pkg/util/exec/error.go new file mode 100644 index 000000000..c7e158716 --- /dev/null +++ b/pkg/util/exec/error.go @@ -0,0 +1,32 @@ +package exec + +import ( + "fmt" + "os/exec" + "syscall" +) + +type CommandError struct { + ActionName string + Err error +} + +func (e *CommandError) Error() string { + return fmt.Sprintf("Failed to complete %s: %+v", e.ActionName, e.Err) +} + +func (e *CommandError) ExitStatus() int { + exitStatus := -1 + exitErr, ok := e.Err.(*exec.ExitError) + if ok { + waitStatus, ok := exitErr.ProcessState.Sys().(syscall.WaitStatus) + if ok { + exitStatus = waitStatus.ExitStatus() + } + } + return exitStatus +} + +func createCommandError(err error, actionName string) error { + return &CommandError{ActionName: actionName, Err: err} +} diff --git a/pkg/util/exec/exec.go b/pkg/util/exec/exec.go index 64b10e703..e4c8fcd11 100644 --- a/pkg/util/exec/exec.go +++ b/pkg/util/exec/exec.go @@ -2,16 +2,14 @@ package exec import ( "bufio" - "errors" - "fmt" "io" "log" "os/exec" "strings" - "syscall" ) type Executor interface { + StartExecuteCommand(actionName string, command string, arg ...string) (*exec.Cmd, error) ExecuteCommand(actionName string, command string, arg ...string) error ExecuteCommandPipeline(actionName string, command string) (string, error) ExecuteCommandWithOutput(actionName string, command string, arg ...string) (string, error) @@ -20,23 +18,26 @@ type Executor interface { type CommandExecutor struct { } +// Start a process and return immediately +func (*CommandExecutor) StartExecuteCommand(actionName string, command string, arg ...string) (*exec.Cmd, error) { + cmd, stdout, stderr, err := startCommand(command, arg...) + if err != nil { + return cmd, createCommandError(err, actionName) + } + + go logOutput(stdout, stderr) + + return cmd, nil +} + +// Start a process and wait for its completion func (*CommandExecutor) ExecuteCommand(actionName string, command string, arg ...string) error { - cmd := exec.Command(command, arg...) - stdout, _ := cmd.StdoutPipe() - stderr, _ := cmd.StderrPipe() - - logCommand(command, arg...) - if err := cmd.Start(); err != nil { - return errors.New(fmt.Sprintf("Failed to start action '%s': %+v", actionName, err)) + cmd, stdout, stderr, err := startCommand(command, arg...) + if err != nil { + return createCommandError(err, actionName) } - // read command's stdout line by line and write it to the log - in := bufio.NewScanner(io.MultiReader(stdout, stderr)) - lastLine := "" - for in.Scan() { - lastLine = in.Text() - log.Printf(lastLine) - } + logOutput(stdout, stderr) if err := cmd.Wait(); err != nil { return createCommandError(err, actionName) @@ -46,18 +47,44 @@ func (*CommandExecutor) ExecuteCommand(actionName string, command string, arg .. } func (*CommandExecutor) ExecuteCommandWithOutput(actionName string, command string, arg ...string) (string, error) { - cmd := exec.Command(command, arg...) logCommand(command, arg...) + cmd := exec.Command(command, arg...) return runCommandWithOutput(actionName, cmd) } func (*CommandExecutor) ExecuteCommandPipeline(actionName string, command string) (string, error) { - cmd := exec.Command("bash", "-c", command) - logCommand(command) + cmd := exec.Command("bash", "-c", command) return runCommandWithOutput(actionName, cmd) } +func startCommand(command string, arg ...string) (*exec.Cmd, io.ReadCloser, io.ReadCloser, error) { + logCommand(command, arg...) + + cmd := exec.Command(command, arg...) + stdout, _ := cmd.StdoutPipe() + stderr, _ := cmd.StderrPipe() + + err := cmd.Start() + + return cmd, stdout, stderr, err +} + +func logOutput(stdout, stderr io.ReadCloser) { + if stdout == nil || stderr == nil { + log.Printf("failed to collect stdout and stderr") + return + } + + // read command's stdout line by line and write it to the log + in := bufio.NewScanner(io.MultiReader(stdout, stderr)) + lastLine := "" + for in.Scan() { + lastLine = in.Text() + log.Printf(lastLine) + } +} + func runCommandWithOutput(actionName string, cmd *exec.Cmd) (string, error) { output, err := cmd.Output() if err != nil { @@ -70,28 +97,3 @@ func runCommandWithOutput(actionName string, cmd *exec.Cmd) (string, error) { func logCommand(command string, arg ...string) { log.Printf("Running command: %s %s", command, strings.Join(arg, " ")) } - -type CommandError struct { - ActionName string - Err error -} - -func (e *CommandError) Error() string { - return fmt.Sprintf("Failed to complete %s: %+v", e.ActionName, e.Err) -} - -func (e *CommandError) ExitStatus() int { - exitStatus := -1 - exitErr, ok := e.Err.(*exec.ExitError) - if ok { - waitStatus, ok := exitErr.ProcessState.Sys().(syscall.WaitStatus) - if ok { - exitStatus = waitStatus.ExitStatus() - } - } - return exitStatus -} - -func createCommandError(err error, actionName string) error { - return &CommandError{ActionName: actionName, Err: err} -} diff --git a/pkg/util/exec/test/mockexec.go b/pkg/util/exec/test/mockexec.go index d9dbdeaff..bb387f568 100644 --- a/pkg/util/exec/test/mockexec.go +++ b/pkg/util/exec/test/mockexec.go @@ -1,8 +1,11 @@ package test +import "os/exec" + // ******************** MockExecutor ******************** type MockExecutor struct { MockExecuteCommand func(actionName string, command string, arg ...string) error + MockStartExecuteCommand func(actionName string, command string, arg ...string) (*exec.Cmd, error) MockExecuteCommandPipeline func(actionName string, command string) (string, error) MockExecuteCommandWithOutput func(actionName string, command string, arg ...string) (string, error) } @@ -15,6 +18,14 @@ func (e *MockExecutor) ExecuteCommand(actionName string, command string, arg ... return nil } +func (e *MockExecutor) StartExecuteCommand(actionName string, command string, arg ...string) (*exec.Cmd, error) { + if e.MockStartExecuteCommand != nil { + return e.MockStartExecuteCommand(actionName, command, arg...) + } + + return &exec.Cmd{}, nil +} + func (e *MockExecutor) ExecuteCommandPipeline(actionName string, command string) (string, error) { if e.MockExecuteCommandPipeline != nil { return e.MockExecuteCommandPipeline(actionName, command) diff --git a/pkg/util/proc/monitoredproc.go b/pkg/util/proc/monitoredproc.go index 7dd131106..6f792b5e1 100644 --- a/pkg/util/proc/monitoredproc.go +++ b/pkg/util/proc/monitoredproc.go @@ -1,6 +1,7 @@ package proc import ( + "fmt" "log" "math" "os/exec" @@ -49,11 +50,8 @@ func (p *MonitoredProc) Monitor() { 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) + p.cmd, err = p.parent.executor.StartExecuteCommand("restart", p.cmd.Args[0], p.cmd.Args[1:]...) 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++ @@ -86,7 +84,19 @@ func (p *MonitoredProc) waitForProcessExit() { func (p *MonitoredProc) Stop() error { p.monitor = false - return p.parent.stopChildProcess(p.cmd) + if p.cmd == nil || p.cmd.Process == nil { + return nil + } + + pid := p.cmd.Process.Pid + fmt.Printf("stopping child process %d\n", pid) + if err := p.cmd.Process.Kill(); err != nil { + fmt.Printf("failed to stop child process %d: %+v\n", pid, err) + return err + } + + fmt.Printf("child process %d stopped successfully\n", pid) + return nil } func calcRetryDelay(base float64, power int) float64 { diff --git a/pkg/util/proc/monitoredproc_test.go b/pkg/util/proc/monitoredproc_test.go index 36520a04e..4cd84511a 100644 --- a/pkg/util/proc/monitoredproc_test.go +++ b/pkg/util/proc/monitoredproc_test.go @@ -6,6 +6,7 @@ import ( "os/exec" "testing" + "github.com/rook/rook/pkg/util/exec/test" "github.com/stretchr/testify/assert" ) @@ -20,22 +21,24 @@ func TestRestartDelay(t *testing.T) { } func TestMonitoredRestart(t *testing.T) { - procMgr := &ProcManager{} + executor := &test.MockExecutor{} + procMgr := New(executor) 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++ + commands := 0 + executor.MockStartExecuteCommand = func(name string, command string, args ...string) (*exec.Cmd, error) { + commands++ + cmd := &exec.Cmd{Args: append([]string{command}, args...)} switch { - case traps == 1: - return errors.New("test failure") - case traps == 2: + case commands == 1: + return cmd, errors.New("test failure") + case commands == 2: assert.Equal(t, 1, proc.retries) break } - return nil + return cmd, nil } iter := 0 diff --git a/pkg/util/proc/procmanager.go b/pkg/util/proc/procmanager.go index 0ca9948c3..70184e9ff 100644 --- a/pkg/util/proc/procmanager.go +++ b/pkg/util/proc/procmanager.go @@ -4,9 +4,10 @@ import ( "fmt" "log" "os" - "os/exec" "sync" "syscall" + + "github.com/rook/rook/pkg/util/exec" ) type ProcStartPolicy int @@ -22,16 +23,20 @@ const ( type ProcManager struct { sync.RWMutex - procs []*MonitoredProc - - // test override method to handle starting processes - Trap func(action string, cmd *exec.Cmd) error + procs []*MonitoredProc + executor exec.Executor } +// Create a new proc manager +func New(executor exec.Executor) *ProcManager { + return &ProcManager{executor: executor} +} + +// Start a child process and wait for its completion func (p *ProcManager) Run(daemon string, args ...string) error { log.Printf("Running process %s with args: %v", daemon, args) - err := p.runChildProcess(daemon, args...) + err := p.executor.ExecuteCommand("daemon "+daemon, os.Args[0], createDaemonArgs(daemon, args...)...) if err != nil { return fmt.Errorf("failed to run %s: %+v", daemon, err) } @@ -56,15 +61,15 @@ func (p *ProcManager) Start(daemon, procSearchPattern string, policy ProcStartPo } log.Printf("Starting process %s with args: %v", daemon, args) - process, err := p.startChildProcess(daemon, args...) + cmd, err := p.executor.StartExecuteCommand("daemon "+daemon, os.Args[0], createDaemonArgs(daemon, args...)...) if err != nil { return nil, fmt.Errorf("failed to start daemon %s: %+v", daemon, err) } - proc := newMonitoredProc(p, process) + proc := newMonitoredProc(p, cmd) // monitor the process if it was not mocked - if process != nil && process.Process != nil { + if cmd != nil && cmd.Process != nil { go proc.Monitor() } @@ -74,24 +79,6 @@ func (p *ProcManager) Start(daemon, procSearchPattern string, policy ProcStartPo return proc, 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 := cmd.Process.Kill(); err != nil { - fmt.Printf("failed to stop child process %d: %+v\n", pid, err) - return err - } - - fmt.Printf("child process %d stopped successfully\n", pid) - return nil -} - func (p *ProcManager) Shutdown() { p.RLock() for _, proc := range p.procs { @@ -173,37 +160,8 @@ func (p *ProcManager) removeManagedProc(pid int) { p.Unlock() } -func (p *ProcManager) runChildProcess(daemon string, args ...string) error { - cmd := createCmd(daemon, args...) - if p.Trap != nil { - // mock running the process - return p.Trap(RunAction, cmd) - } - - return cmd.Run() -} - -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) startChildProcessCmd(cmd *exec.Cmd) (err error) { - if p.Trap != nil { - // mock starting the process - return p.Trap(StartAction, cmd) - } - - return cmd.Start() -} - -func createCmd(daemon string, args ...string) *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 createDaemonArgs(daemon string, args ...string) []string { + return append( + []string{"daemon", fmt.Sprintf("--type=%s", daemon), "--"}, + args...) }