runtime/linux/shim -> runtime/shim

Signed-off-by: Evan Hazlett <ejhazlett@gmail.com>

runtime/shim/client/client.go

@@ -0,0 +1,278 @@
+// +build !windows
+
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package client
+
+import (
+	"context"
+	"io"
+	"net"
+	"os"
+	"os/exec"
+	"strings"
+	"sync"
+	"syscall"
+	"time"
+
+	"golang.org/x/sys/unix"
+
+	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
+	"github.com/stevvooe/ttrpc"
+
+	"github.com/containerd/containerd/events"
+	"github.com/containerd/containerd/log"
+	"github.com/containerd/containerd/runtime/shim"
+	shimapi "github.com/containerd/containerd/runtime/shim/v1"
+	"github.com/containerd/containerd/sys"
+	ptypes "github.com/gogo/protobuf/types"
+)
+
+var empty = &ptypes.Empty{}
+
+// Opt is an option for a shim client configuration
+type Opt func(context.Context, shim.Config) (shimapi.ShimService, io.Closer, error)
+
+// WithStart executes a new shim process
+func WithStart(binary, address, daemonAddress, cgroup string, debug bool, exitHandler func()) Opt {
+	return func(ctx context.Context, config shim.Config) (_ shimapi.ShimService, _ io.Closer, err error) {
+		socket, err := newSocket(address)
+		if err != nil {
+			return nil, nil, err
+		}
+		defer socket.Close()
+		f, err := socket.File()
+		if err != nil {
+			return nil, nil, errors.Wrapf(err, "failed to get fd for socket %s", address)
+		}
+		defer f.Close()
+
+		cmd, err := newCommand(binary, daemonAddress, debug, config, f)
+		if err != nil {
+			return nil, nil, err
+		}
+		if err := cmd.Start(); err != nil {
+			return nil, nil, errors.Wrapf(err, "failed to start shim")
+		}
+		defer func() {
+			if err != nil {
+				cmd.Process.Kill()
+			}
+		}()
+		go func() {
+			cmd.Wait()
+			exitHandler()
+		}()
+		log.G(ctx).WithFields(logrus.Fields{
+			"pid":     cmd.Process.Pid,
+			"address": address,
+			"debug":   debug,
+		}).Infof("shim %s started", binary)
+		// set shim in cgroup if it is provided
+		if cgroup != "" {
+			if err := setCgroup(cgroup, cmd); err != nil {
+				return nil, nil, err
+			}
+			log.G(ctx).WithFields(logrus.Fields{
+				"pid":     cmd.Process.Pid,
+				"address": address,
+			}).Infof("shim placed in cgroup %s", cgroup)
+		}
+		if err = sys.SetOOMScore(cmd.Process.Pid, sys.OOMScoreMaxKillable); err != nil {
+			return nil, nil, errors.Wrap(err, "failed to set OOM Score on shim")
+		}
+		c, clo, err := WithConnect(address, func() {})(ctx, config)
+		if err != nil {
+			return nil, nil, errors.Wrap(err, "failed to connect")
+		}
+		return c, clo, nil
+	}
+}
+
+func newCommand(binary, daemonAddress string, debug bool, config shim.Config, socket *os.File) (*exec.Cmd, error) {
+	selfExe, err := os.Executable()
+	if err != nil {
+		return nil, err
+	}
+	args := []string{
+		"-namespace", config.Namespace,
+		"-workdir", config.WorkDir,
+		"-address", daemonAddress,
+		"-containerd-binary", selfExe,
+	}
+
+	if config.Criu != "" {
+		args = append(args, "-criu-path", config.Criu)
+	}
+	if config.RuntimeRoot != "" {
+		args = append(args, "-runtime-root", config.RuntimeRoot)
+	}
+	if config.SystemdCgroup {
+		args = append(args, "-systemd-cgroup")
+	}
+	if debug {
+		args = append(args, "-debug")
+	}
+
+	cmd := exec.Command(binary, args...)
+	cmd.Dir = config.Path
+	// make sure the shim can be re-parented to system init
+	// and is cloned in a new mount namespace because the overlay/filesystems
+	// will be mounted by the shim
+	cmd.SysProcAttr = getSysProcAttr()
+	cmd.ExtraFiles = append(cmd.ExtraFiles, socket)
+	if debug {
+		cmd.Stdout = os.Stdout
+		cmd.Stderr = os.Stderr
+	}
+	return cmd, nil
+}
+
+func newSocket(address string) (*net.UnixListener, error) {
+	if len(address) > 106 {
+		return nil, errors.Errorf("%q: unix socket path too long (> 106)", address)
+	}
+	l, err := net.Listen("unix", "\x00"+address)
+	if err != nil {
+		return nil, errors.Wrapf(err, "failed to listen to abstract unix socket %q", address)
+	}
+
+	return l.(*net.UnixListener), nil
+}
+
+func connect(address string, d func(string, time.Duration) (net.Conn, error)) (net.Conn, error) {
+	return d(address, 100*time.Second)
+}
+
+func annonDialer(address string, timeout time.Duration) (net.Conn, error) {
+	address = strings.TrimPrefix(address, "unix://")
+	return net.DialTimeout("unix", "\x00"+address, timeout)
+}
+
+// WithConnect connects to an existing shim
+func WithConnect(address string, onClose func()) Opt {
+	return func(ctx context.Context, config shim.Config) (shimapi.ShimService, io.Closer, error) {
+		conn, err := connect(address, annonDialer)
+		if err != nil {
+			return nil, nil, err
+		}
+		client := ttrpc.NewClient(conn)
+		client.OnClose(onClose)
+		return shimapi.NewShimClient(client), conn, nil
+	}
+}
+
+// WithLocal uses an in process shim
+func WithLocal(publisher events.Publisher) func(context.Context, shim.Config) (shimapi.ShimService, io.Closer, error) {
+	return func(ctx context.Context, config shim.Config) (shimapi.ShimService, io.Closer, error) {
+		service, err := shim.NewService(config, publisher)
+		if err != nil {
+			return nil, nil, err
+		}
+		return shim.NewLocal(service), nil, nil
+	}
+}
+
+// New returns a new shim client
+func New(ctx context.Context, config shim.Config, opt Opt) (*Client, error) {
+	s, c, err := opt(ctx, config)
+	if err != nil {
+		return nil, err
+	}
+	return &Client{
+		ShimService: s,
+		c:           c,
+		exitCh:      make(chan struct{}),
+	}, nil
+}
+
+// Client is a shim client containing the connection to a shim
+type Client struct {
+	shimapi.ShimService
+
+	c        io.Closer
+	exitCh   chan struct{}
+	exitOnce sync.Once
+}
+
+// IsAlive returns true if the shim can be contacted.
+// NOTE: a negative answer doesn't mean that the process is gone.
+func (c *Client) IsAlive(ctx context.Context) (bool, error) {
+	_, err := c.ShimInfo(ctx, empty)
+	if err != nil {
+		// TODO(stevvooe): There are some error conditions that need to be
+		// handle with unix sockets existence to give the right answer here.
+		return false, err
+	}
+	return true, nil
+}
+
+// StopShim signals the shim to exit and wait for the process to disappear
+func (c *Client) StopShim(ctx context.Context) error {
+	return c.signalShim(ctx, unix.SIGTERM)
+}
+
+// KillShim kills the shim forcefully and wait for the process to disappear
+func (c *Client) KillShim(ctx context.Context) error {
+	return c.signalShim(ctx, unix.SIGKILL)
+}
+
+// Close the cient connection
+func (c *Client) Close() error {
+	if c.c == nil {
+		return nil
+	}
+	return c.c.Close()
+}
+
+func (c *Client) signalShim(ctx context.Context, sig syscall.Signal) error {
+	info, err := c.ShimInfo(ctx, empty)
+	if err != nil {
+		return err
+	}
+	pid := int(info.ShimPid)
+	// make sure we don't kill ourselves if we are running a local shim
+	if os.Getpid() == pid {
+		return nil
+	}
+	if err := unix.Kill(pid, sig); err != nil && err != unix.ESRCH {
+		return err
+	}
+	// wait for shim to die after being signaled
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-c.waitForExit(pid):
+		return nil
+	}
+}
+
+func (c *Client) waitForExit(pid int) <-chan struct{} {
+	c.exitOnce.Do(func() {
+		for {
+			// use kill(pid, 0) here because the shim could have been reparented
+			// and we are no longer able to waitpid(pid, ...) on the shim
+			if err := unix.Kill(pid, 0); err == unix.ESRCH {
+				close(c.exitCh)
+				return
+			}
+			time.Sleep(10 * time.Millisecond)
+		}
+	})
+	return c.exitCh
+}

runtime/shim/client/client_linux.go

@@ -0,0 +1,46 @@
+// +build linux
+
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package client
+
+import (
+	"os/exec"
+	"syscall"
+
+	"github.com/containerd/cgroups"
+	"github.com/pkg/errors"
+)
+
+func getSysProcAttr() *syscall.SysProcAttr {
+	return &syscall.SysProcAttr{
+		Setpgid: true,
+	}
+}
+
+func setCgroup(cgroupPath string, cmd *exec.Cmd) error {
+	cg, err := cgroups.Load(cgroups.V1, cgroups.StaticPath(cgroupPath))
+	if err != nil {
+		return errors.Wrapf(err, "failed to load cgroup %s", cgroupPath)
+	}
+	if err := cg.Add(cgroups.Process{
+		Pid: cmd.Process.Pid,
+	}); err != nil {
+		return errors.Wrapf(err, "failed to join cgroup %s", cgroupPath)
+	}
+	return nil
+}

runtime/shim/client/client_unix.go

@@ -0,0 +1,34 @@
+// +build !linux,!windows
+
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package client
+
+import (
+	"os/exec"
+	"syscall"
+)
+
+func getSysProcAttr() *syscall.SysProcAttr {
+	return &syscall.SysProcAttr{
+		Setpgid: true,
+	}
+}
+
+func setCgroup(cgroupPath string, cmd *exec.Cmd) error {
+	return nil
+}

runtime/shim/local.go

@@ -0,0 +1,107 @@
+// +build !windows
+
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package shim
+
+import (
+	"context"
+	"path/filepath"
+
+	"github.com/containerd/containerd/mount"
+	shimapi "github.com/containerd/containerd/runtime/shim/v1"
+	ptypes "github.com/gogo/protobuf/types"
+)
+
+// NewLocal returns a shim client implementation for issue commands to a shim
+func NewLocal(s *Service) shimapi.ShimService {
+	return &local{
+		s: s,
+	}
+}
+
+type local struct {
+	s *Service
+}
+
+func (c *local) Create(ctx context.Context, in *shimapi.CreateTaskRequest) (*shimapi.CreateTaskResponse, error) {
+	return c.s.Create(ctx, in)
+}
+
+func (c *local) Start(ctx context.Context, in *shimapi.StartRequest) (*shimapi.StartResponse, error) {
+	return c.s.Start(ctx, in)
+}
+
+func (c *local) Delete(ctx context.Context, in *ptypes.Empty) (*shimapi.DeleteResponse, error) {
+	// make sure we unmount the containers rootfs for this local
+	if err := mount.Unmount(filepath.Join(c.s.config.Path, "rootfs"), 0); err != nil {
+		return nil, err
+	}
+	return c.s.Delete(ctx, in)
+}
+
+func (c *local) DeleteProcess(ctx context.Context, in *shimapi.DeleteProcessRequest) (*shimapi.DeleteResponse, error) {
+	return c.s.DeleteProcess(ctx, in)
+}
+
+func (c *local) Exec(ctx context.Context, in *shimapi.ExecProcessRequest) (*ptypes.Empty, error) {
+	return c.s.Exec(ctx, in)
+}
+
+func (c *local) ResizePty(ctx context.Context, in *shimapi.ResizePtyRequest) (*ptypes.Empty, error) {
+	return c.s.ResizePty(ctx, in)
+}
+
+func (c *local) State(ctx context.Context, in *shimapi.StateRequest) (*shimapi.StateResponse, error) {
+	return c.s.State(ctx, in)
+}
+
+func (c *local) Pause(ctx context.Context, in *ptypes.Empty) (*ptypes.Empty, error) {
+	return c.s.Pause(ctx, in)
+}
+
+func (c *local) Resume(ctx context.Context, in *ptypes.Empty) (*ptypes.Empty, error) {
+	return c.s.Resume(ctx, in)
+}
+
+func (c *local) Kill(ctx context.Context, in *shimapi.KillRequest) (*ptypes.Empty, error) {
+	return c.s.Kill(ctx, in)
+}
+
+func (c *local) ListPids(ctx context.Context, in *shimapi.ListPidsRequest) (*shimapi.ListPidsResponse, error) {
+	return c.s.ListPids(ctx, in)
+}
+
+func (c *local) CloseIO(ctx context.Context, in *shimapi.CloseIORequest) (*ptypes.Empty, error) {
+	return c.s.CloseIO(ctx, in)
+}
+
+func (c *local) Checkpoint(ctx context.Context, in *shimapi.CheckpointTaskRequest) (*ptypes.Empty, error) {
+	return c.s.Checkpoint(ctx, in)
+}
+
+func (c *local) ShimInfo(ctx context.Context, in *ptypes.Empty) (*shimapi.ShimInfoResponse, error) {
+	return c.s.ShimInfo(ctx, in)
+}
+
+func (c *local) Update(ctx context.Context, in *shimapi.UpdateTaskRequest) (*ptypes.Empty, error) {
+	return c.s.Update(ctx, in)
+}
+
+func (c *local) Wait(ctx context.Context, in *shimapi.WaitRequest) (*shimapi.WaitResponse, error) {
+	return c.s.Wait(ctx, in)
+}

runtime/shim/reaper.go

@@ -0,0 +1,110 @@
+// +build !windows
+
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package shim
+
+import (
+	"os/exec"
+	"sync"
+	"time"
+
+	"github.com/containerd/containerd/sys"
+	runc "github.com/containerd/go-runc"
+	"github.com/pkg/errors"
+)
+
+// ErrNoSuchProcess is returned when the process no longer exists
+var ErrNoSuchProcess = errors.New("no such process")
+
+const bufferSize = 32
+
+// Reap should be called when the process receives an SIGCHLD.  Reap will reap
+// all exited processes and close their wait channels
+func Reap() error {
+	now := time.Now()
+	exits, err := sys.Reap(false)
+	Default.Lock()
+	for c := range Default.subscribers {
+		for _, e := range exits {
+			c <- runc.Exit{
+				Timestamp: now,
+				Pid:       e.Pid,
+				Status:    e.Status,
+			}
+		}
+
+	}
+	Default.Unlock()
+	return err
+}
+
+// Default is the default monitor initialized for the package
+var Default = &Monitor{
+	subscribers: make(map[chan runc.Exit]struct{}),
+}
+
+// Monitor monitors the underlying system for process status changes
+type Monitor struct {
+	sync.Mutex
+
+	subscribers map[chan runc.Exit]struct{}
+}
+
+// Start starts the command a registers the process with the reaper
+func (m *Monitor) Start(c *exec.Cmd) (chan runc.Exit, error) {
+	ec := m.Subscribe()
+	if err := c.Start(); err != nil {
+		m.Unsubscribe(ec)
+		return nil, err
+	}
+	return ec, nil
+}
+
+// Wait blocks until a process is signal as dead.
+// User should rely on the value of the exit status to determine if the
+// command was successful or not.
+func (m *Monitor) Wait(c *exec.Cmd, ec chan runc.Exit) (int, error) {
+	for e := range ec {
+		if e.Pid == c.Process.Pid {
+			// make sure we flush all IO
+			c.Wait()
+			m.Unsubscribe(ec)
+			return e.Status, nil
+		}
+	}
+	// return no such process if the ec channel is closed and no more exit
+	// events will be sent
+	return -1, ErrNoSuchProcess
+}
+
+// Subscribe to process exit changes
+func (m *Monitor) Subscribe() chan runc.Exit {
+	c := make(chan runc.Exit, bufferSize)
+	m.Lock()
+	m.subscribers[c] = struct{}{}
+	m.Unlock()
+	return c
+}
+
+// Unsubscribe to process exit changes
+func (m *Monitor) Unsubscribe(c chan runc.Exit) {
+	m.Lock()
+	delete(m.subscribers, c)
+	close(c)
+	m.Unlock()
+}

runtime/shim/service.go

@@ -0,0 +1,546 @@
+// +build !windows
+
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package shim
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"sync"
+
+	"github.com/containerd/console"
+	eventstypes "github.com/containerd/containerd/api/events"
+	"github.com/containerd/containerd/api/types/task"
+	"github.com/containerd/containerd/errdefs"
+	"github.com/containerd/containerd/events"
+	"github.com/containerd/containerd/log"
+	"github.com/containerd/containerd/namespaces"
+	"github.com/containerd/containerd/runtime"
+	"github.com/containerd/containerd/runtime/linux/proc"
+	"github.com/containerd/containerd/runtime/linux/runctypes"
+	shimapi "github.com/containerd/containerd/runtime/shim/v1"
+	runc "github.com/containerd/go-runc"
+	"github.com/containerd/typeurl"
+	ptypes "github.com/gogo/protobuf/types"
+	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+var (
+	empty   = &ptypes.Empty{}
+	bufPool = sync.Pool{
+		New: func() interface{} {
+			buffer := make([]byte, 32<<10)
+			return &buffer
+		},
+	}
+)
+
+// Config contains shim specific configuration
+type Config struct {
+	Path          string
+	Namespace     string
+	WorkDir       string
+	Criu          string
+	RuntimeRoot   string
+	SystemdCgroup bool
+}
+
+// NewService returns a new shim service that can be used via GRPC
+func NewService(config Config, publisher events.Publisher) (*Service, error) {
+	if config.Namespace == "" {
+		return nil, fmt.Errorf("shim namespace cannot be empty")
+	}
+	ctx := namespaces.WithNamespace(context.Background(), config.Namespace)
+	ctx = log.WithLogger(ctx, logrus.WithFields(logrus.Fields{
+		"namespace": config.Namespace,
+		"path":      config.Path,
+		"pid":       os.Getpid(),
+	}))
+	s := &Service{
+		config:    config,
+		context:   ctx,
+		processes: make(map[string]proc.Process),
+		events:    make(chan interface{}, 128),
+		ec:        Default.Subscribe(),
+	}
+	go s.processExits()
+	if err := s.initPlatform(); err != nil {
+		return nil, errors.Wrap(err, "failed to initialized platform behavior")
+	}
+	go s.forward(publisher)
+	return s, nil
+}
+
+// Service is the shim implementation of a remote shim over GRPC
+type Service struct {
+	mu sync.Mutex
+
+	config    Config
+	context   context.Context
+	processes map[string]proc.Process
+	events    chan interface{}
+	platform  proc.Platform
+	ec        chan runc.Exit
+
+	// Filled by Create()
+	id     string
+	bundle string
+}
+
+// Create a new initial process and container with the underlying OCI runtime
+func (s *Service) Create(ctx context.Context, r *shimapi.CreateTaskRequest) (*shimapi.CreateTaskResponse, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	var mounts []proc.Mount
+	for _, m := range r.Rootfs {
+		mounts = append(mounts, proc.Mount{
+			Type:    m.Type,
+			Source:  m.Source,
+			Target:  m.Target,
+			Options: m.Options,
+		})
+	}
+	process, err := proc.New(
+		ctx,
+		s.config.Path,
+		s.config.WorkDir,
+		s.config.RuntimeRoot,
+		s.config.Namespace,
+		s.config.Criu,
+		s.config.SystemdCgroup,
+		s.platform,
+		&proc.CreateConfig{
+			ID:               r.ID,
+			Bundle:           r.Bundle,
+			Runtime:          r.Runtime,
+			Rootfs:           mounts,
+			Terminal:         r.Terminal,
+			Stdin:            r.Stdin,
+			Stdout:           r.Stdout,
+			Stderr:           r.Stderr,
+			Checkpoint:       r.Checkpoint,
+			ParentCheckpoint: r.ParentCheckpoint,
+			Options:          r.Options,
+		},
+	)
+	if err != nil {
+		return nil, errdefs.ToGRPC(err)
+	}
+	// save the main task id and bundle to the shim for additional requests
+	s.id = r.ID
+	s.bundle = r.Bundle
+	pid := process.Pid()
+	s.processes[r.ID] = process
+	return &shimapi.CreateTaskResponse{
+		Pid: uint32(pid),
+	}, nil
+}
+
+// Start a process
+func (s *Service) Start(ctx context.Context, r *shimapi.StartRequest) (*shimapi.StartResponse, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	p := s.processes[r.ID]
+	if p == nil {
+		return nil, errdefs.ToGRPCf(errdefs.ErrNotFound, "process %s", r.ID)
+	}
+	if err := p.Start(ctx); err != nil {
+		return nil, err
+	}
+	return &shimapi.StartResponse{
+		ID:  p.ID(),
+		Pid: uint32(p.Pid()),
+	}, nil
+}
+
+// Delete the initial process and container
+func (s *Service) Delete(ctx context.Context, r *ptypes.Empty) (*shimapi.DeleteResponse, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	p := s.processes[s.id]
+	if p == nil {
+		return nil, errdefs.ToGRPCf(errdefs.ErrFailedPrecondition, "container must be created")
+	}
+	if err := p.Delete(ctx); err != nil {
+		return nil, err
+	}
+	delete(s.processes, s.id)
+	s.platform.Close()
+	return &shimapi.DeleteResponse{
+		ExitStatus: uint32(p.ExitStatus()),
+		ExitedAt:   p.ExitedAt(),
+		Pid:        uint32(p.Pid()),
+	}, nil
+}
+
+// DeleteProcess deletes an exec'd process
+func (s *Service) DeleteProcess(ctx context.Context, r *shimapi.DeleteProcessRequest) (*shimapi.DeleteResponse, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if r.ID == s.id {
+		return nil, status.Errorf(codes.InvalidArgument, "cannot delete init process with DeleteProcess")
+	}
+	p := s.processes[r.ID]
+	if p == nil {
+		return nil, errors.Wrapf(errdefs.ErrNotFound, "process %s", r.ID)
+	}
+	if err := p.Delete(ctx); err != nil {
+		return nil, err
+	}
+	delete(s.processes, r.ID)
+	return &shimapi.DeleteResponse{
+		ExitStatus: uint32(p.ExitStatus()),
+		ExitedAt:   p.ExitedAt(),
+		Pid:        uint32(p.Pid()),
+	}, nil
+}
+
+// Exec an additional process inside the container
+func (s *Service) Exec(ctx context.Context, r *shimapi.ExecProcessRequest) (*ptypes.Empty, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if p := s.processes[r.ID]; p != nil {
+		return nil, errdefs.ToGRPCf(errdefs.ErrAlreadyExists, "id %s", r.ID)
+	}
+
+	p := s.processes[s.id]
+	if p == nil {
+		return nil, errdefs.ToGRPCf(errdefs.ErrFailedPrecondition, "container must be created")
+	}
+
+	process, err := p.(*proc.Init).Exec(ctx, s.config.Path, &proc.ExecConfig{
+		ID:       r.ID,
+		Terminal: r.Terminal,
+		Stdin:    r.Stdin,
+		Stdout:   r.Stdout,
+		Stderr:   r.Stderr,
+		Spec:     r.Spec,
+	})
+	if err != nil {
+		return nil, errdefs.ToGRPC(err)
+	}
+	s.processes[r.ID] = process
+	return empty, nil
+}
+
+// ResizePty of a process
+func (s *Service) ResizePty(ctx context.Context, r *shimapi.ResizePtyRequest) (*ptypes.Empty, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if r.ID == "" {
+		return nil, errdefs.ToGRPCf(errdefs.ErrInvalidArgument, "id not provided")
+	}
+	ws := console.WinSize{
+		Width:  uint16(r.Width),
+		Height: uint16(r.Height),
+	}
+	p := s.processes[r.ID]
+	if p == nil {
+		return nil, errors.Errorf("process does not exist %s", r.ID)
+	}
+	if err := p.Resize(ws); err != nil {
+		return nil, errdefs.ToGRPC(err)
+	}
+	return empty, nil
+}
+
+// State returns runtime state information for a process
+func (s *Service) State(ctx context.Context, r *shimapi.StateRequest) (*shimapi.StateResponse, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	p := s.processes[r.ID]
+	if p == nil {
+		return nil, errdefs.ToGRPCf(errdefs.ErrNotFound, "process id %s", r.ID)
+	}
+	st, err := p.Status(ctx)
+	if err != nil {
+		return nil, err
+	}
+	status := task.StatusUnknown
+	switch st {
+	case "created":
+		status = task.StatusCreated
+	case "running":
+		status = task.StatusRunning
+	case "stopped":
+		status = task.StatusStopped
+	case "paused":
+		status = task.StatusPaused
+	case "pausing":
+		status = task.StatusPausing
+	}
+	sio := p.Stdio()
+	return &shimapi.StateResponse{
+		ID:         p.ID(),
+		Bundle:     s.bundle,
+		Pid:        uint32(p.Pid()),
+		Status:     status,
+		Stdin:      sio.Stdin,
+		Stdout:     sio.Stdout,
+		Stderr:     sio.Stderr,
+		Terminal:   sio.Terminal,
+		ExitStatus: uint32(p.ExitStatus()),
+		ExitedAt:   p.ExitedAt(),
+	}, nil
+}
+
+// Pause the container
+func (s *Service) Pause(ctx context.Context, r *ptypes.Empty) (*ptypes.Empty, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	p := s.processes[s.id]
+	if p == nil {
+		return nil, errdefs.ToGRPCf(errdefs.ErrFailedPrecondition, "container must be created")
+	}
+	if err := p.(*proc.Init).Pause(ctx); err != nil {
+		return nil, err
+	}
+	return empty, nil
+}
+
+// Resume the container
+func (s *Service) Resume(ctx context.Context, r *ptypes.Empty) (*ptypes.Empty, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	p := s.processes[s.id]
+	if p == nil {
+		return nil, errdefs.ToGRPCf(errdefs.ErrFailedPrecondition, "container must be created")
+	}
+	if err := p.(*proc.Init).Resume(ctx); err != nil {
+		return nil, err
+	}
+	return empty, nil
+}
+
+// Kill a process with the provided signal
+func (s *Service) Kill(ctx context.Context, r *shimapi.KillRequest) (*ptypes.Empty, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if r.ID == "" {
+		p := s.processes[s.id]
+		if p == nil {
+			return nil, errdefs.ToGRPCf(errdefs.ErrFailedPrecondition, "container must be created")
+		}
+		if err := p.Kill(ctx, r.Signal, r.All); err != nil {
+			return nil, errdefs.ToGRPC(err)
+		}
+		return empty, nil
+	}
+
+	p := s.processes[r.ID]
+	if p == nil {
+		return nil, errdefs.ToGRPCf(errdefs.ErrNotFound, "process id %s not found", r.ID)
+	}
+	if err := p.Kill(ctx, r.Signal, r.All); err != nil {
+		return nil, errdefs.ToGRPC(err)
+	}
+	return empty, nil
+}
+
+// ListPids returns all pids inside the container
+func (s *Service) ListPids(ctx context.Context, r *shimapi.ListPidsRequest) (*shimapi.ListPidsResponse, error) {
+	pids, err := s.getContainerPids(ctx, r.ID)
+	if err != nil {
+		return nil, errdefs.ToGRPC(err)
+	}
+	var processes []*task.ProcessInfo
+	for _, pid := range pids {
+		pInfo := task.ProcessInfo{
+			Pid: pid,
+		}
+		for _, p := range s.processes {
+			if p.Pid() == int(pid) {
+				d := &runctypes.ProcessDetails{
+					ExecID: p.ID(),
+				}
+				a, err := typeurl.MarshalAny(d)
+				if err != nil {
+					return nil, errors.Wrapf(err, "failed to marshal process %d info", pid)
+				}
+				pInfo.Info = a
+				break
+			}
+		}
+		processes = append(processes, &pInfo)
+	}
+	return &shimapi.ListPidsResponse{
+		Processes: processes,
+	}, nil
+}
+
+// CloseIO of a process
+func (s *Service) CloseIO(ctx context.Context, r *shimapi.CloseIORequest) (*ptypes.Empty, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	p := s.processes[r.ID]
+	if p == nil {
+		return nil, errdefs.ToGRPCf(errdefs.ErrNotFound, "process does not exist %s", r.ID)
+	}
+	if stdin := p.Stdin(); stdin != nil {
+		if err := stdin.Close(); err != nil {
+			return nil, errors.Wrap(err, "close stdin")
+		}
+	}
+	return empty, nil
+}
+
+// Checkpoint the container
+func (s *Service) Checkpoint(ctx context.Context, r *shimapi.CheckpointTaskRequest) (*ptypes.Empty, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	p := s.processes[s.id]
+	if p == nil {
+		return nil, errdefs.ToGRPCf(errdefs.ErrFailedPrecondition, "container must be created")
+	}
+	if err := p.(*proc.Init).Checkpoint(ctx, &proc.CheckpointConfig{
+		Path:    r.Path,
+		Options: r.Options,
+	}); err != nil {
+		return nil, errdefs.ToGRPC(err)
+	}
+	return empty, nil
+}
+
+// ShimInfo returns shim information such as the shim's pid
+func (s *Service) ShimInfo(ctx context.Context, r *ptypes.Empty) (*shimapi.ShimInfoResponse, error) {
+	return &shimapi.ShimInfoResponse{
+		ShimPid: uint32(os.Getpid()),
+	}, nil
+}
+
+// Update a running container
+func (s *Service) Update(ctx context.Context, r *shimapi.UpdateTaskRequest) (*ptypes.Empty, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	p := s.processes[s.id]
+	if p == nil {
+		return nil, errdefs.ToGRPCf(errdefs.ErrFailedPrecondition, "container must be created")
+	}
+	if err := p.(*proc.Init).Update(ctx, r.Resources); err != nil {
+		return nil, errdefs.ToGRPC(err)
+	}
+	return empty, nil
+}
+
+// Wait for a process to exit
+func (s *Service) Wait(ctx context.Context, r *shimapi.WaitRequest) (*shimapi.WaitResponse, error) {
+	s.mu.Lock()
+	p := s.processes[r.ID]
+	s.mu.Unlock()
+	if p == nil {
+		return nil, errdefs.ToGRPCf(errdefs.ErrFailedPrecondition, "container must be created")
+	}
+	p.Wait()
+
+	return &shimapi.WaitResponse{
+		ExitStatus: uint32(p.ExitStatus()),
+		ExitedAt:   p.ExitedAt(),
+	}, nil
+}
+
+func (s *Service) processExits() {
+	for e := range s.ec {
+		s.checkProcesses(e)
+	}
+}
+
+func (s *Service) checkProcesses(e runc.Exit) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	for _, p := range s.processes {
+		if p.Pid() == e.Pid {
+			if ip, ok := p.(*proc.Init); ok {
+				// Ensure all children are killed
+				if err := ip.KillAll(s.context); err != nil {
+					log.G(s.context).WithError(err).WithField("id", ip.ID()).
+						Error("failed to kill init's children")
+				}
+			}
+			p.SetExited(e.Status)
+			s.events <- &eventstypes.TaskExit{
+				ContainerID: s.id,
+				ID:          p.ID(),
+				Pid:         uint32(e.Pid),
+				ExitStatus:  uint32(e.Status),
+				ExitedAt:    p.ExitedAt(),
+			}
+			return
+		}
+	}
+}
+
+func (s *Service) getContainerPids(ctx context.Context, id string) ([]uint32, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	p := s.processes[s.id]
+	if p == nil {
+		return nil, errors.Wrapf(errdefs.ErrFailedPrecondition, "container must be created")
+	}
+
+	ps, err := p.(*proc.Init).Runtime().Ps(ctx, id)
+	if err != nil {
+		return nil, err
+	}
+	pids := make([]uint32, 0, len(ps))
+	for _, pid := range ps {
+		pids = append(pids, uint32(pid))
+	}
+	return pids, nil
+}
+
+func (s *Service) forward(publisher events.Publisher) {
+	for e := range s.events {
+		if err := publisher.Publish(s.context, getTopic(s.context, e), e); err != nil {
+			log.G(s.context).WithError(err).Error("post event")
+		}
+	}
+}
+
+func getTopic(ctx context.Context, e interface{}) string {
+	switch e.(type) {
+	case *eventstypes.TaskCreate:
+		return runtime.TaskCreateEventTopic
+	case *eventstypes.TaskStart:
+		return runtime.TaskStartEventTopic
+	case *eventstypes.TaskOOM:
+		return runtime.TaskOOMEventTopic
+	case *eventstypes.TaskExit:
+		return runtime.TaskExitEventTopic
+	case *eventstypes.TaskDelete:
+		return runtime.TaskDeleteEventTopic
+	case *eventstypes.TaskExecAdded:
+		return runtime.TaskExecAddedEventTopic
+	case *eventstypes.TaskExecStarted:
+		return runtime.TaskExecStartedEventTopic
+	case *eventstypes.TaskPaused:
+		return runtime.TaskPausedEventTopic
+	case *eventstypes.TaskResumed:
+		return runtime.TaskResumedEventTopic
+	case *eventstypes.TaskCheckpointed:
+		return runtime.TaskCheckpointedEventTopic
+	default:
+		logrus.Warnf("no topic for type %#v", e)
+	}
+	return runtime.TaskUnknownTopic
+}

runtime/shim/service_linux.go

@@ -0,0 +1,111 @@
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package shim
+
+import (
+	"context"
+	"io"
+	"sync"
+	"syscall"
+
+	"github.com/containerd/console"
+	"github.com/containerd/fifo"
+	"github.com/pkg/errors"
+)
+
+type linuxPlatform struct {
+	epoller *console.Epoller
+}
+
+func (p *linuxPlatform) CopyConsole(ctx context.Context, console console.Console, stdin, stdout, stderr string, wg, cwg *sync.WaitGroup) (console.Console, error) {
+	if p.epoller == nil {
+		return nil, errors.New("uninitialized epoller")
+	}
+
+	epollConsole, err := p.epoller.Add(console)
+	if err != nil {
+		return nil, err
+	}
+
+	if stdin != "" {
+		in, err := fifo.OpenFifo(ctx, stdin, syscall.O_RDONLY, 0)
+		if err != nil {
+			return nil, err
+		}
+		cwg.Add(1)
+		go func() {
+			cwg.Done()
+			p := bufPool.Get().(*[]byte)
+			defer bufPool.Put(p)
+			io.CopyBuffer(epollConsole, in, *p)
+		}()
+	}
+
+	outw, err := fifo.OpenFifo(ctx, stdout, syscall.O_WRONLY, 0)
+	if err != nil {
+		return nil, err
+	}
+	outr, err := fifo.OpenFifo(ctx, stdout, syscall.O_RDONLY, 0)
+	if err != nil {
+		return nil, err
+	}
+	wg.Add(1)
+	cwg.Add(1)
+	go func() {
+		cwg.Done()
+		p := bufPool.Get().(*[]byte)
+		defer bufPool.Put(p)
+		io.CopyBuffer(outw, epollConsole, *p)
+		epollConsole.Close()
+		outr.Close()
+		outw.Close()
+		wg.Done()
+	}()
+	return epollConsole, nil
+}
+
+func (p *linuxPlatform) ShutdownConsole(ctx context.Context, cons console.Console) error {
+	if p.epoller == nil {
+		return errors.New("uninitialized epoller")
+	}
+	epollConsole, ok := cons.(*console.EpollConsole)
+	if !ok {
+		return errors.Errorf("expected EpollConsole, got %#v", cons)
+	}
+	return epollConsole.Shutdown(p.epoller.CloseConsole)
+}
+
+func (p *linuxPlatform) Close() error {
+	return p.epoller.Close()
+}
+
+// initialize a single epoll fd to manage our consoles. `initPlatform` should
+// only be called once.
+func (s *Service) initPlatform() error {
+	if s.platform != nil {
+		return nil
+	}
+	epoller, err := console.NewEpoller()
+	if err != nil {
+		return errors.Wrap(err, "failed to initialize epoller")
+	}
+	s.platform = &linuxPlatform{
+		epoller: epoller,
+	}
+	go epoller.Wait()
+	return nil
+}

runtime/shim/service_unix.go

@@ -0,0 +1,84 @@
+// +build !windows,!linux
+
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package shim
+
+import (
+	"context"
+	"io"
+	"sync"
+	"syscall"
+
+	"github.com/containerd/console"
+	"github.com/containerd/fifo"
+)
+
+type unixPlatform struct {
+}
+
+func (p *unixPlatform) CopyConsole(ctx context.Context, console console.Console, stdin, stdout, stderr string, wg, cwg *sync.WaitGroup) (console.Console, error) {
+	if stdin != "" {
+		in, err := fifo.OpenFifo(ctx, stdin, syscall.O_RDONLY, 0)
+		if err != nil {
+			return nil, err
+		}
+		cwg.Add(1)
+		go func() {
+			cwg.Done()
+			p := bufPool.Get().(*[]byte)
+			defer bufPool.Put(p)
+
+			io.CopyBuffer(console, in, *p)
+		}()
+	}
+	outw, err := fifo.OpenFifo(ctx, stdout, syscall.O_WRONLY, 0)
+	if err != nil {
+		return nil, err
+	}
+	outr, err := fifo.OpenFifo(ctx, stdout, syscall.O_RDONLY, 0)
+	if err != nil {
+		return nil, err
+	}
+	wg.Add(1)
+	cwg.Add(1)
+	go func() {
+		cwg.Done()
+		p := bufPool.Get().(*[]byte)
+		defer bufPool.Put(p)
+
+		io.CopyBuffer(outw, console, *p)
+		console.Close()
+		outr.Close()
+		outw.Close()
+		wg.Done()
+	}()
+	return console, nil
+}
+
+func (p *unixPlatform) ShutdownConsole(ctx context.Context, cons console.Console) error {
+	return nil
+}
+
+func (p *unixPlatform) Close() error {
+	return nil
+}
+
+func (s *Service) initPlatform() error {
+	s.platform = &unixPlatform{}
+	return nil
+}

runtime/shim/v1/doc.go

@@ -0,0 +1,17 @@
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package shim

runtime/shim/v1/shim.proto

@@ -0,0 +1,165 @@
+syntax = "proto3";
+
+package containerd.runtime.linux.shim.v1;
+
+import "google/protobuf/any.proto";
+import "google/protobuf/empty.proto";
+import weak "gogoproto/gogo.proto";
+import "google/protobuf/timestamp.proto";
+import "github.com/containerd/containerd/api/types/mount.proto";
+import "github.com/containerd/containerd/api/types/task/task.proto";
+
+option go_package = "github.com/containerd/containerd/runtime/shim/v1;shim";
+
+// Shim service is launched for each container and is responsible for owning the IO
+// for the container and its additional processes.  The shim is also the parent of
+// each container and allows reattaching to the IO and receiving the exit status
+// for the container processes.
+service Shim {
+	// State returns shim and task state information.
+	rpc State(StateRequest) returns (StateResponse);
+
+	rpc Create(CreateTaskRequest) returns (CreateTaskResponse);
+
+	rpc Start(StartRequest) returns (StartResponse);
+
+	rpc Delete(google.protobuf.Empty) returns (DeleteResponse);
+
+	rpc DeleteProcess(DeleteProcessRequest) returns (DeleteResponse);
+
+	rpc ListPids(ListPidsRequest) returns (ListPidsResponse);
+
+	rpc Pause(google.protobuf.Empty) returns (google.protobuf.Empty);
+
+	rpc Resume(google.protobuf.Empty) returns (google.protobuf.Empty);
+
+	rpc Checkpoint(CheckpointTaskRequest) returns (google.protobuf.Empty);
+
+	rpc Kill(KillRequest) returns (google.protobuf.Empty);
+
+	rpc Exec(ExecProcessRequest) returns (google.protobuf.Empty);
+
+	rpc ResizePty(ResizePtyRequest) returns (google.protobuf.Empty);
+
+	rpc CloseIO(CloseIORequest) returns (google.protobuf.Empty);
+
+	// ShimInfo returns information about the shim.
+	rpc ShimInfo(google.protobuf.Empty) returns (ShimInfoResponse);
+
+	rpc Update(UpdateTaskRequest) returns (google.protobuf.Empty);
+
+	rpc Wait(WaitRequest) returns (WaitResponse);
+}
+
+message CreateTaskRequest {
+	string id = 1;
+	string bundle = 2;
+	string runtime = 3;
+	repeated containerd.types.Mount rootfs = 4;
+	bool terminal = 5;
+	string stdin = 6;
+	string stdout = 7;
+	string stderr = 8;
+	string checkpoint = 9;
+	string parent_checkpoint = 10;
+	google.protobuf.Any options = 11;
+}
+
+message CreateTaskResponse {
+	uint32 pid = 1;
+}
+
+message DeleteResponse {
+	uint32 pid = 1;
+	uint32 exit_status = 2;
+	google.protobuf.Timestamp exited_at = 3 [(gogoproto.stdtime) = true, (gogoproto.nullable) = false];
+}
+
+message DeleteProcessRequest {
+	string id = 1;
+}
+
+message ExecProcessRequest {
+	string id = 1;
+	bool terminal = 2;
+	string stdin = 3;
+	string stdout = 4;
+	string stderr = 5;
+	google.protobuf.Any spec = 6;
+}
+
+message ExecProcessResponse {
+}
+
+message ResizePtyRequest {
+	string id = 1;
+	uint32 width = 2;
+	uint32 height = 3;
+}
+
+message StateRequest {
+	string id = 1;
+}
+
+message StateResponse {
+	string id = 1;
+	string bundle = 2;
+	uint32 pid = 3;
+	containerd.v1.types.Status status = 4;
+	string stdin = 5;
+	string stdout = 6;
+	string stderr = 7;
+	bool terminal = 8;
+	uint32 exit_status = 9;
+	google.protobuf.Timestamp exited_at = 10 [(gogoproto.stdtime) = true, (gogoproto.nullable) = false];
+}
+
+message KillRequest {
+	string id = 1;
+	uint32 signal = 2;
+	bool all = 3;
+}
+
+message CloseIORequest {
+	string id = 1;
+	bool stdin = 2;
+}
+
+message ListPidsRequest {
+	string id = 1;
+}
+
+message ListPidsResponse {
+	repeated containerd.v1.types.ProcessInfo processes = 1;
+}
+
+message CheckpointTaskRequest {
+	string path = 1;
+	google.protobuf.Any options = 2;
+}
+
+message ShimInfoResponse {
+	uint32 shim_pid = 1;
+}
+
+message UpdateTaskRequest {
+	google.protobuf.Any resources = 1;
+}
+
+message StartRequest {
+	string id = 1;
+}
+
+message StartResponse {
+	string id = 1;
+	uint32 pid = 2;
+}
+
+message WaitRequest {
+	string id = 1;
+}
+
+message WaitResponse {
+	uint32 exit_status = 1;
+	google.protobuf.Timestamp exited_at = 2 [(gogoproto.stdtime) = true, (gogoproto.nullable) = false];
+}