API: add sandbox waitprocess api

It waits a process inside the container of a sandbox.

Signed-off-by: Peng Tao <bergwolf@gmail.com>

virtcontainers/agent.go

@@ -168,6 +168,9 @@ type agent interface {
 	// processListContainer will list the processes running inside the container
 	processListContainer(sandbox *Sandbox, c Container, options ProcessListOptions) (ProcessList, error)
 
+	// waitProcess will wait for the exit code of a process
+	waitProcess(c *Container, processID string) (int32, error)
+
 	// onlineCPUMem will online CPUs and Memory inside the Sandbox.
 	// This function should be called after hot adding vCPUs or Memory.
 	// cpus specifies the number of CPUs that were added and the agent should online

virtcontainers/container.go

@@ -724,6 +724,16 @@ func (c *Container) enter(cmd Cmd) (*Process, error) {
 	return process, nil
 }
 
+func (c *Container) wait(processID string) (int32, error) {
+	if c.state.State != StateReady &&
+		c.state.State != StateRunning {
+		return 0, fmt.Errorf("Container not ready or running, " +
+			"impossible to wait")
+	}
+
+	return c.sandbox.agent.waitProcess(c, processID)
+}
+
 func (c *Container) kill(signal syscall.Signal, all bool) error {
 	if c.sandbox.state.State != StateReady && c.sandbox.state.State != StateRunning {
 		return fmt.Errorf("Sandbox not ready or running, impossible to signal the container")

virtcontainers/container_test.go

@@ -359,3 +359,29 @@ func TestContainerEnterErrorsOnContainerStates(t *testing.T) {
 	_, err = c.enter(cmd)
 	assert.Error(err)
 }
+
+func TestContainerWaitErrorState(t *testing.T) {
+	assert := assert.New(t)
+	c := &Container{
+		sandbox: &Sandbox{
+			state: State{
+				State: StateRunning,
+			},
+		},
+	}
+	processID := "foobar"
+
+	// Container state undefined
+	_, err := c.wait(processID)
+	assert.Error(err)
+
+	// Container paused
+	c.state.State = StatePaused
+	_, err = c.wait(processID)
+	assert.Error(err)
+
+	// Container stopped
+	c.state.State = StateStopped
+	_, err = c.wait(processID)
+	assert.Error(err)
+}

virtcontainers/hyperstart_agent.go

@@ -806,3 +806,8 @@ func (h *hyper) check() error {
 	// cc-agent does not support check
 	return nil
 }
+
+func (h *hyper) waitProcess(c *Container, processID string) (int32, error) {
+	// cc-agent does not support wait process
+	return 0, nil
+}

virtcontainers/interfaces.go

@@ -57,6 +57,7 @@ type VCSandbox interface {
 	StartContainer(containerID string) (VCContainer, error)
 	StatusContainer(containerID string) (ContainerStatus, error)
 	EnterContainer(containerID string, cmd Cmd) (VCContainer, *Process, error)
+	WaitProcess(containerID, processID string) (int32, error)
 }
 
 // VCContainer is the Container interface

virtcontainers/kata_agent.go

@@ -937,6 +937,18 @@ func (k *kataAgent) check() error {
 	return err
 }
 
+func (k *kataAgent) waitProcess(c *Container, processID string) (int32, error) {
+	resp, err := k.sendReq(&grpc.WaitProcessRequest{
+		ContainerId: c.id,
+		ExecId:      processID,
+	})
+	if err != nil {
+		return 0, err
+	}
+
+	return resp.(*grpc.WaitProcessResponse).Status, nil
+}
+
 type reqFunc func(context.Context, interface{}, ...golangGrpc.CallOption) (interface{}, error)
 
 func (k *kataAgent) installReqFunc(c *kataclient.AgentClient) {
@@ -979,6 +991,9 @@ func (k *kataAgent) installReqFunc(c *kataclient.AgentClient) {
 	k.reqHandlers["grpc.ListProcessesRequest"] = func(ctx context.Context, req interface{}, opts ...golangGrpc.CallOption) (interface{}, error) {
 		return k.client.ListProcesses(ctx, req.(*grpc.ListProcessesRequest), opts...)
 	}
+	k.reqHandlers["grpc.WaitProcessRequest"] = func(ctx context.Context, req interface{}, opts ...golangGrpc.CallOption) (interface{}, error) {
+		return k.client.WaitProcess(ctx, req.(*grpc.WaitProcessRequest), opts...)
+	}
 }
 
 func (k *kataAgent) sendReq(request interface{}) (interface{}, error) {
@@ -992,7 +1007,7 @@ func (k *kataAgent) sendReq(request interface{}) (interface{}, error) {
 	msgName := proto.MessageName(request.(proto.Message))
 	handler := k.reqHandlers[msgName]
 	if msgName == "" || handler == nil {
-		return nil, fmt.Errorf("Invalid request type")
+		return nil, errors.New("Invalid request type")
 	}
 
 	return handler(context.Background(), request)

virtcontainers/kata_agent_test.go

@@ -217,6 +217,7 @@ var reqList = []interface{}{
 	&pb.RemoveContainerRequest{},
 	&pb.SignalProcessRequest{},
 	&pb.CheckRequest{},
+	&pb.WaitProcessRequest{},
 }
 
 func TestKataAgentSendReq(t *testing.T) {

virtcontainers/noop_agent.go

@@ -83,3 +83,8 @@ func (n *noopAgent) onlineCPUMem(cpus uint32) error {
 func (n *noopAgent) check() error {
 	return nil
 }
+
+// waitProcess is the Noop agent process waiter. It does nothing.
+func (n *noopAgent) waitProcess(c *Container, processID string) (int32, error) {
+	return 0, nil
+}

virtcontainers/pkg/vcmock/sandbox.go

@@ -104,3 +104,8 @@ func (p *Sandbox) EnterContainer(containerID string, cmd vc.Cmd) (vc.VCContainer
 func (p *Sandbox) Monitor() (chan error, error) {
 	return nil, nil
 }
+
+// WaitProcess implements the VCSandbox function of the same name.
+func (p *Sandbox) WaitProcess(containerID, processID string) (int32, error) {
+	return 0, nil
+}

virtcontainers/sandbox.go

@@ -587,6 +587,20 @@ func (s *Sandbox) Monitor() (chan error, error) {
 	return s.monitor.newWatcher()
 }
 
+// WaitProcess waits on a container process and return its exit code
+func (s *Sandbox) WaitProcess(containerID, processID string) (int32, error) {
+	if s.state.State != StateRunning {
+		return 0, fmt.Errorf("Sandbox not running")
+	}
+
+	c, err := s.findContainer(containerID)
+	if err != nil {
+		return 0, err
+	}
+
+	return c.wait(processID)
+}
+
 func createAssets(sandboxConfig *SandboxConfig) error {
 	kernel, err := newAsset(sandboxConfig, kernelAsset)
 	if err != nil {

virtcontainers/sandbox_test.go

@@ -1408,3 +1408,33 @@ func TestMonitor(t *testing.T) {
 
 	s.monitor.stop()
 }
+
+func TestWaitProcess(t *testing.T) {
+	s, err := testCreateSandbox(t, testSandboxID, MockHypervisor, newHypervisorConfig(nil, nil), NoopAgentType, NoopNetworkModel, NetworkConfig{}, nil, nil)
+	assert.Nil(t, err, "VirtContainers should not allow empty sandboxes")
+	defer cleanUp()
+
+	contID := "foo"
+	execID := "bar"
+	_, err = s.WaitProcess(contID, execID)
+	assert.NotNil(t, err, "Wait process in stopped sandbox should fail")
+
+	err = s.start()
+	assert.Nil(t, err, "Failed to start sandbox: %v", err)
+
+	_, err = s.WaitProcess(contID, execID)
+	assert.NotNil(t, err, "Wait process in non-existing container should fail")
+
+	contConfig := newTestContainerConfigNoop(contID)
+	_, err = s.CreateContainer(contConfig)
+	assert.Nil(t, err, "Failed to create container %+v in sandbox %+v: %v", contConfig, s, err)
+
+	_, err = s.WaitProcess(contID, execID)
+	assert.Nil(t, err, "Wait process in ready container failed: %v", err)
+
+	_, err = s.StartContainer(contID)
+	assert.Nil(t, err, "Start container failed: %v", err)
+
+	_, err = s.WaitProcess(contID, execID)
+	assert.Nil(t, err, "Wait process failed: %v", err)
+}