vbatts/containerd
1
0
Fork 0
Code Pull requests Releases Activity

Pass context down to executors

Browse source 
Signed-off-by: Kenfe-Mickael Laventure <mickael.laventure@gmail.com>
This commit is contained in:
Kenfe-Mickael Laventure 2016-12-09 10:00:30 -08:00
parent 0aad42f5cf
commit 4157d1adfd
3 changed files with 53 additions and 51 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

23
execution/executor.go
View file

@ -1,6 +1,7 @@
package execution package execution
import ( import (
"context"
"os" "os"
"github.com/opencontainers/runtime-spec/specs-go" "github.com/opencontainers/runtime-spec/specs-go"
@ -23,16 +24,16 @@ type StartProcessOpts struct {
} }
type Executor interface { type Executor interface {
Create(id string, o CreateOpts) (*Container, error) Create(ctx context.Context, id string, o CreateOpts) (*Container, error)
Pause(*Container) error Pause(context.Context, *Container) error
Resume(*Container) error Resume(context.Context, *Container) error
Status(*Container) (Status, error) Status(context.Context, *Container) (Status, error)
List() ([]*Container, error) List(context.Context) ([]*Container, error)
Load(id string) (*Container, error) Load(ctx context.Context, id string) (*Container, error)
Delete(*Container) error Delete(context.Context, *Container) error
Start(*Container) error Start(context.Context, *Container) error
StartProcess(*Container, StartProcessOpts) (Process, error) StartProcess(context.Context, *Container, StartProcessOpts) (Process, error)
SignalProcess(*Container, string, os.Signal) error SignalProcess(ctx context.Context, c *Container, id string, sig os.Signal) error
DeleteProcess(*Container, string) error DeleteProcess(ctx context.Context, c *Container, id string) error
} }

45
execution/executors/oci/oci.go
View file

@ -1,6 +1,7 @@
package oci package oci
import ( import (
"context"
"errors" "errors"
"fmt" "fmt"
"io" "io"
@ -77,7 +78,7 @@ func setupConsole(rio runc.IO) (*os.File, string, error) {
return master, console, nil return master, console, nil
} }
func (r *OCIRuntime) Create(id string, o execution.CreateOpts) (container *execution.Container, err error) { func (r *OCIRuntime) Create(ctx context.Context, id string, o execution.CreateOpts) (container *execution.Container, err error) {
rio, err := getRuncIO(o.Stdin, o.Stdout, o.Stderr) rio, err := getRuncIO(o.Stdin, o.Stdout, o.Stderr)
if err != nil { if err != nil {
return nil, err return nil, err
@ -115,7 +116,7 @@ func (r *OCIRuntime) Create(id string, o execution.CreateOpts) (container *execu
return nil, err return nil, err
} }
pidFile := filepath.Join(initDir, "pid") pidFile := filepath.Join(initDir, "pid")
err = r.runc.Create(id, o.Bundle, &runc.CreateOpts{ err = r.runc.Create(ctx, id, o.Bundle, &runc.CreateOpts{
PidFile: pidFile, PidFile: pidFile,
Console: consolePath, Console: consolePath,
IO: rio, IO: rio,
@ -125,8 +126,8 @@ func (r *OCIRuntime) Create(id string, o execution.CreateOpts) (container *execu
} }
defer func() { defer func() {
if err != nil { if err != nil {
r.runc.Kill(id, int(syscall.SIGKILL)) r.runc.Kill(ctx, id, int(syscall.SIGKILL))
r.runc.Delete(id) r.runc.Delete(ctx, id)
} }
}() }()
@ -146,12 +147,12 @@ func (r *OCIRuntime) Create(id string, o execution.CreateOpts) (container *execu
return container, nil return container, nil
} }
func (r *OCIRuntime) Start(c *execution.Container) error { func (r *OCIRuntime) Start(ctx context.Context, c *execution.Container) error {
return r.runc.Start(c.ID()) return r.runc.Start(ctx, c.ID())
} }
func (r *OCIRuntime) Status(c *execution.Container) (execution.Status, error) { func (r *OCIRuntime) Status(ctx context.Context, c *execution.Container) (execution.Status, error) {
state, err := r.runc.State(c.ID()) state, err := r.runc.State(ctx, c.ID())
if err != nil { if err != nil {
return "", err return "", err
} }
@ -190,8 +191,8 @@ func (r *OCIRuntime) load(runcC *runc.Container) (*execution.Container, error) {
return container, nil return container, nil
} }
func (r *OCIRuntime) List() ([]*execution.Container, error) { func (r *OCIRuntime) List(ctx context.Context) ([]*execution.Container, error) {
runcCs, err := r.runc.List() runcCs, err := r.runc.List(ctx)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -208,8 +209,8 @@ func (r *OCIRuntime) List() ([]*execution.Container, error) {
return containers, nil return containers, nil
} }
func (r *OCIRuntime) Load(id string) (*execution.Container, error) { func (r *OCIRuntime) Load(ctx context.Context, id string) (*execution.Container, error) {
runcC, err := r.runc.State(id) runcC, err := r.runc.State(ctx, id)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -217,9 +218,9 @@ func (r *OCIRuntime) Load(id string) (*execution.Container, error) {
return r.load(runcC) return r.load(runcC)
} }
func (r *OCIRuntime) Delete(c *execution.Container) error { func (r *OCIRuntime) Delete(ctx context.Context, c *execution.Container) error {
id := c.ID() id := c.ID()
if err := r.runc.Delete(id); err != nil { if err := r.runc.Delete(ctx, id); err != nil {
return err return err
} }
c.StateDir().Delete() c.StateDir().Delete()
@ -228,15 +229,15 @@ func (r *OCIRuntime) Delete(c *execution.Container) error {
return nil return nil
} }
func (r *OCIRuntime) Pause(c *execution.Container) error { func (r *OCIRuntime) Pause(ctx context.Context, c *execution.Container) error {
return r.runc.Pause(c.ID()) return r.runc.Pause(ctx, c.ID())
} }
func (r *OCIRuntime) Resume(c *execution.Container) error { func (r *OCIRuntime) Resume(ctx context.Context, c *execution.Container) error {
return r.runc.Resume(c.ID()) return r.runc.Resume(ctx, c.ID())
} }
func (r *OCIRuntime) StartProcess(c *execution.Container, o execution.StartProcessOpts) (p execution.Process, err error) { func (r *OCIRuntime) StartProcess(ctx context.Context, c *execution.Container, o execution.StartProcessOpts) (p execution.Process, err error) {
rio, err := getRuncIO(o.Stdin, o.Stdout, o.Stderr) rio, err := getRuncIO(o.Stdin, o.Stdout, o.Stderr)
if err != nil { if err != nil {
return nil, err return nil, err
@ -271,7 +272,7 @@ func (r *OCIRuntime) StartProcess(c *execution.Container, o execution.StartProce
}() }()
pidFile := filepath.Join(processStateDir, "pid") pidFile := filepath.Join(processStateDir, "pid")
if err := r.runc.ExecProcess(c.ID(), o.Spec, &runc.ExecOpts{ if err := r.runc.Exec(ctx, c.ID(), o.Spec, &runc.ExecOpts{
PidFile: pidFile, PidFile: pidFile,
Detach: false, Detach: false,
Console: consolePath, Console: consolePath,
@ -297,7 +298,7 @@ func (r *OCIRuntime) StartProcess(c *execution.Container, o execution.StartProce
return process, nil return process, nil
} }
func (r *OCIRuntime) SignalProcess(c *execution.Container, id string, sig os.Signal) error { func (r *OCIRuntime) SignalProcess(ctx context.Context, c *execution.Container, id string, sig os.Signal) error {
process := c.GetProcess(id) process := c.GetProcess(id)
if process == nil { if process == nil {
return fmt.Errorf("Make a Process Not Found error") return fmt.Errorf("Make a Process Not Found error")
@ -305,7 +306,7 @@ func (r *OCIRuntime) SignalProcess(c *execution.Container, id string, sig os.Sig
return syscall.Kill(int(process.Pid()), sig.(syscall.Signal)) return syscall.Kill(int(process.Pid()), sig.(syscall.Signal))
} }
func (r *OCIRuntime) DeleteProcess(c *execution.Container, id string) error { func (r *OCIRuntime) DeleteProcess(ctx context.Context, c *execution.Container, id string) error {
ioID := fmt.Sprintf("%s-%s", c.ID(), id) ioID := fmt.Sprintf("%s-%s", c.ID(), id)
closeRuncIO(r.ios[ioID]) closeRuncIO(r.ios[ioID])
delete(r.ios, ioID) delete(r.ios, ioID)

36
execution/service.go
View file

@ -30,7 +30,7 @@ func (s *Service) Create(ctx context.Context, r *api.CreateContainerRequest) (*a
// TODO: write io and bundle path to dir // TODO: write io and bundle path to dir
var err error var err error
container, err := s.executor.Create(r.ID, CreateOpts{ container, err := s.executor.Create(ctx, r.ID, CreateOpts{
Bundle: r.BundlePath, Bundle: r.BundlePath,
Console: r.Console, Console: r.Console,
Stdin: r.Stdin, Stdin: r.Stdin,
@ -49,19 +49,19 @@ func (s *Service) Create(ctx context.Context, r *api.CreateContainerRequest) (*a
} }
func (s *Service) Delete(ctx context.Context, r *api.DeleteContainerRequest) (*google_protobuf.Empty, error) { func (s *Service) Delete(ctx context.Context, r *api.DeleteContainerRequest) (*google_protobuf.Empty, error) {
container, err := s.executor.Load(r.ID) container, err := s.executor.Load(ctx, r.ID)
if err != nil { if err != nil {
return emptyResponse, err return emptyResponse, err
} }
if err = s.executor.Delete(container); err != nil { if err = s.executor.Delete(ctx, container); err != nil {
return emptyResponse, err return emptyResponse, err
} }
return emptyResponse, nil return emptyResponse, nil
} }
func (s *Service) List(ctx context.Context, r *api.ListContainersRequest) (*api.ListContainersResponse, error) { func (s *Service) List(ctx context.Context, r *api.ListContainersRequest) (*api.ListContainersResponse, error) {
containers, err := s.executor.List() containers, err := s.executor.List(ctx)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -72,7 +72,7 @@ func (s *Service) List(ctx context.Context, r *api.ListContainersRequest) (*api.
return resp, nil return resp, nil
} }
func (s *Service) Get(ctx context.Context, r *api.GetContainerRequest) (*api.GetContainerResponse, error) { func (s *Service) Get(ctx context.Context, r *api.GetContainerRequest) (*api.GetContainerResponse, error) {
container, err := s.executor.Load(r.ID) container, err := s.executor.Load(ctx, r.ID)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -86,31 +86,31 @@ func (s *Service) Update(ctx context.Context, r *api.UpdateContainerRequest) (*g
} }
func (s *Service) Pause(ctx context.Context, r *api.PauseContainerRequest) (*google_protobuf.Empty, error) { func (s *Service) Pause(ctx context.Context, r *api.PauseContainerRequest) (*google_protobuf.Empty, error) {
container, err := s.executor.Load(r.ID) container, err := s.executor.Load(ctx, r.ID)
if err != nil { if err != nil {
return nil, err return nil, err
} }
return emptyResponse, s.executor.Pause(container) return emptyResponse, s.executor.Pause(ctx, container)
} }
func (s *Service) Resume(ctx context.Context, r *api.ResumeContainerRequest) (*google_protobuf.Empty, error) { func (s *Service) Resume(ctx context.Context, r *api.ResumeContainerRequest) (*google_protobuf.Empty, error) {
container, err := s.executor.Load(r.ID) container, err := s.executor.Load(ctx, r.ID)
if err != nil { if err != nil {
return nil, err return nil, err
} }
return emptyResponse, s.executor.Resume(container) return emptyResponse, s.executor.Resume(ctx, container)
} }
func (s *Service) Start(ctx context.Context, r *api.StartContainerRequest) (*google_protobuf.Empty, error) { func (s *Service) Start(ctx context.Context, r *api.StartContainerRequest) (*google_protobuf.Empty, error) {
container, err := s.executor.Load(r.ID) container, err := s.executor.Load(ctx, r.ID)
if err != nil { if err != nil {
return nil, err return nil, err
} }
return emptyResponse, s.executor.Start(container) return emptyResponse, s.executor.Start(ctx, container)
} }
func (s *Service) StartProcess(ctx context.Context, r *api.StartProcessRequest) (*api.StartProcessResponse, error) { func (s *Service) StartProcess(ctx context.Context, r *api.StartProcessRequest) (*api.StartProcessResponse, error) {
container, err := s.executor.Load(r.ContainerId) container, err := s.executor.Load(ctx, r.ContainerId)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -128,7 +128,7 @@ func (s *Service) StartProcess(ctx context.Context, r *api.StartProcessRequest)
NoNewPrivileges: true, NoNewPrivileges: true,
} }
process, err := s.executor.StartProcess(container, StartProcessOpts{ process, err := s.executor.StartProcess(ctx, container, StartProcessOpts{
Spec: spec, Spec: spec,
Console: r.Console, Console: r.Console,
Stdin: r.Stdin, Stdin: r.Stdin,
@ -147,7 +147,7 @@ func (s *Service) StartProcess(ctx context.Context, r *api.StartProcessRequest)
// containerd managed execs + system pids forked in container // containerd managed execs + system pids forked in container
func (s *Service) GetProcess(ctx context.Context, r *api.GetProcessRequest) (*api.GetProcessResponse, error) { func (s *Service) GetProcess(ctx context.Context, r *api.GetProcessRequest) (*api.GetProcessResponse, error) {
container, err := s.executor.Load(r.Container.ID) container, err := s.executor.Load(ctx, r.Container.ID)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -161,7 +161,7 @@ func (s *Service) GetProcess(ctx context.Context, r *api.GetProcessRequest) (*ap
} }
func (s *Service) SignalProcess(ctx context.Context, r *api.SignalProcessRequest) (*google_protobuf.Empty, error) { func (s *Service) SignalProcess(ctx context.Context, r *api.SignalProcessRequest) (*google_protobuf.Empty, error) {
container, err := s.executor.Load(r.Container.ID) container, err := s.executor.Load(ctx, r.Container.ID)
if err != nil { if err != nil {
return emptyResponse, err return emptyResponse, err
} }
@ -173,18 +173,18 @@ func (s *Service) SignalProcess(ctx context.Context, r *api.SignalProcessRequest
} }
func (s *Service) DeleteProcess(ctx context.Context, r *api.DeleteProcessRequest) (*google_protobuf.Empty, error) { func (s *Service) DeleteProcess(ctx context.Context, r *api.DeleteProcessRequest) (*google_protobuf.Empty, error) {
container, err := s.executor.Load(r.Container.ID) container, err := s.executor.Load(ctx, r.Container.ID)
if err != nil { if err != nil {
return emptyResponse, err return emptyResponse, err
} }
if err := s.executor.DeleteProcess(container, r.Process.ID); err != nil { if err := s.executor.DeleteProcess(ctx, container, r.Process.ID); err != nil {
return emptyResponse, err return emptyResponse, err
} }
return emptyResponse, nil return emptyResponse, nil
} }
func (s *Service) ListProcesses(ctx context.Context, r *api.ListProcessesRequest) (*api.ListProcessesResponse, error) { func (s *Service) ListProcesses(ctx context.Context, r *api.ListProcessesRequest) (*api.ListProcessesResponse, error) {
container, err := s.executor.Load(r.Container.ID) container, err := s.executor.Load(ctx, r.Container.ID)
if err != nil { if err != nil {
return nil, err return nil, err
} }