Merge component 'engine' from git@github.com:moby/moby master

components/engine/Dockerfile

@@ -233,6 +233,7 @@ COPY --from=docker-py /docker-py /docker-py
 # above.
 RUN cd /docker-py \
 	&& pip install docker-pycreds==0.2.1 \
+	&& pip install yamllint==1.5.0 \
 	&& pip install -r test-requirements.txt
 
 ENV PATH=/usr/local/cli:$PATH

components/engine/api/server/router/system/system_routes.go

@@ -73,7 +73,7 @@ func (s *systemRouter) getDiskUsage(ctx context.Context, w http.ResponseWriter,
 	if err != nil {
 		return err
 	}
-	builderSize, err := s.builder.DiskUsage()
+	builderSize, err := s.builder.DiskUsage(ctx)
 	if err != nil {
 		return pkgerrors.Wrap(err, "error getting build cache usage")
 	}

components/engine/builder/fscache/fscache.go

@@ -154,8 +154,8 @@ func (fsc *FSCache) SyncFrom(ctx context.Context, id RemoteIdentifier) (builder.
 }
 
 // DiskUsage reports how much data is allocated by the cache
-func (fsc *FSCache) DiskUsage() (int64, error) {
-	return fsc.store.DiskUsage()
+func (fsc *FSCache) DiskUsage(ctx context.Context) (int64, error) {
+	return fsc.store.DiskUsage(ctx)
 }
 
 // Prune allows manually cleaning up the cache
@@ -382,14 +382,14 @@ func (s *fsCacheStore) Get(id string) (*cachedSourceRef, error) {
 }
 
 // DiskUsage reports how much data is allocated by the cache
-func (s *fsCacheStore) DiskUsage() (int64, error) {
+func (s *fsCacheStore) DiskUsage(ctx context.Context) (int64, error) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	var size int64
 
 	for _, snap := range s.sources {
 		if len(snap.refs) == 0 {
-			ss, err := snap.getSize()
+			ss, err := snap.getSize(ctx)
 			if err != nil {
 				return 0, err
 			}
@@ -414,7 +414,7 @@ func (s *fsCacheStore) Prune(ctx context.Context) (uint64, error) {
 		default:
 		}
 		if len(snap.refs) == 0 {
-			ss, err := snap.getSize()
+			ss, err := snap.getSize(ctx)
 			if err != nil {
 				return size, err
 			}
@@ -433,6 +433,7 @@ func (s *fsCacheStore) GC() error {
 	defer s.mu.Unlock()
 	var size uint64
 
+	ctx := context.Background()
 	cutoff := time.Now().Add(-s.gcPolicy.MaxKeepDuration)
 	var blacklist []*cachedSource
 
@@ -443,7 +444,7 @@ func (s *fsCacheStore) GC() error {
 					return errors.Wrapf(err, "failed to delete %s", id)
 				}
 			} else {
-				ss, err := snap.getSize()
+				ss, err := snap.getSize(ctx)
 				if err != nil {
 					return err
 				}
@@ -458,7 +459,7 @@ func (s *fsCacheStore) GC() error {
 		if size <= s.gcPolicy.MaxSize {
 			break
 		}
-		ss, err := snap.getSize()
+		ss, err := snap.getSize(ctx)
 		if err != nil {
 			return err
 		}
@@ -521,9 +522,9 @@ func (cs *cachedSource) getRef() *cachedSourceRef {
 }
 
 // hold storage lock before calling
-func (cs *cachedSource) getSize() (int64, error) {
+func (cs *cachedSource) getSize(ctx context.Context) (int64, error) {
 	if cs.sourceMeta.Size < 0 {
-		ss, err := directory.Size(cs.dir)
+		ss, err := directory.Size(ctx, cs.dir)
 		if err != nil {
 			return 0, err
 		}

components/engine/builder/fscache/fscache_test.go

@@ -59,13 +59,13 @@ func TestFSCache(t *testing.T) {
 	assert.Check(t, err)
 	assert.Check(t, is.Equal(string(dt), "data2"))
 
-	s, err := fscache.DiskUsage()
+	s, err := fscache.DiskUsage(context.TODO())
 	assert.Check(t, err)
 	assert.Check(t, is.Equal(s, int64(0)))
 
 	assert.Check(t, src3.Close())
 
-	s, err = fscache.DiskUsage()
+	s, err = fscache.DiskUsage(context.TODO())
 	assert.Check(t, err)
 	assert.Check(t, is.Equal(s, int64(5)))
 
@@ -80,7 +80,7 @@ func TestFSCache(t *testing.T) {
 	assert.Check(t, is.Equal(src4.Root().Path(), src3.Root().Path()))
 	assert.Check(t, src4.Close())
 
-	s, err = fscache.DiskUsage()
+	s, err = fscache.DiskUsage(context.TODO())
 	assert.Check(t, err)
 	assert.Check(t, is.Equal(s, int64(10)))
 
@@ -93,7 +93,7 @@ func TestFSCache(t *testing.T) {
 	time.Sleep(100 * time.Millisecond)
 
 	// only last insertion after GC
-	s, err = fscache.DiskUsage()
+	s, err = fscache.DiskUsage(context.TODO())
 	assert.Check(t, err)
 	assert.Check(t, is.Equal(s, int64(8)))
 
@@ -102,7 +102,7 @@ func TestFSCache(t *testing.T) {
 	assert.Check(t, err)
 	assert.Check(t, is.Equal(released, uint64(8)))
 
-	s, err = fscache.DiskUsage()
+	s, err = fscache.DiskUsage(context.TODO())
 	assert.Check(t, err)
 	assert.Check(t, is.Equal(s, int64(0)))
 }

components/engine/cmd/dockerd/daemon.go

@@ -257,6 +257,8 @@ func (cli *DaemonCli) start(opts *daemonOptions) (err error) {
 		PluginBackend:          d.PluginManager(),
 		NetworkSubnetsProvider: d,
 		DefaultAdvertiseAddr:   cli.Config.SwarmDefaultAdvertiseAddr,
+		RaftHeartbeatTick:      cli.Config.SwarmRaftHeartbeatTick,
+		RaftElectionTick:       cli.Config.SwarmRaftElectionTick,
 		RuntimeRoot:            cli.getSwarmRunRoot(),
 		WatchStream:            watchStream,
 	})

components/engine/daemon/cluster/cluster.go

@@ -96,6 +96,13 @@ type Config struct {
 
 	// WatchStream is a channel to pass watch API notifications to daemon
 	WatchStream chan *swarmapi.WatchMessage
+
+	// RaftHeartbeatTick is the number of ticks for heartbeat of quorum members
+	RaftHeartbeatTick uint32
+
+	// RaftElectionTick is the number of ticks to elapse before followers propose a new round of leader election
+	// This value should be 10x that of RaftHeartbeatTick
+	RaftElectionTick uint32
 }
 
 // Cluster provides capabilities to participate in a cluster as a worker or a
@@ -134,6 +141,14 @@ func New(config Config) (*Cluster, error) {
 	if config.RuntimeRoot == "" {
 		config.RuntimeRoot = root
 	}
+	if config.RaftHeartbeatTick == 0 {
+		config.RaftHeartbeatTick = 1
+	}
+	if config.RaftElectionTick == 0 {
+		// 10X heartbeat tick is the recommended ratio according to etcd docs.
+		config.RaftElectionTick = 10 * config.RaftHeartbeatTick
+	}
+
 	if err := os.MkdirAll(config.RuntimeRoot, 0700); err != nil {
 		return nil, err
 	}

components/engine/daemon/cluster/noderunner.go

@@ -124,11 +124,11 @@ func (n *nodeRunner) start(conf nodeStartConfig) error {
 			n.cluster.config.Backend,
 			n.cluster.config.PluginBackend,
 			n.cluster.config.ImageBackend),
-		HeartbeatTick: 1,
+		HeartbeatTick: n.cluster.config.RaftHeartbeatTick,
 		// Recommended value in etcd/raft is 10 x (HeartbeatTick).
 		// Lower values were seen to have caused instability because of
 		// frequent leader elections when running on flakey networks.
-		ElectionTick:     10,
+		ElectionTick:     n.cluster.config.RaftElectionTick,
 		UnlockKey:        conf.lockKey,
 		AutoLockManagers: conf.autolock,
 		PluginGetter:     n.cluster.config.Backend.PluginGetter(),

components/engine/daemon/config/config.go

@@ -158,7 +158,18 @@ type CommonConfig struct {
 	// given to the /swarm/init endpoint and no advertise address is
 	// specified.
 	SwarmDefaultAdvertiseAddr string `json:"swarm-default-advertise-addr"`
-	MetricsAddress            string `json:"metrics-addr"`
+
+	// SwarmRaftHeartbeatTick is the number of ticks in time for swarm mode raft quorum heartbeat
+	// Typical value is 1
+	SwarmRaftHeartbeatTick uint32 `json:"swarm-raft-heartbeat-tick"`
+
+	// SwarmRaftElectionTick is the number of ticks to elapse before followers in the quorum can propose
+	// a new round of leader election.  Default, recommended value is at least 10X that of Heartbeat tick.
+	// Higher values can make the quorum less sensitive to transient faults in the environment, but this also
+	// means it takes longer for the managers to detect a down leader.
+	SwarmRaftElectionTick uint32 `json:"swarm-raft-election-tick"`
+
+	MetricsAddress string `json:"metrics-addr"`
 
 	LogConfig
 	BridgeConfig // bridgeConfig holds bridge network specific configuration.

components/engine/daemon/disk_usage.go

@@ -53,7 +53,7 @@ func (daemon *Daemon) SystemDiskUsage(ctx context.Context) (*types.DiskUsage, er
 			refs := daemon.volumes.Refs(v)
 
 			tv := volumeToAPIType(v)
-			sz, err := directory.Size(v.Path())
+			sz, err := directory.Size(ctx, v.Path())
 			if err != nil {
 				logrus.Warnf("failed to determine size of volume %v", name)
 				sz = -1

components/engine/daemon/graphdriver/aufs/aufs.go

@@ -24,6 +24,7 @@ package aufs // import "github.com/docker/docker/daemon/graphdriver/aufs"
 
 import (
 	"bufio"
+	"context"
 	"fmt"
 	"io"
 	"io/ioutil"
@@ -502,7 +503,7 @@ func (a *Driver) DiffSize(id, parent string) (size int64, err error) {
 		return a.naiveDiff.DiffSize(id, parent)
 	}
 	// AUFS doesn't need the parent layer to calculate the diff size.
-	return directory.Size(path.Join(a.rootPath(), "diff", id))
+	return directory.Size(context.TODO(), path.Join(a.rootPath(), "diff", id))
 }
 
 // ApplyDiff extracts the changeset from the given diff into the

components/engine/daemon/graphdriver/copy/copy_test.go

@@ -12,32 +12,12 @@ import (
 	"testing"
 	"time"
 
-	"github.com/docker/docker/pkg/parsers/kernel"
 	"github.com/docker/docker/pkg/system"
 	"github.com/gotestyourself/gotestyourself/assert"
 	is "github.com/gotestyourself/gotestyourself/assert/cmp"
 	"golang.org/x/sys/unix"
 )
 
-func TestIsCopyFileRangeSyscallAvailable(t *testing.T) {
-	// Verifies:
-	// 1. That copyFileRangeEnabled is being set to true when copy_file_range syscall is available
-	// 2. That isCopyFileRangeSyscallAvailable() works on "new" kernels
-	v, err := kernel.GetKernelVersion()
-	assert.NilError(t, err)
-
-	copyWithFileRange := true
-	copyWithFileClone := false
-	doCopyTest(t, &copyWithFileRange, &copyWithFileClone)
-
-	if kernel.CompareKernelVersion(*v, kernel.VersionInfo{Kernel: 4, Major: 5, Minor: 0}) < 0 {
-		assert.Check(t, !copyWithFileRange)
-	} else {
-		assert.Check(t, copyWithFileRange)
-	}
-
-}
-
 func TestCopy(t *testing.T) {
 	copyWithFileRange := true
 	copyWithFileClone := true

components/engine/daemon/graphdriver/overlay2/overlay.go

@@ -4,6 +4,7 @@ package overlay2 // import "github.com/docker/docker/daemon/graphdriver/overlay2
 
 import (
 	"bufio"
+	"context"
 	"errors"
 	"fmt"
 	"io"
@@ -706,7 +707,7 @@ func (d *Driver) ApplyDiff(id string, parent string, diff io.Reader) (size int64
 		return 0, err
 	}
 
-	return directory.Size(applyDir)
+	return directory.Size(context.TODO(), applyDir)
 }
 
 func (d *Driver) getDiffPath(id string) string {
@@ -722,7 +723,7 @@ func (d *Driver) DiffSize(id, parent string) (size int64, err error) {
 	if useNaiveDiff(d.home) || !d.isParent(id, parent) {
 		return d.naiveDiff.DiffSize(id, parent)
 	}
-	return directory.Size(d.getDiffPath(id))
+	return directory.Size(context.TODO(), d.getDiffPath(id))
 }
 
 // Diff produces an archive of the changes between the specified

components/engine/daemon/prune.go

@@ -125,7 +125,7 @@ func (daemon *Daemon) VolumesPrune(ctx context.Context, pruneFilters filters.Arg
 					return nil
 				}
 			}
-			vSize, err := directory.Size(v.Path())
+			vSize, err := directory.Size(ctx, v.Path())
 			if err != nil {
 				logrus.Warnf("could not determine size of volume %s: %v", name, err)
 			}

components/engine/libcontainerd/client_daemon.go

@@ -205,7 +205,7 @@ func (c *client) Create(ctx context.Context, id string, ociSpec *specs.Spec, run
 		// TODO(mlaventure): when containerd support lcow, revisit runtime value
 		containerd.WithRuntime(fmt.Sprintf("io.containerd.runtime.v1.%s", runtime.GOOS), runtimeOptions))
 	if err != nil {
-		return err
+		return wrapError(err)
 	}
 
 	c.Lock()
@@ -286,7 +286,7 @@ func (c *client) Start(ctx context.Context, id, checkpointDir string, withStdin
 			rio.Cancel()
 			rio.Close()
 		}
-		return -1, err
+		return -1, wrapError(err)
 	}
 
 	ctr.setTask(t)
@@ -300,7 +300,7 @@ func (c *client) Start(ctx context.Context, id, checkpointDir string, withStdin
 				Error("failed to delete task after fail start")
 		}
 		ctr.setTask(nil)
-		return -1, err
+		return -1, wrapError(err)
 	}
 
 	return int(t.Pid()), nil
@@ -344,7 +344,7 @@ func (c *client) Exec(ctx context.Context, containerID, processID string, spec *
 	})
 	if err != nil {
 		close(stdinCloseSync)
-		return -1, err
+		return -1, wrapError(err)
 	}
 
 	ctr.addProcess(processID, p)
@@ -355,7 +355,7 @@ func (c *client) Exec(ctx context.Context, containerID, processID string, spec *
 	if err = p.Start(ctx); err != nil {
 		p.Delete(context.Background())
 		ctr.deleteProcess(processID)
-		return -1, err
+		return -1, wrapError(err)
 	}
 
 	return int(p.Pid()), nil
@@ -393,7 +393,7 @@ func (c *client) Pause(ctx context.Context, containerID string) error {
 		return err
 	}
 
-	return p.(containerd.Task).Pause(ctx)
+	return wrapError(p.(containerd.Task).Pause(ctx))
 }
 
 func (c *client) Resume(ctx context.Context, containerID string) error {
@@ -493,7 +493,7 @@ func (c *client) Delete(ctx context.Context, containerID string) error {
 	}
 
 	if err := ctr.ctr.Delete(ctx); err != nil {
-		return err
+		return wrapError(err)
 	}
 
 	if os.Getenv("LIBCONTAINERD_NOCLEAN") != "1" {
@@ -523,7 +523,7 @@ func (c *client) Status(ctx context.Context, containerID string) (Status, error)
 
 	s, err := t.Status(ctx)
 	if err != nil {
-		return StatusUnknown, err
+		return StatusUnknown, wrapError(err)
 	}
 
 	return Status(s.Status), nil
@@ -537,7 +537,7 @@ func (c *client) CreateCheckpoint(ctx context.Context, containerID, checkpointDi
 
 	img, err := p.(containerd.Task).Checkpoint(ctx)
 	if err != nil {
-		return err
+		return wrapError(err)
 	}
 	// Whatever happens, delete the checkpoint from containerd
 	defer func() {

components/engine/pkg/directory/directory_test.go

@@ -1,6 +1,7 @@
 package directory // import "github.com/docker/docker/pkg/directory"
 
 import (
+	"context"
 	"io/ioutil"
 	"os"
 	"path/filepath"
@@ -18,7 +19,7 @@ func TestSizeEmpty(t *testing.T) {
 	}
 
 	var size int64
-	if size, _ = Size(dir); size != 0 {
+	if size, _ = Size(context.Background(), dir); size != 0 {
 		t.Fatalf("empty directory has size: %d", size)
 	}
 }
@@ -37,7 +38,7 @@ func TestSizeEmptyFile(t *testing.T) {
 	}
 
 	var size int64
-	if size, _ = Size(file.Name()); size != 0 {
+	if size, _ = Size(context.Background(), file.Name()); size != 0 {
 		t.Fatalf("directory with one file has size: %d", size)
 	}
 }
@@ -59,7 +60,7 @@ func TestSizeNonemptyFile(t *testing.T) {
 	file.Write(d)
 
 	var size int64
-	if size, _ = Size(file.Name()); size != 5 {
+	if size, _ = Size(context.Background(), file.Name()); size != 5 {
 		t.Fatalf("directory with one 5-byte file has size: %d", size)
 	}
 }
@@ -76,7 +77,7 @@ func TestSizeNestedDirectoryEmpty(t *testing.T) {
 	}
 
 	var size int64
-	if size, _ = Size(dir); size != 0 {
+	if size, _ = Size(context.Background(), dir); size != 0 {
 		t.Fatalf("directory with one empty directory has size: %d", size)
 	}
 }
@@ -101,7 +102,7 @@ func TestSizeFileAndNestedDirectoryEmpty(t *testing.T) {
 	file.Write(d)
 
 	var size int64
-	if size, _ = Size(dir); size != 6 {
+	if size, _ = Size(context.Background(), dir); size != 6 {
 		t.Fatalf("directory with 6-byte file and empty directory has size: %d", size)
 	}
 }
@@ -134,7 +135,7 @@ func TestSizeFileAndNestedDirectoryNonempty(t *testing.T) {
 	nestedFile.Write(nestedData)
 
 	var size int64
-	if size, _ = Size(dir); size != 12 {
+	if size, _ = Size(context.Background(), dir); size != 12 {
 		t.Fatalf("directory with 6-byte file and nested directory with 6-byte file has size: %d", size)
 	}
 }
@@ -186,7 +187,7 @@ func TestMoveToSubdir(t *testing.T) {
 
 // Test a non-existing directory
 func TestSizeNonExistingDirectory(t *testing.T) {
-	if _, err := Size("/thisdirectoryshouldnotexist/TestSizeNonExistingDirectory"); err == nil {
+	if _, err := Size(context.Background(), "/thisdirectoryshouldnotexist/TestSizeNonExistingDirectory"); err == nil {
 		t.Fatalf("error is expected")
 	}
 }

components/engine/pkg/directory/directory_unix.go

@@ -3,13 +3,14 @@
 package directory // import "github.com/docker/docker/pkg/directory"
 
 import (
+	"context"
 	"os"
 	"path/filepath"
 	"syscall"
 )
 
 // Size walks a directory tree and returns its total size in bytes.
-func Size(dir string) (size int64, err error) {
+func Size(ctx context.Context, dir string) (size int64, err error) {
 	data := make(map[uint64]struct{})
 	err = filepath.Walk(dir, func(d string, fileInfo os.FileInfo, err error) error {
 		if err != nil {
@@ -20,6 +21,11 @@ func Size(dir string) (size int64, err error) {
 			}
 			return err
 		}
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		default:
+		}
 
 		// Ignore directory sizes
 		if fileInfo == nil {

components/engine/pkg/directory/directory_windows.go

@@ -1,12 +1,13 @@
 package directory // import "github.com/docker/docker/pkg/directory"
 
 import (
+	"context"
 	"os"
 	"path/filepath"
 )
 
 // Size walks a directory tree and returns its total size in bytes.
-func Size(dir string) (size int64, err error) {
+func Size(ctx context.Context, dir string) (size int64, err error) {
 	err = filepath.Walk(dir, func(d string, fileInfo os.FileInfo, err error) error {
 		if err != nil {
 			// if dir does not exist, Size() returns the error.
@@ -17,6 +18,12 @@ func Size(dir string) (size int64, err error) {
 			return err
 		}
 
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		default:
+		}
+
 		// Ignore directory sizes
 		if fileInfo == nil {
 			return nil

components/engine/plugin/executor/containerd/containerd.go

@@ -5,6 +5,7 @@ import (
 	"io"
 	"path/filepath"
 	"sync"
+	"time"
 
 	"github.com/containerd/containerd/cio"
 	"github.com/containerd/containerd/linux/runctypes"
@@ -15,21 +16,34 @@ import (
 	"github.com/sirupsen/logrus"
 )
 
-// PluginNamespace is the name used for the plugins namespace
-var PluginNamespace = "plugins.moby"
+// pluginNamespace is the name used for the plugins namespace
+const pluginNamespace = "plugins.moby"
 
 // ExitHandler represents an object that is called when the exit event is received from containerd
 type ExitHandler interface {
 	HandleExitEvent(id string) error
 }
 
+// Client is used by the exector to perform operations.
+// TODO(@cpuguy83): This should really just be based off the containerd client interface.
+// However right now this whole package is tied to github.com/docker/docker/libcontainerd
+type Client interface {
+	Create(ctx context.Context, containerID string, spec *specs.Spec, runtimeOptions interface{}) error
+	Restore(ctx context.Context, containerID string, attachStdio libcontainerd.StdioCallback) (alive bool, pid int, err error)
+	Status(ctx context.Context, containerID string) (libcontainerd.Status, error)
+	Delete(ctx context.Context, containerID string) error
+	DeleteTask(ctx context.Context, containerID string) (uint32, time.Time, error)
+	Start(ctx context.Context, containerID, checkpointDir string, withStdin bool, attachStdio libcontainerd.StdioCallback) (pid int, err error)
+	SignalProcess(ctx context.Context, containerID, processID string, signal int) error
+}
+
 // New creates a new containerd plugin executor
 func New(rootDir string, remote libcontainerd.Remote, exitHandler ExitHandler) (*Executor, error) {
 	e := &Executor{
 		rootDir:     rootDir,
 		exitHandler: exitHandler,
 	}
-	client, err := remote.NewClient(PluginNamespace, e)
+	client, err := remote.NewClient(pluginNamespace, e)
 	if err != nil {
 		return nil, errors.Wrap(err, "error creating containerd exec client")
 	}
@@ -40,7 +54,7 @@ func New(rootDir string, remote libcontainerd.Remote, exitHandler ExitHandler) (
 // Executor is the containerd client implementation of a plugin executor
 type Executor struct {
 	rootDir     string
-	client      libcontainerd.Client
+	client      Client
 	exitHandler ExitHandler
 }
 
@@ -52,10 +66,34 @@ func (e *Executor) Create(id string, spec specs.Spec, stdout, stderr io.WriteClo
 	ctx := context.Background()
 	err := e.client.Create(ctx, id, &spec, &opts)
 	if err != nil {
-		return err
+		status, err2 := e.client.Status(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")
+			}
+		} else {
+			if status != libcontainerd.StatusRunning && status != libcontainerd.StatusUnknown {
+				if err2 := e.client.Delete(ctx, id); err2 != nil && !errdefs.IsNotFound(err2) {
+					logrus.WithError(err2).WithField("plugin", id).Error("Error cleaning up containerd container")
+				}
+				err = e.client.Create(ctx, id, &spec, &opts)
+			}
+		}
+
+		if err != nil {
+			return errors.Wrap(err, "error creating containerd container")
+		}
 	}
 
 	_, err = e.client.Start(ctx, id, "", false, attachStreamsFunc(stdout, stderr))
+	if err != nil {
+		if _, _, err2 := e.client.DeleteTask(ctx, id); err2 != nil && !errdefs.IsNotFound(err2) {
+			logrus.WithError(err2).WithField("id", id).Warn("Received an error while attempting to clean up containerd plugin task after failed start")
+		}
+		if err2 := e.client.Delete(ctx, id); err2 != nil && !errdefs.IsNotFound(err2) {
+			logrus.WithError(err2).WithField("id", id).Warn("Received an error while attempting to clean up containerd plugin container after failed start")
+		}
+	}
 	return err
 }
 
@@ -69,13 +107,11 @@ func (e *Executor) Restore(id string, stdout, stderr io.WriteCloser) error {
 		_, _, 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

components/engine/plugin/executor/containerd/containerd_test.go

@@ -0,0 +1,148 @@
+package containerd
+
+import (
+	"context"
+	"io/ioutil"
+	"os"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/docker/docker/libcontainerd"
+	"github.com/gotestyourself/gotestyourself/assert"
+	specs "github.com/opencontainers/runtime-spec/specs-go"
+	"github.com/pkg/errors"
+)
+
+func TestLifeCycle(t *testing.T) {
+	t.Parallel()
+
+	mock := newMockClient()
+	exec, cleanup := setupTest(t, mock, mock)
+	defer cleanup()
+
+	id := "test-create"
+	mock.simulateStartError(true, id)
+	err := exec.Create(id, specs.Spec{}, nil, nil)
+	assert.Assert(t, err != nil)
+	mock.simulateStartError(false, id)
+
+	err = exec.Create(id, specs.Spec{}, nil, nil)
+	assert.Assert(t, err)
+	running, _ := exec.IsRunning(id)
+	assert.Assert(t, running)
+
+	// create with the same ID
+	err = exec.Create(id, specs.Spec{}, nil, nil)
+	assert.Assert(t, err != nil)
+
+	mock.HandleExitEvent(id) // simulate a plugin that exits
+
+	err = exec.Create(id, specs.Spec{}, nil, nil)
+	assert.Assert(t, err)
+}
+
+func setupTest(t *testing.T, client Client, eh ExitHandler) (*Executor, func()) {
+	rootDir, err := ioutil.TempDir("", "test-daemon")
+	assert.Assert(t, err)
+	assert.Assert(t, client != nil)
+	assert.Assert(t, eh != nil)
+
+	return &Executor{
+			rootDir:     rootDir,
+			client:      client,
+			exitHandler: eh,
+		}, func() {
+			assert.Assert(t, os.RemoveAll(rootDir))
+		}
+}
+
+type mockClient struct {
+	mu           sync.Mutex
+	containers   map[string]bool
+	errorOnStart map[string]bool
+}
+
+func newMockClient() *mockClient {
+	return &mockClient{
+		containers:   make(map[string]bool),
+		errorOnStart: make(map[string]bool),
+	}
+}
+
+func (c *mockClient) Create(ctx context.Context, id string, _ *specs.Spec, _ interface{}) error {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	if _, ok := c.containers[id]; ok {
+		return errors.New("exists")
+	}
+
+	c.containers[id] = false
+	return nil
+}
+
+func (c *mockClient) Restore(ctx context.Context, id string, attachStdio libcontainerd.StdioCallback) (alive bool, pid int, err error) {
+	return false, 0, nil
+}
+
+func (c *mockClient) Status(ctx context.Context, id string) (libcontainerd.Status, error) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	running, ok := c.containers[id]
+	if !ok {
+		return libcontainerd.StatusUnknown, errors.New("not found")
+	}
+	if running {
+		return libcontainerd.StatusRunning, nil
+	}
+	return libcontainerd.StatusStopped, nil
+}
+
+func (c *mockClient) Delete(ctx context.Context, id string) error {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	delete(c.containers, id)
+	return nil
+}
+
+func (c *mockClient) DeleteTask(ctx context.Context, id string) (uint32, time.Time, error) {
+	return 0, time.Time{}, nil
+}
+
+func (c *mockClient) Start(ctx context.Context, id, checkpointDir string, withStdin bool, attachStdio libcontainerd.StdioCallback) (pid int, err error) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	if _, ok := c.containers[id]; !ok {
+		return 0, errors.New("not found")
+	}
+
+	if c.errorOnStart[id] {
+		return 0, errors.New("some startup error")
+	}
+	c.containers[id] = true
+	return 1, nil
+}
+
+func (c *mockClient) SignalProcess(ctx context.Context, containerID, processID string, signal int) error {
+	return nil
+}
+
+func (c *mockClient) simulateStartError(sim bool, id string) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if sim {
+		c.errorOnStart[id] = sim
+		return
+	}
+	delete(c.errorOnStart, id)
+}
+
+func (c *mockClient) HandleExitEvent(id string) error {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	delete(c.containers, id)
+	return nil
+}