Start work on Container and Process model
Signed-off-by: Michael Crosby <crosbymichael@gmail.com>
This commit is contained in:
Michael Crosby
parent
6641888667
commit
32bf0f69fd
38 changed files with 525 additions and 353 deletions
543
execution/executors/shim/process.go
Normal file
543
execution/executors/shim/process.go
Normal file
|
|
@ -0,0 +1,543 @@
|
|||
package shim
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"io/ioutil"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
"github.com/Sirupsen/logrus"
|
||||
"github.com/docker/containerd"
|
||||
specs "github.com/opencontainers/runtime-spec/specs-go"
|
||||
"golang.org/x/sys/unix"
|
||||
)
|
||||
|
||||
var (
|
||||
ErrContainerStartTimeout = errors.New("shim: container did not start before the specified timeout")
|
||||
ErrContainerNotStarted = errors.New("shim: container not started")
|
||||
ErrProcessNotExited = errors.New("containerd: process has not exited")
|
||||
ErrShimExited = errors.New("containerd: shim exited before container process was started")
|
||||
errInvalidPidInt = errors.New("shim: process pid is invalid")
|
||||
)
|
||||
|
||||
const UnknownStatus = 255
|
||||
|
||||
type processOpts struct {
|
||||
root string
|
||||
noPivotRoot bool
|
||||
checkpoint string
|
||||
c *containerd.Container
|
||||
cmd *exec.Cmd
|
||||
exec bool
|
||||
spec specs.Process
|
||||
stdin io.Reader
|
||||
stdout io.Writer
|
||||
stderr io.Writer
|
||||
}
|
||||
|
||||
func newProcess(opts processOpts) (*process, error) {
|
||||
var (
|
||||
spec = opts.c.Spec()
|
||||
stdin, stdout, stderr string
|
||||
)
|
||||
uid, gid, err := getRootIDs(spec)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for _, t := range []struct {
|
||||
path *string
|
||||
v interface{}
|
||||
}{
|
||||
{
|
||||
path: &stdin,
|
||||
v: opts.stdin,
|
||||
},
|
||||
{
|
||||
path: &stdout,
|
||||
v: opts.stdout,
|
||||
},
|
||||
{
|
||||
path: &stderr,
|
||||
v: opts.stderr,
|
||||
},
|
||||
} {
|
||||
p, err := getFifoPath(t.v)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
*t.path = p
|
||||
}
|
||||
p := &process{
|
||||
root: opts.root,
|
||||
cmd: opts.cmd,
|
||||
done: make(chan struct{}),
|
||||
spec: opts.spec,
|
||||
exec: opts.exec,
|
||||
rootUid: uid,
|
||||
rootGid: gid,
|
||||
noPivotRoot: opts.noPivotRoot,
|
||||
checkpoint: opts.checkpoint,
|
||||
stdin: stdin,
|
||||
stdout: stdout,
|
||||
stderr: stderr,
|
||||
}
|
||||
f, err := os.Create(filepath.Join(opts.root, "process.json"))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
err = json.NewEncoder(f).Encode(p)
|
||||
f.Close()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
exit, err := getExitPipe(filepath.Join(opts.root, "exit"))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
control, err := getControlPipe(filepath.Join(opts.root, "control"))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
p.exit, p.control = exit, control
|
||||
return p, nil
|
||||
}
|
||||
|
||||
type process struct {
|
||||
root string
|
||||
cmd *exec.Cmd
|
||||
done chan struct{}
|
||||
success bool
|
||||
startTime string
|
||||
mu sync.Mutex
|
||||
pid int
|
||||
exit *os.File
|
||||
control *os.File
|
||||
|
||||
spec specs.Process
|
||||
noPivotRoot bool
|
||||
exec bool
|
||||
rootUid int
|
||||
rootGid int
|
||||
checkpoint string
|
||||
stdin string
|
||||
stdout string
|
||||
stderr string
|
||||
}
|
||||
|
||||
type processState struct {
|
||||
specs.Process
|
||||
Exec bool `json:"exec"`
|
||||
RootUID int `json:"rootUID"`
|
||||
RootGID int `json:"rootGID"`
|
||||
Checkpoint string `json:"checkpoint"`
|
||||
NoPivotRoot bool `json:"noPivotRoot"`
|
||||
RuntimeArgs []string `json:"runtimeArgs"`
|
||||
Root string `json:"root"`
|
||||
StartTime string `json:"startTime"`
|
||||
// Stdin fifo filepath
|
||||
Stdin string `json:"stdin"`
|
||||
// Stdout fifo filepath
|
||||
Stdout string `json:"stdout"`
|
||||
// Stderr fifo filepath
|
||||
Stderr string `json:"stderr"`
|
||||
}
|
||||
|
||||
func (p *process) MarshalJSON() ([]byte, error) {
|
||||
ps := processState{
|
||||
Process: p.spec,
|
||||
NoPivotRoot: p.noPivotRoot,
|
||||
Checkpoint: p.checkpoint,
|
||||
RootUID: p.rootUid,
|
||||
RootGID: p.rootGid,
|
||||
Exec: p.exec,
|
||||
Stdin: p.stdin,
|
||||
Stdout: p.stdout,
|
||||
Stderr: p.stderr,
|
||||
Root: p.root,
|
||||
StartTime: p.startTime,
|
||||
}
|
||||
return json.Marshal(ps)
|
||||
}
|
||||
|
||||
func (p *process) UnmarshalJSON(b []byte) error {
|
||||
var ps processState
|
||||
if err := json.Unmarshal(b, &ps); err != nil {
|
||||
return err
|
||||
}
|
||||
p.spec = ps.Process
|
||||
p.noPivotRoot = ps.NoPivotRoot
|
||||
p.rootGid = ps.RootGID
|
||||
p.rootUid = ps.RootUID
|
||||
p.checkpoint = ps.Checkpoint
|
||||
p.exec = ps.Exec
|
||||
p.stdin = ps.Stdin
|
||||
p.stdout = ps.Stdout
|
||||
p.stderr = ps.Stderr
|
||||
p.root = ps.Root
|
||||
p.startTime = ps.StartTime
|
||||
p.done = make(chan struct{})
|
||||
pid, err := readPid(filepath.Join(p.root, "pid"))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
p.pid = pid
|
||||
exit, err := getExitPipe(filepath.Join(p.root, "exit"))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
control, err := getControlPipe(filepath.Join(p.root, "control"))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
p.exit, p.control = exit, control
|
||||
return nil
|
||||
}
|
||||
|
||||
func (p *process) Pid() int {
|
||||
return p.pid
|
||||
}
|
||||
|
||||
func (p *process) FD() int {
|
||||
return int(p.exit.Fd())
|
||||
}
|
||||
|
||||
func (p *process) Close() error {
|
||||
return p.exit.Close()
|
||||
}
|
||||
|
||||
func (p *process) Remove() bool {
|
||||
return true
|
||||
}
|
||||
|
||||
func (p *process) Wait() (rst uint32, rerr error) {
|
||||
<-p.done
|
||||
data, err := ioutil.ReadFile(filepath.Join(p.root, "exitStatus"))
|
||||
defer func() {
|
||||
if rerr != nil {
|
||||
rst, rerr = p.handleSigkilledShim(rst, rerr)
|
||||
}
|
||||
}()
|
||||
if err != nil {
|
||||
if os.IsNotExist(err) {
|
||||
return UnknownStatus, ErrProcessNotExited
|
||||
}
|
||||
return UnknownStatus, err
|
||||
}
|
||||
if len(data) == 0 {
|
||||
return UnknownStatus, ErrProcessNotExited
|
||||
}
|
||||
i, err := strconv.ParseUint(string(data), 10, 32)
|
||||
return uint32(i), err
|
||||
}
|
||||
|
||||
func (p *process) Signal(s os.Signal) error {
|
||||
_, err := fmt.Fprintf(p.control, "%d %d %d\n", 2, s, 0)
|
||||
return err
|
||||
}
|
||||
|
||||
// same checks if the process is the same process originally launched
|
||||
func (p *process) same() (bool, error) {
|
||||
/// for backwards compat assume true if it is not set
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
if p.startTime == "" {
|
||||
return true, nil
|
||||
}
|
||||
pid, err := readPid(filepath.Join(p.root, "pid"))
|
||||
if err != nil {
|
||||
return false, nil
|
||||
}
|
||||
started, err := readProcessStartTime(pid)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
return p.startTime == started, nil
|
||||
}
|
||||
|
||||
func (p *process) checkExited() {
|
||||
err := p.cmd.Wait()
|
||||
if err == nil {
|
||||
p.mu.Lock()
|
||||
if p.success {
|
||||
p.mu.Unlock()
|
||||
return
|
||||
}
|
||||
p.success = true
|
||||
p.mu.Unlock()
|
||||
}
|
||||
if same, _ := p.same(); same && p.hasPid() {
|
||||
// The process changed its PR_SET_PDEATHSIG, so force kill it
|
||||
logrus.Infof("containerd: (pid %v) has become an orphan, killing it", p.pid)
|
||||
if err := unix.Kill(p.pid, syscall.SIGKILL); err != nil && err != syscall.ESRCH {
|
||||
logrus.Errorf("containerd: unable to SIGKILL (pid %v): %v", p.pid, err)
|
||||
return
|
||||
}
|
||||
// wait for the container process to exit
|
||||
for {
|
||||
if err := unix.Kill(p.pid, 0); err != nil {
|
||||
break
|
||||
}
|
||||
time.Sleep(5 * time.Millisecond)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (p *process) hasPid() bool {
|
||||
p.mu.Lock()
|
||||
r := p.pid > 0
|
||||
p.mu.Unlock()
|
||||
return r
|
||||
}
|
||||
|
||||
type pidResponse struct {
|
||||
pid int
|
||||
err error
|
||||
}
|
||||
|
||||
func (p *process) waitForCreate(timeout time.Duration) error {
|
||||
r := make(chan pidResponse, 1)
|
||||
go p.readContainerPid(r)
|
||||
|
||||
select {
|
||||
case resp := <-r:
|
||||
if resp.err != nil {
|
||||
return resp.err
|
||||
}
|
||||
p.mu.Lock()
|
||||
p.pid = resp.pid
|
||||
started, err := readProcessStartTime(resp.pid)
|
||||
if err != nil {
|
||||
if os.IsNotExist(err) {
|
||||
// process already exited
|
||||
p.success = true
|
||||
p.mu.Unlock()
|
||||
return nil
|
||||
}
|
||||
logrus.Warnf("shim: unable to save starttime: %v", err)
|
||||
}
|
||||
p.startTime = started
|
||||
f, err := os.Create(filepath.Join(p.root, "process.json"))
|
||||
if err != nil {
|
||||
logrus.Warnf("shim: unable to create process.json: %v", err)
|
||||
p.mu.Unlock()
|
||||
return nil
|
||||
}
|
||||
defer f.Close()
|
||||
if err := json.NewEncoder(f).Encode(p); err != nil {
|
||||
logrus.Warnf("shim: unable to encode process: %v", err)
|
||||
}
|
||||
p.mu.Unlock()
|
||||
return nil
|
||||
case <-time.After(timeout):
|
||||
p.cmd.Process.Kill()
|
||||
p.cmd.Wait()
|
||||
return ErrContainerStartTimeout
|
||||
}
|
||||
}
|
||||
|
||||
func (p *process) readContainerPid(r chan pidResponse) {
|
||||
pidFile := filepath.Join(p.root, "pid")
|
||||
for {
|
||||
pid, err := readPid(pidFile)
|
||||
if err != nil {
|
||||
if os.IsNotExist(err) || err == errInvalidPidInt {
|
||||
if serr := checkErrorLogs(p.cmd,
|
||||
filepath.Join(p.root, "shim-log.json"),
|
||||
filepath.Join(p.root, "log.json")); serr != nil && !os.IsNotExist(serr) {
|
||||
r <- pidResponse{
|
||||
err: serr,
|
||||
}
|
||||
break
|
||||
}
|
||||
time.Sleep(15 * time.Millisecond)
|
||||
continue
|
||||
}
|
||||
r <- pidResponse{
|
||||
err: err,
|
||||
}
|
||||
break
|
||||
}
|
||||
r <- pidResponse{
|
||||
pid: pid,
|
||||
}
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
func (p *process) handleSigkilledShim(rst uint32, rerr error) (uint32, error) {
|
||||
if err := unix.Kill(p.pid, 0); err == syscall.ESRCH {
|
||||
logrus.Warnf("containerd: (pid %d) does not exist", p.pid)
|
||||
// The process died while containerd was down (probably of
|
||||
// SIGKILL, but no way to be sure)
|
||||
return UnknownStatus, writeExitStatus(filepath.Join(p.root, "exitStatus"), UnknownStatus)
|
||||
}
|
||||
|
||||
// If it's not the same process, just mark it stopped and set
|
||||
// the status to the UnknownStatus value (i.e. 255)
|
||||
if same, _ := p.same(); !same {
|
||||
// Create the file so we get the exit event generated once monitor kicks in
|
||||
// without having to go through all this process again
|
||||
return UnknownStatus, writeExitStatus(filepath.Join(p.root, "exitStatus"), UnknownStatus)
|
||||
}
|
||||
ppid, err := readProcStatField(p.pid, 4)
|
||||
if err != nil {
|
||||
return rst, fmt.Errorf("could not check process ppid: %v (%v)", err, rerr)
|
||||
}
|
||||
if ppid == "1" {
|
||||
if err := unix.Kill(p.pid, syscall.SIGKILL); err != nil && err != syscall.ESRCH {
|
||||
return UnknownStatus, fmt.Errorf(
|
||||
"containerd: unable to SIGKILL (pid %v): %v", p.pid, err)
|
||||
}
|
||||
// wait for the process to die
|
||||
for {
|
||||
if err := unix.Kill(p.pid, 0); err == syscall.ESRCH {
|
||||
break
|
||||
}
|
||||
time.Sleep(5 * time.Millisecond)
|
||||
}
|
||||
// Create the file so we get the exit event generated once monitor kicks in
|
||||
// without having to go through all this process again
|
||||
status := 128 + uint32(syscall.SIGKILL)
|
||||
return status, writeExitStatus(filepath.Join(p.root, "exitStatus"), status)
|
||||
}
|
||||
return rst, rerr
|
||||
}
|
||||
|
||||
func checkErrorLogs(cmd *exec.Cmd, shimLogPath, runtimeLogPath string) error {
|
||||
alive, err := isAlive(cmd)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if !alive {
|
||||
// runc could have failed to run the container so lets get the error
|
||||
// out of the logs or the shim could have encountered an error
|
||||
messages, err := readLogMessages(shimLogPath)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, m := range messages {
|
||||
if m.Level == "error" {
|
||||
return fmt.Errorf("shim error: %v", m.Msg)
|
||||
}
|
||||
}
|
||||
// no errors reported back from shim, check for runc/runtime errors
|
||||
messages, err = readLogMessages(runtimeLogPath)
|
||||
if err != nil {
|
||||
if os.IsNotExist(err) {
|
||||
err = ErrContainerNotStarted
|
||||
}
|
||||
return err
|
||||
}
|
||||
for _, m := range messages {
|
||||
if m.Level == "error" {
|
||||
return fmt.Errorf("oci runtime error: %v", m.Msg)
|
||||
}
|
||||
}
|
||||
return ErrContainerNotStarted
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func readProcessStartTime(pid int) (string, error) {
|
||||
return readProcStatField(pid, 22)
|
||||
}
|
||||
|
||||
func readProcStatField(pid int, field int) (string, error) {
|
||||
data, err := ioutil.ReadFile(filepath.Join(string(filepath.Separator), "proc", strconv.Itoa(pid), "stat"))
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
if field > 2 {
|
||||
// First, split out the name since he could contains spaces.
|
||||
parts := strings.Split(string(data), ") ")
|
||||
// Now split out the rest, we end up with 2 fields less
|
||||
parts = strings.Split(parts[1], " ")
|
||||
return parts[field-2-1], nil // field count start at 1 in manual
|
||||
}
|
||||
parts := strings.Split(string(data), " (")
|
||||
if field == 1 {
|
||||
return parts[0], nil
|
||||
}
|
||||
return strings.Split(parts[1], ") ")[0], nil
|
||||
}
|
||||
|
||||
func readPid(pidFile string) (int, error) {
|
||||
data, err := ioutil.ReadFile(pidFile)
|
||||
if err != nil {
|
||||
return -1, err
|
||||
}
|
||||
i, err := strconv.Atoi(string(data))
|
||||
if err != nil {
|
||||
return -1, errInvalidPidInt
|
||||
}
|
||||
return i, nil
|
||||
}
|
||||
|
||||
// isAlive checks if the shim that launched the container is still alive
|
||||
func isAlive(cmd *exec.Cmd) (bool, error) {
|
||||
if _, err := syscall.Wait4(cmd.Process.Pid, nil, syscall.WNOHANG, nil); err == nil {
|
||||
return true, nil
|
||||
}
|
||||
if err := syscall.Kill(cmd.Process.Pid, 0); err != nil {
|
||||
if err == syscall.ESRCH {
|
||||
return false, nil
|
||||
}
|
||||
return false, err
|
||||
}
|
||||
return true, nil
|
||||
}
|
||||
|
||||
type message struct {
|
||||
Level string `json:"level"`
|
||||
Msg string `json:"msg"`
|
||||
}
|
||||
|
||||
func readLogMessages(path string) ([]message, error) {
|
||||
var out []message
|
||||
f, err := os.Open(path)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer f.Close()
|
||||
dec := json.NewDecoder(f)
|
||||
for {
|
||||
var m message
|
||||
if err := dec.Decode(&m); err != nil {
|
||||
if err == io.EOF {
|
||||
break
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, m)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func getExitPipe(path string) (*os.File, error) {
|
||||
if err := unix.Mkfifo(path, 0755); err != nil && !os.IsExist(err) {
|
||||
return nil, err
|
||||
}
|
||||
// add NONBLOCK in case the other side has already closed or else
|
||||
// this function would never return
|
||||
return os.OpenFile(path, syscall.O_RDONLY|syscall.O_NONBLOCK, 0)
|
||||
}
|
||||
|
||||
func getControlPipe(path string) (*os.File, error) {
|
||||
if err := unix.Mkfifo(path, 0755); err != nil && !os.IsExist(err) {
|
||||
return nil, err
|
||||
}
|
||||
return os.OpenFile(path, syscall.O_RDWR|syscall.O_NONBLOCK, 0)
|
||||
}
|
||||
|
||||
func writeExitStatus(path string, status uint32) error {
|
||||
return ioutil.WriteFile(path, []byte(fmt.Sprintf("%u", status)), 0644)
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue