Add event support to execution subsystem

The implementation relies on nats.io

Signed-off-by: Kenfe-Mickael Laventure <mickael.laventure@gmail.com>

execution/container.go

@@ -2,7 +2,7 @@ package execution
 
 import "fmt"
 
-func NewContainer(stateRoot, id, bundle, status string) (*Container, error) {
+func NewContainer(stateRoot, id, bundle string) (*Container, error) {
 	stateDir, err := NewStateDir(stateRoot, id)
 	if err != nil {
 		return nil, err
@@ -11,7 +11,7 @@ func NewContainer(stateRoot, id, bundle, status string) (*Container, error) {
 		id:        id,
 		bundle:    bundle,
 		stateDir:  stateDir,
-		status:    status,
+		status:    "created",
 		processes: make(map[string]Process),
 	}, nil
 }

execution/events.go

@@ -0,0 +1,21 @@
+package execution
+
+type ContainerEvent struct {
+	ID     string
+	Action string
+}
+
+type ContainerExitEvent struct {
+	ContainerEvent
+	PID        string
+	StatusCode uint32
+}
+
+const (
+	ContainersEventsSubjectSubscriber = "containerd.execution.container.>"
+)
+
+const (
+	containerEventsSubjectFormat        = "containerd.execution.container.%s"
+	containerProcessEventsSubjectFormat = "containerd.execution.container.%s.%s"
+)

execution/executors/oci/io.go

@@ -0,0 +1,61 @@
+package oci
+
+import (
+	"io"
+	"os"
+
+	"github.com/crosbymichael/go-runc"
+)
+
+type OIO struct {
+	master  *os.File // master holds a fd to the created pty if any
+	console string   // console holds the path the the slave linked to master
+	rio     runc.IO  // rio holds the open fifos for stdios
+}
+
+func newOIO(stdin, stdout, stderr string, console bool) (o OIO, err error) {
+	defer func() {
+		if err != nil {
+			o.cleanup()
+		}
+	}()
+
+	if o.rio.Stdin, err = os.OpenFile(stdin, os.O_RDONLY, 0); err != nil {
+		return
+	}
+	if o.rio.Stdout, err = os.OpenFile(stdout, os.O_WRONLY, 0); err != nil {
+		return
+	}
+	if o.rio.Stderr, err = os.OpenFile(stderr, os.O_WRONLY, 0); err != nil {
+		return
+	}
+
+	if console {
+		o.master, o.console, err = newConsole(0, 0)
+		if err != nil {
+			return
+		}
+		go io.Copy(o.master, o.rio.Stdin)
+		go func() {
+			io.Copy(o.rio.Stdout, o.master)
+			o.master.Close()
+		}()
+	}
+
+	return
+}
+
+func (o OIO) cleanup() {
+	if o.master != nil {
+		o.master.Close()
+	}
+	if o.rio.Stdin != nil {
+		o.rio.Stdin.(*os.File).Close()
+	}
+	if o.rio.Stdout != nil {
+		o.rio.Stdout.(*os.File).Close()
+	}
+	if o.rio.Stderr != nil {
+		o.rio.Stderr.(*os.File).Close()
+	}
+}

execution/executors/oci/oci.go

@@ -14,23 +14,24 @@ import (
 
 var ErrRootEmpty = errors.New("oci: runtime root cannot be an empty string")
 
-func New(root string) *OCIRuntime {
+func New(root string) (*OCIRuntime, error) {
+	err := SetSubreaper(1)
+	if err != nil {
+		return nil, err
+	}
 	return &OCIRuntime{
 		root: root,
 		runc: &runc.Runc{
 			Root: filepath.Join(root, "runc"),
 		},
 		ios: make(map[string]OIO),
-	}
+	}, nil
 }
 
 type OCIRuntime struct {
-	// root holds runtime state information for the containers
 	root string
 	runc *runc.Runc
-
-	// We need to keep track of the created IO for
-	ios map[string]OIO
+	ios  map[string]OIO // ios tracks created process io for cleanup purpose on delete
 }
 
 func (r *OCIRuntime) Create(ctx context.Context, id string, o execution.CreateOpts) (container *execution.Container, err error) {
@@ -44,7 +45,7 @@ func (r *OCIRuntime) Create(ctx context.Context, id string, o execution.CreateOp
 		}
 	}()
 
-	if container, err = execution.NewContainer(r.root, id, o.Bundle, "created"); err != nil {
+	if container, err = execution.NewContainer(r.root, id, o.Bundle); err != nil {
 		return nil, err
 	}
 	defer func(c *execution.Container) {
@@ -141,11 +142,16 @@ func (r *OCIRuntime) List(ctx context.Context) ([]*execution.Container, error) {
 
 	var containers []*execution.Container
 	for _, c := range runcCs {
-		container, err := r.load(c)
-		if err != nil {
-			return nil, err
+		select {
+		case <-ctx.Done():
+			return nil, ctx.Err()
+		default:
+			container, err := r.load(c)
+			if err != nil {
+				return nil, err
+			}
+			containers = append(containers, container)
 		}
-		containers = append(containers, container)
 	}
 
 	return containers, nil

execution/executors/oci/reaper.go

@@ -0,0 +1,43 @@
+package oci
+
+import (
+	"syscall"
+	"unsafe"
+)
+
+// PR_SET_CHILD_SUBREAPER allows setting the child subreaper.
+// If arg2 is nonzero, set the "child subreaper" attribute of the
+// calling process; if arg2 is zero, unset the attribute.  When a
+// process is marked as a child subreaper, all of the children
+// that it creates, and their descendants, will be marked as
+// having a subreaper.  In effect, a subreaper fulfills the role
+// of init(1) for its descendant processes.  Upon termination of
+// a process that is orphaned (i.e., its immediate parent has
+// already terminated) and marked as having a subreaper, the
+// nearest still living ancestor subreaper will receive a SIGCHLD
+// signal and be able to wait(2) on the process to discover its
+// termination status.
+const prSetChildSubreaper = 36
+
+// PR_GET_CHILD_SUBREAPER allows retrieving the current child
+// subreaper.
+// Return the "child subreaper" setting of the caller, in the
+// location pointed to by (int *) arg2.
+const prGetChildSubreaper = 37
+
+// GetSubreaper returns the subreaper setting for the calling process
+func GetSubreaper() (int, error) {
+	var i uintptr
+	if _, _, err := syscall.RawSyscall(syscall.SYS_PRCTL, prGetChildSubreaper, uintptr(unsafe.Pointer(&i)), 0); err != 0 {
+		return -1, err
+	}
+	return int(i), nil
+}
+
+// SetSubreaper sets the value i as the subreaper setting for the calling process
+func SetSubreaper(i int) error {
+	if _, _, err := syscall.RawSyscall(syscall.SYS_PRCTL, prSetChildSubreaper, uintptr(i), 0); err != 0 {
+		return err
+	}
+	return nil
+}

execution/service.go

@@ -6,6 +6,7 @@ import (
 
 	api "github.com/docker/containerd/api/execution"
 	google_protobuf "github.com/golang/protobuf/ptypes/empty"
+	"github.com/nats-io/go-nats"
 	"github.com/opencontainers/runtime-spec/specs-go"
 	"golang.org/x/net/context"
 )
@@ -15,19 +16,20 @@ var (
 	ErrProcessNotFound = fmt.Errorf("Process not found")
 )
 
-func New(executor Executor) (*Service, error) {
+func New(executor Executor, nec *nats.EncodedConn) (*Service, error) {
 	return &Service{
 		executor: executor,
+		nec:      nec,
 	}, nil
 }
 
 type Service struct {
 	executor   Executor
 	supervisor *Supervisor
+	nec        *nats.EncodedConn
 }
 
 func (s *Service) Create(ctx context.Context, r *api.CreateContainerRequest) (*api.CreateContainerResponse, error) {
-	// TODO: write io and bundle path to dir
 	var err error
 
 	container, err := s.executor.Create(ctx, r.ID, CreateOpts{
@@ -41,10 +43,14 @@ func (s *Service) Create(ctx context.Context, r *api.CreateContainerRequest) (*a
 		return nil, err
 	}
 
-	s.supervisor.Add(container)
+	procs := container.Processes()
+	initProcess := procs[0]
+
+	s.monitorProcess(container, initProcess)
 
 	return &api.CreateContainerResponse{
-		Container: toGRPCContainer(container),
+		Container:   toGRPCContainer(container),
+		InitProcess: toGRPCProcess(initProcess),
 	}, nil
 }
 
@@ -138,7 +144,8 @@ func (s *Service) StartProcess(ctx context.Context, r *api.StartProcessRequest)
 	if err != nil {
 		return nil, err
 	}
-	s.supervisor.Add(process)
+
+	s.monitorProcess(container, process)
 
 	return &api.StartProcessResponse{
 		Process: toGRPCProcess(process),
@@ -198,6 +205,43 @@ var (
 	_ = (api.ExecutionServiceServer)(&Service{})
 )
 
+func (s *Service) publishEvent(name string, v interface{}) {
+	if s.nec == nil {
+		return
+	}
+
+	err := s.nec.Publish(name, v)
+	if err != nil {
+		// TODO: Use logrus?
+		fmt.Println("Failed to publish '%s:%#v': %v", name, v, err)
+	}
+}
+
+func (s *Service) monitorProcess(container *Container, process Process) {
+	go func() {
+		status, err := process.Wait()
+		if err == nil {
+			subject := GetContainerProcessEventSubject(container.ID(), process.ID())
+			s.publishEvent(subject, &ContainerExitEvent{
+				ContainerEvent: ContainerEvent{
+					ID:     container.ID(),
+					Action: "exit",
+				},
+				PID:        process.ID(),
+				StatusCode: status,
+			})
+		}
+	}()
+}
+
+func GetContainerEventSubject(id string) string {
+	return fmt.Sprintf(containerEventsSubjectFormat, id)
+}
+
+func GetContainerProcessEventSubject(containerID, processID string) string {
+	return fmt.Sprintf(containerProcessEventsSubjectFormat, containerID, processID)
+}
+
 func toGRPCContainer(container *Container) *api.Container {
 	c := &api.Container{
 		ID:         container.ID(),

execution/supervisor.go

@@ -7,6 +7,6 @@ type waiter interface {
 	Wait() (uint32, error)
 }
 
-func (s *Supervisor) Add(w waiter) {
-
+func (s *Supervisor) Monitor(w waiter, cb func(uint32, error)) {
+	go cb(w.Wait())
 }