Move shim process code to package

Signed-off-by: Michael Crosby <crosbymichael@gmail.com>
This commit is contained in:
Michael Crosby 2017-11-09 11:17:53 -05:00
parent 92ca22c997
commit 6e25898ff0
11 changed files with 155 additions and 132 deletions

View File

@ -1,6 +1,6 @@
// +build !windows // +build !windows
package shim package proc
import ( import (
"context" "context"

View File

@ -1,6 +1,6 @@
// +build !windows // +build !windows
package shim package proc
import ( import (
"context" "context"
@ -27,7 +27,7 @@ import (
type execProcess struct { type execProcess struct {
wg sync.WaitGroup wg sync.WaitGroup
processState State
mu sync.Mutex mu sync.Mutex
id string id string
@ -38,15 +38,16 @@ type execProcess struct {
pid int pid int
closers []io.Closer closers []io.Closer
stdin io.Closer stdin io.Closer
stdio stdio stdio Stdio
path string path string
spec specs.Process spec specs.Process
parent *initProcess parent *Init
waitBlock chan struct{} waitBlock chan struct{}
} }
func newExecProcess(context context.Context, path string, r *shimapi.ExecProcessRequest, parent *initProcess, id string) (process, error) { // NewExec returns a new exec'd process
func NewExec(context context.Context, path string, r *shimapi.ExecProcessRequest, parent *Init, id string) (Process, error) {
if err := identifiers.Validate(id); err != nil { if err := identifiers.Validate(id); err != nil {
return nil, errors.Wrapf(err, "invalid exec id") return nil, errors.Wrapf(err, "invalid exec id")
} }
@ -62,15 +63,15 @@ func newExecProcess(context context.Context, path string, r *shimapi.ExecProcess
path: path, path: path,
parent: parent, parent: parent,
spec: spec, spec: spec,
stdio: stdio{ stdio: Stdio{
stdin: r.Stdin, Stdin: r.Stdin,
stdout: r.Stdout, Stdout: r.Stdout,
stderr: r.Stderr, Stderr: r.Stderr,
terminal: r.Terminal, Terminal: r.Terminal,
}, },
waitBlock: make(chan struct{}), waitBlock: make(chan struct{}),
} }
e.processState = &execCreatedState{p: e} e.State = &execCreatedState{p: e}
return e, nil return e, nil
} }
@ -103,7 +104,7 @@ func (e *execProcess) ExitedAt() time.Time {
func (e *execProcess) setExited(status int) { func (e *execProcess) setExited(status int) {
e.status = status e.status = status
e.exited = time.Now() e.exited = time.Now()
e.parent.platform.shutdownConsole(context.Background(), e.console) e.parent.platform.ShutdownConsole(context.Background(), e.console)
close(e.waitBlock) close(e.waitBlock)
} }
@ -142,7 +143,7 @@ func (e *execProcess) Stdin() io.Closer {
return e.stdin return e.stdin
} }
func (e *execProcess) Stdio() stdio { func (e *execProcess) Stdio() Stdio {
return e.stdio return e.stdio
} }
@ -151,12 +152,12 @@ func (e *execProcess) start(ctx context.Context) (err error) {
socket *runc.Socket socket *runc.Socket
pidfile = filepath.Join(e.path, fmt.Sprintf("%s.pid", e.id)) 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 { if socket, err = runc.NewTempConsoleSocket(); err != nil {
return errors.Wrap(err, "failed to create runc console socket") return errors.Wrap(err, "failed to create runc console socket")
} }
defer socket.Close() defer socket.Close()
} else if e.stdio.isNull() { } else if e.stdio.IsNull() {
if e.io, err = runc.NewNullIO(); err != nil { if e.io, err = runc.NewNullIO(); err != nil {
return errors.Wrap(err, "creating new NULL IO") return errors.Wrap(err, "creating new NULL IO")
} }
@ -176,10 +177,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 { if err := e.parent.runtime.Exec(ctx, e.parent.id, e.spec, opts); err != nil {
return e.parent.runtimeError(err, "OCI runtime exec failed") return e.parent.runtimeError(err, "OCI runtime exec failed")
} }
if e.stdio.stdin != "" { if e.stdio.Stdin != "" {
sc, err := fifo.OpenFifo(ctx, e.stdio.stdin, syscall.O_WRONLY|syscall.O_NONBLOCK, 0) sc, err := fifo.OpenFifo(ctx, e.stdio.Stdin, syscall.O_WRONLY|syscall.O_NONBLOCK, 0)
if err != nil { 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.closers = append(e.closers, sc)
e.stdin = sc e.stdin = sc
@ -190,11 +191,11 @@ func (e *execProcess) start(ctx context.Context) (err error) {
if err != nil { if err != nil {
return errors.Wrap(err, "failed to retrieve console master") 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") return errors.Wrap(err, "failed to start console copy")
} }
} else if !e.stdio.isNull() { } 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 { 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") return errors.Wrap(err, "failed to start io pipe copy")
} }
} }

View File

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

View File

@ -1,6 +1,6 @@
// +build !windows // +build !windows
package shim package proc
import ( import (
"context" "context"
@ -30,7 +30,7 @@ import (
// InitPidFile name of the file that contains the init pid // InitPidFile name of the file that contains the init pid
const InitPidFile = "init.pid" const InitPidFile = "init.pid"
type initProcess struct { type Init struct {
wg sync.WaitGroup wg sync.WaitGroup
initState initState
@ -47,7 +47,7 @@ type initProcess struct {
id string id string
bundle string bundle string
console console.Console console console.Console
platform platform platform Platform
io runc.IO io runc.IO
runtime *runc.Runc runtime *runc.Runc
status int status int
@ -55,13 +55,14 @@ type initProcess struct {
pid int pid int
closers []io.Closer closers []io.Closer
stdin io.Closer stdin io.Closer
stdio stdio stdio Stdio
rootfs string rootfs string
IoUID int IoUID int
IoGID int IoGID int
} }
func (s *Service) newInitProcess(context context.Context, r *shimapi.CreateTaskRequest) (*initProcess, error) { // New returns a new init process
func New(context context.Context, path, workDir, runtimeRoot, namespace, criu string, systemdCgroup bool, platform Platform, r *shimapi.CreateTaskRequest) (*Init, error) {
var success bool var success bool
if err := identifiers.Validate(r.ID); err != nil { if err := identifiers.Validate(r.ID); err != nil {
@ -76,7 +77,7 @@ func (s *Service) newInitProcess(context context.Context, r *shimapi.CreateTaskR
options = *v.(*runctypes.CreateOptions) options = *v.(*runctypes.CreateOptions)
} }
rootfs := filepath.Join(s.config.Path, "rootfs") rootfs := filepath.Join(path, "rootfs")
// count the number of successful mounts so we can undo // count the number of successful mounts so we can undo
// what was actually done rather than what should have been // what was actually done rather than what should have been
// done. // done.
@ -98,32 +99,32 @@ func (s *Service) newInitProcess(context context.Context, r *shimapi.CreateTaskR
return nil, errors.Wrapf(err, "failed to mount rootfs component %v", m) return nil, errors.Wrapf(err, "failed to mount rootfs component %v", m)
} }
} }
root := s.config.RuntimeRoot root := runtimeRoot
if root == "" { if root == "" {
root = RuncRoot root = RuncRoot
} }
runtime := &runc.Runc{ runtime := &runc.Runc{
Command: r.Runtime, Command: r.Runtime,
Log: filepath.Join(s.config.Path, "log.json"), Log: filepath.Join(path, "log.json"),
LogFormat: runc.JSON, LogFormat: runc.JSON,
PdeathSignal: syscall.SIGKILL, PdeathSignal: syscall.SIGKILL,
Root: filepath.Join(root, s.config.Namespace), Root: filepath.Join(root, namespace),
Criu: s.config.Criu, Criu: criu,
SystemdCgroup: s.config.SystemdCgroup, SystemdCgroup: systemdCgroup,
} }
p := &initProcess{ p := &Init{
id: r.ID, id: r.ID,
bundle: r.Bundle, bundle: r.Bundle,
runtime: runtime, runtime: runtime,
platform: s.platform, platform: platform,
stdio: stdio{ stdio: Stdio{
stdin: r.Stdin, Stdin: r.Stdin,
stdout: r.Stdout, Stdout: r.Stdout,
stderr: r.Stderr, Stderr: r.Stderr,
terminal: r.Terminal, Terminal: r.Terminal,
}, },
rootfs: rootfs, rootfs: rootfs,
workDir: s.config.WorkDir, workDir: workDir,
status: 0, status: 0,
waitBlock: make(chan struct{}), waitBlock: make(chan struct{}),
IoUID: int(options.IoUid), IoUID: int(options.IoUid),
@ -148,7 +149,7 @@ func (s *Service) newInitProcess(context context.Context, r *shimapi.CreateTaskR
return nil, errors.Wrap(err, "failed to create OCI runtime io pipes") return nil, errors.Wrap(err, "failed to create OCI runtime io pipes")
} }
} }
pidFile := filepath.Join(s.config.Path, InitPidFile) pidFile := filepath.Join(path, InitPidFile)
if r.Checkpoint != "" { if r.Checkpoint != "" {
opts := &runc.RestoreOpts{ opts := &runc.RestoreOpts{
CheckpointOpts: runc.CheckpointOpts{ CheckpointOpts: runc.CheckpointOpts{
@ -195,7 +196,7 @@ func (s *Service) newInitProcess(context context.Context, r *shimapi.CreateTaskR
if err != nil { if err != nil {
return nil, errors.Wrap(err, "failed to retrieve console master") return nil, errors.Wrap(err, "failed to retrieve console master")
} }
console, err = s.platform.copyConsole(context, console, r.Stdin, r.Stdout, r.Stderr, &p.wg, &copyWaitGroup) console, err = platform.CopyConsole(context, console, r.Stdin, r.Stdout, r.Stderr, &p.wg, &copyWaitGroup)
if err != nil { if err != nil {
return nil, errors.Wrap(err, "failed to start console copy") return nil, errors.Wrap(err, "failed to start console copy")
} }
@ -216,31 +217,31 @@ func (s *Service) newInitProcess(context context.Context, r *shimapi.CreateTaskR
return p, nil return p, nil
} }
func (p *initProcess) Wait() { func (p *Init) Wait() {
<-p.waitBlock <-p.waitBlock
} }
func (p *initProcess) ID() string { func (p *Init) ID() string {
return p.id return p.id
} }
func (p *initProcess) Pid() int { func (p *Init) Pid() int {
return p.pid return p.pid
} }
func (p *initProcess) ExitStatus() int { func (p *Init) ExitStatus() int {
p.mu.Lock() p.mu.Lock()
defer p.mu.Unlock() defer p.mu.Unlock()
return p.status return p.status
} }
func (p *initProcess) ExitedAt() time.Time { func (p *Init) ExitedAt() time.Time {
p.mu.Lock() p.mu.Lock()
defer p.mu.Unlock() defer p.mu.Unlock()
return p.exited return p.exited
} }
func (p *initProcess) Status(ctx context.Context) (string, error) { func (p *Init) Status(ctx context.Context) (string, error) {
p.mu.Lock() p.mu.Lock()
defer p.mu.Unlock() defer p.mu.Unlock()
c, err := p.runtime.State(ctx, p.id) c, err := p.runtime.State(ctx, p.id)
@ -253,20 +254,20 @@ func (p *initProcess) Status(ctx context.Context) (string, error) {
return c.Status, nil return c.Status, nil
} }
func (p *initProcess) start(context context.Context) error { func (p *Init) start(context context.Context) error {
err := p.runtime.Start(context, p.id) err := p.runtime.Start(context, p.id)
return p.runtimeError(err, "OCI runtime start failed") return p.runtimeError(err, "OCI runtime start failed")
} }
func (p *initProcess) setExited(status int) { func (p *Init) setExited(status int) {
p.exited = time.Now() p.exited = time.Now()
p.status = status p.status = status
p.platform.shutdownConsole(context.Background(), p.console) p.platform.ShutdownConsole(context.Background(), p.console)
close(p.waitBlock) close(p.waitBlock)
} }
func (p *initProcess) delete(context context.Context) error { func (p *Init) delete(context context.Context) error {
p.killAll(context) p.KillAll(context)
p.wg.Wait() p.wg.Wait()
err := p.runtime.Delete(context, p.id, nil) err := p.runtime.Delete(context, p.id, nil)
// ignore errors if a runtime has already deleted the process // ignore errors if a runtime has already deleted the process
@ -296,42 +297,47 @@ func (p *initProcess) delete(context context.Context) error {
return err return err
} }
func (p *initProcess) resize(ws console.WinSize) error { func (p *Init) resize(ws console.WinSize) error {
if p.console == nil { if p.console == nil {
return nil return nil
} }
return p.console.Resize(ws) return p.console.Resize(ws)
} }
func (p *initProcess) pause(context context.Context) error { func (p *Init) pause(context context.Context) error {
err := p.runtime.Pause(context, p.id) err := p.runtime.Pause(context, p.id)
return p.runtimeError(err, "OCI runtime pause failed") return p.runtimeError(err, "OCI runtime pause failed")
} }
func (p *initProcess) resume(context context.Context) error { func (p *Init) resume(context context.Context) error {
err := p.runtime.Resume(context, p.id) err := p.runtime.Resume(context, p.id)
return p.runtimeError(err, "OCI runtime resume failed") return p.runtimeError(err, "OCI runtime resume failed")
} }
func (p *initProcess) kill(context context.Context, signal uint32, all bool) error { func (p *Init) kill(context context.Context, signal uint32, all bool) error {
err := p.runtime.Kill(context, p.id, int(signal), &runc.KillOpts{ err := p.runtime.Kill(context, p.id, int(signal), &runc.KillOpts{
All: all, All: all,
}) })
return checkKillError(err) return checkKillError(err)
} }
func (p *initProcess) killAll(context context.Context) error { func (p *Init) KillAll(context context.Context) error {
err := p.runtime.Kill(context, p.id, int(syscall.SIGKILL), &runc.KillOpts{ err := p.runtime.Kill(context, p.id, int(syscall.SIGKILL), &runc.KillOpts{
All: true, All: true,
}) })
return p.runtimeError(err, "OCI runtime killall failed") return p.runtimeError(err, "OCI runtime killall failed")
} }
func (p *initProcess) Stdin() io.Closer { func (p *Init) Stdin() io.Closer {
return p.stdin return p.stdin
} }
func (p *initProcess) checkpoint(context context.Context, r *shimapi.CheckpointTaskRequest) error { // Runtime returns the OCI runtime configured for the init process
func (p *Init) Runtime() *runc.Runc {
return p.runtime
}
func (p *Init) checkpoint(context context.Context, r *shimapi.CheckpointTaskRequest) error {
var options runctypes.CheckpointOptions var options runctypes.CheckpointOptions
if r.Options != nil { if r.Options != nil {
v, err := typeurl.UnmarshalAny(r.Options) v, err := typeurl.UnmarshalAny(r.Options)
@ -364,7 +370,7 @@ func (p *initProcess) checkpoint(context context.Context, r *shimapi.CheckpointT
return nil return nil
} }
func (p *initProcess) update(context context.Context, r *shimapi.UpdateTaskRequest) error { func (p *Init) update(context context.Context, r *shimapi.UpdateTaskRequest) error {
var resources specs.LinuxResources var resources specs.LinuxResources
if err := json.Unmarshal(r.Resources.Value, &resources); err != nil { if err := json.Unmarshal(r.Resources.Value, &resources); err != nil {
return err return err
@ -372,11 +378,11 @@ func (p *initProcess) update(context context.Context, r *shimapi.UpdateTaskReque
return p.runtime.Update(context, p.id, &resources) return p.runtime.Update(context, p.id, &resources)
} }
func (p *initProcess) Stdio() stdio { func (p *Init) Stdio() Stdio {
return p.stdio return p.stdio
} }
func (p *initProcess) runtimeError(rErr error, msg string) error { func (p *Init) runtimeError(rErr error, msg string) error {
if rErr == nil { if rErr == nil {
return nil return nil
} }

View File

@ -1,6 +1,6 @@
// +build !windows // +build !windows
package shim package proc
import ( import (
"context" "context"
@ -16,7 +16,7 @@ import (
) )
type initState interface { type initState interface {
processState State
Pause(context.Context) error Pause(context.Context) error
Resume(context.Context) error Resume(context.Context) error
@ -25,7 +25,7 @@ type initState interface {
} }
type createdState struct { type createdState struct {
p *initProcess p *Init
} }
func (s *createdState) transition(name string) error { func (s *createdState) transition(name string) error {
@ -114,7 +114,7 @@ func (s *createdState) SetExited(status int) {
} }
type createdCheckpointState struct { type createdCheckpointState struct {
p *initProcess p *Init
opts *runc.RestoreOpts opts *runc.RestoreOpts
} }
@ -175,17 +175,17 @@ func (s *createdCheckpointState) Start(ctx context.Context) error {
return p.runtimeError(err, "OCI runtime restore failed") return p.runtimeError(err, "OCI runtime restore failed")
} }
sio := p.stdio sio := p.stdio
if sio.stdin != "" { if sio.Stdin != "" {
sc, err := fifo.OpenFifo(ctx, sio.stdin, syscall.O_WRONLY|syscall.O_NONBLOCK, 0) sc, err := fifo.OpenFifo(ctx, sio.Stdin, syscall.O_WRONLY|syscall.O_NONBLOCK, 0)
if err != nil { if err != nil {
return errors.Wrapf(err, "failed to open stdin fifo %s", sio.stdin) return errors.Wrapf(err, "failed to open stdin fifo %s", sio.Stdin)
} }
p.stdin = sc p.stdin = sc
p.closers = append(p.closers, sc) p.closers = append(p.closers, sc)
} }
var copyWaitGroup sync.WaitGroup var copyWaitGroup sync.WaitGroup
if !sio.isNull() { if !sio.IsNull() {
if err := copyPipes(ctx, p.io, sio.stdin, sio.stdout, sio.stderr, &p.wg, &copyWaitGroup); err != nil { if err := copyPipes(ctx, p.io, sio.Stdin, sio.Stdout, sio.Stderr, &p.wg, &copyWaitGroup); err != nil {
return errors.Wrap(err, "failed to start io pipe copy") return errors.Wrap(err, "failed to start io pipe copy")
} }
} }
@ -228,7 +228,7 @@ func (s *createdCheckpointState) SetExited(status int) {
} }
type runningState struct { type runningState struct {
p *initProcess p *Init
} }
func (s *runningState) transition(name string) error { func (s *runningState) transition(name string) error {
@ -313,7 +313,7 @@ func (s *runningState) SetExited(status int) {
} }
type pausedState struct { type pausedState struct {
p *initProcess p *Init
} }
func (s *pausedState) transition(name string) error { func (s *pausedState) transition(name string) error {
@ -400,7 +400,7 @@ func (s *pausedState) SetExited(status int) {
} }
type stoppedState struct { type stoppedState struct {
p *initProcess p *Init
} }
func (s *stoppedState) transition(name string) error { func (s *stoppedState) transition(name string) error {

View File

@ -1,6 +1,6 @@
// +build !windows // +build !windows
package shim package proc
import ( import (
"context" "context"

View File

@ -1,30 +1,36 @@
// +build !windows // +build !windows
package shim package proc
import ( import (
"context" "context"
"io" "io"
"sync"
"time" "time"
"github.com/containerd/console" "github.com/containerd/console"
"github.com/pkg/errors" "github.com/pkg/errors"
) )
type stdio struct { // RuncRoot is the path to the root runc state directory
stdin string const RuncRoot = "/run/containerd/runc"
stdout string
stderr string // Stdio of a process
terminal bool type Stdio struct {
Stdin string
Stdout string
Stderr string
Terminal bool
} }
func (s stdio) isNull() bool { // IsNull returns true if the stdio is not defined
return s.stdin == "" && s.stdout == "" && s.stderr == "" func (s Stdio) IsNull() bool {
return s.Stdin == "" && s.Stdout == "" && s.Stderr == ""
} }
type process interface { // Process on a linux system
processState type Process interface {
State
// ID returns the id for the process // ID returns the id for the process
ID() string ID() string
// Pid returns the pid for the process // Pid returns the pid for the process
@ -36,14 +42,15 @@ type process interface {
// Stdin returns the process STDIN // Stdin returns the process STDIN
Stdin() io.Closer Stdin() io.Closer
// Stdio returns io information for the container // Stdio returns io information for the container
Stdio() stdio Stdio() Stdio
// Status returns the process status // Status returns the process status
Status(context.Context) (string, error) Status(context.Context) (string, error)
// Wait blocks until the process has exited // Wait blocks until the process has exited
Wait() Wait()
} }
type processState interface { // State of a process
type State interface {
// Resize resizes the process console // Resize resizes the process console
Resize(ws console.WinSize) error Resize(ws console.WinSize) error
// Start execution of the process // Start execution of the process
@ -71,3 +78,12 @@ func stateName(v interface{}) string {
} }
panic(errors.Errorf("invalid state %v", v)) panic(errors.Errorf("invalid state %v", v))
} }
// Platform handles platform-specific behavior that may differs across
// platform implementations
type Platform interface {
CopyConsole(ctx context.Context, console console.Console, stdin, stdout, stderr string,
wg, cwg *sync.WaitGroup) (console.Console, error)
ShutdownConsole(ctx context.Context, console console.Console) error
Close() error
}

View File

@ -1,6 +1,6 @@
// +build !windows // +build !windows
package shim package proc
import ( import (
"encoding/json" "encoding/json"

View File

@ -15,6 +15,7 @@ import (
"github.com/containerd/containerd/api/types/task" "github.com/containerd/containerd/api/types/task"
"github.com/containerd/containerd/errdefs" "github.com/containerd/containerd/errdefs"
"github.com/containerd/containerd/events" "github.com/containerd/containerd/events"
"github.com/containerd/containerd/linux/proc"
"github.com/containerd/containerd/linux/runctypes" "github.com/containerd/containerd/linux/runctypes"
shimapi "github.com/containerd/containerd/linux/shim/v1" shimapi "github.com/containerd/containerd/linux/shim/v1"
"github.com/containerd/containerd/log" "github.com/containerd/containerd/log"
@ -31,9 +32,6 @@ import (
var empty = &google_protobuf.Empty{} var empty = &google_protobuf.Empty{}
// RuncRoot is the path to the root runc state directory
const RuncRoot = "/run/containerd/runc"
// Config contains shim specific configuration // Config contains shim specific configuration
type Config struct { type Config struct {
Path string Path string
@ -58,7 +56,7 @@ func NewService(config Config, publisher events.Publisher) (*Service, error) {
s := &Service{ s := &Service{
config: config, config: config,
context: context, context: context,
processes: make(map[string]process), processes: make(map[string]proc.Process),
events: make(chan interface{}, 128), events: make(chan interface{}, 128),
ec: reaper.Default.Subscribe(), ec: reaper.Default.Subscribe(),
} }
@ -70,23 +68,15 @@ func NewService(config Config, publisher events.Publisher) (*Service, error) {
return s, nil return s, nil
} }
// platform handles platform-specific behavior that may differs across
// platform implementations
type platform interface {
copyConsole(ctx context.Context, console console.Console, stdin, stdout, stderr string, wg, cwg *sync.WaitGroup) (console.Console, error)
shutdownConsole(ctx context.Context, console console.Console) error
close() error
}
// Service is the shim implementation of a remote shim over GRPC // Service is the shim implementation of a remote shim over GRPC
type Service struct { type Service struct {
mu sync.Mutex mu sync.Mutex
config Config config Config
context context.Context context context.Context
processes map[string]process processes map[string]proc.Process
events chan interface{} events chan interface{}
platform platform platform proc.Platform
ec chan runc.Exit ec chan runc.Exit
// Filled by Create() // Filled by Create()
@ -98,7 +88,17 @@ type Service struct {
func (s *Service) Create(ctx context.Context, r *shimapi.CreateTaskRequest) (*shimapi.CreateTaskResponse, error) { func (s *Service) Create(ctx context.Context, r *shimapi.CreateTaskRequest) (*shimapi.CreateTaskResponse, error) {
s.mu.Lock() s.mu.Lock()
defer s.mu.Unlock() defer s.mu.Unlock()
process, err := s.newInitProcess(ctx, r) 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,
r,
)
if err != nil { if err != nil {
return nil, errdefs.ToGRPC(err) return nil, errdefs.ToGRPC(err)
} }
@ -168,7 +168,7 @@ func (s *Service) Delete(ctx context.Context, r *google_protobuf.Empty) (*shimap
return nil, err return nil, err
} }
delete(s.processes, s.id) delete(s.processes, s.id)
s.platform.close() s.platform.Close()
s.events <- &eventsapi.TaskDelete{ s.events <- &eventsapi.TaskDelete{
ContainerID: s.id, ContainerID: s.id,
ExitStatus: uint32(p.ExitStatus()), ExitStatus: uint32(p.ExitStatus()),
@ -218,7 +218,7 @@ func (s *Service) Exec(ctx context.Context, r *shimapi.ExecProcessRequest) (*goo
return nil, errdefs.ToGRPCf(errdefs.ErrFailedPrecondition, "container must be created") return nil, errdefs.ToGRPCf(errdefs.ErrFailedPrecondition, "container must be created")
} }
process, err := newExecProcess(ctx, s.config.Path, r, p.(*initProcess), r.ID) process, err := proc.NewExec(ctx, s.config.Path, r, p.(*proc.Init), r.ID)
if err != nil { if err != nil {
return nil, errdefs.ToGRPC(err) return nil, errdefs.ToGRPC(err)
} }
@ -283,10 +283,10 @@ func (s *Service) State(ctx context.Context, r *shimapi.StateRequest) (*shimapi.
Bundle: s.bundle, Bundle: s.bundle,
Pid: uint32(p.Pid()), Pid: uint32(p.Pid()),
Status: status, Status: status,
Stdin: sio.stdin, Stdin: sio.Stdin,
Stdout: sio.stdout, Stdout: sio.Stdout,
Stderr: sio.stderr, Stderr: sio.Stderr,
Terminal: sio.terminal, Terminal: sio.Terminal,
ExitStatus: uint32(p.ExitStatus()), ExitStatus: uint32(p.ExitStatus()),
ExitedAt: p.ExitedAt(), ExitedAt: p.ExitedAt(),
}, nil }, nil
@ -300,7 +300,7 @@ func (s *Service) Pause(ctx context.Context, r *google_protobuf.Empty) (*google_
if p == nil { if p == nil {
return nil, errdefs.ToGRPCf(errdefs.ErrFailedPrecondition, "container must be created") return nil, errdefs.ToGRPCf(errdefs.ErrFailedPrecondition, "container must be created")
} }
if err := p.(*initProcess).Pause(ctx); err != nil { if err := p.(*proc.Init).Pause(ctx); err != nil {
return nil, err return nil, err
} }
s.events <- &eventsapi.TaskPaused{ s.events <- &eventsapi.TaskPaused{
@ -317,7 +317,7 @@ func (s *Service) Resume(ctx context.Context, r *google_protobuf.Empty) (*google
if p == nil { if p == nil {
return nil, errdefs.ToGRPCf(errdefs.ErrFailedPrecondition, "container must be created") return nil, errdefs.ToGRPCf(errdefs.ErrFailedPrecondition, "container must be created")
} }
if err := p.(*initProcess).Resume(ctx); err != nil { if err := p.(*proc.Init).Resume(ctx); err != nil {
return nil, err return nil, err
} }
s.events <- &eventsapi.TaskResumed{ s.events <- &eventsapi.TaskResumed{
@ -406,7 +406,7 @@ func (s *Service) Checkpoint(ctx context.Context, r *shimapi.CheckpointTaskReque
if p == nil { if p == nil {
return nil, errdefs.ToGRPCf(errdefs.ErrFailedPrecondition, "container must be created") return nil, errdefs.ToGRPCf(errdefs.ErrFailedPrecondition, "container must be created")
} }
if err := p.(*initProcess).Checkpoint(ctx, r); err != nil { if err := p.(*proc.Init).Checkpoint(ctx, r); err != nil {
return nil, errdefs.ToGRPC(err) return nil, errdefs.ToGRPC(err)
} }
s.events <- &eventsapi.TaskCheckpointed{ s.events <- &eventsapi.TaskCheckpointed{
@ -430,7 +430,7 @@ func (s *Service) Update(ctx context.Context, r *shimapi.UpdateTaskRequest) (*go
if p == nil { if p == nil {
return nil, errdefs.ToGRPCf(errdefs.ErrFailedPrecondition, "container must be created") return nil, errdefs.ToGRPCf(errdefs.ErrFailedPrecondition, "container must be created")
} }
if err := p.(*initProcess).Update(ctx, r); err != nil { if err := p.(*proc.Init).Update(ctx, r); err != nil {
return nil, errdefs.ToGRPC(err) return nil, errdefs.ToGRPC(err)
} }
return empty, nil return empty, nil
@ -463,9 +463,9 @@ func (s *Service) checkProcesses(e runc.Exit) {
defer s.mu.Unlock() defer s.mu.Unlock()
for _, p := range s.processes { for _, p := range s.processes {
if p.Pid() == e.Pid { if p.Pid() == e.Pid {
if ip, ok := p.(*initProcess); ok { if ip, ok := p.(*proc.Init); ok {
// Ensure all children are killed // Ensure all children are killed
if err := ip.killAll(s.context); err != nil { if err := ip.KillAll(s.context); err != nil {
log.G(s.context).WithError(err).WithField("id", ip.ID()). log.G(s.context).WithError(err).WithField("id", ip.ID()).
Error("failed to kill init's children") Error("failed to kill init's children")
} }
@ -491,7 +491,7 @@ func (s *Service) getContainerPids(ctx context.Context, id string) ([]uint32, er
return nil, errors.Wrapf(errdefs.ErrFailedPrecondition, "container must be created") return nil, errors.Wrapf(errdefs.ErrFailedPrecondition, "container must be created")
} }
ps, err := p.(*initProcess).runtime.Ps(ctx, id) ps, err := p.(*proc.Init).Runtime().Ps(ctx, id)
if err != nil { if err != nil {
return nil, err return nil, err
} }

View File

@ -1,6 +1,7 @@
package shim package shim
import ( import (
"context"
"io" "io"
"sync" "sync"
"syscall" "syscall"
@ -8,14 +9,13 @@ import (
"github.com/containerd/console" "github.com/containerd/console"
"github.com/containerd/fifo" "github.com/containerd/fifo"
"github.com/pkg/errors" "github.com/pkg/errors"
"golang.org/x/net/context"
) )
type linuxPlatform struct { type linuxPlatform struct {
epoller *console.Epoller epoller *console.Epoller
} }
func (p *linuxPlatform) copyConsole(ctx context.Context, console console.Console, stdin, stdout, stderr string, wg, cwg *sync.WaitGroup) (console.Console, error) { 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 { if p.epoller == nil {
return nil, errors.New("uninitialized epoller") return nil, errors.New("uninitialized epoller")
} }
@ -58,7 +58,7 @@ func (p *linuxPlatform) copyConsole(ctx context.Context, console console.Console
return epollConsole, nil return epollConsole, nil
} }
func (p *linuxPlatform) shutdownConsole(ctx context.Context, cons console.Console) error { func (p *linuxPlatform) ShutdownConsole(ctx context.Context, cons console.Console) error {
if p.epoller == nil { if p.epoller == nil {
return errors.New("uninitialized epoller") return errors.New("uninitialized epoller")
} }
@ -69,7 +69,7 @@ func (p *linuxPlatform) shutdownConsole(ctx context.Context, cons console.Consol
return epollConsole.Shutdown(p.epoller.CloseConsole) return epollConsole.Shutdown(p.epoller.CloseConsole)
} }
func (p *linuxPlatform) close() error { func (p *linuxPlatform) Close() error {
return p.epoller.Close() return p.epoller.Close()
} }

View File

@ -15,7 +15,7 @@ import (
type unixPlatform struct { type unixPlatform struct {
} }
func (p *unixPlatform) copyConsole(ctx context.Context, console console.Console, stdin, stdout, stderr string, wg, cwg *sync.WaitGroup) (console.Console, error) { func (p *unixPlatform) CopyConsole(ctx context.Context, console console.Console, stdin, stdout, stderr string, wg, cwg *sync.WaitGroup) (console.Console, error) {
if stdin != "" { if stdin != "" {
in, err := fifo.OpenFifo(ctx, stdin, syscall.O_RDONLY, 0) in, err := fifo.OpenFifo(ctx, stdin, syscall.O_RDONLY, 0)
if err != nil { if err != nil {
@ -48,11 +48,11 @@ func (p *unixPlatform) copyConsole(ctx context.Context, console console.Console,
return console, nil return console, nil
} }
func (p *unixPlatform) shutdownConsole(ctx context.Context, cons console.Console) error { func (p *unixPlatform) ShutdownConsole(ctx context.Context, cons console.Console) error {
return nil return nil
} }
func (p *unixPlatform) close() error { func (p *unixPlatform) Close() error {
return nil return nil
} }