Merge pull request #1739 from crosbymichael/shim-redux

Move shim process code into subpackage

cmd/containerd-shim/main_unix.go

@@ -18,6 +18,7 @@ import (
 	"github.com/containerd/containerd/dialer"
 	"github.com/containerd/containerd/errdefs"
 	"github.com/containerd/containerd/events"
+	"github.com/containerd/containerd/linux/proc"
 	"github.com/containerd/containerd/linux/shim"
 	shimapi "github.com/containerd/containerd/linux/shim/v1"
 	"github.com/containerd/containerd/reaper"
@@ -56,7 +57,7 @@ func init() {
 	flag.StringVar(&socketFlag, "socket", "", "abstract socket path to serve")
 	flag.StringVar(&addressFlag, "address", "", "grpc address back to main containerd")
 	flag.StringVar(&workdirFlag, "workdir", "", "path used to storge large temporary data")
-	flag.StringVar(&runtimeRootFlag, "runtime-root", shim.RuncRoot, "root directory for the runtime")
+	flag.StringVar(&runtimeRootFlag, "runtime-root", proc.RuncRoot, "root directory for the runtime")
 	flag.StringVar(&criuFlag, "criu", "", "path to criu binary")
 	flag.BoolVar(&systemdCgroupFlag, "systemd-cgroup", false, "set runtime to use systemd-cgroup")
 	flag.Parse()

debug/debug.go

@@ -0,0 +1,66 @@
+package debug
+
+import (
+	"bufio"
+	"fmt"
+	"os"
+	"sort"
+	"strconv"
+	"strings"
+)
+
+// Smaps prints the smaps to a file
+func Smaps(note, file string) error {
+	smaps, err := getMaps(os.Getpid())
+	if err != nil {
+		return err
+	}
+	f, err := os.OpenFile(file, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0666)
+	if err != nil {
+		return err
+	}
+	defer f.Close()
+	fmt.Fprintf(f, "%s: rss %d\n", note, smaps["rss"])
+	fmt.Fprintf(f, "%s: pss %d\n", note, smaps["pss"])
+	return nil
+}
+
+func getMaps(pid int) (map[string]int, error) {
+	f, err := os.Open(fmt.Sprintf("/proc/%d/smaps", pid))
+	if err != nil {
+		return nil, err
+	}
+	defer f.Close()
+	var (
+		smaps = make(map[string]int)
+		s     = bufio.NewScanner(f)
+	)
+	for s.Scan() {
+		var (
+			fields = strings.Fields(s.Text())
+			name   = fields[0]
+		)
+		name = strings.TrimSuffix(strings.ToLower(name), ":")
+		if len(fields) < 2 {
+			continue
+		}
+		n, err := strconv.Atoi(fields[1])
+		if err != nil {
+			continue
+		}
+		smaps[name] += n
+	}
+	if err := s.Err(); err != nil {
+		return nil, err
+	}
+	return smaps, nil
+}
+
+func keys(smaps map[string]int) []string {
+	var o []string
+	for k := range smaps {
+		o = append(o, k)
+	}
+	sort.Strings(o)
+	return o
+}

errdefs/grpc.go

@@ -4,7 +4,6 @@ import (
 	"strings"
 
 	"github.com/pkg/errors"
-	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 )
@@ -61,7 +60,7 @@ func FromGRPC(err error) error {
 
 	var cls error // divide these into error classes, becomes the cause
 
-	switch grpc.Code(err) {
+	switch code(err) {
 	case codes.InvalidArgument:
 		cls = ErrInvalidArgument
 	case codes.AlreadyExists:
@@ -94,7 +93,7 @@ func FromGRPC(err error) error {
 // Effectively, we just remove the string of cls from the end of err if it
 // appears there.
 func rebaseMessage(cls error, err error) string {
-	desc := grpc.ErrorDesc(err)
+	desc := errDesc(err)
 	clss := cls.Error()
 	if desc == clss {
 		return ""
@@ -107,3 +106,17 @@ func isGRPCError(err error) bool {
 	_, ok := status.FromError(err)
 	return ok
 }
+
+func code(err error) codes.Code {
+	if s, ok := status.FromError(err); ok {
+		return s.Code()
+	}
+	return codes.Unknown
+}
+
+func errDesc(err error) string {
+	if s, ok := status.FromError(err); ok {
+		return s.Message()
+	}
+	return err.Error()
+}

linux/shim/deleted_state.go → linux/proc/deleted_state.go

@@ -1,12 +1,12 @@
 // +build !windows
 
-package shim
+package proc
 
 import (
 	"context"
 
 	"github.com/containerd/console"
-	shimapi "github.com/containerd/containerd/linux/shim/v1"
+	google_protobuf "github.com/gogo/protobuf/types"
 	"github.com/pkg/errors"
 )
 
@@ -21,11 +21,11 @@ func (s *deletedState) Resume(ctx context.Context) error {
 	return errors.Errorf("cannot resume a deleted process")
 }
 
-func (s *deletedState) Update(context context.Context, r *shimapi.UpdateTaskRequest) error {
+func (s *deletedState) Update(context context.Context, r *google_protobuf.Any) error {
 	return errors.Errorf("cannot update a deleted process")
 }
 
-func (s *deletedState) Checkpoint(ctx context.Context, r *shimapi.CheckpointTaskRequest) error {
+func (s *deletedState) Checkpoint(ctx context.Context, r *CheckpointConfig) error {
 	return errors.Errorf("cannot checkpoint a deleted process")
 }
 

linux/shim/exec.go → linux/proc/exec.go

@@ -1,10 +1,9 @@
 // +build !windows
 
-package shim
+package proc
 
 import (
 	"context"
-	"encoding/json"
 	"fmt"
 	"io"
 	"os"
@@ -16,8 +15,6 @@ import (
 	"golang.org/x/sys/unix"
 
 	"github.com/containerd/console"
-	"github.com/containerd/containerd/identifiers"
-	shimapi "github.com/containerd/containerd/linux/shim/v1"
 	"github.com/containerd/fifo"
 	runc "github.com/containerd/go-runc"
 	specs "github.com/opencontainers/runtime-spec/specs-go"
@@ -27,7 +24,7 @@ import (
 type execProcess struct {
 	wg sync.WaitGroup
 
-	processState
+	State
 
 	mu      sync.Mutex
 	id      string
@@ -38,42 +35,14 @@ type execProcess struct {
 	pid     int
 	closers []io.Closer
 	stdin   io.Closer
-	stdio   stdio
+	stdio   Stdio
 	path    string
 	spec    specs.Process
 
-	parent    *initProcess
+	parent    *Init
 	waitBlock chan struct{}
 }
 
-func newExecProcess(context context.Context, path string, r *shimapi.ExecProcessRequest, parent *initProcess, id string) (process, error) {
-	if err := identifiers.Validate(id); err != nil {
-		return nil, errors.Wrapf(err, "invalid exec id")
-	}
-	// process exec request
-	var spec specs.Process
-	if err := json.Unmarshal(r.Spec.Value, &spec); err != nil {
-		return nil, err
-	}
-	spec.Terminal = r.Terminal
-
-	e := &execProcess{
-		id:     id,
-		path:   path,
-		parent: parent,
-		spec:   spec,
-		stdio: stdio{
-			stdin:    r.Stdin,
-			stdout:   r.Stdout,
-			stderr:   r.Stderr,
-			terminal: r.Terminal,
-		},
-		waitBlock: make(chan struct{}),
-	}
-	e.processState = &execCreatedState{p: e}
-	return e, nil
-}
-
 func (e *execProcess) Wait() {
 	<-e.waitBlock
 }
@@ -103,7 +72,7 @@ func (e *execProcess) ExitedAt() time.Time {
 func (e *execProcess) setExited(status int) {
 	e.status = status
 	e.exited = time.Now()
-	e.parent.platform.shutdownConsole(context.Background(), e.console)
+	e.parent.platform.ShutdownConsole(context.Background(), e.console)
 	close(e.waitBlock)
 }
 
@@ -142,7 +111,7 @@ func (e *execProcess) Stdin() io.Closer {
 	return e.stdin
 }
 
-func (e *execProcess) Stdio() stdio {
+func (e *execProcess) Stdio() Stdio {
 	return e.stdio
 }
 
@@ -151,12 +120,12 @@ func (e *execProcess) start(ctx context.Context) (err error) {
 		socket  *runc.Socket
 		pidfile = filepath.Join(e.path, fmt.Sprintf("%s.pid", e.id))
 	)
-	if e.stdio.terminal {
+	if e.stdio.Terminal {
 		if socket, err = runc.NewTempConsoleSocket(); err != nil {
 			return errors.Wrap(err, "failed to create runc console socket")
 		}
 		defer socket.Close()
-	} else if e.stdio.isNull() {
+	} else if e.stdio.IsNull() {
 		if e.io, err = runc.NewNullIO(); err != nil {
 			return errors.Wrap(err, "creating new NULL IO")
 		}
@@ -176,10 +145,10 @@ func (e *execProcess) start(ctx context.Context) (err error) {
 	if err := e.parent.runtime.Exec(ctx, e.parent.id, e.spec, opts); err != nil {
 		return e.parent.runtimeError(err, "OCI runtime exec failed")
 	}
-	if e.stdio.stdin != "" {
-		sc, err := fifo.OpenFifo(ctx, e.stdio.stdin, syscall.O_WRONLY|syscall.O_NONBLOCK, 0)
+	if e.stdio.Stdin != "" {
+		sc, err := fifo.OpenFifo(ctx, e.stdio.Stdin, syscall.O_WRONLY|syscall.O_NONBLOCK, 0)
 		if err != nil {
-			return errors.Wrapf(err, "failed to open stdin fifo %s", e.stdio.stdin)
+			return errors.Wrapf(err, "failed to open stdin fifo %s", e.stdio.Stdin)
 		}
 		e.closers = append(e.closers, sc)
 		e.stdin = sc
@@ -190,11 +159,11 @@ func (e *execProcess) start(ctx context.Context) (err error) {
 		if err != nil {
 			return errors.Wrap(err, "failed to retrieve console master")
 		}
-		if e.console, err = e.parent.platform.copyConsole(ctx, console, e.stdio.stdin, e.stdio.stdout, e.stdio.stderr, &e.wg, &copyWaitGroup); err != nil {
+		if e.console, err = e.parent.platform.CopyConsole(ctx, console, e.stdio.Stdin, e.stdio.Stdout, e.stdio.Stderr, &e.wg, &copyWaitGroup); err != nil {
 			return errors.Wrap(err, "failed to start console copy")
 		}
-	} else if !e.stdio.isNull() {
-		if err := copyPipes(ctx, e.io, e.stdio.stdin, e.stdio.stdout, e.stdio.stderr, &e.wg, &copyWaitGroup); err != nil {
+	} else if !e.stdio.IsNull() {
+		if err := copyPipes(ctx, e.io, e.stdio.Stdin, e.stdio.Stdout, e.stdio.Stderr, &e.wg, &copyWaitGroup); err != nil {
 			return errors.Wrap(err, "failed to start io pipe copy")
 		}
 	}

linux/shim/exec_state.go → linux/proc/exec_state.go

@@ -1,6 +1,6 @@
 // +build !windows
 
-package shim
+package proc
 
 import (
 	"context"
@@ -16,11 +16,11 @@ type execCreatedState struct {
 func (s *execCreatedState) transition(name string) error {
 	switch name {
 	case "running":
-		s.p.processState = &execRunningState{p: s.p}
+		s.p.State = &execRunningState{p: s.p}
 	case "stopped":
-		s.p.processState = &execStoppedState{p: s.p}
+		s.p.State = &execStoppedState{p: s.p}
 	case "deleted":
-		s.p.processState = &deletedState{}
+		s.p.State = &deletedState{}
 	default:
 		return errors.Errorf("invalid state transition %q to %q", stateName(s), name)
 	}
@@ -77,7 +77,7 @@ type execRunningState struct {
 func (s *execRunningState) transition(name string) error {
 	switch name {
 	case "stopped":
-		s.p.processState = &execStoppedState{p: s.p}
+		s.p.State = &execStoppedState{p: s.p}
 	default:
 		return errors.Errorf("invalid state transition %q to %q", stateName(s), name)
 	}
@@ -130,7 +130,7 @@ type execStoppedState struct {
 func (s *execStoppedState) transition(name string) error {
 	switch name {
 	case "deleted":
-		s.p.processState = &deletedState{}
+		s.p.State = &deletedState{}
 	default:
 		return errors.Errorf("invalid state transition %q to %q", stateName(s), name)
 	}