Update libcontainerd to use containerd 1.0

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

1 files changed, 86 insertions, 17 deletions

diff --git a/plugin/executor/containerd/containerd.go b/plugin/executor/containerd/containerd.go
index 74cf530cf1..d93b8b75ec 100644
--- a/plugin/executor/containerd/containerd.go
+++ b/plugin/executor/containerd/containerd.go
@@ -1,22 +1,35 @@
 package containerd
 
 import (
+	"context"
 	"io"
+	"path/filepath"
+	"sync"
 
+	"github.com/containerd/containerd"
+	"github.com/containerd/containerd/linux/runcopts"
+	"github.com/docker/docker/api/errdefs"
 	"github.com/docker/docker/libcontainerd"
 	"github.com/opencontainers/runtime-spec/specs-go"
 	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
 )
 
+// PluginNamespace is the name used for the plugins namespace
+var PluginNamespace = "moby-plugins"
+
 // ExitHandler represents an object that is called when the exit event is received from containerd
 type ExitHandler interface {
 	HandleExitEvent(id string) error
 }
 
 // New creates a new containerd plugin executor
-func New(remote libcontainerd.Remote, exitHandler ExitHandler) (*Executor, error) {
-	e := &Executor{exitHandler: exitHandler}
-	client, err := remote.Client(e)
+func New(rootDir string, remote libcontainerd.Remote, exitHandler ExitHandler) (*Executor, error) {
+	e := &Executor{
+		rootDir:     rootDir,
+		exitHandler: exitHandler,
+	}
+	client, err := remote.NewClient(PluginNamespace, e)
 	if err != nil {
 		return nil, errors.Wrap(err, "error creating containerd exec client")
 	}
@@ -26,52 +39,108 @@ func New(remote libcontainerd.Remote, exitHandler ExitHandler) (*Executor, error
 
 // Executor is the containerd client implementation of a plugin executor
 type Executor struct {
+	rootDir     string
 	client      libcontainerd.Client
 	exitHandler ExitHandler
 }
 
 // Create creates a new container
 func (e *Executor) Create(id string, spec specs.Spec, stdout, stderr io.WriteCloser) error {
-	return e.client.Create(id, "", "", spec, attachStreamsFunc(stdout, stderr))
+	opts := runcopts.RuncOptions{
+		RuntimeRoot: filepath.Join(e.rootDir, "runtime-root"),
+	}
+	ctx := context.Background()
+	err := e.client.Create(ctx, id, &spec, &opts)
+	if err != nil {
+		return err
+	}
+
+	_, err = e.client.Start(ctx, id, "", false, attachStreamsFunc(stdout, stderr))
+	return err
 }
 
 // Restore restores a container
 func (e *Executor) Restore(id string, stdout, stderr io.WriteCloser) error {
-	return e.client.Restore(id, attachStreamsFunc(stdout, stderr))
+	alive, _, err := e.client.Restore(context.Background(), id, attachStreamsFunc(stdout, stderr))
+	if err != nil && !errdefs.IsNotFound(err) {
+		return err
+	}
+	if !alive {
+		_, _, err = e.client.DeleteTask(context.Background(), id)
+		if err != nil && !errdefs.IsNotFound(err) {
+			logrus.WithError(err).Errorf("failed to delete container plugin %s task from containerd", id)
+			return err
+		}
+
+		err = e.client.Delete(context.Background(), id)
+		if err != nil && !errdefs.IsNotFound(err) {
+			logrus.WithError(err).Errorf("failed to delete container plugin %s from containerd", id)
+			return err
+		}
+	}
+	return nil
 }
 
 // IsRunning returns if the container with the given id is running
 func (e *Executor) IsRunning(id string) (bool, error) {
-	pids, err := e.client.GetPidsForContainer(id)
-	return len(pids) > 0, err
+	status, err := e.client.Status(context.Background(), id)
+	return status == libcontainerd.StatusRunning, err
 }
 
 // Signal sends the specified signal to the container
 func (e *Executor) Signal(id string, signal int) error {
-	return e.client.Signal(id, signal)
+	return e.client.SignalProcess(context.Background(), id, libcontainerd.InitProcessName, signal)
 }
 
-// StateChanged handles state changes from containerd
+// ProcessEvent handles events from containerd
 // All events are ignored except the exit event, which is sent of to the stored handler
-func (e *Executor) StateChanged(id string, event libcontainerd.StateInfo) error {
-	switch event.State {
-	case libcontainerd.StateExit:
-		return e.exitHandler.HandleExitEvent(id)
+func (e *Executor) ProcessEvent(id string, et libcontainerd.EventType, ei libcontainerd.EventInfo) error {
+	switch et {
+	case libcontainerd.EventExit:
+		// delete task and container
+		if _, _, err := e.client.DeleteTask(context.Background(), id); err != nil {
+			logrus.WithError(err).Errorf("failed to delete container plugin %s task from containerd", id)
+		}
+
+		if err := e.client.Delete(context.Background(), id); err != nil {
+			logrus.WithError(err).Errorf("failed to delete container plugin %s from containerd", id)
+		}
+		return e.exitHandler.HandleExitEvent(ei.ContainerID)
 	}
 	return nil
 }
 
-func attachStreamsFunc(stdout, stderr io.WriteCloser) func(libcontainerd.IOPipe) error {
-	return func(iop libcontainerd.IOPipe) error {
-		iop.Stdin.Close()
+type cio struct {
+	containerd.IO
+
+	wg sync.WaitGroup
+}
+
+func (c *cio) Wait() {
+	c.wg.Wait()
+	c.IO.Wait()
+}
+
+func attachStreamsFunc(stdout, stderr io.WriteCloser) libcontainerd.StdioCallback {
+	return func(iop *libcontainerd.IOPipe) (containerd.IO, error) {
+		if iop.Stdin != nil {
+			iop.Stdin.Close()
+			// closing stdin shouldn't be needed here, it should never be open
+			panic("plugin stdin shouldn't have been created!")
+		}
+
+		cio := &cio{IO: iop}
+		cio.wg.Add(2)
 		go func() {
 			io.Copy(stdout, iop.Stdout)
 			stdout.Close()
+			cio.wg.Done()
 		}()
 		go func() {
 			io.Copy(stderr, iop.Stderr)
 			stderr.Close()
+			cio.wg.Done()
 		}()
-		return nil
+		return cio, nil
 	}
 }