containerd/linux/shim/service.go

package shim

import (
	"os"
	"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"
	"github.com/crosbymichael/console"
	google_protobuf "github.com/golang/protobuf/ptypes/empty"
	"github.com/pkg/errors"
	"golang.org/x/net/context"
)

var empty = &google_protobuf.Empty{}

// New returns a new shim service that can be used via GRPC
func New(path string) *Service {
	return &Service{
		path:      path,
		processes: make(map[int]process),
		events:    make(chan *container.Event, 4096),
	}
}

type Service struct {
	initProcess *initProcess
	path        string
	id          string
	bundle      string
	mu          sync.Mutex
	processes   map[int]process
	events      chan *container.Event
	execID      int
}

func (s *Service) Create(ctx context.Context, r *shimapi.CreateRequest) (*shimapi.CreateResponse, error) {
	process, err := newInitProcess(ctx, s.path, r)
	if err != nil {
		return nil, err
	}
	s.mu.Lock()
	s.id = r.ID
	s.bundle = r.Bundle
	s.initProcess = process
	pid := process.Pid()
	s.processes[pid] = process
	s.mu.Unlock()
	cmd := &reaper.Cmd{
		ExitCh: make(chan int, 1),
	}
	reaper.Default.Register(pid, cmd)
	go func() {
		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,
		Pid:  uint32(pid),
	}
	return &shimapi.CreateResponse{
		Pid: uint32(pid),
	}, nil
}

func (s *Service) Start(ctx context.Context, r *shimapi.StartRequest) (*google_protobuf.Empty, error) {
	if err := s.initProcess.Start(ctx); err != nil {
		return nil, err
	}
	s.events <- &container.Event{
		Type: container.Event_START,
		ID:   s.id,
		Pid:  uint32(s.initProcess.Pid()),
	}
	return empty, nil
}

func (s *Service) Delete(ctx context.Context, r *shimapi.DeleteRequest) (*shimapi.DeleteResponse, error) {
	s.mu.Lock()
	p, ok := s.processes[int(r.Pid)]
	s.mu.Unlock()
	if !ok {
		p = s.initProcess
	}
	if err := p.Delete(ctx); err != nil {
		return nil, err
	}
	s.mu.Lock()
	delete(s.processes, p.Pid())
	s.mu.Unlock()
	return &shimapi.DeleteResponse{
		ExitStatus: uint32(p.Status()),
	}, nil
}

func (s *Service) Exec(ctx context.Context, r *shimapi.ExecRequest) (*shimapi.ExecResponse, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.execID++
	process, err := newExecProcess(ctx, s.path, r, s.initProcess, s.execID)
	if err != nil {
		return nil, err
	}
	pid := process.Pid()
	s.processes[pid] = process
	s.events <- &container.Event{
		Type: container.Event_EXEC_ADDED,
		ID:   s.id,
		Pid:  uint32(pid),
	}
	return &shimapi.ExecResponse{
		Pid: uint32(pid),
	}, nil
}

func (s *Service) Pty(ctx context.Context, r *shimapi.PtyRequest) (*google_protobuf.Empty, error) {
	if r.Pid == 0 {
		return nil, errors.Errorf("pid not provided in request")
	}
	ws := console.WinSize{
		Width:  uint16(r.Width),
		Height: uint16(r.Height),
	}
	s.mu.Lock()
	p, ok := s.processes[int(r.Pid)]
	s.mu.Unlock()
	if !ok {
		return nil, errors.Errorf("process does not exist %d", r.Pid)
	}
	if err := p.Resize(ws); err != nil {
		return nil, err
	}
	return empty, nil
}

func (s *Service) Events(r *shimapi.EventsRequest, stream shimapi.Shim_EventsServer) error {
	for e := range s.events {
		if err := stream.Send(e); err != nil {
			return err
		}
	}
	return nil
}

func (s *Service) State(ctx context.Context, r *shimapi.StateRequest) (*shimapi.StateResponse, error) {
	st, err := s.initProcess.ContainerStatus(ctx)
	if err != nil {
		return nil, err
	}
	status := container.Status_UNKNOWN
	switch st {
	case "created":
		status = container.Status_CREATED
	case "running":
		status = container.Status_RUNNING
	case "stopped":
		status = container.Status_STOPPED
	case "paused":
		status = container.Status_PAUSED
	}
	o := &shimapi.StateResponse{
		ID:        s.id,
		Bundle:    s.bundle,
		Pid:       uint32(s.initProcess.Pid()),
		Status:    status,
		Processes: []*container.Process{},
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	for _, p := range s.processes {
		status := container.Status_RUNNING
		if err := syscall.Kill(p.Pid(), 0); err != nil {
			if err != syscall.ESRCH {
				return nil, err
			}
			status = container.Status_STOPPED
		}
		o.Processes = append(o.Processes, &container.Process{
			Pid:    uint32(p.Pid()),
			Status: status,
		})
	}
	return o, nil
}

func (s *Service) Pause(ctx context.Context, r *shimapi.PauseRequest) (*google_protobuf.Empty, error) {
	if err := s.initProcess.Pause(ctx); err != nil {
		return nil, err
	}
	return empty, nil
}

func (s *Service) Resume(ctx context.Context, r *shimapi.ResumeRequest) (*google_protobuf.Empty, error) {
	if err := s.initProcess.Resume(ctx); err != nil {
		return nil, err
	}
	return empty, nil
}

func (s *Service) Exit(ctx context.Context, r *shimapi.ExitRequest) (*google_protobuf.Empty, error) {
	// signal ourself to exit
	if err := syscall.Kill(os.Getpid(), syscall.SIGTERM); err != nil {
		return nil, err
	}
	return empty, nil
}

func (s *Service) Kill(ctx context.Context, r *shimapi.KillRequest) (*google_protobuf.Empty, error) {
	if err := s.initProcess.Kill(ctx, r.Signal, r.All); err != nil {
		return nil, err
	}
	return empty, nil
}