libcontainerd: fix leaking container/exec state

Signed-off-by: Tonis Tiigi <tonistiigi@gmail.com>

integration-cli/docker_api_exec_test.go

@@ -8,6 +8,8 @@ import (
 	"fmt"
 	"io/ioutil"
 	"net/http"
+	"os"
+	"strings"
 	"time"
 
 	"github.com/docker/docker/api/types"
@@ -198,6 +200,45 @@ func (s *DockerSuite) TestExecAPIStartInvalidCommand(c *check.C) {
 	c.Assert(inspectJSON.ExecIDs, checker.IsNil)
 }
 
+func (s *DockerSuite) TestExecStateCleanup(c *check.C) {
+	testRequires(c, DaemonIsLinux, SameHostDaemon)
+
+	// This test checks accidental regressions. Not part of stable API.
+
+	name := "exec_cleanup"
+	cid, _ := dockerCmd(c, "run", "-d", "-t", "--name", name, "busybox", "/bin/sh")
+	cid = strings.TrimSpace(cid)
+
+	stateDir := "/var/run/docker/containerd/" + cid
+
+	checkReadDir := func(c *check.C) (interface{}, check.CommentInterface) {
+		fi, err := ioutil.ReadDir(stateDir)
+		c.Assert(err, checker.IsNil)
+		return len(fi), nil
+	}
+
+	fi, err := ioutil.ReadDir(stateDir)
+	c.Assert(err, checker.IsNil)
+	c.Assert(len(fi), checker.GreaterThan, 1)
+
+	id := createExecCmd(c, name, "ls")
+	startExec(c, id, http.StatusOK)
+	waitForExec(c, id)
+
+	waitAndAssert(c, 5*time.Second, checkReadDir, checker.Equals, len(fi))
+
+	id = createExecCmd(c, name, "invalid")
+	startExec(c, id, http.StatusBadRequest)
+	waitForExec(c, id)
+
+	waitAndAssert(c, 5*time.Second, checkReadDir, checker.Equals, len(fi))
+
+	dockerCmd(c, "stop", name)
+	_, err = os.Stat(stateDir)
+	c.Assert(err, checker.NotNil)
+	c.Assert(os.IsNotExist(err), checker.True)
+}
+
 func createExec(c *check.C, name string) string {
 	return createExecCmd(c, name, "true")
 }

libcontainerd/client_daemon.go

@@ -28,7 +28,7 @@ import (
 	"github.com/containerd/typeurl"
 	"github.com/docker/docker/pkg/ioutils"
 	"github.com/opencontainers/image-spec/specs-go/v1"
-	"github.com/opencontainers/runtime-spec/specs-go"
+	specs "github.com/opencontainers/runtime-spec/specs-go"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 )
@@ -202,7 +202,8 @@ func (c *client) Start(ctx context.Context, id, checkpointDir string, withStdin
 	uid, gid := getSpecUser(spec)
 	t, err = ctr.ctr.NewTask(ctx,
 		func(id string) (containerd.IO, error) {
-			cio, err = c.createIO(ctr.bundleDir, id, InitProcessName, stdinCloseSync, withStdin, spec.Process.Terminal, attachStdio)
+			fifos := newFIFOSet(ctr.bundleDir, id, InitProcessName, withStdin, spec.Process.Terminal)
+			cio, err = c.createIO(fifos, id, InitProcessName, stdinCloseSync, attachStdio)
 			return cio, err
 		},
 		func(_ context.Context, _ *containerd.Client, info *containerd.TaskInfo) error {
@@ -260,17 +261,21 @@ func (c *client) Exec(ctx context.Context, containerID, processID string, spec *
 		err            error
 		stdinCloseSync = make(chan struct{})
 	)
+
+	fifos := newFIFOSet(ctr.bundleDir, containerID, processID, withStdin, spec.Terminal)
+
 	defer func() {
 		if err != nil {
 			if cio != nil {
 				cio.Cancel()
 				cio.Close()
 			}
+			rmFIFOSet(fifos)
 		}
 	}()
 
 	p, err = ctr.task.Exec(ctx, processID, spec, func(id string) (containerd.IO, error) {
-		cio, err = c.createIO(ctr.bundleDir, containerID, processID, stdinCloseSync, withStdin, spec.Terminal, attachStdio)
+		cio, err = c.createIO(fifos, containerID, processID, stdinCloseSync, attachStdio)
 		return cio, err
 	})
 	if err != nil {
@@ -441,7 +446,7 @@ func (c *client) Delete(ctx context.Context, containerID string) error {
 		return err
 	}
 
-	if os.Getenv("LIBCONTAINERD_NOCLEAN") == "1" {
+	if os.Getenv("LIBCONTAINERD_NOCLEAN") != "1" {
 		if err := os.RemoveAll(ctr.bundleDir); err != nil {
 			c.logger.WithError(err).WithFields(logrus.Fields{
 				"container": containerID,
@@ -562,8 +567,7 @@ func (c *client) getProcess(containerID, processID string) (containerd.Process,
 
 // createIO creates the io to be used by a process
 // This needs to get a pointer to interface as upon closure the process may not have yet been registered
-func (c *client) createIO(bundleDir, containerID, processID string, stdinCloseSync chan struct{}, withStdin, withTerminal bool, attachStdio StdioCallback) (containerd.IO, error) {
-	fifos := newFIFOSet(bundleDir, containerID, processID, withStdin, withTerminal)
+func (c *client) createIO(fifos *containerd.FIFOSet, containerID, processID string, stdinCloseSync chan struct{}, attachStdio StdioCallback) (containerd.IO, error) {
 	io, err := newIOPipe(fifos)
 	if err != nil {
 		return nil, err
@@ -639,6 +643,14 @@ func (c *client) processEvent(ctr *container, et EventType, ei EventInfo) {
 			c.Lock()
 			delete(ctr.execs, ei.ProcessID)
 			c.Unlock()
+			ctr := c.getContainer(ei.ContainerID)
+			if ctr == nil {
+				c.logger.WithFields(logrus.Fields{
+					"container": ei.ContainerID,
+				}).Error("failed to find container")
+			} else {
+				rmFIFOSet(newFIFOSet(ctr.bundleDir, ei.ContainerID, ei.ProcessID, true, false))
+			}
 		}
 	})
 }

libcontainerd/client_daemon_linux.go

@@ -10,6 +10,7 @@ import (
 	"github.com/containerd/containerd"
 	"github.com/docker/docker/pkg/idtools"
 	specs "github.com/opencontainers/runtime-spec/specs-go"
+	"github.com/sirupsen/logrus"
 )
 
 func summaryFromInterface(i interface{}) (*Summary, error) {
@@ -94,3 +95,13 @@ func newFIFOSet(bundleDir, containerID, processID string, withStdin, withTermina
 
 	return fifos
 }
+
+func rmFIFOSet(fset *containerd.FIFOSet) {
+	for _, fn := range []string{fset.Out, fset.In, fset.Err} {
+		if fn != "" {
+			if err := os.RemoveAll(fn); err != nil {
+				logrus.Warnf("libcontainerd: failed to remove fifo %v: %v", fn, err)
+			}
+		}
+	}
+}