420 lines
11 KiB
Go
420 lines
11 KiB
Go
package shim
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"io/ioutil"
|
|
"os"
|
|
"os/exec"
|
|
"path/filepath"
|
|
"sync"
|
|
"syscall"
|
|
|
|
"github.com/docker/containerd/execution"
|
|
"github.com/docker/containerd/log"
|
|
"github.com/opencontainers/runtime-spec/specs-go"
|
|
"github.com/pkg/errors"
|
|
"github.com/sirupsen/logrus"
|
|
)
|
|
|
|
const (
|
|
DefaultShimBinary = "containerd-shim"
|
|
|
|
pidFilename = "pid"
|
|
startTimeFilename = "starttime"
|
|
exitPipeFilename = "exit"
|
|
controlPipeFilename = "control"
|
|
initProcessID = "init"
|
|
exitStatusFilename = "exitStatus"
|
|
)
|
|
|
|
func New(ctx context.Context, root, shim, runtime string, runtimeArgs []string) (*ShimRuntime, error) {
|
|
fd, err := syscall.EpollCreate1(0)
|
|
if err != nil {
|
|
return nil, errors.Wrap(err, "epollcreate1 failed")
|
|
}
|
|
s := &ShimRuntime{
|
|
ctx: ctx,
|
|
epollFd: fd,
|
|
root: root,
|
|
binaryName: shim,
|
|
runtime: runtime,
|
|
runtimeArgs: runtimeArgs,
|
|
exitChannels: make(map[int]*process),
|
|
containers: make(map[string]*execution.Container),
|
|
}
|
|
|
|
s.loadContainers()
|
|
|
|
go s.monitor()
|
|
|
|
return s, nil
|
|
}
|
|
|
|
type ShimRuntime struct {
|
|
ctx context.Context
|
|
|
|
mutex sync.Mutex
|
|
exitChannels map[int]*process
|
|
containers map[string]*execution.Container
|
|
|
|
epollFd int
|
|
root string
|
|
binaryName string
|
|
runtime string
|
|
runtimeArgs []string
|
|
}
|
|
|
|
type ProcessOpts struct {
|
|
Bundle string
|
|
Terminal bool
|
|
Stdin string
|
|
Stdout string
|
|
Stderr string
|
|
}
|
|
|
|
type processState struct {
|
|
specs.Process
|
|
Exec bool `json:"exec"`
|
|
Stdin string `json:"containerdStdin"`
|
|
Stdout string `json:"containerdStdout"`
|
|
Stderr string `json:"containerdStderr"`
|
|
RuntimeArgs []string `json:"runtimeArgs"`
|
|
NoPivotRoot bool `json:"noPivotRoot"`
|
|
CheckpointPath string `json:"checkpoint"`
|
|
RootUID int `json:"rootUID"`
|
|
RootGID int `json:"rootGID"`
|
|
}
|
|
|
|
func (s *ShimRuntime) Create(ctx context.Context, id string, o execution.CreateOpts) (*execution.Container, error) {
|
|
log.G(s.ctx).WithFields(logrus.Fields{"container-id": id, "options": o}).Debug("Create()")
|
|
|
|
if s.getContainer(id) != nil {
|
|
return nil, execution.ErrContainerExists
|
|
}
|
|
|
|
container, err := execution.NewContainer(s.root, id, o.Bundle)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer func() {
|
|
if err != nil {
|
|
container.StateDir().Delete()
|
|
}
|
|
}()
|
|
|
|
err = ioutil.WriteFile(filepath.Join(string(container.StateDir()), "bundle"), []byte(o.Bundle), 0600)
|
|
if err != nil {
|
|
return nil, errors.Wrap(err, "failed to save bundle path to disk")
|
|
}
|
|
|
|
// extract Process spec from bundle's config.json
|
|
var spec specs.Spec
|
|
f, err := os.Open(filepath.Join(o.Bundle, "config.json"))
|
|
if err != nil {
|
|
return nil, errors.Wrap(err, "failed to open config.json")
|
|
}
|
|
defer f.Close()
|
|
if err := json.NewDecoder(f).Decode(&spec); err != nil {
|
|
return nil, errors.Wrap(err, "failed to decode container OCI specs")
|
|
}
|
|
|
|
processOpts := newProcessOpts{
|
|
shimBinary: s.binaryName,
|
|
runtime: s.runtime,
|
|
runtimeArgs: s.runtimeArgs,
|
|
container: container,
|
|
exec: false,
|
|
StartProcessOpts: execution.StartProcessOpts{
|
|
ID: initProcessID,
|
|
Spec: spec.Process,
|
|
Console: o.Console,
|
|
Stdin: o.Stdin,
|
|
Stdout: o.Stdout,
|
|
Stderr: o.Stderr,
|
|
},
|
|
}
|
|
|
|
process, err := newProcess(ctx, processOpts)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
process.ctx = log.WithModule(log.WithModule(s.ctx, "container"), id)
|
|
|
|
s.monitorProcess(process)
|
|
container.AddProcess(process, true)
|
|
|
|
s.addContainer(container)
|
|
|
|
return container, nil
|
|
}
|
|
|
|
func (s *ShimRuntime) Start(ctx context.Context, c *execution.Container) error {
|
|
log.G(s.ctx).WithFields(logrus.Fields{"container": c}).Debug("Start()")
|
|
|
|
cmd := exec.CommandContext(ctx, s.runtime, append(s.runtimeArgs, "start", c.ID())...)
|
|
out, err := cmd.CombinedOutput()
|
|
if err != nil {
|
|
return errors.Wrapf(err, "'%s start' failed with output: %v", s.runtime, string(out))
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *ShimRuntime) List(ctx context.Context) ([]*execution.Container, error) {
|
|
log.G(s.ctx).Debug("List()")
|
|
|
|
containers := make([]*execution.Container, 0)
|
|
s.mutex.Lock()
|
|
for _, c := range s.containers {
|
|
containers = append(containers, c)
|
|
}
|
|
s.mutex.Unlock()
|
|
|
|
return containers, nil
|
|
}
|
|
|
|
func (s *ShimRuntime) Load(ctx context.Context, id string) (*execution.Container, error) {
|
|
log.G(s.ctx).WithFields(logrus.Fields{"container-id": id}).Debug("Start()")
|
|
|
|
s.mutex.Lock()
|
|
c, ok := s.containers[id]
|
|
s.mutex.Unlock()
|
|
|
|
if !ok {
|
|
return nil, errors.New(execution.ErrContainerNotFound.Error())
|
|
}
|
|
|
|
return c, nil
|
|
}
|
|
|
|
func (s *ShimRuntime) Delete(ctx context.Context, c *execution.Container) error {
|
|
log.G(s.ctx).WithFields(logrus.Fields{"container": c}).Debug("Delete()")
|
|
|
|
if c.Status() != execution.Stopped {
|
|
return errors.Errorf("cannot delete a container in the '%s' state", c.Status())
|
|
}
|
|
|
|
c.StateDir().Delete()
|
|
s.removeContainer(c)
|
|
return nil
|
|
}
|
|
|
|
func (s *ShimRuntime) Pause(ctx context.Context, c *execution.Container) error {
|
|
log.G(s.ctx).WithFields(logrus.Fields{"container": c}).Debug("Pause()")
|
|
|
|
cmd := exec.CommandContext(ctx, s.runtime, append(s.runtimeArgs, "pause", c.ID())...)
|
|
out, err := cmd.CombinedOutput()
|
|
if err != nil {
|
|
return errors.Wrapf(err, "'%s pause' failed with output: %v", s.runtime, string(out))
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *ShimRuntime) Resume(ctx context.Context, c *execution.Container) error {
|
|
log.G(s.ctx).WithFields(logrus.Fields{"container": c}).Debug("Resume()")
|
|
|
|
cmd := exec.CommandContext(ctx, s.runtime, append(s.runtimeArgs, "resume", c.ID())...)
|
|
out, err := cmd.CombinedOutput()
|
|
if err != nil {
|
|
return errors.Wrapf(err, "'%s resume' failed with output: %v", s.runtime, string(out))
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *ShimRuntime) StartProcess(ctx context.Context, c *execution.Container, o execution.StartProcessOpts) (p execution.Process, err error) {
|
|
log.G(s.ctx).WithFields(logrus.Fields{"container": c, "options": o}).Debug("StartProcess()")
|
|
|
|
processOpts := newProcessOpts{
|
|
shimBinary: s.binaryName,
|
|
runtime: s.runtime,
|
|
runtimeArgs: s.runtimeArgs,
|
|
container: c,
|
|
exec: true,
|
|
StartProcessOpts: o,
|
|
}
|
|
process, err := newProcess(ctx, processOpts)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
process.status = execution.Running
|
|
s.monitorProcess(process)
|
|
|
|
c.AddProcess(process, false)
|
|
return process, nil
|
|
}
|
|
|
|
func (s *ShimRuntime) SignalProcess(ctx context.Context, c *execution.Container, id string, sig os.Signal) error {
|
|
log.G(s.ctx).WithFields(logrus.Fields{"container": c, "process-id": id, "signal": sig}).
|
|
Debug("SignalProcess()")
|
|
|
|
process := c.GetProcess(id)
|
|
if process == nil {
|
|
return errors.Errorf("no such process %s", id)
|
|
}
|
|
err := syscall.Kill(int(process.Pid()), sig.(syscall.Signal))
|
|
if err != nil {
|
|
return errors.Wrapf(err, "failed to send %v signal to process %v", sig, process.Pid())
|
|
}
|
|
return err
|
|
}
|
|
|
|
func (s *ShimRuntime) DeleteProcess(ctx context.Context, c *execution.Container, id string) error {
|
|
log.G(s.ctx).WithFields(logrus.Fields{"container": c, "process-id": id}).
|
|
Debug("DeleteProcess()")
|
|
|
|
c.RemoveProcess(id)
|
|
return c.StateDir().DeleteProcess(id)
|
|
}
|
|
|
|
//
|
|
//
|
|
//
|
|
|
|
func (s *ShimRuntime) monitor() {
|
|
var events [128]syscall.EpollEvent
|
|
for {
|
|
n, err := syscall.EpollWait(s.epollFd, events[:], -1)
|
|
if err != nil {
|
|
if err == syscall.EINTR {
|
|
continue
|
|
}
|
|
log.G(s.ctx).Error("epollwait failed:", err)
|
|
}
|
|
|
|
for i := 0; i < n; i++ {
|
|
fd := int(events[i].Fd)
|
|
|
|
s.mutex.Lock()
|
|
p := s.exitChannels[fd]
|
|
delete(s.exitChannels, fd)
|
|
s.mutex.Unlock()
|
|
|
|
if err = syscall.EpollCtl(s.epollFd, syscall.EPOLL_CTL_DEL, fd, &syscall.EpollEvent{
|
|
Events: syscall.EPOLLHUP,
|
|
Fd: int32(fd),
|
|
}); err != nil {
|
|
log.G(s.ctx).Error("epollctl deletion failed:", err)
|
|
}
|
|
|
|
close(p.exitChan)
|
|
}
|
|
}
|
|
}
|
|
|
|
func (s *ShimRuntime) addContainer(c *execution.Container) {
|
|
s.mutex.Lock()
|
|
s.containers[c.ID()] = c
|
|
s.mutex.Unlock()
|
|
}
|
|
|
|
func (s *ShimRuntime) removeContainer(c *execution.Container) {
|
|
s.mutex.Lock()
|
|
delete(s.containers, c.ID())
|
|
s.mutex.Unlock()
|
|
}
|
|
|
|
func (s *ShimRuntime) getContainer(id string) *execution.Container {
|
|
s.mutex.Lock()
|
|
c := s.containers[id]
|
|
s.mutex.Unlock()
|
|
|
|
return c
|
|
}
|
|
|
|
// monitorProcess adds a process to the list of monitored process if
|
|
// we fail to do so, we closed the exitChan channel used by Wait().
|
|
// Since service always call on Wait() for generating "exit" events,
|
|
// this will ensure the process gets killed
|
|
func (s *ShimRuntime) monitorProcess(p *process) {
|
|
if p.status == execution.Stopped {
|
|
close(p.exitChan)
|
|
return
|
|
}
|
|
|
|
fd := int(p.exitPipe.Fd())
|
|
event := syscall.EpollEvent{
|
|
Fd: int32(fd),
|
|
Events: syscall.EPOLLHUP,
|
|
}
|
|
s.mutex.Lock()
|
|
s.exitChannels[fd] = p
|
|
s.mutex.Unlock()
|
|
if err := syscall.EpollCtl(s.epollFd, syscall.EPOLL_CTL_ADD, fd, &event); err != nil {
|
|
s.mutex.Lock()
|
|
delete(s.exitChannels, fd)
|
|
s.mutex.Unlock()
|
|
close(p.exitChan)
|
|
return
|
|
}
|
|
|
|
// TODO: take care of the OOM handler
|
|
}
|
|
|
|
func (s *ShimRuntime) unmonitorProcess(p *process) {
|
|
s.mutex.Lock()
|
|
for fd, proc := range s.exitChannels {
|
|
if proc == p {
|
|
delete(s.exitChannels, fd)
|
|
break
|
|
}
|
|
}
|
|
s.mutex.Unlock()
|
|
}
|
|
|
|
func (s *ShimRuntime) loadContainers() {
|
|
cs, err := ioutil.ReadDir(s.root)
|
|
if err != nil {
|
|
log.G(s.ctx).WithField("statedir", s.root).
|
|
Warn("failed to load containers, state dir cannot be listed:", err)
|
|
return
|
|
}
|
|
|
|
for _, c := range cs {
|
|
if !c.IsDir() {
|
|
continue
|
|
}
|
|
|
|
stateDir, err := execution.LoadStateDir(s.root, c.Name())
|
|
if err != nil {
|
|
// We should never fail the above call unless someone
|
|
// delete the directory while we're loading
|
|
log.G(s.ctx).WithFields(logrus.Fields{"container": c.Name(), "statedir": s.root}).
|
|
Warn("failed to load container statedir:", err)
|
|
continue
|
|
}
|
|
bundle, err := ioutil.ReadFile(filepath.Join(string(stateDir), "bundle"))
|
|
if err != nil {
|
|
log.G(s.ctx).WithField("container", c.Name()).
|
|
Warn("failed to load container bundle path:", err)
|
|
continue
|
|
}
|
|
|
|
container := execution.LoadContainer(stateDir, c.Name(), string(bundle), execution.Unknown)
|
|
s.addContainer(container)
|
|
|
|
processDirs, err := stateDir.Processes()
|
|
if err != nil {
|
|
log.G(s.ctx).WithField("container", c.Name()).
|
|
Warn("failed to retrieve container processes:", err)
|
|
continue
|
|
}
|
|
|
|
for _, procStateRoot := range processDirs {
|
|
id := filepath.Base(procStateRoot)
|
|
proc, err := loadProcess(procStateRoot, id)
|
|
if err != nil {
|
|
log.G(s.ctx).WithFields(logrus.Fields{"container": c.Name(), "process": id}).
|
|
Warn("failed to load process:", err)
|
|
s.removeContainer(container)
|
|
for _, p := range container.Processes() {
|
|
s.unmonitorProcess(p.(*process))
|
|
}
|
|
break
|
|
}
|
|
proc.ctx = log.WithModule(log.WithModule(s.ctx, "container"), container.ID())
|
|
container.AddProcess(proc, proc.ID() == initProcessID)
|
|
s.monitorProcess(proc)
|
|
}
|
|
}
|
|
}
|