Merge pull request #259 from bergwolf/sandbox_api_2

add sandbox process operation relay API support

virtcontainers/agent.go

@@ -160,14 +160,32 @@ type agent interface {
 	// stopContainer will tell the agent to stop a container related to a Sandbox.
 	stopContainer(sandbox *Sandbox, c Container) error
 
-	// killContainer will tell the agent to send a signal to a
-	// container related to a Sandbox. If all is true, all processes in
+	// signalProcess will tell the agent to send a signal to a
+	// container or a process related to a Sandbox. If all is true, all processes in
 	// the container will be sent the signal.
-	killContainer(sandbox *Sandbox, c Container, signal syscall.Signal, all bool) error
+	signalProcess(c *Container, processID string, signal syscall.Signal, all bool) error
+
+	// winsizeProcess will tell the agent to set a process' tty size
+	winsizeProcess(c *Container, processID string, height, width uint32) error
+
+	// writeProcessStdin will tell the agent to write a process stdin
+	writeProcessStdin(c *Container, ProcessID string, data []byte) (int, error)
+
+	// closeProcessStdin will tell the agent to close a process stdin
+	closeProcessStdin(c *Container, ProcessID string) error
+
+	// readProcessStdout will tell the agent to read a process stdout
+	readProcessStdout(c *Container, processID string, data []byte) (int, error)
+
+	// readProcessStderr will tell the agent to read a process stderr
+	readProcessStderr(c *Container, processID string, data []byte) (int, error)
 
 	// 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

@@ -8,6 +8,7 @@ package virtcontainers
 import (
 	"encoding/hex"
 	"fmt"
+	"io"
 	"os"
 	"path/filepath"
 	"syscall"
@@ -663,7 +664,7 @@ func (c *Container) stop() error {
 	// return an error, but instead try to kill it forcefully.
 	if err := waitForShim(c.process.Pid); err != nil {
 		// Force the container to be killed.
-		if err := c.sandbox.agent.killContainer(c.sandbox, *c, syscall.SIGKILL, true); err != nil {
+		if err := c.kill(syscall.SIGKILL, true); err != nil {
 			return err
 		}
 
@@ -684,7 +685,7 @@ func (c *Container) stop() error {
 	// this signal will ensure the container will get killed to match
 	// the state of the shim. This will allow the following call to
 	// stopContainer() to succeed in such particular case.
-	c.sandbox.agent.killContainer(c.sandbox, *c, syscall.SIGKILL, true)
+	c.kill(syscall.SIGKILL, true)
 
 	if err := c.sandbox.agent.stopContainer(c.sandbox, *c); err != nil {
 		return err
@@ -724,7 +725,21 @@ 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 {
+	return c.signalProcess(c.process.Token, signal, all)
+}
+
+func (c *Container) signalProcess(processID string, 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")
 	}
@@ -733,7 +748,25 @@ func (c *Container) kill(signal syscall.Signal, all bool) error {
 		return fmt.Errorf("Container not ready or running, impossible to signal the container")
 	}
 
-	return c.sandbox.agent.killContainer(c.sandbox, *c, signal, all)
+	return c.sandbox.agent.signalProcess(c, processID, signal, all)
+}
+
+func (c *Container) winsizeProcess(processID string, height, width uint32) error {
+	if c.state.State != StateReady && c.state.State != StateRunning {
+		return fmt.Errorf("Container not ready or running, impossible to signal the container")
+	}
+
+	return c.sandbox.agent.winsizeProcess(c, processID, height, width)
+}
+
+func (c *Container) ioStream(processID string) (io.WriteCloser, io.Reader, io.Reader, error) {
+	if c.state.State != StateReady && c.state.State != StateRunning {
+		return nil, nil, nil, fmt.Errorf("Container not ready or running, impossible to signal the container")
+	}
+
+	stream := newIOStream(c.sandbox, c, processID)
+
+	return stream.stdin(), stream.stdout(), stream.stderr(), nil
 }
 
 func (c *Container) processList(options ProcessListOptions) (ProcessList, error) {

virtcontainers/container_test.go

@@ -359,3 +359,105 @@ 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)
+}
+
+func TestKillContainerErrorState(t *testing.T) {
+	assert := assert.New(t)
+	c := &Container{
+		sandbox: &Sandbox{
+			state: State{
+				State: StateRunning,
+			},
+		},
+	}
+	// Container state undefined
+	err := c.kill(syscall.SIGKILL, true)
+	assert.Error(err)
+
+	// Container paused
+	c.state.State = StatePaused
+	err = c.kill(syscall.SIGKILL, false)
+	assert.Error(err)
+
+	// Container stopped
+	c.state.State = StateStopped
+	err = c.kill(syscall.SIGKILL, true)
+	assert.Error(err)
+}
+
+func TestWinsizeProcessErrorState(t *testing.T) {
+	assert := assert.New(t)
+	c := &Container{
+		sandbox: &Sandbox{
+			state: State{
+				State: StateRunning,
+			},
+		},
+	}
+	processID := "foobar"
+
+	// Container state undefined
+	err := c.winsizeProcess(processID, 100, 200)
+	assert.Error(err)
+
+	// Container paused
+	c.state.State = StatePaused
+	err = c.winsizeProcess(processID, 100, 200)
+	assert.Error(err)
+
+	// Container stopped
+	c.state.State = StateStopped
+	err = c.winsizeProcess(processID, 100, 200)
+	assert.Error(err)
+}
+
+func TestProcessIOStream(t *testing.T) {
+	assert := assert.New(t)
+	c := &Container{
+		sandbox: &Sandbox{
+			state: State{
+				State: StateRunning,
+			},
+		},
+	}
+	processID := "foobar"
+
+	// Container state undefined
+	_, _, _, err := c.ioStream(processID)
+	assert.Error(err)
+
+	// Container paused
+	c.state.State = StatePaused
+	_, _, _, err = c.ioStream(processID)
+	assert.Error(err)
+
+	// Container stopped
+	c.state.State = StateStopped
+	_, _, _, err = c.ioStream(processID)
+	assert.Error(err)
+}

virtcontainers/hyperstart_agent.go

@@ -571,8 +571,8 @@ func (h *hyper) stopOneContainer(sandboxID string, c Container) error {
 	return nil
 }
 
-// killContainer is the agent process signal implementation for hyperstart.
-func (h *hyper) killContainer(sandbox *Sandbox, c Container, signal syscall.Signal, all bool) error {
+// signalProcess is the agent process signal implementation for hyperstart.
+func (h *hyper) signalProcess(c *Container, processID string, signal syscall.Signal, all bool) error {
 	// Send the signal to the shim directly in case the container has not
 	// been started yet.
 	if c.state.State == StateReady {
@@ -798,11 +798,41 @@ func (h *hyper) sendCmd(proxyCmd hyperstartProxyCmd) (interface{}, error) {
 }
 
 func (h *hyper) onlineCPUMem(cpus uint32) error {
-	// cc-agent uses udev to online CPUs automatically
+	// hyperstart-agent uses udev to online CPUs automatically
 	return nil
 }
 
 func (h *hyper) check() error {
-	// cc-agent does not support check
+	// hyperstart-agent does not support check
 	return nil
 }
+
+func (h *hyper) waitProcess(c *Container, processID string) (int32, error) {
+	// hyperstart-agent does not support wait process
+	return 0, nil
+}
+
+func (h *hyper) winsizeProcess(c *Container, processID string, height, width uint32) error {
+	// hyperstart-agent does not support winsize process
+	return nil
+}
+
+func (h *hyper) writeProcessStdin(c *Container, ProcessID string, data []byte) (int, error) {
+	// hyperstart-agent does not support stdin write request
+	return 0, nil
+}
+
+func (h *hyper) closeProcessStdin(c *Container, ProcessID string) error {
+	// hyperstart-agent does not support stdin close request
+	return nil
+}
+
+func (h *hyper) readProcessStdout(c *Container, processID string, data []byte) (int, error) {
+	// hyperstart-agent does not support stdout read request
+	return 0, nil
+}
+
+func (h *hyper) readProcessStderr(c *Container, processID string, data []byte) (int, error) {
+	// hyperstart-agent does not support stderr read request
+	return 0, nil
+}

virtcontainers/interfaces.go

@@ -6,6 +6,7 @@
 package virtcontainers
 
 import (
+	"io"
 	"syscall"
 
 	"github.com/sirupsen/logrus"
@@ -57,6 +58,10 @@ 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)
+	SignalProcess(containerID, processID string, signal syscall.Signal, all bool) error
+	WinsizeProcess(containerID, processID string, height, width uint32) error
+	IOStream(containerID, processID string) (io.WriteCloser, io.Reader, io.Reader, error)
 }
 
 // VCContainer is the Container interface

virtcontainers/iostream.go

@@ -0,0 +1,91 @@
+// Copyright (c) 2018 HyperHQ Inc.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+
+package virtcontainers
+
+import (
+	"errors"
+	"io"
+)
+
+type iostream struct {
+	sandbox   *Sandbox
+	container *Container
+	process   string
+	closed    bool
+}
+
+// io.WriteCloser
+type stdinStream struct {
+	*iostream
+}
+
+// io.Reader
+type stdoutStream struct {
+	*iostream
+}
+
+// io.Reader
+type stderrStream struct {
+	*iostream
+}
+
+func newIOStream(s *Sandbox, c *Container, proc string) *iostream {
+	return &iostream{
+		sandbox:   s,
+		container: c,
+		process:   proc,
+		closed:    false, // needed to workaround buggy structcheck
+	}
+}
+
+func (s *iostream) stdin() io.WriteCloser {
+	return &stdinStream{s}
+}
+
+func (s *iostream) stdout() io.Reader {
+	return &stdoutStream{s}
+}
+
+func (s *iostream) stderr() io.Reader {
+	return &stderrStream{s}
+}
+
+func (s *stdinStream) Write(data []byte) (n int, err error) {
+	if s.closed {
+		return 0, errors.New("stream closed")
+	}
+
+	return s.sandbox.agent.writeProcessStdin(s.container, s.process, data)
+}
+
+func (s *stdinStream) Close() error {
+	if s.closed {
+		return errors.New("stream closed")
+	}
+
+	err := s.sandbox.agent.closeProcessStdin(s.container, s.process)
+	if err == nil {
+		s.closed = true
+	}
+
+	return err
+}
+
+func (s *stdoutStream) Read(data []byte) (n int, err error) {
+	if s.closed {
+		return 0, errors.New("stream closed")
+	}
+
+	return s.sandbox.agent.readProcessStdout(s.container, s.process, data)
+}
+
+func (s *stderrStream) Read(data []byte) (n int, err error) {
+	if s.closed {
+		return 0, errors.New("stream closed")
+	}
+
+	return s.sandbox.agent.readProcessStderr(s.container, s.process, data)
+}