Share logic to create-or-replace a container

The existing logic to handle container ID conflicts when attempting to
create a plugin container is not nearly as robust as the implementation
in daemon for user containers. Extract and refine the logic from daemon
and use it in the plugin executor.

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

daemon/start.go

@@ -9,6 +9,7 @@ import (
 	containertypes "github.com/docker/docker/api/types/container"
 	"github.com/docker/docker/container"
 	"github.com/docker/docker/errdefs"
+	"github.com/docker/docker/libcontainerd"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 )
@@ -178,16 +179,9 @@ func (daemon *Daemon) containerStart(container *container.Container, checkpoint
 
 	ctx := context.TODO()
 
-	ctr, err := daemon.containerd.NewContainer(ctx, container.ID, spec, shim, createOptions)
+	ctr, err := libcontainerd.ReplaceContainer(ctx, daemon.containerd, container.ID, spec, shim, createOptions)
 	if err != nil {
-		if errdefs.IsConflict(err) {
-			logrus.WithError(err).WithField("container", container.ID).Error("Container not cleaned up from containerd from previous run")
-			daemon.cleanupStaleContainer(ctx, container.ID)
-			ctr, err = daemon.containerd.NewContainer(ctx, container.ID, spec, shim, createOptions)
-		}
-		if err != nil {
-			return translateContainerdStartErr(container.Path, container.SetExitCode, err)
-		}
+		return translateContainerdStartErr(container.Path, container.SetExitCode, err)
 	}
 
 	// TODO(mlaventure): we need to specify checkpoint options here
@@ -220,31 +214,6 @@ func (daemon *Daemon) containerStart(container *container.Container, checkpoint
 	return nil
 }
 
-func (daemon *Daemon) cleanupStaleContainer(ctx context.Context, id string) {
-	// best effort to clean up old container object
-	log := logrus.WithContext(ctx).WithField("container", id)
-	ctr, err := daemon.containerd.LoadContainer(ctx, id)
-	if err != nil {
-		// Log an error no matter the kind. A container existed with the
-		// ID, so a NotFound error would be an exceptional situation
-		// worth logging.
-		log.WithError(err).Error("Error loading stale containerd container object")
-		return
-	}
-	if tsk, err := ctr.Task(ctx); err != nil {
-		if !errdefs.IsNotFound(err) {
-			log.WithError(err).Error("Error loading stale containerd task object")
-		}
-	} else {
-		if err := tsk.ForceDelete(ctx); err != nil {
-			log.WithError(err).Error("Error cleaning up stale containerd task object")
-		}
-	}
-	if err := ctr.Delete(ctx); err != nil && !errdefs.IsNotFound(err) {
-		log.WithError(err).Error("Error cleaning up stale containerd container object")
-	}
-}
-
 // Cleanup releases any network resources allocated to the container along with any rules
 // around how containers are linked together.  It also unmounts the container's root filesystem.
 func (daemon *Daemon) Cleanup(container *container.Container) {

libcontainerd/replace.go

@@ -0,0 +1,62 @@
+package libcontainerd // import "github.com/docker/docker/libcontainerd"
+
+import (
+	"context"
+
+	"github.com/containerd/containerd"
+	"github.com/opencontainers/runtime-spec/specs-go"
+	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
+
+	"github.com/docker/docker/errdefs"
+	"github.com/docker/docker/libcontainerd/types"
+)
+
+// ReplaceContainer creates a new container, replacing any existing container
+// with the same id if necessary.
+func ReplaceContainer(ctx context.Context, client types.Client, id string, spec *specs.Spec, shim string, runtimeOptions interface{}, opts ...containerd.NewContainerOpts) (types.Container, error) {
+	newContainer := func() (types.Container, error) {
+		return client.NewContainer(ctx, id, spec, shim, runtimeOptions, opts...)
+	}
+	ctr, err := newContainer()
+	if err == nil || !errdefs.IsConflict(err) {
+		return ctr, err
+	}
+
+	log := logrus.WithContext(ctx).WithField("container", id)
+	log.Debug("A container already exists with the same ID. Attempting to clean up the old container.")
+	ctr, err = client.LoadContainer(ctx, id)
+	if err != nil {
+		if errdefs.IsNotFound(err) {
+			// Task failed successfully: the container no longer exists,
+			// despite us not doing anything. May as well try to create
+			// the container again. It might succeed.
+			return newContainer()
+		}
+		return nil, errors.Wrap(err, "could not load stale containerd container object")
+	}
+	tsk, err := ctr.Task(ctx)
+	if err != nil {
+		if errdefs.IsNotFound(err) {
+			goto deleteContainer
+		}
+		// There is no point in trying to delete the container if we
+		// cannot determine whether or not it has a task. The containerd
+		// client would just try to load the task itself, get the same
+		// error, and give up.
+		return nil, errors.Wrap(err, "could not load stale containerd task object")
+	}
+	if err := tsk.ForceDelete(ctx); err != nil {
+		if !errdefs.IsNotFound(err) {
+			return nil, errors.Wrap(err, "could not delete stale containerd task object")
+		}
+		// The task might have exited on its own. Proceed with
+		// attempting to delete the container.
+	}
+deleteContainer:
+	if err := ctr.Delete(ctx); err != nil && !errdefs.IsNotFound(err) {
+		return nil, errors.Wrap(err, "could not delete stale containerd container object")
+	}
+
+	return newContainer()
+}

plugin/executor/containerd/containerd.go

@@ -75,39 +75,9 @@ func (p c8dPlugin) deleteTaskAndContainer(ctx context.Context) {
 func (e *Executor) Create(id string, spec specs.Spec, stdout, stderr io.WriteCloser) error {
 	ctx := context.Background()
 	log := logrus.WithField("plugin", id)
-	ctr, err := e.client.NewContainer(ctx, id, &spec, e.runtime.Shim.Binary, e.runtime.Shim.Opts)
+	ctr, err := libcontainerd.ReplaceContainer(ctx, e.client, id, &spec, e.runtime.Shim.Binary, e.runtime.Shim.Opts)
 	if err != nil {
-		ctr2, err2 := e.client.LoadContainer(ctx, id)
-		if err2 != nil {
-			if !errdefs.IsNotFound(err2) {
-				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 := ctr2.Delete(ctx); err2 != nil && !errdefs.IsNotFound(err2) {
-					log.WithError(err2).Error("Error cleaning up containerd container")
-				}
-				ctr, err = e.client.NewContainer(ctx, id, &spec, e.runtime.Shim.Binary, e.runtime.Shim.Opts)
-			}
-		}
-
-		if err != nil {
-			return errors.Wrap(err, "error creating containerd container")
-		}
+		return errors.Wrap(err, "error creating containerd container for plugin")
 	}
 
 	p := c8dPlugin{log: log, ctr: ctr}