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

Remove runtime.Event types

Browse Source 
This uses the events service types for runtime events

Signed-off-by: Michael Crosby <crosbymichael@gmail.com>
This commit is contained in:
Michael Crosby
2017-06-28 10:33:36 -07:00
parent 7c8acca29a
commit 040558cf81
15 changed files with 271 additions and 596 deletions
Download Patch File Download Diff File Expand all files Collapse all files

47
windows/container.go
View File

@@ -9,8 +9,9 @@ import (
"sync"
"time"
events "github.com/containerd/containerd/api/services/events/v1"
"github.com/containerd/containerd/log"
"github.com/containerd/containerd/plugin"
"github.com/containerd/containerd/runtime"
"github.com/containerd/containerd/windows/hcs"
specs "github.com/opencontainers/runtime-spec/specs-go"
"github.com/pkg/errors"
@@ -19,7 +20,7 @@ import (
var ErrLoadedContainer = errors.New("loaded container can only be terminated")
type eventCallback func(id string, evType plugin.EventType, pid, exitStatus uint32, exitedAt time.Time)
type eventCallback func(id string, evType events.RuntimeEvent_EventType, pid, exitStatus uint32, exitedAt time.Time)
func loadContainers(ctx context.Context, h *hcs.HCS, sendEvent eventCallback) ([]*container, error) {
hCtr, err := h.LoadContainers(ctx)
@@ -31,7 +32,7 @@ func loadContainers(ctx context.Context, h *hcs.HCS, sendEvent eventCallback) ([
for _, c := range hCtr {
containers = append(containers, &container{
ctr: c,
status: plugin.RunningStatus,
status: runtime.RunningStatus,
sendEvent: sendEvent,
})
}
@@ -39,7 +40,7 @@ func loadContainers(ctx context.Context, h *hcs.HCS, sendEvent eventCallback) ([
return containers, nil
}
func newContainer(ctx context.Context, h *hcs.HCS, id string, spec RuntimeSpec, io plugin.IO, sendEvent eventCallback) (*container, error) {
func newContainer(ctx context.Context, h *hcs.HCS, id string, spec RuntimeSpec, io runtime.IO, sendEvent eventCallback) (*container, error) {
cio, err := hcs.NewIO(io.Stdin, io.Stdout, io.Stderr, io.Terminal)
if err != nil {
return nil, err
@@ -49,11 +50,11 @@ func newContainer(ctx context.Context, h *hcs.HCS, id string, spec RuntimeSpec,
if err != nil {
return nil, err
}
sendEvent(id, plugin.CreateEvent, hcsCtr.Pid(), 0, time.Time{})
sendEvent(id, events.RuntimeEvent_CREATE, hcsCtr.Pid(), 0, time.Time{})
return &container{
ctr: hcsCtr,
status: plugin.CreatedStatus,
status: runtime.CreatedStatus,
sendEvent: sendEvent,
}, nil
}
@@ -62,12 +63,12 @@ type container struct {
sync.Mutex
ctr *hcs.Container
status plugin.Status
status runtime.Status
sendEvent eventCallback
}
func (c *container) Info() plugin.TaskInfo {
return plugin.TaskInfo{
func (c *container) Info() runtime.TaskInfo {
return runtime.TaskInfo{
ID: c.ctr.ID(),
Runtime: runtimeName,
}
@@ -83,8 +84,8 @@ func (c *container) Start(ctx context.Context) error {
return err
}
c.setStatus(plugin.RunningStatus)
c.sendEvent(c.ctr.ID(), plugin.StartEvent, c.ctr.Pid(), 0, time.Time{})
c.setStatus(runtime.RunningStatus)
c.sendEvent(c.ctr.ID(), events.RuntimeEvent_START, c.ctr.Pid(), 0, time.Time{})
// Wait for our process to terminate
go func() {
@@ -92,8 +93,8 @@ func (c *container) Start(ctx context.Context) error {
if err != nil {
log.G(ctx).Debug(err)
}
c.setStatus(plugin.StoppedStatus)
c.sendEvent(c.ctr.ID(), plugin.ExitEvent, c.ctr.Pid(), ec, c.ctr.Processes()[0].ExitedAt())
c.setStatus(runtime.StoppedStatus)
c.sendEvent(c.ctr.ID(), events.RuntimeEvent_EXIT, c.ctr.Pid(), ec, c.ctr.Processes()[0].ExitedAt())
}()
return nil
@@ -113,8 +114,8 @@ func (c *container) Resume(ctx context.Context) error {
return c.ctr.Resume()
}
func (c *container) State(ctx context.Context) (plugin.State, error) {
return plugin.State{
func (c *container) State(ctx context.Context) (runtime.State, error) {
return runtime.State{
Pid: c.Pid(),
Status: c.Status(),
}, nil
@@ -127,7 +128,7 @@ func (c *container) Kill(ctx context.Context, signal uint32, pid uint32, all boo
return c.ctr.Stop(ctx)
}
func (c *container) Exec(ctx context.Context, opts plugin.ExecOpts) (plugin.Process, error) {
func (c *container) Exec(ctx context.Context, opts runtime.ExecOpts) (runtime.Process, error) {
if c.ctr.Pid() == 0 {
return nil, ErrLoadedContainer
}
@@ -152,7 +153,7 @@ func (c *container) Exec(ctx context.Context, opts plugin.ExecOpts) (plugin.Proc
if err != nil {
log.G(ctx).Debug(err)
}
c.sendEvent(c.ctr.ID(), plugin.ExitEvent, p.Pid(), ec, p.ExitedAt())
c.sendEvent(c.ctr.ID(), events.RuntimeEvent_EXEC_ADDED, p.Pid(), ec, p.ExitedAt())
}()
return &process{p}, nil
@@ -162,11 +163,11 @@ func (c *container) CloseIO(ctx context.Context, pid uint32) error {
return c.ctr.CloseIO(ctx, pid)
}
func (c *container) ResizePty(ctx context.Context, pid uint32, size plugin.ConsoleSize) error {
func (c *container) ResizePty(ctx context.Context, pid uint32, size runtime.ConsoleSize) error {
return c.ctr.ResizePty(ctx, pid, size)
}
func (c *container) Status() plugin.Status {
func (c *container) Status() runtime.Status {
return c.getStatus()
}
@@ -192,7 +193,7 @@ func (c *container) Checkpoint(ctx context.Context, _ string, _ map[string]strin
return fmt.Errorf("Windows containers do not support checkpoint")
}
func (c *container) DeleteProcess(ctx context.Context, pid uint32) (*plugin.Exit, error) {
func (c *container) DeleteProcess(ctx context.Context, pid uint32) (*runtime.Exit, error) {
var process *hcs.Process
for _, p := range c.ctr.Processes() {
if p.Pid() == pid {
@@ -208,7 +209,7 @@ func (c *container) DeleteProcess(ctx context.Context, pid uint32) (*plugin.Exit
return nil, err
}
process.Delete()
return &plugin.Exit{
return &runtime.Exit{
Status: ec,
Timestamp: process.ExitedAt(),
}, nil
@@ -218,13 +219,13 @@ func (c *container) Update(ctx context.Context, spec []byte) error {
return fmt.Errorf("Windows containers do not support update")
}
func (c *container) setStatus(status plugin.Status) {
func (c *container) setStatus(status runtime.Status) {
c.Lock()
c.status = status
c.Unlock()
}
func (c *container) getStatus() plugin.Status {
func (c *container) getStatus() runtime.Status {
c.Lock()
defer c.Unlock()
return c.status

4
windows/hcs/hcs.go
View File

@@ -16,7 +16,7 @@ import (
"github.com/Microsoft/hcsshim"
"github.com/Sirupsen/logrus"
"github.com/containerd/containerd/log"
"github.com/containerd/containerd/plugin"
"github.com/containerd/containerd/runtime"
"github.com/containerd/containerd/windows/pid"
"github.com/opencontainers/runtime-spec/specs-go"
"github.com/pkg/errors"
@@ -216,7 +216,7 @@ func (c *Container) CloseIO(ctx context.Context, pid uint32) error {
return proc.CloseStdin()
}
func (c *Container) ResizePty(ctx context.Context, pid uint32, size plugin.ConsoleSize) error {
func (c *Container) ResizePty(ctx context.Context, pid uint32, size runtime.ConsoleSize) error {
var proc *Process
c.Lock()
for _, p := range c.processes {

8
windows/hcs/process.go
View File

@@ -7,7 +7,7 @@ import (
"time"
"github.com/Microsoft/hcsshim"
"github.com/containerd/containerd/plugin"
"github.com/containerd/containerd/runtime"
"github.com/pkg/errors"
)
@@ -35,14 +35,14 @@ func (p *Process) ExitedAt() time.Time {
return p.exitedAt
}
func (p *Process) Status() plugin.Status {
func (p *Process) Status() runtime.Status {
select {
case <-p.ecSync:
return plugin.StoppedStatus
return runtime.StoppedStatus
default:
}
return plugin.RunningStatus
return runtime.RunningStatus
}
func (p *Process) Delete() error {

8
windows/process.go
View File

@@ -5,7 +5,7 @@ package windows
import (
"context"
"github.com/containerd/containerd/plugin"
"github.com/containerd/containerd/runtime"
"github.com/containerd/containerd/windows/hcs"
)
@@ -14,8 +14,8 @@ type process struct {
*hcs.Process
}
func (p *process) State(ctx context.Context) (plugin.State, error) {
return plugin.State{
func (p *process) State(ctx context.Context) (runtime.State, error) {
return runtime.State{
Pid: p.Pid(),
Status: p.Status(),
}, nil
@@ -25,7 +25,7 @@ func (p *process) Kill(ctx context.Context, sig uint32, all bool) error {
return p.Process.Kill()
}
func (p *process) Status() plugin.Status {
func (p *process) Status() runtime.Status {
return p.Process.Status()
}

27
windows/runtime.go
View File

@@ -11,8 +11,10 @@ import (
"sync"
"time"
events "github.com/containerd/containerd/api/services/events/v1"
"github.com/containerd/containerd/log"
"github.com/containerd/containerd/plugin"
"github.com/containerd/containerd/runtime"
"github.com/containerd/containerd/windows/hcs"
"github.com/containerd/containerd/windows/pid"
specs "github.com/opencontainers/runtime-spec/specs-go"
@@ -24,7 +26,7 @@ const (
owner = "containerd"
)
var _ = (plugin.Runtime)(&Runtime{})
var _ = (runtime.Runtime)(&Runtime{})
func init() {
plugin.Register(&plugin.Registration{
@@ -44,7 +46,7 @@ func New(ic *plugin.InitContext) (interface{}, error) {
r := &Runtime{
pidPool: pid.NewPool(),
containers: make(map[string]*container),
events: make(chan *plugin.Event, 2048),
events: make(chan *events.RuntimeEvent, 2048),
eventsContext: c,
eventsCancel: cancel,
rootDir: rootDir,
@@ -60,7 +62,7 @@ func New(ic *plugin.InitContext) (interface{}, error) {
for _, c := range ctrs {
c.ctr.Delete(ic.Context)
r.sendEvent(c.ctr.ID(), plugin.ExitEvent, c.ctr.Pid(), 255, time.Time{})
r.sendEvent(c.ctr.ID(), events.RuntimeEvent_EXIT, c.ctr.Pid(), 255, time.Time{})
}
// Try to delete the old state dir and recreate it
@@ -87,7 +89,7 @@ type Runtime struct {
containers map[string]*container
events chan *plugin.Event
events chan *events.RuntimeEvent
eventsContext context.Context
eventsCancel func()
}
@@ -104,7 +106,7 @@ func (r *Runtime) ID() string {
return fmt.Sprintf("%s.%s", plugin.RuntimePlugin, runtimeName)
}
func (r *Runtime) Create(ctx context.Context, id string, opts plugin.CreateOpts) (plugin.Task, error) {
func (r *Runtime) Create(ctx context.Context, id string, opts runtime.CreateOpts) (runtime.Task, error) {
var rtSpec RuntimeSpec
if err := json.Unmarshal(opts.Spec, &rtSpec); err != nil {
return nil, errors.Wrap(err, "failed to unmarshal oci spec")
@@ -122,7 +124,7 @@ func (r *Runtime) Create(ctx context.Context, id string, opts plugin.CreateOpts)
return ctr, nil
}
func (r *Runtime) Delete(ctx context.Context, c plugin.Task) (*plugin.Exit, error) {
func (r *Runtime) Delete(ctx context.Context, c runtime.Task) (*runtime.Exit, error) {
wc, ok := c.(*container)
if !ok {
return nil, fmt.Errorf("container cannot be cast as *windows.container")
@@ -138,16 +140,16 @@ func (r *Runtime) Delete(ctx context.Context, c plugin.Task) (*plugin.Exit, erro
delete(r.containers, wc.ctr.ID())
r.Unlock()
return &plugin.Exit{
return &runtime.Exit{
Status: ec,
Timestamp: wc.ctr.Processes()[0].ExitedAt(),
}, nil
}
func (r *Runtime) Tasks(ctx context.Context) ([]plugin.Task, error) {
func (r *Runtime) Tasks(ctx context.Context) ([]runtime.Task, error) {
r.Lock()
defer r.Unlock()
list := make([]plugin.Task, len(r.containers))
list := make([]runtime.Task, len(r.containers))
for _, c := range r.containers {
select {
case <-ctx.Done():
@@ -159,7 +161,7 @@ func (r *Runtime) Tasks(ctx context.Context) ([]plugin.Task, error) {
return list, nil
}
func (r *Runtime) Get(ctx context.Context, id string) (plugin.Task, error) {
func (r *Runtime) Get(ctx context.Context, id string) (runtime.Task, error) {
r.Lock()
defer r.Unlock()
c, ok := r.containers[id]
@@ -169,10 +171,9 @@ func (r *Runtime) Get(ctx context.Context, id string) (plugin.Task, error) {
return c, nil
}
func (r *Runtime) sendEvent(id string, evType plugin.EventType, pid, exitStatus uint32, exitedAt time.Time) {
r.events <- &plugin.Event{
func (r *Runtime) sendEvent(id string, evType events.RuntimeEvent_EventType, pid, exitStatus uint32, exitedAt time.Time) {
r.events <- &events.RuntimeEvent{
Timestamp: time.Now(),
Runtime: runtimeName,
Type: evType,
Pid: pid,
ID: id,