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

move work on execution service

Browse source 
Signed-off-by: Kenfe-Mickael Laventure <mickael.laventure@gmail.com>
This commit is contained in:
Kenfe-Mickael Laventure 2016-12-05 14:15:03 -08:00
parent dd5f74edec
commit c857213b4c
15 changed files with 596 additions and 1016 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

719
api/execution/container.pb.go
View file

File diff suppressed because it is too large Load diff

29
api/execution/container.proto
View file

@ -11,15 +11,14 @@ service ContainerService {
rpc Pause(PauseContainerRequest) returns (google.protobuf.Empty); rpc Pause(PauseContainerRequest) returns (google.protobuf.Empty);
rpc Resume(ResumeContainerRequest) returns (google.protobuf.Empty); rpc Resume(ResumeContainerRequest) returns (google.protobuf.Empty);
rpc CreateProcess(CreateProcessRequest) returns (CreateProcessResponse);
rpc GetProcess(GetProcessRequest) returns (GetProcessResponse);
rpc StartProcess(StartProcessRequest) returns (StartProcessResponse); rpc StartProcess(StartProcessRequest) returns (StartProcessResponse);
rpc GetProcess(GetProcessRequest) returns (GetProcessResponse);
rpc SignalProcess(SignalProcessRequest) returns (google.protobuf.Empty); rpc SignalProcess(SignalProcessRequest) returns (google.protobuf.Empty);
rpc DeleteProcess(DeleteProcessRequest) returns (google.protobuf.Empty); rpc DeleteProcess(DeleteProcessRequest) returns (google.protobuf.Empty);
rpc ListProcesses(ListProcessesRequest) returns (ListProcessesResponse); rpc ListProcesses(ListProcessesRequest) returns (ListProcessesResponse);
} }
message CreateProcessRequest { message StartProcessRequest {
string container_id = 1; string container_id = 1;
Process process = 2; Process process = 2;
string stdin = 3; string stdin = 3;
@ -27,7 +26,7 @@ message CreateProcessRequest {
string stderr = 5; string stderr = 5;
} }
message CreateProcessResponse { message StartProcessResponse {
Process process = 1; Process process = 1;
} }
@ -40,7 +39,7 @@ message Container {
message Process { message Process {
string id = 1 [(gogoproto.customname) = "ID"]; string id = 1 [(gogoproto.customname) = "ID"];
uint64 pid = 2; int64 pid = 2;
repeated string args = 3; repeated string args = 3;
repeated string env = 4; repeated string env = 4;
User user = 5; User user = 5;
@ -84,17 +83,9 @@ message ResumeContainerRequest {
string id = 1 [(gogoproto.customname) = "ID"]; string id = 1 [(gogoproto.customname) = "ID"];
} }
message StartProcessRequest {
Process process = 1;
}
message StartProcessResponse {
Process process = 1;
}
message GetProcessRequest { message GetProcessRequest {
Container container = 1; Container container = 1;
string process_id = 2; int64 pid = 2;
} }
message GetProcessResponse { message GetProcessResponse {
@ -102,12 +93,14 @@ message GetProcessResponse {
} }
message SignalProcessRequest { message SignalProcessRequest {
Process process = 1; Container container = 1;
uint32 signal = 2; Process process = 2;
uint32 signal = 3;
} }
message DeleteProcessRequest { message DeleteProcessRequest {
Process process = 1; Container container = 1;
Process process = 2;
} }
message ListProcessesRequest { message ListProcessesRequest {
@ -117,5 +110,3 @@ message ListProcessesRequest {
message ListProcessesResponse { message ListProcessesResponse {
repeated Process processes = 1; repeated Process processes = 1;
} }

6
api/execution/execution.pb.go
View file

@ -16,8 +16,8 @@
DeleteContainerResponse DeleteContainerResponse
ListContainersRequest ListContainersRequest
ListContainersResponse ListContainersResponse
CreateProcessRequest StartProcessRequest
CreateProcessResponse StartProcessResponse
Container Container
Process Process
User User
@ -26,8 +26,6 @@
UpdateContainerRequest UpdateContainerRequest
PauseContainerRequest PauseContainerRequest
ResumeContainerRequest ResumeContainerRequest
StartProcessRequest
StartProcessResponse
GetProcessRequest GetProcessRequest
GetProcessResponse GetProcessResponse
SignalProcessRequest SignalProcessRequest

2
api/execution/execution.proto
View file

@ -38,5 +38,3 @@ message ListContainersRequest {
message ListContainersResponse { message ListContainersResponse {
repeated Container containers = 1; repeated Container containers = 1;
} }

23
cmd/containerd/main.go
View file

@ -3,7 +3,6 @@ package main
import ( import (
"fmt" "fmt"
"net" "net"
"net/http"
"os" "os"
"os/signal" "os/signal"
"path/filepath" "path/filepath"
@ -15,7 +14,7 @@ import (
"github.com/docker/containerd" "github.com/docker/containerd"
api "github.com/docker/containerd/api/execution" api "github.com/docker/containerd/api/execution"
"github.com/docker/containerd/services/execution" "github.com/docker/containerd/services/execution"
metrics "github.com/docker/go-metrics" // metrics "github.com/docker/go-metrics"
"github.com/urfave/cli" "github.com/urfave/cli"
) )
@ -68,9 +67,9 @@ high performance container runtime
signals := make(chan os.Signal, 2048) signals := make(chan os.Signal, 2048)
signal.Notify(signals, syscall.SIGTERM, syscall.SIGINT) signal.Notify(signals, syscall.SIGTERM, syscall.SIGINT)
if address := context.GlobalString("metrics-address"); address != "" { // if address := context.GlobalString("metrics-address"); address != "" {
go serveMetrics(address) // go serveMetrics(address)
} // }
path := context.GlobalString("socket") path := context.GlobalString("socket")
if path == "" { if path == "" {
@ -120,13 +119,13 @@ func createUnixSocket(path string) (net.Listener, error) {
return net.Listen("unix", path) return net.Listen("unix", path)
} }
func serveMetrics(address string) { // func serveMetrics(address string) {
m := http.NewServeMux() // m := http.NewServeMux()
m.Handle("/metrics", metrics.Handler()) // m.Handle("/metrics", metrics.Handler())
if err := http.ListenAndServe(address, m); err != nil { // if err := http.ListenAndServe(address, m); err != nil {
logrus.WithError(err).Fatal("containerd: metrics server failure") // logrus.WithError(err).Fatal("containerd: metrics server failure")
} // }
} // }
func serveGRPC(server *grpc.Server, l net.Listener) { func serveGRPC(server *grpc.Server, l net.Listener) {
defer l.Close() defer l.Close()

213
container.go
View file

@ -1,213 +0,0 @@
package containerd
import (
"encoding/json"
"io"
"os"
"path/filepath"
"sync"
specs "github.com/opencontainers/runtime-spec/specs-go"
)
type ContainerConfig interface {
ID() string
Root() string // bundle path
Spec() (*specs.Spec, error)
Runtime() (Runtime, error)
}
func NewContainer(config ContainerConfig) (*Container, error) {
var (
id = config.ID()
root = config.Root()
path = filepath.Join(root, id)
)
s, err := config.Spec()
if err != nil {
return nil, err
}
// HACK: for runc to allow to use this path without a premounted rootfs
if err := os.MkdirAll(filepath.Join(path, s.Root.Path), 0711); err != nil {
return nil, err
}
f, err := os.Create(filepath.Join(path, "config.json"))
if err != nil {
return nil, err
}
// write the spec file to the container's directory
err = json.NewEncoder(f).Encode(s)
f.Close()
if err != nil {
return nil, err
}
r, err := config.Runtime()
if err != nil {
return nil, err
}
return &Container{
id: id,
path: path,
s: s,
driver: r,
}, nil
}
func LoadContainer(config ContainerConfig) (*Container, error) {
var (
id = config.ID()
root = config.Root()
path = filepath.Join(root, id)
)
spec, err := loadSpec(path)
if err != nil {
return nil, err
}
r, err := config.Runtime()
if err != nil {
return nil, err
}
process, err := r.Load(id)
if err != nil {
return nil, err
}
// TODO: load exec processes
return &Container{
id: id,
path: path,
s: spec,
driver: r,
init: &Process{
d: process,
driver: r,
},
}, nil
}
func loadSpec(path string) (*specs.Spec, error) {
f, err := os.Open(filepath.Join(path, "config.json"))
if err != nil {
return nil, err
}
var s specs.Spec
err = json.NewDecoder(f).Decode(&s)
f.Close()
if err != nil {
return nil, err
}
return &s, nil
}
type Container struct {
mu sync.Mutex
id string
path string
s *specs.Spec
driver Runtime
Stdin io.Reader
Stdout io.Writer
Stderr io.Writer
// init is the container's init processes
init *Process
// processes is a list of additional processes executed inside the container
// via the NewProcess method on the container
processes []*Process
}
// ID returns the id of the container
func (c *Container) ID() string {
return c.id
}
// Path returns the fully qualified path to the container on disk
func (c *Container) Path() string {
return c.path
}
// Spec returns the OCI runtime spec for the container
func (c *Container) Spec() *specs.Spec {
return c.s
}
// Create will create the container on the system by running the runtime's
// initial setup and process waiting for the user process to be started
func (c *Container) Create() error {
c.mu.Lock()
defer c.mu.Unlock()
d, err := c.driver.Create(c)
if err != nil {
return err
}
c.init = &Process{
d: d,
driver: c.driver,
Stdin: c.Stdin,
Stdout: c.Stdout,
Stderr: c.Stderr,
}
return nil
}
// Start will start the container's user specified process
func (c *Container) Start() error {
c.mu.Lock()
defer c.mu.Unlock()
return c.driver.Start(c)
}
// NewProcess will create a new process that will be executed inside the
// container and tied to the init processes lifecycle
func (c *Container) NewProcess(spec *specs.Process) (*Process, error) {
c.mu.Lock()
defer c.mu.Unlock()
process := &Process{
s: spec,
c: c,
exec: true,
driver: c.driver,
}
c.processes = append(c.processes, process)
return process, nil
}
// Pid returns the pid of the init or main process hosted inside the container
func (c *Container) Pid() int {
c.mu.Lock()
if c.init == nil {
c.mu.Unlock()
return -1
}
pid := c.init.Pid()
c.mu.Unlock()
return pid
}
// Wait will perform a blocking wait on the init process of the container
func (c *Container) Wait() (uint32, error) {
c.mu.Lock()
proc := c.init
c.mu.Unlock()
return proc.Wait()
}
// Signal will send the provided signal to the init process of the container
func (c *Container) Signal(s os.Signal) error {
c.mu.Lock()
defer c.mu.Unlock()
return c.init.Signal(s)
}
// Delete will delete the container if it no long has any processes running
// inside the container and removes all state on disk for the container
func (c *Container) Delete() error {
c.mu.Lock()
defer c.mu.Unlock()
err := c.driver.Delete(c)
if rerr := os.RemoveAll(c.path); err == nil {
err = rerr
}
return err
}

76
execution/container.go
View file

@ -1,85 +1,29 @@
package execution package execution
import (
"fmt"
"os"
"path/filepath"
specs "github.com/opencontainers/runtime-spec/specs-go"
)
type ContainerController interface {
Pause(*Container) error
Resume(*Container) error
Status(*Container) (Status, error)
Process(c *Container, pid int) (*Process, error)
Processes(*Container) ([]*Process, error)
}
func NewContainer(c ContainerController) *Container {
return &Container{
controller: c,
}
}
type Container struct { type Container struct {
ID string ID string
Bundle string Bundle string
Root string StateDir StateDir
controller ContainerController
processes map[int]*Process processes map[string]Process
} }
func (c *Container) Process(pid int) (*Process, error) { func (c *Container) AddProcess(p Process) {
for _, p := range c.processes { c.processes[p.ID()] = p
if p.Pid == pid {
return p, nil
}
}
return nil, fmt.Errorf("todo make real error")
} }
func (c *Container) CreateProcess(spec *specs.Process) (*Process, error) { func (c *Container) GetProcess(id string) Process {
if err := os.MkdirAll(filepath.Join(c.Root, c.getNextProcessID()), 0660); err != nil { return c.processes[id]
return nil, err
}
process := &Process{
Spec: spec,
controller: c.controller,
}
c.processes = append(c.processes, process)
return process, nil
} }
func (c *Container) DeleteProcess(pid int) error { func (c *Container) RemoveProcess(id string) {
process, ok := c.processes[pid] delete(c.processes, id)
if !ok {
return fmt.Errorf("it no here")
}
if process.Status() != Stopped {
return fmt.Errorf("tototoit not stopped ok?")
}
delete(c.processes, pid)
return os.RemoveAll(p.Root)
} }
func (c *Container) Processes() []*Process { func (c *Container) Processes() []Process {
var out []*Process var out []Process
for _, p := range c.processes { for _, p := range c.processes {
out = append(out, p) out = append(out, p)
} }
return out return out
} }
func (c *Container) Pause() error {
return c.controller.Pause(c)
}
func (c *Container) Resume() error {
return c.controller.Resume(c)
}
func (c *Container) Status() (Status, error) {
return c.controller.Status(c)
}

23
execution/executor.go
View file

@ -1,6 +1,11 @@
package execution package execution
import "io" import (
"io"
"os"
"github.com/opencontainers/runtime-spec/specs-go"
)
type CreateOpts struct { type CreateOpts struct {
Bundle string Bundle string
@ -9,9 +14,23 @@ type CreateOpts struct {
Stderr io.Writer Stderr io.Writer
} }
type CreateProcessOpts struct {
Spec specs.Process
Stdin io.Reader
Stdout io.Writer
Stderr io.Writer
}
type Executor interface { type Executor interface {
Create(id string, o CreateOpts) (*Container, error) Create(id string, o CreateOpts) (*Container, error)
Pause(*Container) error
Resume(*Container) error
Status(*Container) (Status, error)
List() ([]*Container, error) List() ([]*Container, error)
Load(id string) (*Container, error) Load(id string) (*Container, error)
Delete(string) error Delete(*Container) error
StartProcess(*Container, CreateProcessOpts) (Process, error)
SignalProcess(*Container, os.Signal) error
DeleteProcess(*Container, string) error
} }

8
execution/executors/executors.go
View file

@ -1,13 +1,13 @@
package executors package executors
import "github.com/docker/containerd" import "github.com/docker/containerd/execution"
var executors = make(map[string]func() containerd.Executor) var executors = make(map[string]func() execution.Executor)
func Register(name string, e func() containerd.Executor) { func Register(name string, e func() execution.Executor) {
executors[name] = e executors[name] = e
} }
func Get(name string) func() containerd.Executor { func Get(name string) func() execution.Executor {
return executors[name] return executors[name]
} }

23
execution/executors/oci/container.go
View file

@ -1,23 +0,0 @@
package oci
import "github.com/docker/containerd/execution"
type containerController struct {
root string
}
func (c *containerController) Process(container *execution.Container, pid int) (*execution.Process, error) {
}
func (c *containerController) Processes(container *execution.Container) ([]*execution.Process, error) {
}
func (c *containerController) Pause(container *execution.Container) error {
return command(c.root, "pause", container.ID).Run()
}
func (c *containerController) Resume(container *execution.Container) error {
return command(c.root, "resume", container.ID).Run()
}

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

@ -1,16 +1,14 @@
package oci package oci
import ( import (
"encoding/json"
"errors" "errors"
"fmt" "fmt"
"io/ioutil" "io/ioutil"
"os/exec" "os"
"path/filepath" "path/filepath"
"strconv" "syscall"
"time"
"github.com/docker/containerd" "github.com/crosbymichael/go-runc"
"github.com/docker/containerd/execution" "github.com/docker/containerd/execution"
"github.com/docker/containerd/executors" "github.com/docker/containerd/executors"
) )
@ -25,107 +23,178 @@ func init() {
func New(root string) *OCIRuntime { func New(root string) *OCIRuntime {
return &OCIRuntime{ return &OCIRuntime{
root: root, root: root,
Runc: &runc.Runc{
Root: filepath.Join(root, "runc"),
},
} }
} }
type OCIRuntime struct { type OCIRuntime struct {
// root holds runtime state information for the containers // root holds runtime state information for the containers
// launched by the runtime
root string root string
runc *runc.Runc
} }
func (r *OCIRuntime) Create(id string, o execution.CreateOpts) (*execution.Container, error) { func (r *OCIRuntime) Create(id string, o execution.CreateOpts) (*execution.Container, error) {
var err error
stateDir, err := NewStateDir(r.root, id)
if err != nil {
return nil, err
}
initStateDir, err := stateDir.NewProcess()
if err != nil {
return nil, err
}
// /run/runc/redis/1/pid // /run/runc/redis/1/pid
pidFile := filepath.Join(r.root, id, "1", "pid") pidFile := filepath.Join(initStateDir, "pid")
cmd := command(r.root, "create", err = r.runc.Create(id, o.Bundle, &runc.CreateOpts{
"--pid-file", pidFile, Pidfile: pidfile,
"--bundle", o.Bundle, Stdin: o.Stdin,
id, Stdout: o.Stdout,
) Stderr: o.Stderr,
cmd.Stdin, cmd.Stdout, cmd.Stderr = o.Stdin, o.Stdout, o.Stderr })
if err := cmd.Run(); err != nil {
return nil, err
}
// TODO: kill on error
data, err := ioutil.ReadFile(pidFile)
if err != nil { if err != nil {
return nil, err return nil, err
} }
pid, err := strconv.Atoi(string(data)) pid, err := runc.ReadPifFile(pidfile)
if err != nil { if err != nil {
return nil, err return nil, err
} }
container := execution.NewContainer(r) process, err := newProcess(pid)
container.ID = id
container.Root = filepath.Join(r.root, id)
container.Bundle = o.Bundle
process, err := container.CreateProcess(nil)
if err != nil { if err != nil {
return nil, err return nil, err
} }
process.Pid = pid
process.Stdin = o.Stdin container := &execution.Container{
process.Stdout = o.Stdout ID: id,
process.Stderr = o.Stderr Bundle: o.Bundle,
StateDir: stateDir,
}
container.AddProcess(process)
return container, nil return container, nil
} }
func (r *OCIRuntime) Load(id string) (containerd.ProcessDelegate, error) { func (r *OCIRuntime) load(runcC *runc.Container) (*execution.Container, error) {
data, err := r.Command("state", id).Output() container := &execution.Container{
if err != nil { ID: runcC.ID,
return nil, err Bundle: runcC.Bundle,
} StateDir: StateDir(filepath.Join(r.root, runcC.ID)),
var s state
if err := json.Unmarshal(data, &s); err != nil {
return nil, err
}
return newProcess(s.Pid)
} }
func (r *OCIRuntime) Delete(id string) error { process, err := newProcess(runcC.Pid)
return command(r.root, "delete", id).Run() if err != nil {
return nil, err
}
container.AddProcess(process)
// /run/containerd/container-id/processess/process-id
dirs, err := ioutil.ReadDir(filepath.Join(container.Root))
if err != nil {
return nil, err
} }
func (r *OCIRuntime) Exec(c *containerd.Container, p *containerd.Process) (containerd.ProcessDelegate, error) { return container, nil
f, err := ioutil.TempFile(filepath.Join(r.root, c.ID()), "process")
if err != nil {
return nil, err
}
path := f.Name()
pidFile := fmt.Sprintf("%s/%s.pid", filepath.Join(r.root, c.ID()), filepath.Base(path))
err = json.NewEncoder(f).Encode(p.Spec())
f.Close()
if err != nil {
return nil, err
}
cmd := r.Command("exec", "--detach", "--process", path, "--pid-file", pidFile, c.ID())
cmd.Stdin, cmd.Stdout, cmd.Stderr = p.Stdin, p.Stdout, p.Stderr
if err := cmd.Run(); err != nil {
return nil, err
}
data, err := ioutil.ReadFile(pidFile)
if err != nil {
return nil, err
}
i, err := strconv.Atoi(string(data))
if err != nil {
return nil, err
}
return newProcess(i)
} }
type state struct { func (r *OCIRuntime) List() ([]*execution.Container, error) {
ID string `json:"id"` runcCs, err := r.runc.List()
Pid int `json:"pid"` if err != nil {
Status string `json:"status"` return nil, err
Bundle string `json:"bundle"`
Rootfs string `json:"rootfs"`
Created time.Time `json:"created"`
Annotations map[string]string `json:"annotations"`
} }
func command(root, args ...string) *exec.Cmd { containers := make([]*execution.Container)
return exec.Command("runc", append( for _, c := range runcCs {
[]string{"--root", root}, container, err := r.load(c)
args...)...) if err != nil {
return nil, err
}
containers = append(containers, container)
}
return containers, nil
}
func (r *OCIRuntime) Load(id string) (*execution.Container, error) {
runcC, err := r.runc.State(id)
if err != nil {
return nil, err
}
return r.load(runcC)
}
func (r *OCIRuntime) Delete(c *execution.Container) error {
if err := r.runc.Delete(c.ID); err != nil {
return err
}
c.StateDir.Delete()
return nil
}
func (r *OCIRuntime) Pause(c *execution.Container) error {
return r.runc.Pause(c.ID)
}
func (r *OCIRuntime) Resume(c *execution.Container) error {
return r.runc.Resume(c.ID)
}
func (r *OCIRuntime) StartProcess(c *execution.Container, o CreateProcessOpts) (execution.Process, error) {
var err error
processStateDir, err := c.StateDir.NewProcess()
if err != nil {
return nil, err
}
defer func() {
if err != nil {
c.StateDir.DeleteProcess(filepath.Base(processStateDir))
}
}()
pidFile := filepath.Join(processStateDir, id)
err := r.runc.ExecProcess(c.ID, o.spec, &runc.ExecOpts{
PidFile: pidfile,
Detach: true,
Stdin: o.stdin,
Stdout: o.stdout,
Stderr: o.stderr,
})
if err != nil {
return nil, err
}
pid, err := runc.ReadPidFile(pidfile)
if err != nil {
return nil, err
}
process, err := newProcess(pid)
if err != nil {
return nil, err
}
container.AddProcess(process)
return process, nil
}
func (r *OCIRuntime) SignalProcess(c *execution.Container, id string, sig os.Signal) error {
process := c.GetProcess(id)
if process == nil {
return fmt.Errorf("Make a Process Not Found error")
}
return syscall.Kill(int(process.Pid()), os.Signal)
}
func (r *OCIRuntime) GetProcess(c *execution.Container, id string) process {
return c.GetProcess(id)
}
func (r *OCIRuntime) DeleteProcess(c *execution.Container, id string) error {
c.StateDir.DeleteProcess(id)
return nil
} }

4
execution/executors/oci/process.go
View file

@ -3,9 +3,11 @@ package oci
import ( import (
"os" "os"
"syscall" "syscall"
"github.com/docker/containerd/execution"
) )
func newProcess(pid int) (*process, error) { func newProcess(pid int) (execution.Process, error) {
proc, err := os.FindProcess(pid) proc, err := os.FindProcess(pid)
if err != nil { if err != nil {
return nil, err return nil, err

41
execution/process.go
View file

@ -1,41 +1,18 @@
package execution package execution
import ( import (
"io"
"os" "os"
specs "github.com/opencontainers/runtime-spec/specs-go" "github.com/opencontainers/runtime-spec/specs-go"
) )
type ProcessController interface { type Process interface {
Start(*Process) error ID() string
Status(*Process) (Status, error) Pid() int64
Wait(*Process) (uint32, error) Spec() *specs.Process
Signal(*Process, os.Signal) error
}
type Process struct { Start() error
Pid int Status() (Status, error)
Spec *specs.Process Wait() (uint32, error)
Stdin io.Reader Signal(os.Signal) error
Stdout io.Writer
Stderr io.Writer
controller ProcessController
}
func (p *Process) Status() (Status, error) {
return p.controller.Status(p)
}
func (p *Process) Wait() (uint32, error) {
return p.controller.Wait(p)
}
func (p *Process) Signal(s os.Signal) error {
return p.controller.Signal(p, s)
}
func (p *Process) Start() error {
return p.controller.Start(p)
} }

68
execution/statedir.go Normal file
View file

@ -0,0 +1,68 @@
package execution
import (
"io/ioutil"
"os"
"path/filepath"
)
const processesDir = "processes"
type StateDir string
func NewStateDir(root, id string) (StateDir, error) {
path := filepath.Join(root, id)
err := os.Mkdir(path, 0700)
if err != nil {
return "", err
}
err = os.Mkdir(filepath.Join(path, processesDir), 0700)
if err != nil {
os.RemoveAll(path)
return "", err
}
return StateDir(path), err
}
func (s StateDir) Delete() error {
return os.RemoveAll(string(s))
}
func (s StateDir) NewProcess(id string) (string, error) {
// TODO: generate id
newPath := filepath.Join(string(s), "1")
err := os.Mkdir(newPath, 0755)
if err != nil {
return "", err
}
return newPath, nil
}
func (s StateDir) ProcessDir(id string) string {
return filepath.Join(string(s), id)
}
func (s StateDir) DeleteProcess(id string) error {
return os.RemoveAll(filepath.Join(string(s), id))
}
func (s StateDir) Processes() ([]string, error) {
basepath := filepath.Join(string(s), processesDir)
dirs, err := ioutil.ReadDir(basepath)
if err != nil {
return nil, err
}
paths := make([]string, 0)
for _, d := range dirs {
if d.IsDir() {
paths = append(paths, filepath.Join(basepath, d.Name()))
}
}
return paths, nil
}

116
services/execution/service.go
View file

@ -1,11 +1,14 @@
package execution package execution
import ( import (
"context" "fmt"
api "github.com/docker/containerd/api/execution" api "github.com/docker/containerd/api/execution"
"github.com/docker/containerd/execution" "github.com/docker/containerd/execution"
"github.com/docker/containerd/executors" "github.com/docker/containerd/execution/executors"
google_protobuf "github.com/golang/protobuf/ptypes/empty"
"github.com/opencontainers/runtime-spec/specs-go"
"golang.org/x/net/context"
) )
type Opts struct { type Opts struct {
@ -14,7 +17,7 @@ type Opts struct {
} }
func New(o Opts) (*Service, error) { func New(o Opts) (*Service, error) {
executor := executors.Get(o.Runtime)(o.Root) executor := executors.Get(o.Runtime)()
return &Service{ return &Service{
o: o, o: o,
executor: executor, executor: executor,
@ -28,34 +31,46 @@ type Service struct {
func (s *Service) Create(ctx context.Context, r *api.CreateContainerRequest) (*api.CreateContainerResponse, error) { func (s *Service) Create(ctx context.Context, r *api.CreateContainerRequest) (*api.CreateContainerResponse, error) {
// TODO: write io and bundle path to dir // TODO: write io and bundle path to dir
container, err := s.executor.Create(r.ID, r.BundlePath, &IO{}) // TODO: open IOs
container, err := s.executor.Create(r.ID, execution.CreateOpts{
Bundle: r.BundlePath,
// Stdin: r.Stdin,
// Stdout: r.Stdout,
// Stderr: r.Stderr,
})
if err != nil { if err != nil {
return nil, err return nil, err
} }
s.supervisor.Add(container.InitProcess()) s.supervisor.Add(container)
return &api.CreateContainerResponse{ return &api.CreateContainerResponse{
Container: toGRPCContainer(container), Container: toGRPCContainer(container),
}, nil }, nil
} }
func (s *Service) Delete(ctx context.Context, r *api.DeleteContainerRequest) (*api.Empty, error) { func (s *Service) Delete(ctx context.Context, r *api.DeleteContainerRequest) (*google_protobuf.Empty, error) {
if err := s.executor.Delete(r.ID); err != nil { container, err := s.executor.Load(r.ID)
if err != nil {
return nil, err
}
if err = s.executor.Delete(container); err != nil {
return nil, err return nil, err
} }
return nil, nil return nil, nil
} }
func (s *Service) List(ctx context.Context, r *api.ListContainerRequest) (*api.ListContainerResponse, error) { func (s *Service) List(ctx context.Context, r *api.ListContainersRequest) (*api.ListContainersResponse, error) {
containers, err := s.executor.List() containers, err := s.executor.List()
if err != nil { if err != nil {
return nil, err return nil, err
} }
resp := &api.ListContainersResponse{}
for _, c := range containers { for _, c := range containers {
r.Containers = append(r.Containers, toGRPCContainer(c)) resp.Containers = append(resp.Containers, toGRPCContainer(c))
} }
return r, 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(r.ID)
@ -67,111 +82,94 @@ func (s *Service) Get(ctx context.Context, r *api.GetContainerRequest) (*api.Get
}, nil }, nil
} }
func (s *Service) Update(ctx context.Context, r *api.UpdateContainerRequest) (*api.Empty, error) { func (s *Service) Update(ctx context.Context, r *api.UpdateContainerRequest) (*google_protobuf.Empty, error) {
return nil, nil return nil, nil
} }
func (s *Service) Pause(ctx context.Context, r *api.PauseContainerRequest) (*api.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(r.ID)
if err != nil { if err != nil {
return nil, err return nil, err
} }
return nil, container.Pause() return nil, s.executor.Pause(container)
} }
func (s *Service) Resume(ctx context.Context, r *api.ResumeContainerRequest) (*api.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(r.ID)
if err != nil { if err != nil {
return nil, err return nil, err
} }
return nil, container.Resume() return nil, s.executor.Resume(container)
} }
func (s *Service) CreateProcess(ctx context.Context, r *api.CreateProcessRequest) (*api.CreateProcessResponse, error) { func (s *Service) StartProcess(ctx context.Context, r *api.StartProcessRequest) (*api.StartProcessResponse, error) {
container, err := s.executor.Load(r.ID) container, err := s.executor.Load(r.ContainerId)
if err != nil { if err != nil {
return nil, err return nil, err
} }
process, err := container.CreateProcess(r.Process) // TODO: generate spec
var spec specs.Process
// TODO: open IOs
process, err := s.executor.StartProcess(container, execution.CreateProcessOpts{
Spec: spec,
// Stdin: r.Stdin,
// Stdout: r.Stdout,
// Stderr: r.Stderr,
})
if err != nil { if err != nil {
return nil, err return nil, err
} }
s.supervisor.Add(process) s.supervisor.Add(process)
r.Process.Pid = process.Pid() return &api.StartProcessResponse{
return &api.CreateProcessResponse{ Process: toGRPCProcess(process),
Process: r.Process,
}, nil }, nil
} }
// 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.ID) container, err := s.executor.Load(r.Container.ID)
if err != nil { if err != nil {
return nil, err return nil, err
} }
process, err := container.Process(r.ProcessId) process := s.executor.GetProcess(r.Pid)
if err != nil { if process == nil {
return nil, err return nil, fmt.Errorf("Make me a constant! Process not foumd!")
} }
return &api.GetProcessResponse{ return &api.GetProcessResponse{
Process: process, Process: toGRPCProcess(process),
}, nil }, nil
} }
func (s *Service) StartProcess(ctx context.Context, r *api.StartProcessRequest) (*api.StartProcessResponse, error) { func (s *Service) SignalProcess(ctx context.Context, r *api.SignalProcessRequest) (*google_protobuf.Empty, error) {
container, err := s.executor.Load(r.ID) container, err := s.executor.Load(r.Container.ID)
if err != nil { if err != nil {
return nil, err return nil, err
} }
process, err := container.Process(r.Process.ID) return nil, s.executor.SignalProcess(container, r.Process.ID, r.Signal)
if err != nil {
return nil, err
}
if err := process.Start(); err != nil {
return nil, err
}
return &api.StartProcessRequest{
Process: process,
}, nil
} }
func (s *Service) SignalProcess(ctx context.Context, r *api.SignalProcessRequest) (*api.Empty, error) { func (s *Service) DeleteProcess(ctx context.Context, r *api.DeleteProcessRequest) (*google_protobuf.Empty, error) {
container, err := s.executor.Load(r.ID) container, err := s.executor.Load(r.Container.ID)
if err != nil { if err != nil {
return nil, err return nil, err
} }
process, err := container.Process(r.Process.ID) if err := s.executor.DeleteProcess(container, r.Process.ID); err != nil {
if err != nil {
return nil, err
}
return nil, process.Signal(r.Signal)
}
func (s *Service) DeleteProcess(ctx context.Context, r *api.DeleteProcessRequest) (*api.Empty, error) {
container, err := s.executor.Load(r.ID)
if err != nil {
return nil, err
}
if err := container.DeleteProcess(r.Process.ID); err != nil {
return nil, err return nil, err
} }
return nil, nil return nil, 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.ID) container, err := s.executor.Load(r.Container.ID)
if err != nil {
return nil, err
}
processes, err := container.Processes()
if err != nil { if err != nil {
return nil, err return nil, err
} }
processes := container.Processes()
return &api.ListProcessesResponse{ return &api.ListProcessesResponse{
Processes: processes, Processes: toGRPCProcesses(processes),
}, nil }, nil
} }