github/containerd
4
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 2 Wiki Activity

Add Exec IDs

Browse Source 
Signed-off-by: Michael Crosby <crosbymichael@gmail.com>
This commit is contained in:
Michael Crosby
2017-06-28 12:54:10 -07:00
parent e283b3802d
commit f93bfb6233
36 changed files with 1441 additions and 2055 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
linux/shim/exec.go
View File

@@ -26,7 +26,7 @@ import (
type execProcess struct {
sync.WaitGroup
id int
id string
console console.Console
io runc.IO
status int
@@ -34,29 +34,27 @@ type execProcess struct {
pid int
closers []io.Closer
stdin io.Closer
stdio stdio
parent *initProcess
stdinPath string
stdoutPath string
stderrPath string
terminal bool
}
func newExecProcess(context context.Context, path string, r *shimapi.ExecProcessRequest, parent *initProcess, id int) (process, error) {
func newExecProcess(context context.Context, path string, r *shimapi.ExecProcessRequest, parent *initProcess, id string) (process, error) {
e := &execProcess{
id: id,
parent: parent,
stdinPath: r.Stdin,
stdoutPath: r.Stdout,
stderrPath: r.Stderr,
terminal: r.Terminal,
id: id,
parent: parent,
stdio: stdio{
stdin: r.Stdin,
stdout: r.Stdout,
stderr: r.Stderr,
terminal: r.Terminal,
},
}
var (
err error
socket *runc.Socket
io runc.IO
pidfile = filepath.Join(path, fmt.Sprintf("%d.pid", id))
pidfile = filepath.Join(path, fmt.Sprintf("%s.pid", id))
)
if r.Terminal {
if socket, err = runc.NewConsoleSocket(filepath.Join(path, "pty.sock")); err != nil {
@@ -120,6 +118,10 @@ func newExecProcess(context context.Context, path string, r *shimapi.ExecProcess
return e, nil
}
func (e *execProcess) ID() string {
return e.id
}
func (e *execProcess) Pid() int {
return e.pid
}
@@ -155,7 +157,7 @@ func (e *execProcess) Resize(ws console.WinSize) error {
return e.console.Resize(ws)
}
func (e *execProcess) Signal(sig int) error {
func (e *execProcess) Kill(ctx context.Context, sig uint32, _ bool) error {
if err := unix.Kill(e.pid, syscall.Signal(sig)); err != nil {
return checkKillError(err)
}
@@ -165,3 +167,7 @@ func (e *execProcess) Signal(sig int) error {
func (e *execProcess) Stdin() io.Closer {
return e.stdin
}
func (e *execProcess) Stdio() stdio {
return e.stdio
}

34
linux/shim/init.go
View File

@@ -48,11 +48,7 @@ type initProcess struct {
pid int
closers []io.Closer
stdin io.Closer
stdinPath string
stdoutPath string
stderrPath string
terminal bool
stdio stdio
}
func newInitProcess(context context.Context, path, namespace string, r *shimapi.CreateTaskRequest) (*initProcess, error) {
@@ -82,13 +78,15 @@ func newInitProcess(context context.Context, path, namespace string, r *shimapi.
Root: filepath.Join(RuncRoot, namespace),
}
p := &initProcess{
id: r.ID,
bundle: r.Bundle,
runtime: runtime,
stdinPath: r.Stdin,
stdoutPath: r.Stdout,
stderrPath: r.Stderr,
terminal: r.Terminal,
id: r.ID,
bundle: r.Bundle,
runtime: runtime,
stdio: stdio{
stdin: r.Stdin,
stdout: r.Stdout,
stderr: r.Stderr,
terminal: r.Terminal,
},
}
var (
err error
@@ -171,6 +169,10 @@ func newInitProcess(context context.Context, path, namespace string, r *shimapi.
return p, nil
}
func (p *initProcess) ID() string {
return p.id
}
func (p *initProcess) Pid() int {
return p.pid
}
@@ -257,10 +259,6 @@ func (p *initProcess) killAll(context context.Context) error {
return p.runtimeError(err, "OCI runtime killall failed")
}
func (p *initProcess) Signal(sig int) error {
return checkKillError(unix.Kill(p.pid, syscall.Signal(sig)))
}
func (p *initProcess) Stdin() io.Closer {
return p.stdin
}
@@ -306,6 +304,10 @@ func (p *initProcess) Update(context context.Context, r *shimapi.UpdateTaskReque
return p.runtime.Update(context, p.id, &resources)
}
func (p *initProcess) Stdio() stdio {
return p.stdio
}
// TODO(mlaventure): move to runc package?
func getLastRuntimeError(r *runc.Runc) (string, error) {
if r.Log == "" {

2
linux/shim/local.go
View File

@@ -60,7 +60,7 @@ func (c *local) Stream(ctx context.Context, in *shimapi.StreamEventsRequest, opt
}, nil
}
func (c *local) State(ctx context.Context, in *google_protobuf.Empty, opts ...grpc.CallOption) (*shimapi.StateResponse, error) {
func (c *local) State(ctx context.Context, in *shimapi.StateRequest, opts ...grpc.CallOption) (*shimapi.StateResponse, error) {
return c.s.State(ctx, in)
}

15
linux/shim/process.go
View File

@@ -10,7 +10,16 @@ import (
"github.com/containerd/console"
)
type stdio struct {
stdin string
stdout string
stderr string
terminal bool
}
type process interface {
// ID returns the id for the process
ID() string
// Pid returns the pid for the process
Pid() int
// Resize resizes the process console
@@ -23,8 +32,10 @@ type process interface {
ExitedAt() time.Time
// Delete deletes the process and its resourcess
Delete(context.Context) error
// Signal directly signals the process
Signal(int) error
// Stdin returns the process STDIN
Stdin() io.Closer
// Kill kills the process
Kill(context.Context, uint32, bool) error
// Stdio returns io information for the container
Stdio() stdio
}

154
linux/shim/service.go
View File

@@ -6,7 +6,9 @@ import (
"fmt"
"os"
"sync"
"syscall"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"github.com/containerd/console"
events "github.com/containerd/containerd/api/services/events/v1"
@@ -16,7 +18,6 @@ import (
google_protobuf "github.com/golang/protobuf/ptypes/empty"
"github.com/pkg/errors"
"golang.org/x/net/context"
"golang.org/x/sys/unix"
)
const (
@@ -34,7 +35,7 @@ func NewService(path, namespace string) (*Service, error) {
}
return &Service{
path: path,
processes: make(map[int]process),
processes: make(map[string]process),
events: make(chan *events.RuntimeEvent, 4096),
namespace: namespace,
}, nil
@@ -46,34 +47,38 @@ type Service struct {
id string
bundle string
mu sync.Mutex
processes map[int]process
processes map[string]process
events chan *events.RuntimeEvent
eventsMu sync.Mutex
deferredEvent *events.RuntimeEvent
execID int
namespace string
}
func (s *Service) Create(ctx context.Context, r *shimapi.CreateTaskRequest) (*shimapi.CreateTaskResponse, error) {
if r.ID == "" {
return nil, grpc.Errorf(codes.InvalidArgument, "task id cannot be empty")
}
process, err := newInitProcess(ctx, s.path, s.namespace, r)
if err != nil {
return nil, err
}
s.mu.Lock()
// save the main task id and bundle to the shim for additional requests
s.id = r.ID
s.bundle = r.Bundle
s.initProcess = process
pid := process.Pid()
s.processes[pid] = process
s.processes[r.ID] = process
s.mu.Unlock()
cmd := &reaper.Cmd{
ExitCh: make(chan int, 1),
}
reaper.Default.Register(pid, cmd)
s.events <- &events.RuntimeEvent{
Type: events.RuntimeEvent_CREATE,
ID: r.ID,
Pid: uint32(pid),
Type: events.RuntimeEvent_CREATE,
ID: r.ID,
ContainerID: s.id,
Pid: uint32(pid),
}
go s.waitExit(process, pid, cmd)
return &shimapi.CreateTaskResponse{
@@ -89,9 +94,10 @@ func (s *Service) Start(ctx context.Context, r *google_protobuf.Empty) (*google_
return nil, err
}
s.events <- &events.RuntimeEvent{
Type: events.RuntimeEvent_START,
ID: s.id,
Pid: uint32(s.initProcess.Pid()),
Type: events.RuntimeEvent_START,
ID: s.id,
ContainerID: s.id,
Pid: uint32(s.initProcess.Pid()),
}
return empty, nil
}
@@ -104,7 +110,7 @@ func (s *Service) Delete(ctx context.Context, r *google_protobuf.Empty) (*shimap
// TODO (@crosbymichael): how to handle errors here
p.Delete(ctx)
s.mu.Lock()
delete(s.processes, p.Pid())
delete(s.processes, p.ID())
s.mu.Unlock()
return &shimapi.DeleteResponse{
ExitStatus: uint32(p.Status()),
@@ -117,19 +123,19 @@ func (s *Service) DeleteProcess(ctx context.Context, r *shimapi.DeleteProcessReq
if s.initProcess == nil {
return nil, errors.New(ErrContainerNotCreated)
}
if int(r.Pid) == s.initProcess.pid {
return nil, fmt.Errorf("cannot delete init process with DeleteProcess")
if r.ID == s.initProcess.id {
return nil, grpc.Errorf(codes.InvalidArgument, "cannot delete init process with DeleteProcess")
}
s.mu.Lock()
p, ok := s.processes[int(r.Pid)]
p, ok := s.processes[r.ID]
s.mu.Unlock()
if !ok {
return nil, fmt.Errorf("process %d not found", r.Pid)
return nil, fmt.Errorf("process %s not found", r.ID)
}
// TODO (@crosbymichael): how to handle errors here
p.Delete(ctx)
s.mu.Lock()
delete(s.processes, p.Pid())
delete(s.processes, p.ID())
s.mu.Unlock()
return &shimapi.DeleteResponse{
ExitStatus: uint32(p.Status()),
@@ -144,9 +150,8 @@ func (s *Service) Exec(ctx context.Context, r *shimapi.ExecProcessRequest) (*shi
}
s.mu.Lock()
defer s.mu.Unlock()
s.execID++
process, err := newExecProcess(ctx, s.path, r, s.initProcess, s.execID)
process, err := newExecProcess(ctx, s.path, r, s.initProcess, r.ID)
if err != nil {
return nil, err
}
@@ -155,12 +160,13 @@ func (s *Service) Exec(ctx context.Context, r *shimapi.ExecProcessRequest) (*shi
ExitCh: make(chan int, 1),
}
reaper.Default.Register(pid, cmd)
s.processes[pid] = process
s.processes[r.ID] = process
s.events <- &events.RuntimeEvent{
Type: events.RuntimeEvent_EXEC_ADDED,
ID: s.id,
Pid: uint32(pid),
Type: events.RuntimeEvent_EXEC_ADDED,
ID: r.ID,
ContainerID: s.id,
Pid: uint32(pid),
}
go s.waitExit(process, pid, cmd)
return &shimapi.ExecProcessResponse{
@@ -169,18 +175,18 @@ func (s *Service) Exec(ctx context.Context, r *shimapi.ExecProcessRequest) (*shi
}
func (s *Service) ResizePty(ctx context.Context, r *shimapi.ResizePtyRequest) (*google_protobuf.Empty, error) {
if r.Pid == 0 {
return nil, errors.Errorf("pid not provided in request")
if r.ID == "" {
return nil, grpc.Errorf(codes.InvalidArgument, "id not provided")
}
ws := console.WinSize{
Width: uint16(r.Width),
Height: uint16(r.Height),
}
s.mu.Lock()
p, ok := s.processes[int(r.Pid)]
p, ok := s.processes[r.ID]
s.mu.Unlock()
if !ok {
return nil, errors.Errorf("process does not exist %d", r.Pid)
return nil, errors.Errorf("process does not exist %s", r.ID)
}
if err := p.Resize(ws); err != nil {
return nil, err
@@ -212,10 +218,14 @@ func (s *Service) Stream(r *shimapi.StreamEventsRequest, stream shimapi.Shim_Str
}
}
func (s *Service) State(ctx context.Context, r *google_protobuf.Empty) (*shimapi.StateResponse, error) {
func (s *Service) State(ctx context.Context, r *shimapi.StateRequest) (*shimapi.StateResponse, error) {
if s.initProcess == nil {
return nil, errors.New(ErrContainerNotCreated)
}
p, ok := s.processes[r.ID]
if !ok {
return nil, grpc.Errorf(codes.NotFound, "process id %s not found", r.ID)
}
st, err := s.initProcess.ContainerStatus(ctx)
if err != nil {
return nil, err
@@ -231,38 +241,17 @@ func (s *Service) State(ctx context.Context, r *google_protobuf.Empty) (*shimapi
case "paused":
status = task.StatusPaused
}
o := &shimapi.StateResponse{
ID: s.id,
Bundle: s.bundle,
Pid: uint32(s.initProcess.Pid()),
Status: status,
Processes: []*task.Process{},
Stdin: s.initProcess.stdinPath,
Stdout: s.initProcess.stdoutPath,
Stderr: s.initProcess.stderrPath,
}
s.mu.Lock()
defer s.mu.Unlock()
for _, p := range s.processes {
status := task.StatusRunning
if err := unix.Kill(p.Pid(), 0); err != nil {
if err != syscall.ESRCH {
return nil, err
}
status = task.StatusStopped
}
pp := &task.Process{
Pid: uint32(p.Pid()),
Status: status,
}
if ep, ok := p.(*execProcess); ok {
pp.Stdin = ep.stdinPath
pp.Stdout = ep.stdoutPath
pp.Stderr = ep.stderrPath
}
o.Processes = append(o.Processes, pp)
}
return o, nil
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,
}, nil
}
func (s *Service) Pause(ctx context.Context, r *google_protobuf.Empty) (*google_protobuf.Empty, error) {
@@ -289,35 +278,19 @@ func (s *Service) Kill(ctx context.Context, r *shimapi.KillRequest) (*google_pro
if s.initProcess == nil {
return nil, errors.New(ErrContainerNotCreated)
}
if r.Pid == 0 {
if r.ID == "" {
if err := s.initProcess.Kill(ctx, r.Signal, r.All); err != nil {
return nil, err
}
return empty, nil
}
if int(r.Pid) == s.initProcess.pid {
if err := s.initProcess.Kill(ctx, r.Signal, r.All); err != nil {
return nil, err
}
return empty, nil
p, ok := s.processes[r.ID]
if !ok {
return nil, grpc.Errorf(codes.NotFound, "process id %s not found", r.ID)
}
pids, err := s.getContainerPids(ctx, s.initProcess.id)
if err != nil {
if err := p.Kill(ctx, r.Signal, r.All); err != nil {
return nil, err
}
valid := false
for _, p := range pids {
if r.Pid == p {
valid = true
break
}
}
if !valid {
return nil, errors.Errorf("process %d does not exist in container", r.Pid)
}
if err := unix.Kill(int(r.Pid), syscall.Signal(r.Signal)); err != nil {
return nil, checkKillError(err)
}
return empty, nil
}
@@ -332,9 +305,9 @@ func (s *Service) ListPids(ctx context.Context, r *shimapi.ListPidsRequest) (*sh
}
func (s *Service) CloseIO(ctx context.Context, r *shimapi.CloseIORequest) (*google_protobuf.Empty, error) {
p, ok := s.processes[int(r.Pid)]
p, ok := s.processes[r.ID]
if !ok {
return nil, fmt.Errorf("process does not exist %d", r.Pid)
return nil, grpc.Errorf(codes.NotFound, "process does not exist %s", r.ID)
}
if err := p.Stdin().Close(); err != nil {
return nil, err
@@ -374,11 +347,12 @@ func (s *Service) waitExit(p process, pid int, cmd *reaper.Cmd) {
reaper.Default.Delete(pid)
s.events <- &events.RuntimeEvent{
Type: events.RuntimeEvent_EXIT,
ID: s.id,
Pid: uint32(pid),
ExitStatus: uint32(status),
ExitedAt: p.ExitedAt(),
Type: events.RuntimeEvent_EXIT,
ID: p.ID(),
ContainerID: s.id,
Pid: uint32(pid),
ExitStatus: uint32(status),
ExitedAt: p.ExitedAt(),
}
}
@@ -387,11 +361,9 @@ func (s *Service) getContainerPids(ctx context.Context, id string) ([]uint32, er
if err != nil {
return nil, err
}
pids := make([]uint32, 0, len(p))
for _, pid := range p {
pids = append(pids, uint32(pid))
}
return pids, nil
}

635
linux/shim/v1/shim.pb.go
View File

File diff suppressed because it is too large Load Diff

38
linux/shim/v1/shim.proto
View File

@@ -18,7 +18,7 @@ option go_package = "github.com/containerd/containerd/linux/shim/v1;shim";
// for the container processes.
service Shim {
// State returns shim and task state information.
rpc State(google.protobuf.Empty) returns (StateResponse);
rpc State(StateRequest) returns (StateResponse);
rpc Create(CreateTaskRequest) returns (CreateTaskResponse);
@@ -77,15 +77,16 @@ message DeleteResponse {
}
message DeleteProcessRequest {
uint32 pid = 1;
string id = 1;
}
message ExecProcessRequest {
bool terminal = 1;
string stdin = 2;
string stdout = 3;
string stderr = 4;
google.protobuf.Any spec = 5;
string id = 1;
bool terminal = 2;
string stdin = 3;
string stdout = 4;
string stderr = 5;
google.protobuf.Any spec = 6;
}
message ExecProcessResponse {
@@ -93,31 +94,34 @@ message ExecProcessResponse {
}
message ResizePtyRequest {
uint32 pid = 1;
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;
repeated containerd.v1.types.Process processes = 5;
string stdin = 6;
string stdout = 7;
string stderr = 8;
bool terminal = 9;
string stdin = 5;
string stdout = 6;
string stderr = 7;
bool terminal = 8;
}
message KillRequest {
uint32 signal = 1;
bool all = 2;
uint32 pid = 3;
string id = 1;
uint32 signal = 2;
bool all = 3;
}
message CloseIORequest {
uint32 pid = 1;
string id = 1;
bool stdin = 2;
}