0ct0pu5/moby
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Merge pull request #30344 from cpuguy83/refactor_attach

Browse source 
Refactor attach
This commit is contained in:
Alexander Morozov 2017-01-27 07:35:44 -08:00 committed by GitHub
commit 77ec3aa8f2
5 changed files with 304 additions and 223 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

188
container/container.go
View file

@ -31,7 +31,6 @@ import (
"github.com/docker/docker/opts"
"github.com/docker/docker/pkg/idtools"
"github.com/docker/docker/pkg/ioutils"
"github.com/docker/docker/pkg/promise"
"github.com/docker/docker/pkg/signal"
"github.com/docker/docker/pkg/symlink"
"github.com/docker/docker/restartmanager"
@ -58,13 +57,6 @@ var (
errInvalidNetwork = fmt.Errorf("invalid network settings while building port map info")
)
// DetachError is special error which returned in case of container detach.
type DetachError struct{}
func (DetachError) Error() string {
return "detached from container"
}
// CommonContainer holds the fields for a container which are
// applicable across all platforms supported by the daemon.
type CommonContainer struct {
@ -373,183 +365,17 @@ func (container *Container) GetExecIDs() []string {
return container.ExecCommands.List()
}
// Attach connects to the container's TTY, delegating to standard
// streams or websockets depending on the configuration.
func (container *Container) Attach(stdin io.ReadCloser, stdout io.Writer, stderr io.Writer, keys []byte) chan error {
// Attach connects to the container's stdio to the client streams
func (container *Container) Attach(cfg *stream.AttachConfig) chan error {
ctx := container.InitAttachContext()
return AttachStreams(ctx, container.StreamConfig, container.Config.OpenStdin, container.Config.StdinOnce, container.Config.Tty, stdin, stdout, stderr, keys)
}
// AttachStreams connects streams to a TTY.
// Used by exec too. Should this move somewhere else?
func AttachStreams(ctx context.Context, streamConfig *stream.Config, openStdin, stdinOnce, tty bool, stdin io.ReadCloser, stdout io.Writer, stderr io.Writer, keys []byte) chan error {
var (
cStdout, cStderr io.ReadCloser
cStdin io.WriteCloser
wg sync.WaitGroup
errors = make(chan error, 3)
)
if stdin != nil && openStdin {
cStdin = streamConfig.StdinPipe()
wg.Add(1)
cfg.TTY = container.Config.Tty
if !container.Config.OpenStdin {
cfg.Stdin = nil
}
cfg.CloseStdin = cfg.Stdin != nil && container.Config.StdinOnce
if stdout != nil {
cStdout = streamConfig.StdoutPipe()
wg.Add(1)
}
if stderr != nil {
cStderr = streamConfig.StderrPipe()
wg.Add(1)
}
// Connect stdin of container to the http conn.
go func() {
if stdin == nil || !openStdin {
return
}
logrus.Debug("attach: stdin: begin")
var err error
if tty {
_, err = copyEscapable(cStdin, stdin, keys)
} else {
_, err = io.Copy(cStdin, stdin)
}
if err == io.ErrClosedPipe {
err = nil
}
if err != nil {
logrus.Errorf("attach: stdin: %s", err)
errors <- err
}
if stdinOnce && !tty {
cStdin.Close()
} else {
// No matter what, when stdin is closed (io.Copy unblock), close stdout and stderr
if cStdout != nil {
cStdout.Close()
}
if cStderr != nil {
cStderr.Close()
}
}
logrus.Debug("attach: stdin: end")
wg.Done()
}()
attachStream := func(name string, stream io.Writer, streamPipe io.ReadCloser) {
if stream == nil {
return
}
logrus.Debugf("attach: %s: begin", name)
_, err := io.Copy(stream, streamPipe)
if err == io.ErrClosedPipe {
err = nil
}
if err != nil {
logrus.Errorf("attach: %s: %v", name, err)
errors <- err
}
// Make sure stdin gets closed
if stdin != nil {
stdin.Close()
}
streamPipe.Close()
logrus.Debugf("attach: %s: end", name)
wg.Done()
}
go attachStream("stdout", stdout, cStdout)
go attachStream("stderr", stderr, cStderr)
return promise.Go(func() error {
done := make(chan struct{})
go func() {
wg.Wait()
close(done)
}()
select {
case <-done:
case <-ctx.Done():
// close all pipes
if cStdin != nil {
cStdin.Close()
}
if cStdout != nil {
cStdout.Close()
}
if cStderr != nil {
cStderr.Close()
}
<-done
}
close(errors)
for err := range errors {
if err != nil {
return err
}
}
return nil
})
}
// Code c/c from io.Copy() modified to handle escape sequence
func copyEscapable(dst io.Writer, src io.ReadCloser, keys []byte) (written int64, err error) {
if len(keys) == 0 {
// Default keys : ctrl-p ctrl-q
keys = []byte{16, 17}
}
buf := make([]byte, 32*1024)
for {
nr, er := src.Read(buf)
if nr > 0 {
// ---- Docker addition
preservBuf := []byte{}
for i, key := range keys {
preservBuf = append(preservBuf, buf[0:nr]...)
if nr != 1 || buf[0] != key {
break
}
if i == len(keys)-1 {
src.Close()
return 0, DetachError{}
}
nr, er = src.Read(buf)
}
var nw int
var ew error
if len(preservBuf) > 0 {
nw, ew = dst.Write(preservBuf)
nr = len(preservBuf)
} else {
// ---- End of docker
nw, ew = dst.Write(buf[0:nr])
}
if nw > 0 {
written += int64(nw)
}
if ew != nil {
err = ew
break
}
if nr != nw {
err = io.ErrShortWrite
break
}
}
if er == io.EOF {
break
}
if er != nil {
err = er
break
}
}
return written, err
return container.StreamConfig.Attach(ctx, cfg)
}
// ShouldRestart decides whether the daemon should restart the container or not.

217
container/stream/attach.go Normal file
View file

@ -0,0 +1,217 @@
package stream
import (
"io"
"sync"
"golang.org/x/net/context"
"github.com/Sirupsen/logrus"
"github.com/docker/docker/pkg/promise"
)
var defaultEscapeSequence = []byte{16, 17} // ctrl-p, ctrl-q
// DetachError is special error which returned in case of container detach.
type DetachError struct{}
func (DetachError) Error() string {
return "detached from container"
}
// AttachConfig is the config struct used to attach a client to a stream's stdio
type AttachConfig struct {
// Tells the attach copier that the stream's stdin is a TTY and to look for
// escape sequences in stdin to detach from the stream.
// When true the escape sequence is not passed to the underlying stream
TTY bool
// Specifies the detach keys the client will be using
// Only useful when `TTY` is true
DetachKeys []byte
// CloseStdin signals that once done, stdin for the attached stream should be closed
// For example, this would close the attached container's stdin.
CloseStdin bool
// Provide client streams to wire up to
Stdin io.ReadCloser
Stdout, Stderr io.Writer
}
// Attach attaches the stream config to the streams specified in
// the AttachOptions
func (c *Config) Attach(ctx context.Context, cfg *AttachConfig) chan error {
var (
cStdout, cStderr io.ReadCloser
cStdin io.WriteCloser
wg sync.WaitGroup
errors = make(chan error, 3)
)
if cfg.Stdin != nil {
cStdin = c.StdinPipe()
wg.Add(1)
}
if cfg.Stdout != nil {
cStdout = c.StdoutPipe()
wg.Add(1)
}
if cfg.Stderr != nil {
cStderr = c.StderrPipe()
wg.Add(1)
}
// Connect stdin of container to the attach stdin stream.
go func() {
if cfg.Stdin == nil {
return
}
logrus.Debug("attach: stdin: begin")
var err error
if cfg.TTY {
_, err = copyEscapable(cStdin, cfg.Stdin, cfg.DetachKeys)
} else {
_, err = io.Copy(cStdin, cfg.Stdin)
}
if err == io.ErrClosedPipe {
err = nil
}
if err != nil {
logrus.Errorf("attach: stdin: %s", err)
errors <- err
}
if cfg.CloseStdin && !cfg.TTY {
cStdin.Close()
} else {
// No matter what, when stdin is closed (io.Copy unblock), close stdout and stderr
if cStdout != nil {
cStdout.Close()
}
if cStderr != nil {
cStderr.Close()
}
}
logrus.Debug("attach: stdin: end")
wg.Done()
}()
attachStream := func(name string, stream io.Writer, streamPipe io.ReadCloser) {
if stream == nil {
return
}
logrus.Debugf("attach: %s: begin", name)
_, err := io.Copy(stream, streamPipe)
if err == io.ErrClosedPipe {
err = nil
}
if err != nil {
logrus.Errorf("attach: %s: %v", name, err)
errors <- err
}
// Make sure stdin gets closed
if cfg.Stdin != nil {
cfg.Stdin.Close()
}
streamPipe.Close()
logrus.Debugf("attach: %s: end", name)
wg.Done()
}
go attachStream("stdout", cfg.Stdout, cStdout)
go attachStream("stderr", cfg.Stderr, cStderr)
return promise.Go(func() error {
done := make(chan struct{})
go func() {
wg.Wait()
close(done)
}()
select {
case <-done:
case <-ctx.Done():
// close all pipes
if cStdin != nil {
cStdin.Close()
}
if cStdout != nil {
cStdout.Close()
}
if cStderr != nil {
cStderr.Close()
}
<-done
}
close(errors)
for err := range errors {
if err != nil {
return err
}
}
return nil
})
}
// ttyProxy is used only for attaches with a TTY. It is used to proxy
// stdin keypresses from the underlying reader and look for the passed in
// escape key sequence to signal a detach.
type ttyProxy struct {
escapeKeys []byte
escapeKeyPos int
r io.Reader
}
func (r *ttyProxy) Read(buf []byte) (int, error) {
nr, err := r.r.Read(buf)
preserve := func() {
// this preserves the original key presses in the passed in buffer
nr += r.escapeKeyPos
preserve := make([]byte, 0, r.escapeKeyPos+len(buf))
preserve = append(preserve, r.escapeKeys[:r.escapeKeyPos]...)
preserve = append(preserve, buf...)
r.escapeKeyPos = 0
copy(buf[0:nr], preserve)
}
if nr != 1 || err != nil {
if r.escapeKeyPos > 0 {
preserve()
}
return nr, err
}
if buf[0] != r.escapeKeys[r.escapeKeyPos] {
if r.escapeKeyPos > 0 {
preserve()
}
return nr, nil
}
if r.escapeKeyPos == len(r.escapeKeys)-1 {
return 0, DetachError{}
}
// Looks like we've got an escape key, but we need to match again on the next
// read.
// Store the current escape key we found so we can look for the next one on
// the next read.
// Since this is an escape key, make sure we don't let the caller read it
// If later on we find that this is not the escape sequence, we'll add the
// keys back
r.escapeKeyPos++
return nr - r.escapeKeyPos, nil
}
func copyEscapable(dst io.Writer, src io.ReadCloser, keys []byte) (written int64, err error) {
if len(keys) == 0 {
keys = defaultEscapeSequence
}
pr := &ttyProxy{escapeKeys: keys, r: src}
defer src.Close()
return io.Copy(dst, pr)
}

2
container/stream/streams.go
View file

@ -62,6 +62,7 @@ func (c *Config) StdinPipe() io.WriteCloser {
// StdoutPipe creates a new io.ReadCloser with an empty bytes pipe.
// It adds this new out pipe to the Stdout broadcaster.
// This will block stdout if unconsumed.
func (c *Config) StdoutPipe() io.ReadCloser {
bytesPipe := ioutils.NewBytesPipe()
c.stdout.Add(bytesPipe)
@ -70,6 +71,7 @@ func (c *Config) StdoutPipe() io.ReadCloser {
// StderrPipe creates a new io.ReadCloser with an empty bytes pipe.
// It adds this new err pipe to the Stderr broadcaster.
// This will block stderr if unconsumed.
func (c *Config) StderrPipe() io.ReadCloser {
bytesPipe := ioutils.NewBytesPipe()
c.stderr.Add(bytesPipe)

107
daemon/attach.go
View file

@ -9,11 +9,20 @@ import (
"github.com/docker/docker/api/errors"
"github.com/docker/docker/api/types/backend"
"github.com/docker/docker/container"
"github.com/docker/docker/container/stream"
"github.com/docker/docker/daemon/logger"
"github.com/docker/docker/pkg/stdcopy"
"github.com/docker/docker/pkg/term"
)
type containerAttachConfig struct {
detachKeys []byte
stdin io.ReadCloser
stdout, stderr io.Writer
showHistory bool
stream bool
}
// ContainerAttach attaches to logs according to the config passed in. See ContainerAttachConfig.
func (daemon *Daemon) ContainerAttach(prefixOrName string, c *backend.ContainerAttachConfig) error {
keys := []byte{}
@ -45,20 +54,23 @@ func (daemon *Daemon) ContainerAttach(prefixOrName string, c *backend.ContainerA
outStream = stdcopy.NewStdWriter(outStream, stdcopy.Stdout)
}
var stdin io.ReadCloser
var stdout, stderr io.Writer
var cfg containerAttachConfig
if c.UseStdin {
stdin = inStream
cfg.stdin = inStream
}
if c.UseStdout {
stdout = outStream
cfg.stdout = outStream
}
if c.UseStderr {
stderr = errStream
cfg.stderr = errStream
}
if err := daemon.containerAttach(container, stdin, stdout, stderr, c.Logs, c.Stream, keys); err != nil {
cfg.showHistory = c.Logs
cfg.stream = c.Stream
cfg.detachKeys = keys
if err := daemon.containerAttach(container, &cfg); err != nil {
fmt.Fprintf(outStream, "Error attaching: %s\n", err)
}
return nil
@ -70,11 +82,20 @@ func (daemon *Daemon) ContainerAttachRaw(prefixOrName string, stdin io.ReadClose
if err != nil {
return err
}
return daemon.containerAttach(container, stdin, stdout, stderr, false, stream, nil)
cfg := &containerAttachConfig{
stdin: stdin,
stdout: stdout,
stderr: stderr,
stream: stream,
}
return daemon.containerAttach(container, cfg)
}
func (daemon *Daemon) containerAttach(c *container.Container, stdin io.ReadCloser, stdout, stderr io.Writer, logs, stream bool, keys []byte) error {
if logs {
func (daemon *Daemon) containerAttach(c *container.Container, cfg *containerAttachConfig) error {
stdin := cfg.stdin
stdout := cfg.stdout
stderr := cfg.stderr
if cfg.showHistory {
logDriver, err := daemon.getLogger(c)
if err != nil {
return err
@ -107,41 +128,47 @@ func (daemon *Daemon) containerAttach(c *container.Container, stdin io.ReadClose
daemon.LogContainerEvent(c, "attach")
//stream
if stream {
var stdinPipe io.ReadCloser
if stdin != nil {
r, w := io.Pipe()
go func() {
defer w.Close()
defer logrus.Debug("Closing buffered stdin pipe")
io.Copy(w, stdin)
}()
stdinPipe = r
}
if !cfg.stream {
return nil
}
waitChan := make(chan struct{})
if c.Config.StdinOnce && !c.Config.Tty {
go func() {
c.WaitStop(-1 * time.Second)
close(waitChan)
}()
}
var stdinPipe io.ReadCloser
if stdin != nil {
r, w := io.Pipe()
go func() {
defer w.Close()
defer logrus.Debug("Closing buffered stdin pipe")
io.Copy(w, stdin)
}()
stdinPipe = r
}
err := <-c.Attach(stdinPipe, stdout, stderr, keys)
if err != nil {
if _, ok := err.(container.DetachError); ok {
daemon.LogContainerEvent(c, "detach")
} else {
logrus.Errorf("attach failed with error: %v", err)
}
}
// If we are in stdinonce mode, wait for the process to end
// otherwise, simply return
if c.Config.StdinOnce && !c.Config.Tty {
waitChan := make(chan struct{})
if c.Config.StdinOnce && !c.Config.Tty {
defer func() {
<-waitChan
}()
go func() {
c.WaitStop(-1 * time.Second)
close(waitChan)
}()
}
aCfg := &stream.AttachConfig{
Stdin: stdinPipe,
Stdout: stdout,
Stderr: stderr,
DetachKeys: cfg.detachKeys,
}
err := <-c.Attach(aCfg)
if err != nil {
if _, ok := err.(stream.DetachError); ok {
daemon.LogContainerEvent(c, "detach")
} else {
logrus.Errorf("attach failed with error: %v", err)
}
}
return nil
}

13
daemon/exec.go
View file

@ -13,6 +13,7 @@ import (
"github.com/docker/docker/api/types"
"github.com/docker/docker/api/types/strslice"
"github.com/docker/docker/container"
"github.com/docker/docker/container/stream"
"github.com/docker/docker/daemon/exec"
"github.com/docker/docker/libcontainerd"
"github.com/docker/docker/pkg/pools"
@ -209,7 +210,15 @@ func (d *Daemon) ContainerExecStart(ctx context.Context, name string, stdin io.R
return err
}
attachErr := container.AttachStreams(ctx, ec.StreamConfig, ec.OpenStdin, true, ec.Tty, cStdin, cStdout, cStderr, ec.DetachKeys)
attachConfig := &stream.AttachConfig{
TTY: ec.Tty,
Stdin: cStdin,
Stdout: cStdout,
Stderr: cStderr,
DetachKeys: ec.DetachKeys,
CloseStdin: true,
}
attachErr := ec.StreamConfig.Attach(ctx, attachConfig)
systemPid, err := d.containerd.AddProcess(ctx, c.ID, name, p, ec.InitializeStdio)
if err != nil {
@ -233,7 +242,7 @@ func (d *Daemon) ContainerExecStart(ctx context.Context, name string, stdin io.R
return fmt.Errorf("context cancelled")
case err := <-attachErr:
if err != nil {
if _, ok := err.(container.DetachError); !ok {
if _, ok := err.(stream.DetachError); !ok {
return fmt.Errorf("exec attach failed with error: %v", err)
}
d.LogContainerEvent(c, "exec_detach")