Add ctr exec

Signed-off-by: Michael Crosby <crosbymichael@gmail.com>

cmd/ctr/exec.go

@@ -0,0 +1,128 @@
+package main
+
+import (
+	gocontext "context"
+	"encoding/json"
+	"os"
+	"path/filepath"
+
+	"github.com/containerd/containerd/api/services/execution"
+	"github.com/crosbymichael/console"
+	protobuf "github.com/gogo/protobuf/types"
+	specs "github.com/opencontainers/runtime-spec/specs-go"
+	"github.com/urfave/cli"
+)
+
+var execCommand = cli.Command{
+	Name:  "exec",
+	Usage: "execute additional processes in an existing container",
+	Flags: []cli.Flag{
+		cli.StringFlag{
+			Name:  "id",
+			Usage: "id of the container",
+		},
+		cli.BoolFlag{
+			Name:  "tty,t",
+			Usage: "allocate a TTY for the container",
+		},
+	},
+	Action: func(context *cli.Context) error {
+		var (
+			id  = context.String("id")
+			ctx = gocontext.Background()
+		)
+
+		process := createProcess(context.Args(), "", context.Bool("tty"))
+		data, err := json.Marshal(process)
+		if err != nil {
+			return err
+		}
+		containers, err := getExecutionService(context)
+		if err != nil {
+			return err
+		}
+		events, err := containers.Events(ctx, &execution.EventsRequest{})
+		if err != nil {
+			return err
+		}
+		tmpDir, err := getTempDir(id)
+		if err != nil {
+			return err
+		}
+		defer os.RemoveAll(tmpDir)
+		request := &execution.ExecRequest{
+			ID: id,
+			Spec: &protobuf.Any{
+				TypeUrl: specs.Version,
+				Value:   data,
+			},
+			Terminal: context.Bool("tty"),
+			Stdin:    filepath.Join(tmpDir, "stdin"),
+			Stdout:   filepath.Join(tmpDir, "stdout"),
+			Stderr:   filepath.Join(tmpDir, "stderr"),
+		}
+		if request.Terminal {
+			con := console.Current()
+			defer con.Reset()
+			if err := con.SetRaw(); err != nil {
+				return err
+			}
+		}
+		fwg, err := prepareStdio(request.Stdin, request.Stdout, request.Stderr, request.Terminal)
+		if err != nil {
+			return err
+		}
+		response, err := containers.Exec(ctx, request)
+		if err != nil {
+			return err
+		}
+		// Ensure we read all io only if container started successfully.
+		defer fwg.Wait()
+
+		status, err := waitContainer(events, id, response.Pid)
+		if err != nil {
+			return err
+		}
+		if status != 0 {
+			return cli.NewExitError("", int(status))
+		}
+		return nil
+	},
+}
+
+func createProcess(args []string, cwd string, tty bool) specs.Process {
+	env := []string{
+		"PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin",
+	}
+	if tty {
+		env = append(env, "TERM=xterm")
+	}
+	if cwd == "" {
+		cwd = "/"
+	}
+	return specs.Process{
+		Args:            args,
+		Env:             env,
+		Terminal:        tty,
+		Cwd:             cwd,
+		NoNewPrivileges: true,
+		User: specs.User{
+			UID: 0,
+			GID: 0,
+		},
+		Capabilities: &specs.LinuxCapabilities{
+			Bounding:    capabilities,
+			Permitted:   capabilities,
+			Inheritable: capabilities,
+			Effective:   capabilities,
+			Ambient:     capabilities,
+		},
+		Rlimits: []specs.LinuxRlimit{
+			{
+				Type: "RLIMIT_NOFILE",
+				Hard: uint64(1024),
+				Soft: uint64(1024),
+			},
+		},
+	}
+}

cmd/ctr/main.go

@@ -55,6 +55,7 @@ containerd client
 		killCommand,
 		shimCommand,
 		pprofCommand,
+		execCommand,
 	}
 	app.Before = func(context *cli.Context) error {
 		if context.GlobalBool("debug") {

cmd/ctr/run.go

@@ -377,7 +377,7 @@ var runCommand = cli.Command{
 		// Ensure we read all io only if container started successfully.
 		defer fwg.Wait()
 
-		status, err := waitContainer(events, response)
+		status, err := waitContainer(events, response.ID, response.Pid)
 		if err != nil {
 			return err
 		}

cmd/ctr/utils.go

@@ -155,7 +155,7 @@ func getTempDir(id string) (string, error) {
 	return tmpDir, nil
 }
 
-func waitContainer(events execution.ContainerService_EventsClient, response *execution.CreateResponse) (uint32, error) {
+func waitContainer(events execution.ContainerService_EventsClient, id string, pid uint32) (uint32, error) {
 	for {
 		e, err := events.Recv()
 		if err != nil {
@@ -164,8 +164,7 @@ func waitContainer(events execution.ContainerService_EventsClient, response *exe
 		if e.Type != container.Event_EXIT {
 			continue
 		}
-		if e.ID == response.ID &&
+		if e.ID == id && e.Pid == pid {
-			e.Pid == response.Pid {
 			return e.ExitStatus, nil
 		}
 	}

linux/shim/process.go

@@ -15,7 +15,7 @@ type process interface {
 	Exited(status int)
 	// Status returns the exit status
 	Status() int
-	// Delete delets the process and its resourcess
+	// Delete deletes the process and its resourcess
 	Delete(context.Context) error
 	// Signal directly signals the process
 	Signal(int) error

linux/shim/service.go

@@ -6,7 +6,6 @@ import (
 	"sync"
 	"syscall"
 
-	"github.com/Sirupsen/logrus"
 	shimapi "github.com/containerd/containerd/api/services/shim"
 	"github.com/containerd/containerd/api/types/container"
 	"github.com/containerd/containerd/reaper"
@@ -54,19 +53,7 @@ func (s *Service) Create(ctx context.Context, r *shimapi.CreateRequest) (*shimap
 		ExitCh: make(chan int, 1),
 	}
 	reaper.Default.Register(pid, cmd)
-	go func() {
+	go s.waitExit(process, pid, cmd)
-		status, err := reaper.Default.WaitPid(pid)
-		if err != nil {
-			logrus.WithError(err).Error("waitpid")
-		}
-		process.Exited(status)
-		s.events <- &container.Event{
-			Type:       container.Event_EXIT,
-			ID:         s.id,
-			Pid:        uint32(pid),
-			ExitStatus: uint32(status),
-		}
-	}()
 	s.events <- &container.Event{
 		Type: container.Event_CREATE,
 		ID:   r.ID,
@@ -111,12 +98,20 @@ func (s *Service) Exec(ctx context.Context, r *shimapi.ExecRequest) (*shimapi.Ex
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	s.execID++
+	reaper.Default.Lock()
 	process, err := newExecProcess(ctx, s.path, r, s.initProcess, s.execID)
 	if err != nil {
+		reaper.Default.Unlock()
 		return nil, err
 	}
 	pid := process.Pid()
 	s.processes[pid] = process
+	cmd := &reaper.Cmd{
+		ExitCh: make(chan int, 1),
+	}
+	reaper.Default.RegisterNL(pid, cmd)
+	reaper.Default.Unlock()
+	go s.waitExit(process, pid, cmd)
 	s.events <- &container.Event{
 		Type: container.Event_EXEC_ADDED,
 		ID:   s.id,
@@ -235,3 +230,14 @@ func (s *Service) Kill(ctx context.Context, r *shimapi.KillRequest) (*google_pro
 	}
 	return empty, nil
 }
+
+func (s *Service) waitExit(p process, pid int, cmd *reaper.Cmd) {
+	status := <-cmd.ExitCh
+	p.Exited(status)
+	s.events <- &container.Event{
+		Type:       container.Event_EXIT,
+		ID:         s.id,
+		Pid:        uint32(pid),
+		ExitStatus: uint32(status),
+	}
+}

reaper/reaper.go

@@ -97,6 +97,12 @@ func (m *Monitor) Register(pid int, c *Cmd) {
 	m.Unlock()
 }
 
+// RegisterNL does not grab the lock internally
+// the caller is responsible for locking the monitor
+func (m *Monitor) RegisterNL(pid int, c *Cmd) {
+	m.cmds[pid] = c
+}
+
 func (m *Monitor) WaitPid(pid int) (int, error) {
 	m.Lock()
 	rc, ok := m.cmds[pid]