Merge pull request #178 from Random-Liu/fix-leak-files

Fix leak files

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
+}

vendor.conf

@@ -1,6 +1,6 @@
 github.com/blang/semver v3.1.0
 github.com/boltdb/bolt v1.3.0-58-ge9cf4fa
-github.com/containerd/containerd 360e46ddda1733c8e237b8ce5a24470ffa08d306
+github.com/containerd/containerd cf09e32618398fc59fcb45bcfe9b4c0335972733
 github.com/containerd/continuity cf279e6ac893682272b4479d4c67fd3abf878b4e
 github.com/containerd/fifo fbfb6a11ec671efbe94ad1c12c2e98773f19e1e6
 github.com/containernetworking/cni v0.6.0

vendor/github.com/containerd/containerd/process.go

@@ -23,7 +23,7 @@ type Process interface {
 	// Delete removes the process and any resources allocated returning the exit status
 	Delete(context.Context, ...ProcessDeleteOpts) (*ExitStatus, error)
 	// Kill sends the provided signal to the process
-	Kill(context.Context, syscall.Signal) error
+	Kill(context.Context, syscall.Signal, ...KillOpts) error
 	// Wait asynchronously waits for the process to exit, and sends the exit code to the returned channel
 	Wait(context.Context) (<-chan ExitStatus, error)
 	// CloseIO allows various pipes to be closed on the process
@@ -104,11 +104,18 @@ func (p *process) Start(ctx context.Context) error {
 	return nil
 }
 
-func (p *process) Kill(ctx context.Context, s syscall.Signal) error {
+func (p *process) Kill(ctx context.Context, s syscall.Signal, opts ...KillOpts) error {
+	var i KillInfo
+	for _, o := range opts {
+		if err := o(ctx, p, &i); err != nil {
+			return err
+		}
+	}
 	_, err := p.task.client.TaskService().Kill(ctx, &tasks.KillRequest{
 		Signal:      uint32(s),
 		ContainerID: p.task.id,
 		ExecID:      p.id,
+		All:         i.All,
 	})
 	return errdefs.FromGRPC(err)
 }

vendor/github.com/containerd/containerd/spec_opts_unix.go

@@ -297,8 +297,9 @@ func WithUidGid(uid, gid uint32) SpecOpts {
 }
 
 // WithUserID sets the correct UID and GID for the container based
-// on the image's /etc/passwd contents. If uid is not found in
+// on the image's /etc/passwd contents. If /etc/passwd does not exist,
-// /etc/passwd, it sets uid but leaves gid 0, and not returns error.
+// or uid is not found in /etc/passwd, it sets gid to be the same with
+// uid, and not returns error.
 func WithUserID(uid uint32) SpecOpts {
 	return func(ctx context.Context, client *Client, c *containers.Container, s *specs.Spec) error {
 		if c.Snapshotter == "" {
@@ -329,6 +330,10 @@ func WithUserID(uid uint32) SpecOpts {
 		}
 		f, err := os.Open(ppath)
 		if err != nil {
+			if os.IsNotExist(err) {
+				s.Process.User.UID, s.Process.User.GID = uid, uid
+				return nil
+			}
 			return err
 		}
 		defer f.Close()
@@ -339,7 +344,7 @@ func WithUserID(uid uint32) SpecOpts {
 			return err
 		}
 		if len(users) == 0 {
-			s.Process.User.UID = uid
+			s.Process.User.UID, s.Process.User.GID = uid, uid
 			return nil
 		}
 		u := users[0]
@@ -349,8 +354,9 @@ func WithUserID(uid uint32) SpecOpts {
 }
 
 // WithUsername sets the correct UID and GID for the container
-// based on the the image's /etc/passwd contents. If the username
+// based on the the image's /etc/passwd contents. If /etc/passwd
-// is not found in /etc/passwd, it returns error.
+// does not exist, or the username is not found in /etc/passwd,
+// it returns error.
 func WithUsername(username string) SpecOpts {
 	return func(ctx context.Context, client *Client, c *containers.Container, s *specs.Spec) error {
 		if c.Snapshotter == "" {

vendor/github.com/containerd/containerd/task.go

@@ -163,10 +163,17 @@ func (t *task) Start(ctx context.Context) error {
 	return errdefs.FromGRPC(err)
 }
 
-func (t *task) Kill(ctx context.Context, s syscall.Signal) error {
+func (t *task) Kill(ctx context.Context, s syscall.Signal, opts ...KillOpts) error {
+	var i KillInfo
+	for _, o := range opts {
+		if err := o(ctx, t, &i); err != nil {
+			return err
+		}
+	}
 	_, err := t.client.TaskService().Kill(ctx, &tasks.KillRequest{
 		Signal:      uint32(s),
 		ContainerID: t.id,
+		All:         i.All,
 	})
 	if err != nil {
 		return errdefs.FromGRPC(err)

vendor/github.com/containerd/containerd/task_opts.go

@@ -41,7 +41,7 @@ func WithProcessKill(ctx context.Context, p Process) error {
 	if err != nil {
 		return err
 	}
-	if err := p.Kill(ctx, syscall.SIGKILL); err != nil {
+	if err := p.Kill(ctx, syscall.SIGKILL, WithKillAll); err != nil {
 		if errdefs.IsFailedPrecondition(err) || errdefs.IsNotFound(err) {
 			return nil
 		}
@@ -51,3 +51,17 @@ func WithProcessKill(ctx context.Context, p Process) error {
 	<-s
 	return nil
 }
+
+type KillInfo struct {
+	// All kills all processes inside the task
+	// only valid on tasks, ignored on processes
+	All bool
+}
+
+type KillOpts func(context.Context, Process, *KillInfo) error
+
+// WithKillAll kills all processes for a task
+func WithKillAll(ctx context.Context, p Process, i *KillInfo) error {
+	i.All = true
+	return nil
+}