Refactor libcontainerd to minimize c8d RPCs

The containerd client is very chatty at the best of times. Because the
libcontained API is stateless and references containers and processes by
string ID for every method call, the implementation is essentially
forced to use the containerd client in a way which amplifies the number
of redundant RPCs invoked to perform any operation. The libcontainerd
remote implementation has to reload the containerd container, task
and/or process metadata for nearly every operation. This in turn
amplifies the number of context switches between dockerd and containerd
to perform any container operation or handle a containerd event,
increasing the load on the system which could otherwise be allocated to
workloads.

Overhaul the libcontainerd interface to reduce the impedance mismatch
with the containerd client so that the containerd client can be used
more efficiently. Split the API out into container, task and process
interfaces which the consumer is expected to retain so that
libcontainerd can retain state---especially the analogous containerd
client objects---without having to manage any state-store inside the
libcontainerd client.

Signed-off-by: Cory Snider <csnider@mirantis.com>

1 files changed, 103 insertions, 29 deletions

diff --git a/plugin/executor/containerd/containerd.go b/plugin/executor/containerd/containerd.go
index 92983056d0..2d3b99fe4c 100644
--- a/plugin/executor/containerd/containerd.go
+++ b/plugin/executor/containerd/containerd.go
@@ -2,6 +2,7 @@ package containerd // import "github.com/docker/docker/plugin/executor/container
 
 import (
 	"context"
+	"fmt"
 	"io"
 	"sync"
 	"syscall"
@@ -28,6 +29,7 @@ func New(ctx context.Context, rootDir string, cli *containerd.Client, ns string,
 		rootDir:     rootDir,
 		exitHandler: exitHandler,
 		runtime:     runtime,
+		plugins:     make(map[string]*c8dPlugin),
 	}
 
 	client, err := libcontainerd.NewClient(ctx, cli, rootDir, ns, e)
@@ -44,41 +46,62 @@ type Executor struct {
 	client      libcontainerdtypes.Client
 	exitHandler ExitHandler
 	runtime     types.Runtime
+
+	mu      sync.Mutex // Guards plugins map
+	plugins map[string]*c8dPlugin
+}
+
+type c8dPlugin struct {
+	log *logrus.Entry
+	ctr libcontainerdtypes.Container
+	tsk libcontainerdtypes.Task
 }
 
 // deleteTaskAndContainer deletes plugin task and then plugin container from containerd
-func deleteTaskAndContainer(ctx context.Context, cli libcontainerdtypes.Client, id string, p libcontainerdtypes.Process) {
-	if p != nil {
-		if _, _, err := p.Delete(ctx); err != nil && !errdefs.IsNotFound(err) {
-			logrus.WithError(err).WithField("id", id).Error("failed to delete plugin task from containerd")
-		}
-	} else {
-		if _, _, err := cli.DeleteTask(ctx, id); err != nil && !errdefs.IsNotFound(err) {
-			logrus.WithError(err).WithField("id", id).Error("failed to delete plugin task from containerd")
+func (p c8dPlugin) deleteTaskAndContainer(ctx context.Context) {
+	if p.tsk != nil {
+		if _, err := p.tsk.Delete(ctx); err != nil && !errdefs.IsNotFound(err) {
+			p.log.WithError(err).Error("failed to delete plugin task from containerd")
 		}
 	}
-
-	if err := cli.Delete(ctx, id); err != nil && !errdefs.IsNotFound(err) {
-		logrus.WithError(err).WithField("id", id).Error("failed to delete plugin container from containerd")
+	if p.ctr != nil {
+		if err := p.ctr.Delete(ctx); err != nil && !errdefs.IsNotFound(err) {
+			p.log.WithError(err).Error("failed to delete plugin container from containerd")
+		}
 	}
 }
 
 // Create creates a new container
 func (e *Executor) Create(id string, spec specs.Spec, stdout, stderr io.WriteCloser) error {
 	ctx := context.Background()
-	err := e.client.Create(ctx, id, &spec, e.runtime.Shim.Binary, e.runtime.Shim.Opts)
+	log := logrus.WithField("plugin", id)
+	ctr, err := e.client.NewContainer(ctx, id, &spec, e.runtime.Shim.Binary, e.runtime.Shim.Opts)
 	if err != nil {
-		status, err2 := e.client.Status(ctx, id)
+		ctr2, err2 := e.client.LoadContainer(ctx, id)
 		if err2 != nil {
 			if !errdefs.IsNotFound(err2) {
-				logrus.WithError(err2).WithField("id", id).Warn("Received an error while attempting to read plugin status")
+				log.WithError(err2).Warn("Received an error while attempting to load containerd container for plugin")
 			}
 		} else {
+			status := containerd.Unknown
+			t, err2 := ctr2.Task(ctx)
+			if err2 != nil {
+				if !errdefs.IsNotFound(err2) {
+					log.WithError(err2).Warn("Received an error while attempting to load containerd task for plugin")
+				}
+			} else {
+				s, err2 := t.Status(ctx)
+				if err2 != nil {
+					log.WithError(err2).Warn("Received an error while attempting to read plugin status")
+				} else {
+					status = s.Status
+				}
+			}
 			if status != containerd.Running && status != containerd.Unknown {
-				if err2 := e.client.Delete(ctx, id); err2 != nil && !errdefs.IsNotFound(err2) {
-					logrus.WithError(err2).WithField("plugin", id).Error("Error cleaning up containerd container")
+				if err2 := ctr2.Delete(ctx); err2 != nil && !errdefs.IsNotFound(err2) {
+					log.WithError(err2).Error("Error cleaning up containerd container")
 				}
-				err = e.client.Create(ctx, id, &spec, e.runtime.Shim.Binary, e.runtime.Shim.Opts)
+				ctr, err = e.client.NewContainer(ctx, id, &spec, e.runtime.Shim.Binary, e.runtime.Shim.Opts)
 			}
 		}
 
@@ -87,34 +110,78 @@ func (e *Executor) Create(id string, spec specs.Spec, stdout, stderr io.WriteClo
 		}
 	}
 
-	_, err = e.client.Start(ctx, id, "", false, attachStreamsFunc(stdout, stderr))
+	p := c8dPlugin{log: log, ctr: ctr}
+	p.tsk, err = ctr.Start(ctx, "", false, attachStreamsFunc(stdout, stderr))
 	if err != nil {
-		deleteTaskAndContainer(ctx, e.client, id, nil)
+		p.deleteTaskAndContainer(ctx)
+		return err
 	}
-	return err
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	e.plugins[id] = &p
+	return nil
 }
 
 // Restore restores a container
 func (e *Executor) Restore(id string, stdout, stderr io.WriteCloser) (bool, error) {
-	alive, _, p, err := e.client.Restore(context.Background(), id, attachStreamsFunc(stdout, stderr))
-	if err != nil && !errdefs.IsNotFound(err) {
+	ctx := context.Background()
+	p := c8dPlugin{log: logrus.WithField("plugin", id)}
+	ctr, err := e.client.LoadContainer(ctx, id)
+	if err != nil {
+		if errdefs.IsNotFound(err) {
+			return false, nil
+		}
 		return false, err
 	}
-	if !alive {
-		deleteTaskAndContainer(context.Background(), e.client, id, p)
+	p.tsk, err = ctr.AttachTask(ctx, attachStreamsFunc(stdout, stderr))
+	if err != nil {
+		if errdefs.IsNotFound(err) {
+			p.deleteTaskAndContainer(ctx)
+			return false, nil
+		}
+		return false, err
+	}
+	s, err := p.tsk.Status(ctx)
+	if err != nil {
+		if errdefs.IsNotFound(err) {
+			// Task vanished after attaching?
+			p.tsk = nil
+			p.deleteTaskAndContainer(ctx)
+			return false, nil
+		}
+		return false, err
+	}
+	if s.Status == containerd.Stopped {
+		p.deleteTaskAndContainer(ctx)
+		return false, nil
 	}
-	return alive, nil
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	e.plugins[id] = &p
+	return true, nil
 }
 
 // IsRunning returns if the container with the given id is running
 func (e *Executor) IsRunning(id string) (bool, error) {
-	status, err := e.client.Status(context.Background(), id)
-	return status == containerd.Running, err
+	e.mu.Lock()
+	p := e.plugins[id]
+	e.mu.Unlock()
+	if p == nil {
+		return false, errdefs.NotFound(fmt.Errorf("unknown plugin %q", id))
+	}
+	status, err := p.tsk.Status(context.Background())
+	return status.Status == containerd.Running, err
 }
 
 // Signal sends the specified signal to the container
 func (e *Executor) Signal(id string, signal syscall.Signal) error {
-	return e.client.SignalProcess(context.Background(), id, libcontainerdtypes.InitProcessName, signal)
+	e.mu.Lock()
+	p := e.plugins[id]
+	e.mu.Unlock()
+	if p == nil {
+		return errdefs.NotFound(fmt.Errorf("unknown plugin %q", id))
+	}
+	return p.tsk.Kill(context.Background(), signal)
 }
 
 // ProcessEvent handles events from containerd
@@ -122,7 +189,14 @@ func (e *Executor) Signal(id string, signal syscall.Signal) error {
 func (e *Executor) ProcessEvent(id string, et libcontainerdtypes.EventType, ei libcontainerdtypes.EventInfo) error {
 	switch et {
 	case libcontainerdtypes.EventExit:
-		deleteTaskAndContainer(context.Background(), e.client, id, nil)
+		e.mu.Lock()
+		p := e.plugins[id]
+		e.mu.Unlock()
+		if p == nil {
+			logrus.WithField("id", id).Warn("Received exit event for an unknown plugin")
+		} else {
+			p.deleteTaskAndContainer(context.Background())
+		}
 		return e.exitHandler.HandleExitEvent(ei.ContainerID)
 	}
 	return nil