Fix fifo files leakage.

Signed-off-by: Lantao Liu <lantaol@google.com>

pkg/server/container_create.go

@@ -120,6 +120,7 @@ func (c *criContainerdService) CreateContainer(ctx context.Context, r *runtime.C
 	containerIO, err := cio.NewContainerIO(id,
 		cio.WithStdin(config.GetStdin()),
 		cio.WithTerminal(config.GetTty()),
+		cio.WithRootDir(containerRootDir),
 	)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create container io: %v", err)

pkg/server/container_stop.go

@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/containerd/containerd"
 	"github.com/containerd/containerd/errdefs"
 	"github.com/docker/docker/pkg/signal"
 	"github.com/golang/glog"
@@ -120,7 +121,7 @@ func (c *criContainerdService) stopContainer(ctx context.Context, container cont
 	// Event handler will Delete the container from containerd after it handles the Exited event.
 	glog.V(2).Infof("Kill container %q", id)
 	if task != nil {
-		if err = task.Kill(ctx, unix.SIGKILL); err != nil {
+		if err = task.Kill(ctx, unix.SIGKILL, containerd.WithKillAll); err != nil {
 			if !errdefs.IsNotFound(err) {
 				return fmt.Errorf("failed to kill container %q: %v", id, err)
 			}

pkg/server/events.go

@@ -19,6 +19,7 @@ package server
 import (
 	"time"
 
+	"github.com/containerd/containerd"
 	"github.com/containerd/containerd/api/services/events/v1"
 	"github.com/containerd/containerd/errdefs"
 	"github.com/containerd/containerd/typeurl"
@@ -107,7 +108,12 @@ func (c *criContainerdService) handleEvent(evt *events.Envelope) {
 			// Non-init process died, ignore the event.
 			return
 		}
-		task, err := cntr.Container.Task(context.Background(), nil)
+		// Attach container IO so that `Delete` could cleanup the stream properly.
+		task, err := cntr.Container.Task(context.Background(),
+			func(*containerd.FIFOSet) (containerd.IO, error) {
+				return cntr.IO, nil
+			},
+		)
 		if err != nil {
 			if !errdefs.IsNotFound(err) {
 				glog.Errorf("failed to stop container, task not found for container %q: %v", e.ContainerID, err)

pkg/server/io/io.go

@@ -19,7 +19,9 @@ package agents
 import (
 	"errors"
 	"io"
+	"io/ioutil"
 	"os"
+	"path/filepath"
 	"strings"
 	"sync"
 	"syscall"
@@ -29,7 +31,7 @@ import (
 	"github.com/golang/glog"
 	"golang.org/x/net/context"
 
-	"github.com/kubernetes-incubator/cri-containerd/pkg/ioutil"
+	cioutil "github.com/kubernetes-incubator/cri-containerd/pkg/ioutil"
 	"github.com/kubernetes-incubator/cri-containerd/pkg/util"
 )
 
@@ -81,8 +83,9 @@ type ContainerIO struct {
 	id     string
 	tty    bool
 	stdin  bool
-	stdout *ioutil.WriterGroup
+	stdout *cioutil.WriterGroup
-	stderr *ioutil.WriterGroup
+	stderr *cioutil.WriterGroup
+	root   string
 
 	closer *wgCloser
 }
@@ -125,25 +128,34 @@ func WithTerminal(tty bool) Opts {
 	}
 }
 
+// WithRootDir sets the root directory to create container streams.
+func WithRootDir(root string) Opts {
+	return func(c *ContainerIO) error {
+		c.root = root
+		return nil
+	}
+}
+
 // NewContainerIO creates container io.
 func NewContainerIO(id string, opts ...Opts) (*ContainerIO, error) {
-	fifos, err := containerd.NewFifos(id)
-	if err != nil {
-		return nil, err
-	}
 	c := &ContainerIO{
 		id:     id,
-		dir:        fifos.Dir,
+		stdout: cioutil.NewWriterGroup(),
-		stdoutPath: fifos.Out,
+		stderr: cioutil.NewWriterGroup(),
-		stderrPath: fifos.Err,
+		root:   os.TempDir(),
-		stdout:     ioutil.NewWriterGroup(),
-		stderr:     ioutil.NewWriterGroup(),
 	}
 	for _, opt := range opts {
 		if err := opt(c); err != nil {
 			return nil, err
 		}
 	}
+	fifos, err := newFifos(c.root, id)
+	if err != nil {
+		return nil, err
+	}
+	c.dir = fifos.Dir
+	c.stdoutPath = fifos.Out
+	c.stderrPath = fifos.Err
 	if c.stdin {
 		c.stdinPath = fifos.In
 	}
@@ -277,7 +289,7 @@ func (c *ContainerIO) Attach(stdin io.Reader, stdout, stderr io.WriteCloser) err
 
 	if stdout != nil {
 		wg.Add(1)
-		wc, close := ioutil.NewWriteCloseInformer(stdout)
+		wc, close := cioutil.NewWriteCloseInformer(stdout)
 		if err := c.stdout.Add(stdoutKey, wc); err != nil {
 			return err
 		}
@@ -285,7 +297,7 @@ func (c *ContainerIO) Attach(stdin io.Reader, stdout, stderr io.WriteCloser) err
 	}
 	if !c.tty && stderr != nil {
 		wg.Add(1)
-		wc, close := ioutil.NewWriteCloseInformer(stderr)
+		wc, close := cioutil.NewWriteCloseInformer(stderr)
 		if err := c.stderr.Add(stderrKey, wc); err != nil {
 			return err
 		}
@@ -315,3 +327,21 @@ func (c *ContainerIO) Close() error {
 	}
 	return nil
 }
+
+// newFifos creates fifos directory for a container.
+func newFifos(root, id string) (*containerd.FIFOSet, error) {
+	root = filepath.Join(root, "io")
+	if err := os.MkdirAll(root, 0700); err != nil {
+		return nil, err
+	}
+	dir, err := ioutil.TempDir(root, "")
+	if err != nil {
+		return nil, err
+	}
+	return &containerd.FIFOSet{
+		Dir: dir,
+		In:  filepath.Join(dir, id+"-stdin"),
+		Out: filepath.Join(dir, id+"-stdout"),
+		Err: filepath.Join(dir, id+"-stderr"),
+	}, nil
+}