diff --git a/cio/io.go b/cio/io.go index c5e112d9c..1b4a4dc25 100644 --- a/cio/io.go +++ b/cio/io.go @@ -8,7 +8,7 @@ import ( "sync" ) -// Config holds the io configurations. +// Config holds the IO configurations. type Config struct { // Terminal is true if one has been allocated Terminal bool @@ -33,40 +33,8 @@ type IO interface { Close() error } -// cio is a basic container IO implementation. -type cio struct { - config Config - - closer *wgCloser -} - -func (c *cio) Config() Config { - return c.config -} - -func (c *cio) Cancel() { - if c.closer == nil { - return - } - c.closer.Cancel() -} - -func (c *cio) Wait() { - if c.closer == nil { - return - } - c.closer.Wait() -} - -func (c *cio) Close() error { - if c.closer == nil { - return nil - } - return c.closer.Close() -} - -// Creation creates new IO sets for a task -type Creation func(id string) (IO, error) +// Creator creates new IO sets for a task +type Creator func(id string) (IO, error) // Attach allows callers to reattach to running tasks // @@ -75,123 +43,138 @@ type Creation func(id string) (IO, error) // will be sent only to the first reads type Attach func(*FIFOSet) (IO, error) -// NewIO returns an Creation that will provide IO sets without a terminal -func NewIO(stdin io.Reader, stdout, stderr io.Writer) Creation { - return NewIOWithTerminal(stdin, stdout, stderr, false) -} - -// NewIOWithTerminal creates a new io set with the provied io.Reader/Writers for use with a terminal -func NewIOWithTerminal(stdin io.Reader, stdout, stderr io.Writer, terminal bool) Creation { - return func(id string) (_ IO, err error) { - paths, err := NewFifos(id) - if err != nil { - return nil, err - } - defer func() { - if err != nil && paths.Dir != "" { - os.RemoveAll(paths.Dir) - } - }() - cfg := Config{ - Terminal: terminal, - Stdout: paths.Out, - Stderr: paths.Err, - Stdin: paths.In, - } - i := &cio{config: cfg} - set := &ioSet{ - in: stdin, - out: stdout, - err: stderr, - } - closer, err := copyIO(paths, set, cfg.Terminal) - if err != nil { - return nil, err - } - i.closer = closer - return i, nil - } -} - -// WithAttach attaches the existing io for a task to the provided io.Reader/Writers -func WithAttach(stdin io.Reader, stdout, stderr io.Writer) Attach { - return func(paths *FIFOSet) (IO, error) { - if paths == nil { - return nil, fmt.Errorf("cannot attach to existing fifos") - } - cfg := Config{ - Terminal: paths.Terminal, - Stdout: paths.Out, - Stderr: paths.Err, - Stdin: paths.In, - } - i := &cio{config: cfg} - set := &ioSet{ - in: stdin, - out: stdout, - err: stderr, - } - closer, err := copyIO(paths, set, cfg.Terminal) - if err != nil { - return nil, err - } - i.closer = closer - return i, nil - } -} - -// Stdio returns an IO set to be used for a task -// that outputs the container's IO as the current processes Stdio -func Stdio(id string) (IO, error) { - return NewIO(os.Stdin, os.Stdout, os.Stderr)(id) -} - -// StdioTerminal will setup the IO for the task to use a terminal -func StdioTerminal(id string) (IO, error) { - return NewIOWithTerminal(os.Stdin, os.Stdout, os.Stderr, true)(id) -} - -// NullIO redirects the container's IO into /dev/null -func NullIO(id string) (IO, error) { - return &cio{}, nil -} - -// FIFOSet is a set of fifos for use with tasks +// FIFOSet is a set of file paths to FIFOs for a task's standard IO streams type FIFOSet struct { - // Dir is the directory holding the task fifos - Dir string - // In, Out, and Err fifo paths - In, Out, Err string - // Terminal returns true if a terminal is being used for the task - Terminal bool + Config + close func() error } -type ioSet struct { - in io.Reader - out, err io.Writer -} - -type wgCloser struct { - wg *sync.WaitGroup - dir string - set []io.Closer - cancel context.CancelFunc -} - -func (g *wgCloser) Wait() { - g.wg.Wait() -} - -func (g *wgCloser) Close() error { - for _, f := range g.set { - f.Close() - } - if g.dir != "" { - return os.RemoveAll(g.dir) +// Close the FIFOSet +func (f *FIFOSet) Close() error { + if f.close != nil { + return f.close() } return nil } -func (g *wgCloser) Cancel() { - g.cancel() +// NewFIFOSet returns a new FIFOSet from a Config and a close function +func NewFIFOSet(config Config, close func() error) *FIFOSet { + return &FIFOSet{Config: config, close: close} } + +// Streams used to configure a Creator or Attach +type Streams struct { + Stdin io.Reader + Stdout io.Writer + Stderr io.Writer + Terminal bool +} + +// Opt customize options for creating a Creator or Attach +type Opt func(*Streams) + +// WithStdio sets stream options to the standard input/output streams +func WithStdio(opt *Streams) { + WithStreams(os.Stdin, os.Stdout, os.Stderr)(opt) +} + +// WithTerminal sets the terminal option +func WithTerminal(opt *Streams) { + opt.Terminal = true +} + +// WithStreams sets the stream options to the specified Reader and Writers +func WithStreams(stdin io.Reader, stdout, stderr io.Writer) Opt { + return func(opt *Streams) { + opt.Stdin = stdin + opt.Stdout = stdout + opt.Stderr = stderr + } +} + +// NewCreator returns an IO creator from the options +func NewCreator(opts ...Opt) Creator { + streams := &Streams{} + for _, opt := range opts { + opt(streams) + } + return func(id string) (IO, error) { + // TODO: accept root as a param + root := "/run/containerd/fifo" + fifos, err := NewFIFOSetInDir(root, id, streams.Terminal) + if err != nil { + return nil, err + } + return copyIO(fifos, streams) + } +} + +// NewAttach attaches the existing io for a task to the provided io.Reader/Writers +func NewAttach(opts ...Opt) Attach { + streams := &Streams{} + for _, opt := range opts { + opt(streams) + } + return func(fifos *FIFOSet) (IO, error) { + if fifos == nil { + return nil, fmt.Errorf("cannot attach, missing fifos") + } + return copyIO(fifos, streams) + } +} + +// NullIO redirects the container's IO into /dev/null +func NullIO(_ string) (IO, error) { + return &cio{}, nil +} + +// cio is a basic container IO implementation. +type cio struct { + config Config + wg *sync.WaitGroup + closers []io.Closer + cancel context.CancelFunc +} + +func (c *cio) Config() Config { + return c.config +} + +func (c *cio) Wait() { + if c.wg != nil { + c.wg.Wait() + } +} + +func (c *cio) Close() error { + var lastErr error + for _, closer := range c.closers { + if closer == nil { + continue + } + if err := closer.Close(); err != nil { + lastErr = err + } + } + return lastErr +} + +func (c *cio) Cancel() { + if c.cancel != nil { + c.cancel() + } +} + +type pipes struct { + Stdin io.WriteCloser + Stdout io.ReadCloser + Stderr io.ReadCloser +} + +// DirectIO allows task IO to be handled externally by the caller +type DirectIO struct { + pipes + cio +} + +var _ IO = &DirectIO{} diff --git a/cio/io_test.go b/cio/io_test.go new file mode 100644 index 000000000..434c0af5f --- /dev/null +++ b/cio/io_test.go @@ -0,0 +1,132 @@ +// +build !windows + +package cio + +import ( + "bytes" + "context" + "io" + "io/ioutil" + "os" + "path/filepath" + "runtime" + "strings" + "syscall" + "testing" + + "github.com/containerd/fifo" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func assertHasPrefix(t *testing.T, s, prefix string) { + t.Helper() + if !strings.HasPrefix(s, prefix) { + t.Fatalf("expected %s to start with %s", s, prefix) + } +} + +func TestNewFIFOSetInDir(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("NewFIFOSetInDir has different behaviour on windows") + } + + root, err := ioutil.TempDir("", "test-new-fifo-set") + require.NoError(t, err) + defer os.RemoveAll(root) + + fifos, err := NewFIFOSetInDir(root, "theid", true) + require.NoError(t, err) + + assertHasPrefix(t, fifos.Stdin, root) + assertHasPrefix(t, fifos.Stdout, root) + assertHasPrefix(t, fifos.Stderr, root) + assert.Equal(t, "theid-stdin", filepath.Base(fifos.Stdin)) + assert.Equal(t, "theid-stdout", filepath.Base(fifos.Stdout)) + assert.Equal(t, "theid-stderr", filepath.Base(fifos.Stderr)) + assert.Equal(t, true, fifos.Terminal) + + files, err := ioutil.ReadDir(root) + require.NoError(t, err) + assert.Len(t, files, 1) + + require.NoError(t, fifos.Close()) + files, err = ioutil.ReadDir(root) + require.NoError(t, err) + assert.Len(t, files, 0) +} + +func TestNewAttach(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("setupFIFOProducers not yet implemented on windows") + } + var ( + expectedStdin = "this is the stdin" + expectedStdout = "this is the stdout" + expectedStderr = "this is the stderr" + stdin = bytes.NewBufferString(expectedStdin) + stdout = new(bytes.Buffer) + stderr = new(bytes.Buffer) + ) + + withBytesBuffers := func(streams *Streams) { + *streams = Streams{Stdin: stdin, Stdout: stdout, Stderr: stderr} + } + attacher := NewAttach(withBytesBuffers) + + fifos, err := NewFIFOSetInDir("", "theid", false) + require.NoError(t, err) + + io, err := attacher(fifos) + require.NoError(t, err) + defer io.Close() + + producers := setupFIFOProducers(t, io.Config()) + initProducers(t, producers, expectedStdout, expectedStderr) + + actualStdin, err := ioutil.ReadAll(producers.Stdin) + require.NoError(t, err) + + io.Cancel() + io.Wait() + assert.NoError(t, io.Close()) + + assert.Equal(t, expectedStdout, stdout.String()) + assert.Equal(t, expectedStderr, stderr.String()) + assert.Equal(t, expectedStdin, string(actualStdin)) +} + +type producers struct { + Stdin io.ReadCloser + Stdout io.WriteCloser + Stderr io.WriteCloser +} + +func setupFIFOProducers(t *testing.T, fifos Config) producers { + var ( + err error + pipes producers + ctx = context.Background() + ) + + pipes.Stdin, err = fifo.OpenFifo(ctx, fifos.Stdin, syscall.O_RDONLY, 0) + require.NoError(t, err) + + pipes.Stdout, err = fifo.OpenFifo(ctx, fifos.Stdout, syscall.O_WRONLY, 0) + require.NoError(t, err) + + pipes.Stderr, err = fifo.OpenFifo(ctx, fifos.Stderr, syscall.O_WRONLY, 0) + require.NoError(t, err) + + return pipes +} + +func initProducers(t *testing.T, producers producers, stdout, stderr string) { + _, err := producers.Stdout.Write([]byte(stdout)) + require.NoError(t, err) + require.NoError(t, producers.Stdout.Close()) + + _, err = producers.Stderr.Write([]byte(stderr)) + require.NoError(t, err) + require.NoError(t, producers.Stderr.Close()) +} diff --git a/cio/io_unix.go b/cio/io_unix.go index c18f7ecf9..005fb0ce9 100644 --- a/cio/io_unix.go +++ b/cio/io_unix.go @@ -12,173 +12,115 @@ import ( "syscall" "github.com/containerd/fifo" + "github.com/pkg/errors" ) -// NewFifos returns a new set of fifos for the task -func NewFifos(id string) (*FIFOSet, error) { - root := "/run/containerd/fifo" - if err := os.MkdirAll(root, 0700); err != nil { - return nil, err +// NewFIFOSetInDir returns a new FIFOSet with paths in a temporary directory under root +func NewFIFOSetInDir(root, id string, terminal bool) (*FIFOSet, error) { + if root != "" { + if err := os.MkdirAll(root, 0700); err != nil { + return nil, err + } } dir, err := ioutil.TempDir(root, "") if err != nil { return nil, err } - return &FIFOSet{ - Dir: dir, - In: filepath.Join(dir, id+"-stdin"), - Out: filepath.Join(dir, id+"-stdout"), - Err: filepath.Join(dir, id+"-stderr"), - }, nil + closer := func() error { + return os.RemoveAll(dir) + } + return NewFIFOSet(Config{ + Stdin: filepath.Join(dir, id+"-stdin"), + Stdout: filepath.Join(dir, id+"-stdout"), + Stderr: filepath.Join(dir, id+"-stderr"), + Terminal: terminal, + }, closer), nil } -func copyIO(fifos *FIFOSet, ioset *ioSet, tty bool) (_ *wgCloser, err error) { - var ( - f io.ReadWriteCloser - set []io.Closer - ctx, cancel = context.WithCancel(context.Background()) - wg = &sync.WaitGroup{} - ) - defer func() { - if err != nil { - for _, f := range set { - f.Close() - } - cancel() - } - }() - - if f, err = fifo.OpenFifo(ctx, fifos.In, syscall.O_WRONLY|syscall.O_CREAT|syscall.O_NONBLOCK, 0700); err != nil { - return nil, err - } - set = append(set, f) - go func(w io.WriteCloser) { - io.Copy(w, ioset.in) - w.Close() - }(f) - - if f, err = fifo.OpenFifo(ctx, fifos.Out, syscall.O_RDONLY|syscall.O_CREAT|syscall.O_NONBLOCK, 0700); err != nil { - return nil, err - } - set = append(set, f) - wg.Add(1) - go func(r io.ReadCloser) { - io.Copy(ioset.out, r) - r.Close() - wg.Done() - }(f) - - if f, err = fifo.OpenFifo(ctx, fifos.Err, syscall.O_RDONLY|syscall.O_CREAT|syscall.O_NONBLOCK, 0700); err != nil { - return nil, err - } - set = append(set, f) - - if !tty { - wg.Add(1) - go func(r io.ReadCloser) { - io.Copy(ioset.err, r) - r.Close() - wg.Done() - }(f) - } - return &wgCloser{ - wg: wg, - dir: fifos.Dir, - set: set, - cancel: cancel, - }, nil -} - -// NewDirectIO returns an IO implementation that exposes the pipes directly -func NewDirectIO(ctx context.Context, terminal bool) (*DirectIO, error) { - set, err := NewFifos("") +func copyIO(fifos *FIFOSet, ioset *Streams) (*cio, error) { + var ctx, cancel = context.WithCancel(context.Background()) + pipes, err := openFifos(ctx, fifos) if err != nil { + cancel() return nil, err } - f := &DirectIO{ - set: set, - terminal: terminal, + + if fifos.Stdin != "" { + go func() { + io.Copy(pipes.Stdin, ioset.Stdin) + pipes.Stdin.Close() + }() } + + var wg = &sync.WaitGroup{} + wg.Add(1) + go func() { + io.Copy(ioset.Stdout, pipes.Stdout) + pipes.Stdout.Close() + wg.Done() + }() + + if !fifos.Terminal { + wg.Add(1) + go func() { + io.Copy(ioset.Stderr, pipes.Stderr) + pipes.Stderr.Close() + wg.Done() + }() + } + return &cio{ + config: fifos.Config, + wg: wg, + closers: append(pipes.closers(), fifos), + cancel: cancel, + }, nil +} + +func openFifos(ctx context.Context, fifos *FIFOSet) (pipes, error) { + var err error defer func() { if err != nil { - f.Delete() + fifos.Close() } }() - if f.Stdin, err = fifo.OpenFifo(ctx, set.In, syscall.O_WRONLY|syscall.O_CREAT|syscall.O_NONBLOCK, 0700); err != nil { - return nil, err + + var f pipes + if fifos.Stdin != "" { + if f.Stdin, err = fifo.OpenFifo(ctx, fifos.Stdin, syscall.O_WRONLY|syscall.O_CREAT|syscall.O_NONBLOCK, 0700); err != nil { + return f, errors.Wrapf(err, "failed to open stdin fifo") + } } - if f.Stdout, err = fifo.OpenFifo(ctx, set.Out, syscall.O_RDONLY|syscall.O_CREAT|syscall.O_NONBLOCK, 0700); err != nil { - f.Stdin.Close() - return nil, err + if fifos.Stdout != "" { + if f.Stdout, err = fifo.OpenFifo(ctx, fifos.Stdout, syscall.O_RDONLY|syscall.O_CREAT|syscall.O_NONBLOCK, 0700); err != nil { + f.Stdin.Close() + return f, errors.Wrapf(err, "failed to open stdout fifo") + } } - if f.Stderr, err = fifo.OpenFifo(ctx, set.Err, syscall.O_RDONLY|syscall.O_CREAT|syscall.O_NONBLOCK, 0700); err != nil { - f.Stdin.Close() - f.Stdout.Close() - return nil, err + if fifos.Stderr != "" { + if f.Stderr, err = fifo.OpenFifo(ctx, fifos.Stderr, syscall.O_RDONLY|syscall.O_CREAT|syscall.O_NONBLOCK, 0700); err != nil { + f.Stdin.Close() + f.Stdout.Close() + return f, errors.Wrapf(err, "failed to open stderr fifo") + } } return f, nil } -// DirectIO allows task IO to be handled externally by the caller -type DirectIO struct { - Stdin io.WriteCloser - Stdout io.ReadCloser - Stderr io.ReadCloser - - set *FIFOSet - terminal bool +// NewDirectIO returns an IO implementation that exposes the IO streams as io.ReadCloser +// and io.WriteCloser. +func NewDirectIO(ctx context.Context, fifos *FIFOSet) (*DirectIO, error) { + ctx, cancel := context.WithCancel(ctx) + pipes, err := openFifos(ctx, fifos) + return &DirectIO{ + pipes: pipes, + cio: cio{ + config: fifos.Config, + closers: append(pipes.closers(), fifos), + cancel: cancel, + }, + }, err } -// IOCreate returns IO avaliable for use with task creation -func (f *DirectIO) IOCreate(id string) (IO, error) { - return f, nil -} - -// IOAttach returns IO avaliable for use with task attachment -func (f *DirectIO) IOAttach(set *FIFOSet) (IO, error) { - return f, nil -} - -// Config returns the Config -func (f *DirectIO) Config() Config { - return Config{ - Terminal: f.terminal, - Stdin: f.set.In, - Stdout: f.set.Out, - Stderr: f.set.Err, - } -} - -// Cancel stops any IO copy operations -// -// Not applicable for DirectIO -func (f *DirectIO) Cancel() { - // nothing to cancel as all operations are handled externally -} - -// Wait on any IO copy operations -// -// Not applicable for DirectIO -func (f *DirectIO) Wait() { - // nothing to wait on as all operations are handled externally -} - -// Close closes all open fds -func (f *DirectIO) Close() error { - err := f.Stdin.Close() - if err2 := f.Stdout.Close(); err == nil { - err = err2 - } - if err2 := f.Stderr.Close(); err == nil { - err = err2 - } - return err -} - -// Delete removes the underlying directory containing fifos -func (f *DirectIO) Delete() error { - if f.set.Dir == "" { - return nil - } - return os.RemoveAll(f.set.Dir) +func (p *pipes) closers() []io.Closer { + return []io.Closer{p.Stdin, p.Stdout, p.Stderr} } diff --git a/cio/io_windows.go b/cio/io_windows.go index 1458c3173..017c9a11f 100644 --- a/cio/io_windows.go +++ b/cio/io_windows.go @@ -13,25 +13,26 @@ import ( const pipeRoot = `\\.\pipe` -// NewFifos returns a new set of fifos for the task -func NewFifos(id string) (*FIFOSet, error) { - return &FIFOSet{ - In: fmt.Sprintf(`%s\ctr-%s-stdin`, pipeRoot, id), - Out: fmt.Sprintf(`%s\ctr-%s-stdout`, pipeRoot, id), - Err: fmt.Sprintf(`%s\ctr-%s-stderr`, pipeRoot, id), - }, nil +// NewFIFOSetInDir returns a new set of fifos for the task +func NewFIFOSetInDir(_, id string, terminal bool) (*FIFOSet, error) { + return NewFIFOSet(Config{ + Terminal: terminal, + Stdin: fmt.Sprintf(`%s\ctr-%s-stdin`, pipeRoot, id), + Stdout: fmt.Sprintf(`%s\ctr-%s-stdout`, pipeRoot, id), + Stderr: fmt.Sprintf(`%s\ctr-%s-stderr`, pipeRoot, id), + }, nil), nil } -func copyIO(fifos *FIFOSet, ioset *ioSet, tty bool) (_ *wgCloser, err error) { +func copyIO(fifos *FIFOSet, ioset *Streams) (*cio, error) { var ( wg sync.WaitGroup set []io.Closer ) - if fifos.In != "" { - l, err := winio.ListenPipe(fifos.In, nil) + if fifos.Stdin != "" { + l, err := winio.ListenPipe(fifos.Stdin, nil) if err != nil { - return nil, errors.Wrapf(err, "failed to create stdin pipe %s", fifos.In) + return nil, errors.Wrapf(err, "failed to create stdin pipe %s", fifos.Stdin) } defer func(l net.Listener) { if err != nil { @@ -43,19 +44,19 @@ func copyIO(fifos *FIFOSet, ioset *ioSet, tty bool) (_ *wgCloser, err error) { go func() { c, err := l.Accept() if err != nil { - log.L.WithError(err).Errorf("failed to accept stdin connection on %s", fifos.In) + log.L.WithError(err).Errorf("failed to accept stdin connection on %s", fifos.Stdin) return } - io.Copy(c, ioset.in) + io.Copy(c, ioset.Stdin) c.Close() l.Close() }() } - if fifos.Out != "" { - l, err := winio.ListenPipe(fifos.Out, nil) + if fifos.Stdout != "" { + l, err := winio.ListenPipe(fifos.Stdout, nil) if err != nil { - return nil, errors.Wrapf(err, "failed to create stdin pipe %s", fifos.Out) + return nil, errors.Wrapf(err, "failed to create stdin pipe %s", fifos.Stdout) } defer func(l net.Listener) { if err != nil { @@ -69,19 +70,19 @@ func copyIO(fifos *FIFOSet, ioset *ioSet, tty bool) (_ *wgCloser, err error) { defer wg.Done() c, err := l.Accept() if err != nil { - log.L.WithError(err).Errorf("failed to accept stdout connection on %s", fifos.Out) + log.L.WithError(err).Errorf("failed to accept stdout connection on %s", fifos.Stdout) return } - io.Copy(ioset.out, c) + io.Copy(ioset.Stdout, c) c.Close() l.Close() }() } - if !tty && fifos.Err != "" { - l, err := winio.ListenPipe(fifos.Err, nil) + if !fifos.Terminal && fifos.Stderr != "" { + l, err := winio.ListenPipe(fifos.Stderr, nil) if err != nil { - return nil, errors.Wrapf(err, "failed to create stderr pipe %s", fifos.Err) + return nil, errors.Wrapf(err, "failed to create stderr pipe %s", fifos.Stderr) } defer func(l net.Listener) { if err != nil { @@ -95,23 +96,29 @@ func copyIO(fifos *FIFOSet, ioset *ioSet, tty bool) (_ *wgCloser, err error) { defer wg.Done() c, err := l.Accept() if err != nil { - log.L.WithError(err).Errorf("failed to accept stderr connection on %s", fifos.Err) + log.L.WithError(err).Errorf("failed to accept stderr connection on %s", fifos.Stderr) return } - io.Copy(ioset.err, c) + io.Copy(ioset.Stderr, c) c.Close() l.Close() }() } - return &wgCloser{ - wg: &wg, - dir: fifos.Dir, - set: set, - cancel: func() { - for _, l := range set { - l.Close() - } - }, - }, nil + return &cio{config: fifos.Config, closers: set}, nil +} + +// NewDirectIO returns an IO implementation that exposes the IO streams as io.ReadCloser +// and io.WriteCloser. +func NewDirectIO(stdin io.WriteCloser, stdout, stderr io.ReadCloser, terminal bool) *DirectIO { + return &DirectIO{ + pipes: pipes{ + Stdin: stdin, + Stdout: stdout, + Stderr: stderr, + }, + cio: cio{ + config: Config{Terminal: terminal}, + }, + } } diff --git a/cmd/ctr/commands/tasks/attach.go b/cmd/ctr/commands/tasks/attach.go index ec4da1005..517bd5764 100644 --- a/cmd/ctr/commands/tasks/attach.go +++ b/cmd/ctr/commands/tasks/attach.go @@ -1,8 +1,6 @@ package tasks import ( - "os" - "github.com/containerd/console" "github.com/containerd/containerd/cio" "github.com/containerd/containerd/cmd/ctr/commands" @@ -39,7 +37,7 @@ var attachCommand = cli.Command{ return err } } - task, err := container.Task(ctx, cio.WithAttach(os.Stdin, os.Stdout, os.Stderr)) + task, err := container.Task(ctx, cio.NewAttach(cio.WithStdio)) if err != nil { return err } diff --git a/cmd/ctr/commands/tasks/exec.go b/cmd/ctr/commands/tasks/exec.go index a7301136b..67011db49 100644 --- a/cmd/ctr/commands/tasks/exec.go +++ b/cmd/ctr/commands/tasks/exec.go @@ -60,9 +60,9 @@ var execCommand = cli.Command{ pspec.Terminal = tty pspec.Args = args - ioCreator := cio.Stdio + ioCreator := cio.NewCreator(cio.WithStdio) if tty { - ioCreator = cio.StdioTerminal + ioCreator = cio.NewCreator(cio.WithStdio, cio.WithTerminal) } process, err := task.Exec(ctx, context.String("exec-id"), pspec, ioCreator) if err != nil { diff --git a/cmd/ctr/commands/tasks/tasks_unix.go b/cmd/ctr/commands/tasks/tasks_unix.go index 0621f9566..2a1d5e832 100644 --- a/cmd/ctr/commands/tasks/tasks_unix.go +++ b/cmd/ctr/commands/tasks/tasks_unix.go @@ -44,10 +44,11 @@ func HandleConsoleResize(ctx gocontext.Context, task resizer, con console.Consol // NewTask creates a new task func NewTask(ctx gocontext.Context, client *containerd.Client, container containerd.Container, checkpoint string, tty, nullIO bool) (containerd.Task, error) { + stdio := cio.NewCreator(cio.WithStdio) if checkpoint == "" { - ioCreator := cio.Stdio + ioCreator := stdio if tty { - ioCreator = cio.StdioTerminal + ioCreator = cio.NewCreator(cio.WithStdio, cio.WithTerminal) } if nullIO { if tty { @@ -61,5 +62,5 @@ func NewTask(ctx gocontext.Context, client *containerd.Client, container contain if err != nil { return nil, err } - return container.NewTask(ctx, cio.Stdio, containerd.WithTaskCheckpoint(im)) + return container.NewTask(ctx, stdio, containerd.WithTaskCheckpoint(im)) } diff --git a/cmd/ctr/commands/tasks/tasks_windows.go b/cmd/ctr/commands/tasks/tasks_windows.go index 083ad49e6..93ca166e1 100644 --- a/cmd/ctr/commands/tasks/tasks_windows.go +++ b/cmd/ctr/commands/tasks/tasks_windows.go @@ -42,9 +42,9 @@ func HandleConsoleResize(ctx gocontext.Context, task resizer, con console.Consol // NewTask creates a new task func NewTask(ctx gocontext.Context, client *containerd.Client, container containerd.Container, _ string, tty, nullIO bool) (containerd.Task, error) { - ioCreator := cio.Stdio + ioCreator := cio.NewCreator(cio.WithStdio) if tty { - ioCreator = cio.StdioTerminal + ioCreator = cio.NewCreator(cio.WithStdio, cio.WithTerminal) } if nullIO { if tty { diff --git a/container.go b/container.go index 716e0456c..ad60c69ea 100644 --- a/container.go +++ b/container.go @@ -3,6 +3,7 @@ package containerd import ( "context" "encoding/json" + "os" "path/filepath" "strings" @@ -26,7 +27,7 @@ type Container interface { // Delete removes the container Delete(context.Context, ...DeleteOpts) error // NewTask creates a new task based on the container metadata - NewTask(context.Context, cio.Creation, ...NewTaskOpts) (Task, error) + NewTask(context.Context, cio.Creator, ...NewTaskOpts) (Task, error) // Spec returns the OCI runtime specification Spec(context.Context) (*specs.Spec, error) // Task returns the current task for the container @@ -162,7 +163,7 @@ func (c *container) Image(ctx context.Context) (Image, error) { }, nil } -func (c *container) NewTask(ctx context.Context, ioCreate cio.Creation, opts ...NewTaskOpts) (_ Task, err error) { +func (c *container) NewTask(ctx context.Context, ioCreate cio.Creator, opts ...NewTaskOpts) (_ Task, err error) { i, err := ioCreate(c.id) if err != nil { return nil, err @@ -288,20 +289,23 @@ func (c *container) get(ctx context.Context) (containers.Container, error) { return c.client.ContainerService().Get(ctx, c.id) } +// get the existing fifo paths from the task information stored by the daemon func attachExistingIO(response *tasks.GetResponse, ioAttach cio.Attach) (cio.IO, error) { - // get the existing fifo paths from the task information stored by the daemon - paths := &cio.FIFOSet{ - Dir: getFifoDir([]string{ - response.Process.Stdin, - response.Process.Stdout, - response.Process.Stderr, - }), - In: response.Process.Stdin, - Out: response.Process.Stdout, - Err: response.Process.Stderr, - Terminal: response.Process.Terminal, + path := getFifoDir([]string{ + response.Process.Stdin, + response.Process.Stdout, + response.Process.Stderr, + }) + closer := func() error { + return os.RemoveAll(path) } - return ioAttach(paths) + fifoSet := cio.NewFIFOSet(cio.Config{ + Stdin: response.Process.Stdin, + Stdout: response.Process.Stdout, + Stderr: response.Process.Stderr, + Terminal: response.Process.Terminal, + }, closer) + return ioAttach(fifoSet) } // getFifoDir looks for any non-empty path for a stdio fifo diff --git a/container_linux_test.go b/container_linux_test.go index 85e934f52..b7d648fba 100644 --- a/container_linux_test.go +++ b/container_linux_test.go @@ -298,7 +298,7 @@ func TestContainerAttach(t *testing.T) { expected := "hello" + newLine - direct, err := cio.NewDirectIO(ctx, false) + direct, err := newDirectIO(ctx) if err != nil { t.Error(err) return @@ -372,6 +372,53 @@ func TestContainerAttach(t *testing.T) { } } +func newDirectIO(ctx context.Context) (*directIO, error) { + fifos, err := cio.NewFIFOSetInDir("", "", false) + if err != nil { + return nil, err + } + dio, err := cio.NewDirectIO(ctx, fifos) + if err != nil { + return nil, err + } + return &directIO{DirectIO: *dio}, nil +} + +type directIO struct { + cio.DirectIO +} + +// ioCreate returns IO avaliable for use with task creation +func (f *directIO) IOCreate(id string) (cio.IO, error) { + return f, nil +} + +// ioAttach returns IO avaliable for use with task attachment +func (f *directIO) IOAttach(set *cio.FIFOSet) (cio.IO, error) { + return f, nil +} + +func (f *directIO) Cancel() { + // nothing to cancel as all operations are handled externally +} + +// Close closes all open fds +func (f *directIO) Close() error { + err := f.Stdin.Close() + if err2 := f.Stdout.Close(); err == nil { + err = err2 + } + if err2 := f.Stderr.Close(); err == nil { + err = err2 + } + return err +} + +// Delete removes the underlying directory containing fifos +func (f *directIO) Delete() error { + return f.DirectIO.Close() +} + func TestContainerUsername(t *testing.T) { t.Parallel() @@ -393,7 +440,7 @@ func TestContainerUsername(t *testing.T) { t.Error(err) return } - direct, err := cio.NewDirectIO(ctx, false) + direct, err := newDirectIO(ctx) if err != nil { t.Error(err) return @@ -486,7 +533,7 @@ func TestContainerAttachProcess(t *testing.T) { expected := "hello" + newLine // creating IO early for easy resource cleanup - direct, err := cio.NewDirectIO(ctx, false) + direct, err := newDirectIO(ctx) if err != nil { t.Error(err) return @@ -602,7 +649,7 @@ func TestContainerUserID(t *testing.T) { t.Error(err) return } - direct, err := cio.NewDirectIO(ctx, false) + direct, err := newDirectIO(ctx) if err != nil { t.Error(err) return @@ -692,7 +739,7 @@ func TestContainerKillAll(t *testing.T) { defer container.Delete(ctx, WithSnapshotCleanup) stdout := bytes.NewBuffer(nil) - task, err := container.NewTask(ctx, cio.NewIO(bytes.NewBuffer(nil), stdout, bytes.NewBuffer(nil))) + task, err := container.NewTask(ctx, cio.NewCreator(withByteBuffers(stdout))) if err != nil { t.Error(err) return @@ -944,7 +991,7 @@ func TestContainerKillInitPidHost(t *testing.T) { defer container.Delete(ctx, WithSnapshotCleanup) stdout := bytes.NewBuffer(nil) - task, err := container.NewTask(ctx, cio.NewIO(bytes.NewBuffer(nil), stdout, bytes.NewBuffer(nil))) + task, err := container.NewTask(ctx, cio.NewCreator(withByteBuffers(stdout))) if err != nil { t.Error(err) return @@ -1043,7 +1090,7 @@ func testUserNamespaces(t *testing.T, readonlyRootFS bool) { } defer container.Delete(ctx, WithSnapshotCleanup) - task, err := container.NewTask(ctx, cio.Stdio, func(_ context.Context, client *Client, r *TaskInfo) error { + task, err := container.NewTask(ctx, cio.NewCreator(cio.WithStdio), func(_ context.Context, client *Client, r *TaskInfo) error { r.Options = &runctypes.CreateOptions{ IoUid: 1000, IoGid: 1000, diff --git a/container_test.go b/container_test.go index 3a1324bfc..8513b0469 100644 --- a/container_test.go +++ b/container_test.go @@ -3,6 +3,7 @@ package containerd import ( "bytes" "context" + "io" "io/ioutil" "os" "runtime" @@ -23,11 +24,11 @@ import ( gogotypes "github.com/gogo/protobuf/types" ) -func empty() cio.Creation { +func empty() cio.Creator { // TODO (@mlaventure) windows searches for pipes // when none are provided if runtime.GOOS == "windows" { - return cio.Stdio + return cio.NewCreator(cio.WithStdio) } return cio.NullIO } @@ -187,7 +188,7 @@ func TestContainerOutput(t *testing.T) { defer container.Delete(ctx, WithSnapshotCleanup) stdout := bytes.NewBuffer(nil) - task, err := container.NewTask(ctx, cio.NewIO(bytes.NewBuffer(nil), stdout, bytes.NewBuffer(nil))) + task, err := container.NewTask(ctx, cio.NewCreator(withByteBuffers(stdout))) if err != nil { t.Error(err) return @@ -223,6 +224,15 @@ func TestContainerOutput(t *testing.T) { } } +func withByteBuffers(stdout io.Writer) cio.Opt { + // TODO: could this use ioutil.Discard? + return func(streams *cio.Streams) { + streams.Stdin = new(bytes.Buffer) + streams.Stdout = stdout + streams.Stderr = new(bytes.Buffer) + } +} + func TestContainerExec(t *testing.T) { t.Parallel() @@ -534,7 +544,7 @@ func TestContainerCloseIO(t *testing.T) { return } - task, err := container.NewTask(ctx, cio.NewIO(r, stdout, ioutil.Discard)) + task, err := container.NewTask(ctx, cio.NewCreator(cio.WithStreams(r, stdout, ioutil.Discard))) if err != nil { t.Error(err) return @@ -1145,7 +1155,7 @@ func TestContainerHostname(t *testing.T) { defer container.Delete(ctx, WithSnapshotCleanup) stdout := bytes.NewBuffer(nil) - task, err := container.NewTask(ctx, cio.NewIO(bytes.NewBuffer(nil), stdout, bytes.NewBuffer(nil))) + task, err := container.NewTask(ctx, cio.NewCreator(withByteBuffers(stdout))) if err != nil { t.Error(err) return diff --git a/task.go b/task.go index 28f49fa6c..121da9af5 100644 --- a/task.go +++ b/task.go @@ -123,7 +123,7 @@ type Task interface { // Resume the execution of the task Resume(context.Context) error // Exec creates a new process inside the task - Exec(context.Context, string, *specs.Process, cio.Creation) (Process, error) + Exec(context.Context, string, *specs.Process, cio.Creator) (Process, error) // Pids returns a list of system specific process ids inside the task Pids(context.Context) ([]ProcessInfo, error) // Checkpoint serializes the runtime and memory information of a task into an @@ -278,7 +278,7 @@ func (t *task) Delete(ctx context.Context, opts ...ProcessDeleteOpts) (*ExitStat return &ExitStatus{code: r.ExitStatus, exitedAt: r.ExitedAt}, nil } -func (t *task) Exec(ctx context.Context, id string, spec *specs.Process, ioCreate cio.Creation) (_ Process, err error) { +func (t *task) Exec(ctx context.Context, id string, spec *specs.Process, ioCreate cio.Creator) (_ Process, err error) { if id == "" { return nil, errors.Wrapf(errdefs.ErrInvalidArgument, "exec id must not be empty") }