Merge pull request #13322 from ncdc/fix-noninteractive-stdin-exec

Godeps/Godeps.json

@@ -224,7 +224,7 @@
 		},
 		{
 			"ImportPath": "github.com/fsouza/go-dockerclient",
-			"Rev": "42d06e2b125654477366c320dcea99107a86e9c2"
+			"Rev": "76fd6c68cf24c48ee6a2b25def997182a29f940e"
 		},
 		{
 			"ImportPath": "github.com/garyburd/redigo/internal",

Godeps/_workspace/src/github.com/fsouza/go-dockerclient/.travis.yml

@@ -3,6 +3,7 @@ sudo: false
 go:
   - 1.3.1
   - 1.4
+  - 1.5
   - tip
 env:
   - GOARCH=amd64

Godeps/_workspace/src/github.com/fsouza/go-dockerclient/AUTHORS

@@ -13,6 +13,7 @@ Brian Lalor <blalor@bravo5.org>
 Brian Palmer <brianp@instructure.com>
 Burke Libbey <burke@libbey.me>
 Carlos Diaz-Padron <cpadron@mozilla.com>
+Cesar Wong <cewong@redhat.com>
 Cezar Sa Espinola <cezar.sa@corp.globo.com>
 Cheah Chu Yeow <chuyeow@gmail.com>
 cheneydeng <cheneydeng@qq.com>
@@ -33,6 +34,7 @@ Fabio Rehm <fgrehm@gmail.com>
 Fatih Arslan <ftharsln@gmail.com>
 Flavia Missi <flaviamissi@gmail.com>
 Francisco Souza <f@souza.cc>
+Grégoire Delattre <gregoire.delattre@gmail.com>
 Guillermo Álvarez Fernández <guillermo@cientifico.net>
 He Simei <hesimei@zju.edu.cn>
 Ivan Mikushin <i.mikushin@gmail.com>
@@ -66,12 +68,14 @@ Paul Morie <pmorie@gmail.com>
 Paul Weil <pweil@redhat.com>
 Peter Edge <peter.edge@gmail.com>
 Peter Jihoon Kim <raingrove@gmail.com>
+Phil Lu <lu@stackengine.com>
 Philippe Lafoucrière <philippe.lafoucriere@tech-angels.com>
 Rafe Colton <rafael.colton@gmail.com>
 Rob Miller <rob@kalistra.com>
 Robert Williamson <williamson.robert@gmail.com>
 Salvador Gironès <salvadorgirones@gmail.com>
 Sam Rijs <srijs@airpost.net>
+Samuel Karp <skarp@amazon.com>
 Simon Eskildsen <sirup@sirupsen.com>
 Simon Menke <simon.menke@gmail.com>
 Skolos <skolos@gopherlab.com>

Godeps/_workspace/src/github.com/fsouza/go-dockerclient/auth.go

@@ -16,7 +16,8 @@ import (
 	"strings"
 )
 
-var AuthParseError error = errors.New("Failed to read authentication from dockercfg")
+// ErrCannotParseDockercfg is the error returned by NewAuthConfigurations when the dockercfg cannot be parsed.
+var ErrCannotParseDockercfg = errors.New("Failed to read authentication from dockercfg")
 
 // AuthConfiguration represents authentication options to use in the PushImage
 // method. It represents the authentication in the Docker index server.
@@ -33,6 +34,10 @@ type AuthConfigurations struct {
 	Configs map[string]AuthConfiguration `json:"configs"`
 }
 
+// AuthConfigurations119 is used to serialize a set of AuthConfigurations
+// for Docker API >= 1.19.
+type AuthConfigurations119 map[string]AuthConfiguration
+
 // dockerConfig represents a registry authentation configuration from the
 // .dockercfg file.
 type dockerConfig struct {
@@ -103,7 +108,7 @@ func authConfigs(confs map[string]dockerConfig) (*AuthConfigurations, error) {
 		}
 		userpass := strings.Split(string(data), ":")
 		if len(userpass) != 2 {
-			return nil, AuthParseError
+			return nil, ErrCannotParseDockercfg
 		}
 		c.Configs[reg] = AuthConfiguration{
 			Email:         conf.Email,
@@ -117,7 +122,7 @@ func authConfigs(confs map[string]dockerConfig) (*AuthConfigurations, error) {
 
 // AuthCheck validates the given credentials. It returns nil if successful.
 //
-// See https://goo.gl/vPoEfJ for more details.
+// See https://goo.gl/m2SleN for more details.
 func (c *Client) AuthCheck(conf *AuthConfiguration) error {
 	if conf == nil {
 		return fmt.Errorf("conf is nil")

Godeps/_workspace/src/github.com/fsouza/go-dockerclient/auth_test.go

@@ -41,7 +41,7 @@ func TestAuthBadConfig(t *testing.T) {
 	auth := base64.StdEncoding.EncodeToString([]byte("userpass"))
 	read := strings.NewReader(fmt.Sprintf(`{"docker.io":{"auth":"%s","email":"user@example.com"}}`, auth))
 	ac, err := NewAuthConfigurations(read)
-	if err != AuthParseError {
+	if err != ErrCannotParseDockercfg {
 		t.Errorf("Incorrect error returned %v\n", err)
 	}
 	if ac != nil {

Godeps/_workspace/src/github.com/fsouza/go-dockerclient/change.go

@@ -23,7 +23,7 @@ const (
 
 // Change represents a change in a container.
 //
-// See http://goo.gl/QkW9sH for more details.
+// See https://goo.gl/9GsTIF for more details.
 type Change struct {
 	Path string
 	Kind ChangeType

Godeps/_workspace/src/github.com/fsouza/go-dockerclient/client.go

@@ -4,7 +4,7 @@
 
 // Package docker provides a client for the Docker remote API.
 //
-// See http://goo.gl/G3plxW for more details on the remote API.
+// See https://goo.gl/G3plxW for more details on the remote API.
 package docker
 
 import (
@@ -45,6 +45,8 @@ var (
 	ErrConnectionRefused = errors.New("cannot connect to Docker endpoint")
 
 	apiVersion112, _ = NewAPIVersion("1.12")
+
+	apiVersion119, _ = NewAPIVersion("1.19")
 )
 
 // APIVersion is an internal representation of a version of the Remote API.
@@ -326,7 +328,7 @@ func (c *Client) checkAPIVersion() error {
 
 // Ping pings the docker server
 //
-// See http://goo.gl/stJENm for more details.
+// See https://goo.gl/kQCfJj for more details.
 func (c *Client) Ping() error {
 	path := "/_ping"
 	body, status, err := c.do("GET", path, doOptions{})
@@ -462,9 +464,13 @@ func (c *Client) stream(method, path string, streamOptions streamOptions) error
 	address := c.endpointURL.Path
 	if streamOptions.stdout == nil {
 		streamOptions.stdout = ioutil.Discard
+	} else if t, ok := streamOptions.stdout.(io.Closer); ok {
+		defer t.Close()
 	}
 	if streamOptions.stderr == nil {
 		streamOptions.stderr = ioutil.Discard
+	} else if t, ok := streamOptions.stderr.(io.Closer); ok {
+		defer t.Close()
 	}
 	if protocol == "unix" {
 		dial, err := net.Dial(protocol, address)
@@ -583,6 +589,8 @@ func (c *Client) hijack(method, path string, hijackOptions hijackOptions) error
 		return err
 	}
 	req.Header.Set("Content-Type", "plain/text")
+	req.Header.Set("Connection", "Upgrade")
+	req.Header.Set("Upgrade", "tcp")
 	protocol := c.endpointURL.Scheme
 	address := c.endpointURL.Path
 	if protocol != "unix" {
@@ -612,13 +620,16 @@ func (c *Client) hijack(method, path string, hijackOptions hijackOptions) error
 	defer rwc.Close()
 	errChanOut := make(chan error, 1)
 	errChanIn := make(chan error, 1)
-	exit := make(chan bool)
 	go func() {
-		defer close(exit)
-		defer close(errChanOut)
+		defer func() {
+			if hijackOptions.in != nil {
+				if closer, ok := hijackOptions.in.(io.Closer); ok {
+					closer.Close()
+				}
+			}
+		}()
 		var err error
 		if hijackOptions.setRawTerminal {
-			// When TTY is ON, use regular copy
 			_, err = io.Copy(hijackOptions.stdout, br)
 		} else {
 			_, err = stdcopy.StdCopy(hijackOptions.stdout, hijackOptions.stderr, br)
@@ -626,17 +637,15 @@ func (c *Client) hijack(method, path string, hijackOptions hijackOptions) error
 		errChanOut <- err
 	}()
 	go func() {
+		var err error
 		if hijackOptions.in != nil {
-			_, err := io.Copy(rwc, hijackOptions.in)
-			errChanIn <- err
-		} else {
-			errChanIn <- nil
+			_, err = io.Copy(rwc, hijackOptions.in)
 		}
+		errChanIn <- err
 		rwc.(interface {
 			CloseWrite() error
 		}).CloseWrite()
 	}()
-	<-exit
 	errIn := <-errChanIn
 	errOut := <-errChanOut
 	if errIn != nil {

Godeps/_workspace/src/github.com/fsouza/go-dockerclient/container_test.go

@@ -1122,10 +1122,7 @@ func TestAttachToContainerRawTerminalFalse(t *testing.T) {
 		Stream:       true,
 		RawTerminal:  false,
 	}
-	err := client.AttachToContainer(opts)
-	if err != nil {
-		t.Fatal(err)
-	}
+	client.AttachToContainer(opts)
 	expected := map[string][]string{
 		"stdin":  {"1"},
 		"stdout": {"1"},

Godeps/_workspace/src/github.com/fsouza/go-dockerclient/exec.go

@@ -2,8 +2,6 @@
 // Use of this source code is governed by a BSD-style
 // license that can be found in the LICENSE file.
 
-// Docs can currently be found at https://github.com/docker/docker/blob/master/docs/sources/reference/api/docker_remote_api_v1.15.md#exec-create
-
 package docker
 
 import (
@@ -15,9 +13,15 @@ import (
 	"strconv"
 )
 
+// Exec is the type representing a `docker exec` instance and containing the
+// instance ID
+type Exec struct {
+	ID string `json:"Id,omitempty" yaml:"Id,omitempty"`
+}
+
 // CreateExecOptions specify parameters to the CreateExecContainer function.
 //
-// See http://goo.gl/8izrzI for more details
+// See https://goo.gl/1KSIb7 for more details
 type CreateExecOptions struct {
 	AttachStdin  bool     `json:"AttachStdin,omitempty" yaml:"AttachStdin,omitempty"`
 	AttachStdout bool     `json:"AttachStdout,omitempty" yaml:"AttachStdout,omitempty"`
@@ -28,9 +32,31 @@ type CreateExecOptions struct {
 	User         string   `json:"User,omitempty" yaml:"User,omitempty"`
 }
 
+// CreateExec sets up an exec instance in a running container `id`, returning the exec
+// instance, or an error in case of failure.
+//
+// See https://goo.gl/1KSIb7 for more details
+func (c *Client) CreateExec(opts CreateExecOptions) (*Exec, error) {
+	path := fmt.Sprintf("/containers/%s/exec", opts.Container)
+	body, status, err := c.do("POST", path, doOptions{data: opts})
+	if status == http.StatusNotFound {
+		return nil, &NoSuchContainer{ID: opts.Container}
+	}
+	if err != nil {
+		return nil, err
+	}
+	var exec Exec
+	err = json.Unmarshal(body, &exec)
+	if err != nil {
+		return nil, err
+	}
+
+	return &exec, nil
+}
+
 // StartExecOptions specify parameters to the StartExecContainer function.
 //
-// See http://goo.gl/JW8Lxl for more details
+// See https://goo.gl/iQCnto for more details
 type StartExecOptions struct {
 	Detach bool `json:"Detach,omitempty" yaml:"Detach,omitempty"`
 
@@ -51,67 +77,11 @@ type StartExecOptions struct {
 	Success chan struct{} `json:"-"`
 }
 
-// Exec is the type representing a `docker exec` instance and containing the
-// instance ID
-type Exec struct {
-	ID string `json:"Id,omitempty" yaml:"Id,omitempty"`
-}
-
-// ExecProcessConfig is a type describing the command associated to a Exec
-// instance. It's used in the ExecInspect type.
-//
-// See http://goo.gl/ypQULN for more details
-type ExecProcessConfig struct {
-	Privileged bool     `json:"privileged,omitempty" yaml:"privileged,omitempty"`
-	User       string   `json:"user,omitempty" yaml:"user,omitempty"`
-	Tty        bool     `json:"tty,omitempty" yaml:"tty,omitempty"`
-	EntryPoint string   `json:"entrypoint,omitempty" yaml:"entrypoint,omitempty"`
-	Arguments  []string `json:"arguments,omitempty" yaml:"arguments,omitempty"`
-}
-
-// ExecInspect is a type with details about a exec instance, including the
-// exit code if the command has finished running. It's returned by a api
-// call to /exec/(id)/json
-//
-// See http://goo.gl/ypQULN for more details
-type ExecInspect struct {
-	ID            string            `json:"ID,omitempty" yaml:"ID,omitempty"`
-	Running       bool              `json:"Running,omitempty" yaml:"Running,omitempty"`
-	ExitCode      int               `json:"ExitCode,omitempty" yaml:"ExitCode,omitempty"`
-	OpenStdin     bool              `json:"OpenStdin,omitempty" yaml:"OpenStdin,omitempty"`
-	OpenStderr    bool              `json:"OpenStderr,omitempty" yaml:"OpenStderr,omitempty"`
-	OpenStdout    bool              `json:"OpenStdout,omitempty" yaml:"OpenStdout,omitempty"`
-	ProcessConfig ExecProcessConfig `json:"ProcessConfig,omitempty" yaml:"ProcessConfig,omitempty"`
-	Container     Container         `json:"Container,omitempty" yaml:"Container,omitempty"`
-}
-
-// CreateExec sets up an exec instance in a running container `id`, returning the exec
-// instance, or an error in case of failure.
-//
-// See http://goo.gl/8izrzI for more details
-func (c *Client) CreateExec(opts CreateExecOptions) (*Exec, error) {
-	path := fmt.Sprintf("/containers/%s/exec", opts.Container)
-	body, status, err := c.do("POST", path, doOptions{data: opts})
-	if status == http.StatusNotFound {
-		return nil, &NoSuchContainer{ID: opts.Container}
-	}
-	if err != nil {
-		return nil, err
-	}
-	var exec Exec
-	err = json.Unmarshal(body, &exec)
-	if err != nil {
-		return nil, err
-	}
-
-	return &exec, nil
-}
-
 // StartExec starts a previously set up exec instance id. If opts.Detach is
 // true, it returns after starting the exec command. Otherwise, it sets up an
 // interactive session with the exec command.
 //
-// See http://goo.gl/JW8Lxl for more details
+// See https://goo.gl/iQCnto for more details
 func (c *Client) StartExec(id string, opts StartExecOptions) error {
 	if id == "" {
 		return &NoSuchExec{ID: id}
@@ -144,7 +114,7 @@ func (c *Client) StartExec(id string, opts StartExecOptions) error {
 // is valid only if Tty was specified as part of creating and starting the exec
 // command.
 //
-// See http://goo.gl/YDSx1f for more details
+// See https://goo.gl/e1JpsA for more details
 func (c *Client) ResizeExecTTY(id string, height, width int) error {
 	params := make(url.Values)
 	params.Set("h", strconv.Itoa(height))
@@ -155,9 +125,35 @@ func (c *Client) ResizeExecTTY(id string, height, width int) error {
 	return err
 }
 
+// ExecProcessConfig is a type describing the command associated to a Exec
+// instance. It's used in the ExecInspect type.
+type ExecProcessConfig struct {
+	Privileged bool     `json:"privileged,omitempty" yaml:"privileged,omitempty"`
+	User       string   `json:"user,omitempty" yaml:"user,omitempty"`
+	Tty        bool     `json:"tty,omitempty" yaml:"tty,omitempty"`
+	EntryPoint string   `json:"entrypoint,omitempty" yaml:"entrypoint,omitempty"`
+	Arguments  []string `json:"arguments,omitempty" yaml:"arguments,omitempty"`
+}
+
+// ExecInspect is a type with details about a exec instance, including the
+// exit code if the command has finished running. It's returned by a api
+// call to /exec/(id)/json
+//
+// See https://goo.gl/gPtX9R for more details
+type ExecInspect struct {
+	ID            string            `json:"ID,omitempty" yaml:"ID,omitempty"`
+	Running       bool              `json:"Running,omitempty" yaml:"Running,omitempty"`
+	ExitCode      int               `json:"ExitCode,omitempty" yaml:"ExitCode,omitempty"`
+	OpenStdin     bool              `json:"OpenStdin,omitempty" yaml:"OpenStdin,omitempty"`
+	OpenStderr    bool              `json:"OpenStderr,omitempty" yaml:"OpenStderr,omitempty"`
+	OpenStdout    bool              `json:"OpenStdout,omitempty" yaml:"OpenStdout,omitempty"`
+	ProcessConfig ExecProcessConfig `json:"ProcessConfig,omitempty" yaml:"ProcessConfig,omitempty"`
+	Container     Container         `json:"Container,omitempty" yaml:"Container,omitempty"`
+}
+
 // InspectExec returns low-level information about the exec command id.
 //
-// See http://goo.gl/ypQULN for more details
+// See https://goo.gl/gPtX9R for more details
 func (c *Client) InspectExec(id string) (*ExecInspect, error) {
 	path := fmt.Sprintf("/exec/%s/json", id)
 	body, status, err := c.do("GET", path, doOptions{})

Godeps/_workspace/src/github.com/fsouza/go-dockerclient/image_test.go

@@ -21,11 +21,13 @@ import (
 func newTestClient(rt *FakeRoundTripper) Client {
 	endpoint := "http://localhost:4243"
 	u, _ := parseEndpoint("http://localhost:4243", false)
+	testAPIVersion, _ := NewAPIVersion("1.17")
 	client := Client{
 		HTTPClient:             &http.Client{Transport: rt},
 		endpoint:               endpoint,
 		endpointURL:            u,
 		SkipServerVersionCheck: true,
+		serverAPIVersion:       testAPIVersion,
 	}
 	return client
 }

Godeps/_workspace/src/github.com/fsouza/go-dockerclient/misc.go

@@ -11,7 +11,7 @@ import (
 
 // Version returns version information about the docker server.
 //
-// See http://goo.gl/BOZrF5 for more details.
+// See https://goo.gl/ND9R8L for more details.
 func (c *Client) Version() (*Env, error) {
 	body, _, err := c.do("GET", "/version", doOptions{})
 	if err != nil {
@@ -26,7 +26,7 @@ func (c *Client) Version() (*Env, error) {
 
 // Info returns system-wide information about the Docker server.
 //
-// See http://goo.gl/wmqZsW for more details.
+// See https://goo.gl/ElTHi2 for more details.
 func (c *Client) Info() (*Env, error) {
 	body, _, err := c.do("GET", "/info", doOptions{})
 	if err != nil {

Godeps/_workspace/src/github.com/fsouza/go-dockerclient/testing/server.go

@@ -12,6 +12,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io/ioutil"
 	mathrand "math/rand"
 	"net"
 	"net/http"
@@ -532,7 +533,7 @@ func (s *DockerServer) startContainer(w http.ResponseWriter, r *http.Request) {
 	}
 	container.HostConfig = &hostConfig
 	if container.State.Running {
-		http.Error(w, "Container already running", http.StatusBadRequest)
+		http.Error(w, "", http.StatusNotModified)
 		return
 	}
 	container.State.Running = true
@@ -610,14 +611,34 @@ func (s *DockerServer) attachContainer(w http.ResponseWriter, r *http.Request) {
 		http.Error(w, err.Error(), http.StatusInternalServerError)
 		return
 	}
+	wg := sync.WaitGroup{}
+	if r.URL.Query().Get("stdin") == "1" {
+		wg.Add(1)
+		go func() {
+			ioutil.ReadAll(conn)
+			wg.Done()
+		}()
+	}
 	outStream := stdcopy.NewStdWriter(conn, stdcopy.Stdout)
 	if container.State.Running {
-		fmt.Fprintf(outStream, "Container %q is running\n", container.ID)
+		fmt.Fprintf(outStream, "Container is running\n")
 	} else {
-		fmt.Fprintf(outStream, "Container %q is not running\n", container.ID)
+		fmt.Fprintf(outStream, "Container is not running\n")
 	}
 	fmt.Fprintln(outStream, "What happened?")
 	fmt.Fprintln(outStream, "Something happened")
+	wg.Wait()
+	if r.URL.Query().Get("stream") == "1" {
+		for {
+			time.Sleep(1e6)
+			s.cMut.RLock()
+			if !container.State.Running {
+				s.cMut.RUnlock()
+				break
+			}
+			s.cMut.RUnlock()
+		}
+	}
 	conn.Close()
 }
 

Godeps/_workspace/src/github.com/fsouza/go-dockerclient/testing/server_test.go

@@ -5,9 +5,11 @@
 package testing
 
 import (
+	"bufio"
 	"bytes"
 	"encoding/json"
 	"fmt"
+	"io/ioutil"
 	"math/rand"
 	"net"
 	"net/http"
@@ -624,8 +626,8 @@ func TestStartContainerAlreadyRunning(t *testing.T) {
 	path := fmt.Sprintf("/containers/%s/start", server.containers[0].ID)
 	request, _ := http.NewRequest("POST", path, bytes.NewBuffer([]byte("null")))
 	server.ServeHTTP(recorder, request)
-	if recorder.Code != http.StatusBadRequest {
-		t.Errorf("StartContainer: wrong status code. Want %d. Got %d.", http.StatusBadRequest, recorder.Code)
+	if recorder.Code != http.StatusNotModified {
+		t.Errorf("StartContainer: wrong status code. Want %d. Got %d.", http.StatusNotModified, recorder.Code)
 	}
 }
 
@@ -845,22 +847,41 @@ func TestWaitContainerNotFound(t *testing.T) {
 	}
 }
 
+type HijackableResponseRecorder struct {
+	httptest.ResponseRecorder
+	readCh chan []byte
+}
+
+func (r *HijackableResponseRecorder) Hijack() (net.Conn, *bufio.ReadWriter, error) {
+	myConn, otherConn := net.Pipe()
+	r.readCh = make(chan []byte)
+	go func() {
+		data, _ := ioutil.ReadAll(myConn)
+		r.readCh <- data
+	}()
+	return otherConn, nil, nil
+}
+
+func (r *HijackableResponseRecorder) HijackBuffer() string {
+	return string(<-r.readCh)
+}
+
 func TestAttachContainer(t *testing.T) {
 	server := DockerServer{}
 	addContainers(&server, 1)
 	server.containers[0].State.Running = true
 	server.buildMuxer()
-	recorder := httptest.NewRecorder()
+	recorder := &HijackableResponseRecorder{}
 	path := fmt.Sprintf("/containers/%s/attach?logs=1", server.containers[0].ID)
 	request, _ := http.NewRequest("POST", path, nil)
 	server.ServeHTTP(recorder, request)
 	lines := []string{
-		fmt.Sprintf("\x01\x00\x00\x00\x03\x00\x00\x00Container %q is running", server.containers[0].ID),
-		"What happened?",
-		"Something happened",
+		"\x01\x00\x00\x00\x00\x00\x00\x15Container is running",
+		"\x01\x00\x00\x00\x00\x00\x00\x0fWhat happened?",
+		"\x01\x00\x00\x00\x00\x00\x00\x13Something happened",
 	}
 	expected := strings.Join(lines, "\n") + "\n"
-	if body := recorder.Body.String(); body == expected {
+	if body := recorder.HijackBuffer(); body != expected {
 		t.Errorf("AttachContainer: wrong body. Want %q. Got %q.", expected, body)
 	}
 }
@@ -868,7 +889,7 @@ func TestAttachContainer(t *testing.T) {
 func TestAttachContainerNotFound(t *testing.T) {
 	server := DockerServer{}
 	server.buildMuxer()
-	recorder := httptest.NewRecorder()
+	recorder := &HijackableResponseRecorder{}
 	path := "/containers/abc123/attach?logs=1"
 	request, _ := http.NewRequest("POST", path, nil)
 	server.ServeHTTP(recorder, request)
@@ -877,6 +898,44 @@ func TestAttachContainerNotFound(t *testing.T) {
 	}
 }
 
+func TestAttachContainerWithStreamBlocks(t *testing.T) {
+	server := DockerServer{}
+	addContainers(&server, 1)
+	server.containers[0].State.Running = true
+	server.buildMuxer()
+	path := fmt.Sprintf("/containers/%s/attach?logs=1&stdout=1&stream=1", server.containers[0].ID)
+	request, _ := http.NewRequest("POST", path, nil)
+	done := make(chan string)
+	go func() {
+		recorder := &HijackableResponseRecorder{}
+		server.ServeHTTP(recorder, request)
+		done <- recorder.HijackBuffer()
+	}()
+	select {
+	case <-done:
+		t.Fatalf("attach stream returned before container is stopped")
+	case <-time.After(500 * time.Millisecond):
+	}
+	server.cMut.Lock()
+	server.containers[0].State.Running = false
+	server.cMut.Unlock()
+	var body string
+	select {
+	case body = <-done:
+	case <-time.After(5 * time.Second):
+		t.Fatalf("timed out waiting for attach to finish")
+	}
+	lines := []string{
+		"\x01\x00\x00\x00\x00\x00\x00\x15Container is running",
+		"\x01\x00\x00\x00\x00\x00\x00\x0fWhat happened?",
+		"\x01\x00\x00\x00\x00\x00\x00\x13Something happened",
+	}
+	expected := strings.Join(lines, "\n") + "\n"
+	if body != expected {
+		t.Errorf("AttachContainer: wrong body. Want %q. Got %q.", expected, body)
+	}
+}
+
 func TestRemoveContainer(t *testing.T) {
 	server := DockerServer{}
 	addContainers(&server, 1)
@@ -1690,7 +1749,7 @@ func addNetworks(server *DockerServer, n int) {
 			ID:   fmt.Sprintf("%x", rand.Int()%10000),
 			Type: "bridge",
 			Endpoints: []*docker.Endpoint{
-				&docker.Endpoint{
+				{
 					Name:    "blah",
 					ID:      fmt.Sprintf("%x", rand.Int()%10000),
 					Network: netid,

Godeps/_workspace/src/github.com/fsouza/go-dockerclient/volume.go

+// Copyright 2015 go-dockerclient authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package docker
+
+import (
+	"encoding/json"
+	"errors"
+	"net/http"
+)
+
+var (
+	// ErrNoSuchVolume is the error returned when the volume does not exist.
+	ErrNoSuchVolume = errors.New("no such volume")
+
+	// ErrVolumeInUse is the error returned when the volume requested to be removed is still in use.
+	ErrVolumeInUse = errors.New("volume in use and cannot be removed")
+)
+
+// Volume represents a volume.
+//
+// See https://goo.gl/FZA4BK for more details.
+type Volume struct {
+	Name       string `json:"Name" yaml:"Name"`
+	Driver     string `json:"Driver,omitempty" yaml:"Driver,omitempty"`
+	Mountpoint string `json:"Mountpoint,omitempty" yaml:"Mountpoint,omitempty"`
+}
+
+// ListVolumesOptions specify parameters to the ListVolumes function.
+//
+// See https://goo.gl/FZA4BK for more details.
+type ListVolumesOptions struct {
+	Filters map[string][]string
+}
+
+// ListVolumes returns a list of available volumes in the server.
+//
+// See https://goo.gl/FZA4BK for more details.
+func (c *Client) ListVolumes(opts ListVolumesOptions) ([]Volume, error) {
+	body, _, err := c.do("GET", "/volumes?"+queryString(opts), doOptions{})
+	if err != nil {
+		return nil, err
+	}
+	m := make(map[string]interface{})
+	if err := json.Unmarshal(body, &m); err != nil {
+		return nil, err
+	}
+	var volumes []Volume
+	volumesJSON, ok := m["Volumes"]
+	if !ok {
+		return volumes, nil
+	}
+	data, err := json.Marshal(volumesJSON)
+	if err != nil {
+		return nil, err
+	}
+	if err := json.Unmarshal(data, &volumes); err != nil {
+		return nil, err
+	}
+	return volumes, nil
+}
+
+// CreateVolumeOptions specify parameters to the CreateVolume function.
+//
+// See https://goo.gl/pBUbZ9 for more details.
+type CreateVolumeOptions struct {
+	Name       string
+	Driver     string
+	DriverOpts map[string]string
+}
+
+// CreateVolume creates a volume on the server.
+//
+// See https://goo.gl/pBUbZ9 for more details.
+func (c *Client) CreateVolume(opts CreateVolumeOptions) (*Volume, error) {
+	body, _, err := c.do("POST", "/volumes", doOptions{data: opts})
+	if err != nil {
+		return nil, err
+	}
+	var volume Volume
+	if err := json.Unmarshal(body, &volume); err != nil {
+		return nil, err
+	}
+	return &volume, nil
+}
+
+// InspectVolume returns a volume by its name.
+//
+// See https://goo.gl/0g9A6i for more details.
+func (c *Client) InspectVolume(name string) (*Volume, error) {
+	body, status, err := c.do("GET", "/volumes/"+name, doOptions{})
+	if status == http.StatusNotFound {
+		return nil, ErrNoSuchVolume
+	}
+	if err != nil {
+		return nil, err
+	}
+	var volume Volume
+	if err := json.Unmarshal(body, &volume); err != nil {
+		return nil, err
+	}
+	return &volume, nil
+}
+
+// RemoveVolume removes a volume by its name.
+//
+// See https://goo.gl/79GNQz for more details.
+func (c *Client) RemoveVolume(name string) error {
+	_, status, err := c.do("DELETE", "/volumes/"+name, doOptions{})
+	if status == http.StatusNotFound {
+		return ErrNoSuchVolume
+	}
+	if status == http.StatusConflict {
+		return ErrVolumeInUse
+	}
+	return err
+}

Godeps/_workspace/src/github.com/fsouza/go-dockerclient/volume_test.go

+// Copyright 2015 go-dockerclient authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package docker
+
+import (
+	"encoding/json"
+	"net/http"
+	"net/url"
+	"reflect"
+	"testing"
+)
+
+func TestListVolumes(t *testing.T) {
+	volumesData := `[
+	{
+		"Name": "tardis",
+		"Driver": "local",
+		"Mountpoint": "/var/lib/docker/volumes/tardis"
+	},
+	{
+		"Name": "foo",
+		"Driver": "bar",
+		"Mountpoint": "/var/lib/docker/volumes/bar"
+	}
+]`
+	body := `{ "Volumes": ` + volumesData + ` }`
+	var expected []Volume
+	if err := json.Unmarshal([]byte(volumesData), &expected); err != nil {
+		t.Fatal(err)
+	}
+	client := newTestClient(&FakeRoundTripper{message: body, status: http.StatusOK})
+	volumes, err := client.ListVolumes(ListVolumesOptions{})
+	if err != nil {
+		t.Error(err)
+	}
+	if !reflect.DeepEqual(volumes, expected) {
+		t.Errorf("ListVolumes: Wrong return value. Want %#v. Got %#v.", expected, volumes)
+	}
+}
+
+func TestCreateVolume(t *testing.T) {
+	body := `{
+		"Name": "tardis",
+		"Driver": "local",
+		"Mountpoint": "/var/lib/docker/volumes/tardis"
+	}`
+	var expected Volume
+	if err := json.Unmarshal([]byte(body), &expected); err != nil {
+		t.Fatal(err)
+	}
+	fakeRT := &FakeRoundTripper{message: body, status: http.StatusOK}
+	client := newTestClient(fakeRT)
+	volume, err := client.CreateVolume(
+		CreateVolumeOptions{
+			Name:   "tardis",
+			Driver: "local",
+			DriverOpts: map[string]string{
+				"foo": "bar",
+			},
+		},
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !reflect.DeepEqual(volume, &expected) {
+		t.Errorf("CreateVolume: Wrong return value. Want %#v. Got %#v.", expected, volume)
+	}
+	req := fakeRT.requests[0]
+	expectedMethod := "POST"
+	if req.Method != expectedMethod {
+		t.Errorf("CreateVolume(): Wrong HTTP method. Want %s. Got %s.", expectedMethod, req.Method)
+	}
+	u, _ := url.Parse(client.getURL("/volumes"))
+	if req.URL.Path != u.Path {
+		t.Errorf("CreateVolume(): Wrong request path. Want %q. Got %q.", u.Path, req.URL.Path)
+	}
+}
+
+func TestInspectVolume(t *testing.T) {
+	body := `{
+		"Name": "tardis",
+		"Driver": "local",
+		"Mountpoint": "/var/lib/docker/volumes/tardis"
+	}`
+	var expected Volume
+	if err := json.Unmarshal([]byte(body), &expected); err != nil {
+		t.Fatal(err)
+	}
+	fakeRT := &FakeRoundTripper{message: body, status: http.StatusOK}
+	client := newTestClient(fakeRT)
+	name := "tardis"
+	volume, err := client.InspectVolume(name)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !reflect.DeepEqual(volume, &expected) {
+		t.Errorf("InspectVolume: Wrong return value. Want %#v. Got %#v.", expected, volume)
+	}
+	req := fakeRT.requests[0]
+	expectedMethod := "GET"
+	if req.Method != expectedMethod {
+		t.Errorf("InspectVolume(%q): Wrong HTTP method. Want %s. Got %s.", name, expectedMethod, req.Method)
+	}
+	u, _ := url.Parse(client.getURL("/volumes/" + name))
+	if req.URL.Path != u.Path {
+		t.Errorf("CreateVolume(%q): Wrong request path. Want %q. Got %q.", name, u.Path, req.URL.Path)
+	}
+}
+
+func TestRemoveVolume(t *testing.T) {
+	name := "test"
+	fakeRT := &FakeRoundTripper{message: "", status: http.StatusNoContent}
+	client := newTestClient(fakeRT)
+	if err := client.RemoveVolume(name); err != nil {
+		t.Fatal(err)
+	}
+	req := fakeRT.requests[0]
+	expectedMethod := "DELETE"
+	if req.Method != expectedMethod {
+		t.Errorf("RemoveVolume(%q): Wrong HTTP method. Want %s. Got %s.", name, expectedMethod, req.Method)
+	}
+	u, _ := url.Parse(client.getURL("/volumes/" + name))
+	if req.URL.Path != u.Path {
+		t.Errorf("RemoveVolume(%q): Wrong request path. Want %q. Got %q.", name, u.Path, req.URL.Path)
+	}
+}
+
+func TestRemoveVolumeNotFound(t *testing.T) {
+	client := newTestClient(&FakeRoundTripper{message: "no such volume", status: http.StatusNotFound})
+	if err := client.RemoveVolume("test:"); err != ErrNoSuchVolume {
+		t.Errorf("RemoveVolume: wrong error. Want %#v. Got %#v.", ErrNoSuchVolume, err)
+	}
+}
+
+func TestRemoveVolumeInUse(t *testing.T) {
+	client := newTestClient(&FakeRoundTripper{message: "volume in use and cannot be removed", status: http.StatusConflict})
+	if err := client.RemoveVolume("test:"); err != ErrVolumeInUse {
+		t.Errorf("RemoveVolume: wrong error. Want %#v. Got %#v.", ErrVolumeInUse, err)
+	}
+}

pkg/client/unversioned/remotecommand/remotecommand.go

@@ -21,12 +21,13 @@ import (
 	"io"
 	"io/ioutil"
 	"net/http"
+	"sync"
 
-	"github.com/golang/glog"
 	"k8s.io/kubernetes/pkg/api"
 	client "k8s.io/kubernetes/pkg/client/unversioned"
 	"k8s.io/kubernetes/pkg/conversion/queryparams"
 	"k8s.io/kubernetes/pkg/runtime"
+	"k8s.io/kubernetes/pkg/util"
 	"k8s.io/kubernetes/pkg/util/httpstream"
 	"k8s.io/kubernetes/pkg/util/httpstream/spdy"
 )
@@ -155,90 +156,110 @@ func (e *Streamer) doStream() error {
 	}
 	defer conn.Close()
 
-	doneChan := make(chan struct{}, 2)
-	errorChan := make(chan error)
-
-	cp := func(s string, dst io.Writer, src io.Reader) {
-		glog.V(4).Infof("Copying %s", s)
-		defer glog.V(4).Infof("Done copying %s", s)
-		if _, err := io.Copy(dst, src); err != nil && err != io.EOF {
-			glog.Errorf("Error copying %s: %v", s, err)
-		}
-		if s == api.StreamTypeStdout || s == api.StreamTypeStderr {
-			doneChan <- struct{}{}
-		}
-	}
-
 	headers := http.Header{}
+
+	// set up error stream
+	errorChan := make(chan error)
 	headers.Set(api.StreamType, api.StreamTypeError)
 	errorStream, err := conn.CreateStream(headers)
 	if err != nil {
 		return err
 	}
+
 	go func() {
 		message, err := ioutil.ReadAll(errorStream)
-		if err != nil && err != io.EOF {
-			errorChan <- fmt.Errorf("Error reading from error stream: %s", err)
-			return
-		}
-		if len(message) > 0 {
-			errorChan <- fmt.Errorf("Error executing remote command: %s", message)
-			return
+		switch {
+		case err != nil && err != io.EOF:
+			errorChan <- fmt.Errorf("error reading from error stream: %s", err)
+		case len(message) > 0:
+			errorChan <- fmt.Errorf("error executing remote command: %s", message)
+		default:
+			errorChan <- nil
 		}
+		close(errorChan)
 	}()
-	defer errorStream.Reset()
 
+	var wg sync.WaitGroup
+	var once sync.Once
+
+	// set up stdin stream
 	if e.stdin != nil {
 		headers.Set(api.StreamType, api.StreamTypeStdin)
 		remoteStdin, err := conn.CreateStream(headers)
 		if err != nil {
 			return err
 		}
-		defer remoteStdin.Reset()
-		// TODO this goroutine will never exit cleanly (the io.Copy never unblocks)
-		// because stdin is not closed until the process exits. If we try to call
-		// stdin.Close(), it returns no error but doesn't unblock the copy. It will
-		// exit when the process exits, instead.
-		go cp(api.StreamTypeStdin, remoteStdin, e.stdin)
+
+		// copy from client's stdin to container's stdin
+		go func() {
+			// if e.stdin is noninteractive, e.g. `echo abc | kubectl exec -i <pod> -- cat`, make sure
+			// we close remoteStdin as soon as the copy from e.stdin to remoteStdin finishes. Otherwise
+			// the executed command will remain running.
+			defer once.Do(func() { remoteStdin.Close() })
+
+			if _, err := io.Copy(remoteStdin, e.stdin); err != nil {
+				util.HandleError(err)
 			}
+		}()
 
-	waitCount := 0
-	completedStreams := 0
+		// read from remoteStdin until the stream is closed. this is essential to
+		// be able to exit interactive sessions cleanly and not leak goroutines or
+		// hang the client's terminal.
+		//
+		// go-dockerclient's current hijack implementation
+		// (https://github.com/fsouza/go-dockerclient/blob/89f3d56d93788dfe85f864a44f85d9738fca0670/client.go#L564)
+		// waits for all three streams (stdin/stdout/stderr) to finish copying
+		// before returning. When hijack finishes copying stdout/stderr, it calls
+		// Close() on its side of remoteStdin, which allows this copy to complete.
+		// When that happens, we must Close() on our side of remoteStdin, to
+		// allow the copy in hijack to complete, and hijack to return.
+		go func() {
+			defer once.Do(func() { remoteStdin.Close() })
+			// this "copy" doesn't actually read anything - it's just here to wait for
+			// the server to close remoteStdin.
+			if _, err := io.Copy(ioutil.Discard, remoteStdin); err != nil {
+				util.HandleError(err)
+			}
+		}()
+	}
 
+	// set up stdout stream
 	if e.stdout != nil {
-		waitCount++
 		headers.Set(api.StreamType, api.StreamTypeStdout)
 		remoteStdout, err := conn.CreateStream(headers)
 		if err != nil {
 			return err
 		}
-		defer remoteStdout.Reset()
-		go cp(api.StreamTypeStdout, e.stdout, remoteStdout)
+
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			if _, err := io.Copy(e.stdout, remoteStdout); err != nil {
+				util.HandleError(err)
+			}
+		}()
 	}
 
+	// set up stderr stream
 	if e.stderr != nil && !e.tty {
-		waitCount++
 		headers.Set(api.StreamType, api.StreamTypeStderr)
 		remoteStderr, err := conn.CreateStream(headers)
 		if err != nil {
 			return err
 		}
-		defer remoteStderr.Reset()
-		go cp(api.StreamTypeStderr, e.stderr, remoteStderr)
-	}
 
-Loop:
-	for {
-		select {
-		case <-doneChan:
-			completedStreams++
-			if completedStreams == waitCount {
-				break Loop
-			}
-		case err := <-errorChan:
-			return err
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			if _, err := io.Copy(e.stderr, remoteStderr); err != nil {
+				util.HandleError(err)
 			}
+		}()
 	}
 
-	return nil
+	// we're waiting for stdout/stderr to finish copying
+	wg.Wait()
+
+	// waits for errorStream to finish reading with an error or nil
+	return <-errorChan
 }

pkg/client/unversioned/remotecommand/remotecommand_test.go

@@ -19,7 +19,7 @@ package remotecommand
 import (
 	"bytes"
 	"fmt"
-	"io/ioutil"
+	"io"
 	"net/http"
 	"net/http/httptest"
 	"net/url"
@@ -32,7 +32,7 @@ import (
 	"k8s.io/kubernetes/pkg/util/httpstream/spdy"
 )
 
-func fakeExecServer(t *testing.T, i int, stdinData, stdoutData, stderrData, errorData string, tty bool) http.HandlerFunc {
+func fakeExecServer(t *testing.T, i int, stdinData, stdoutData, stderrData, errorData string, tty bool, messageCount int) http.HandlerFunc {
 	// error + stdin + stdout
 	expectedStreams := 3
 	if !tty {
@@ -70,7 +70,6 @@ func fakeExecServer(t *testing.T, i int, stdinData, stdoutData, stderrData, erro
 					receivedStreams++
 				case api.StreamTypeStdin:
 					stdinStream = stream
-					stdinStream.Close()
 					receivedStreams++
 				case api.StreamTypeStdout:
 					stdoutStream = stream
@@ -82,8 +81,6 @@ func fakeExecServer(t *testing.T, i int, stdinData, stdoutData, stderrData, erro
 					t.Errorf("%d: unexpected stream type: %q", i, streamType)
 				}
 
-				defer stream.Reset()
-
 				if receivedStreams == expectedStreams {
 					break WaitForStreams
 				}
@@ -91,27 +88,56 @@ func fakeExecServer(t *testing.T, i int, stdinData, stdoutData, stderrData, erro
 		}
 
 		if len(errorData) > 0 {
-			fmt.Fprint(errorStream, errorData)
+			n, err := fmt.Fprint(errorStream, errorData)
+			if err != nil {
+				t.Errorf("%d: error writing to errorStream: %v", i, err)
+			}
+			if e, a := len(errorData), n; e != a {
+				t.Errorf("%d: expected to write %d bytes to errorStream, but only wrote %d", i, e, a)
+			}
 			errorStream.Close()
 		}
 
 		if len(stdoutData) > 0 {
-			fmt.Fprint(stdoutStream, stdoutData)
+			for j := 0; j < messageCount; j++ {
+				n, err := fmt.Fprint(stdoutStream, stdoutData)
+				if err != nil {
+					t.Errorf("%d: error writing to stdoutStream: %v", i, err)
+				}
+				if e, a := len(stdoutData), n; e != a {
+					t.Errorf("%d: expected to write %d bytes to stdoutStream, but only wrote %d", i, e, a)
+				}
+			}
 			stdoutStream.Close()
 		}
 		if len(stderrData) > 0 {
-			fmt.Fprint(stderrStream, stderrData)
+			for j := 0; j < messageCount; j++ {
+				n, err := fmt.Fprint(stderrStream, stderrData)
+				if err != nil {
+					t.Errorf("%d: error writing to stderrStream: %v", i, err)
+				}
+				if e, a := len(stderrData), n; e != a {
+					t.Errorf("%d: expected to write %d bytes to stderrStream, but only wrote %d", i, e, a)
+				}
+			}
 			stderrStream.Close()
 		}
 		if len(stdinData) > 0 {
-			data, err := ioutil.ReadAll(stdinStream)
+			data := make([]byte, len(stdinData))
+			for j := 0; j < messageCount; j++ {
+				n, err := io.ReadFull(stdinStream, data)
 				if err != nil {
 					t.Errorf("%d: error reading stdin stream: %v", i, err)
 				}
+				if e, a := len(stdinData), n; e != a {
+					t.Errorf("%d: expected to read %d bytes from stdinStream, but only read %d", i, e, a)
+				}
 				if e, a := stdinData, string(data); e != a {
 					t.Errorf("%d: stdin: expected %q, got %q", i, e, a)
 				}
 			}
+			stdinStream.Close()
+		}
 	})
 }
 
@@ -122,6 +148,7 @@ func TestRequestExecuteRemoteCommand(t *testing.T) {
 		Stderr       string
 		Error        string
 		Tty          bool
+		MessageCount int
 	}{
 		{
 			Error: "bail",
@@ -130,6 +157,15 @@ func TestRequestExecuteRemoteCommand(t *testing.T) {
 			Stdin:  "a",
 			Stdout: "b",
 			Stderr: "c",
+			// TODO bump this to a larger number such as 100 once
+			// https://github.com/docker/spdystream/issues/55 is fixed and the Godep
+			// is bumped. Sending multiple messages over stdin/stdout/stderr results
+			// in more frames being spread across multiple spdystream frame workers.
+			// This makes it more likely that the spdystream bug will be encountered,
+			// where streams are closed as soon as a goaway frame is received, and
+			// any pending frames that haven't been processed yet may not be
+			// delivered (it's a race).
+			MessageCount: 1,
 		},
 		{
 			Stdin:  "a",
@@ -142,7 +178,7 @@ func TestRequestExecuteRemoteCommand(t *testing.T) {
 		localOut := &bytes.Buffer{}
 		localErr := &bytes.Buffer{}
 
-		server := httptest.NewServer(fakeExecServer(t, i, testCase.Stdin, testCase.Stdout, testCase.Stderr, testCase.Error, testCase.Tty))
+		server := httptest.NewServer(fakeExecServer(t, i, testCase.Stdin, testCase.Stdout, testCase.Stderr, testCase.Error, testCase.Tty, testCase.MessageCount))
 
 		url, _ := url.ParseRequestURI(server.URL)
 		c := client.NewRESTClient(url, "x", nil, -1, -1)
@@ -151,8 +187,7 @@ func TestRequestExecuteRemoteCommand(t *testing.T) {
 		conf := &client.Config{
 			Host: server.URL,
 		}
-		e := New(req, conf, []string{"ls", "/"}, strings.NewReader(testCase.Stdin), localOut, localErr, testCase.Tty)
-		//e.upgrader = testCase.Upgrader
+		e := New(req, conf, []string{"ls", "/"}, strings.NewReader(strings.Repeat(testCase.Stdin, testCase.MessageCount)), localOut, localErr, testCase.Tty)
 		err := e.Execute()
 		hasErr := err != nil
 
@@ -176,13 +211,13 @@ func TestRequestExecuteRemoteCommand(t *testing.T) {
 		}
 
 		if len(testCase.Stdout) > 0 {
-			if e, a := testCase.Stdout, localOut; e != a.String() {
+			if e, a := strings.Repeat(testCase.Stdout, testCase.MessageCount), localOut; e != a.String() {
 				t.Errorf("%d: expected stdout data '%s', got '%s'", i, e, a)
 			}
 		}
 
 		if testCase.Stderr != "" {
-			if e, a := testCase.Stderr, localErr; e != a.String() {
+			if e, a := strings.Repeat(testCase.Stderr, testCase.MessageCount), localErr; e != a.String() {
 				t.Errorf("%d: expected stderr data '%s', got '%s'", i, e, a)
 			}
 		}
@@ -219,7 +254,7 @@ func TestRequestAttachRemoteCommand(t *testing.T) {
 		localOut := &bytes.Buffer{}
 		localErr := &bytes.Buffer{}
 
-		server := httptest.NewServer(fakeExecServer(t, i, testCase.Stdin, testCase.Stdout, testCase.Stderr, testCase.Error, testCase.Tty))
+		server := httptest.NewServer(fakeExecServer(t, i, testCase.Stdin, testCase.Stdout, testCase.Stderr, testCase.Error, testCase.Tty, 1))
 
 		url, _ := url.ParseRequestURI(server.URL)
 		c := client.NewRESTClient(url, "x", nil, -1, -1)
@@ -229,7 +264,6 @@ func TestRequestAttachRemoteCommand(t *testing.T) {
 			Host: server.URL,
 		}
 		e := NewAttach(req, conf, strings.NewReader(testCase.Stdin), localOut, localErr, testCase.Tty)
-		//e.upgrader = testCase.Upgrader
 		err := e.Execute()
 		hasErr := err != nil
 

pkg/kubelet/server.go

@@ -543,7 +543,6 @@ WaitForStreams:
 			switch streamType {
 			case api.StreamTypeError:
 				errorStream = stream
-				defer errorStream.Reset()
 				receivedStreams++
 			case api.StreamTypeStdin:
 				stdinStream = stream
@@ -568,11 +567,6 @@ WaitForStreams:
 		}
 	}
 
-	if stdinStream != nil {
-		// close our half of the input stream, since we won't be writing to it
-		stdinStream.Close()
-	}
-
 	return stdinStream, stdoutStream, stderrStream, errorStream, conn, tty, true
 }
 

pkg/util/httpstream/spdy/connection.go

@@ -78,7 +78,7 @@ const createStreamResponseTimeout = 30 * time.Second
 func (c *connection) Close() error {
 	c.streamLock.Lock()
 	for _, s := range c.streams {
-		s.Reset()
+		s.Close()
 	}
 	c.streams = make([]httpstream.Stream, 0)
 	c.streamLock.Unlock()

test/e2e/kubectl.go

@@ -20,6 +20,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
 	"io/ioutil"
 	"net"
 	"net/http"
@@ -158,11 +159,35 @@ var _ = Describe("Kubectl client", func() {
 		It("should support exec", func() {
 			By("executing a command in the container")
 			execOutput := runKubectl("exec", fmt.Sprintf("--namespace=%v", ns), simplePodName, "echo", "running", "in", "container")
-			expectedExecOutput := "running in container"
-			if execOutput != expectedExecOutput {
-				Failf("Unexpected kubectl exec output. Wanted '%s', got '%s'", execOutput, expectedExecOutput)
+			if e, a := "running in container", execOutput; e != a {
+				Failf("Unexpected kubectl exec output. Wanted %q, got %q", e, a)
+			}
+
+			By("executing a command in the container with noninteractive stdin")
+			execOutput = newKubectlCommand("exec", fmt.Sprintf("--namespace=%v", ns), "-i", simplePodName, "cat").
+				withStdinData("abcd1234").
+				exec()
+			if e, a := "abcd1234", execOutput; e != a {
+				Failf("Unexpected kubectl exec output. Wanted %q, got %q", e, a)
+			}
+
+			// pretend that we're a user in an interactive shell
+			r, c, err := newBlockingReader("echo hi\nexit\n")
+			if err != nil {
+				Failf("Error creating blocking reader: %v", err)
+			}
+			// NOTE this is solely for test cleanup!
+			defer c.Close()
+
+			By("executing a command in the container with pseudo-interactive stdin")
+			execOutput = newKubectlCommand("exec", fmt.Sprintf("--namespace=%v", ns), "-i", simplePodName, "bash").
+				withStdinReader(r).
+				exec()
+			if e, a := "hi", execOutput; e != a {
+				Failf("Unexpected kubectl exec output. Wanted %q, got %q", e, a)
 			}
 		})
+
 		It("should support port-forward", func() {
 			By("forwarding the container port to a local port")
 			cmd := kubectlCmd("port-forward", fmt.Sprintf("--namespace=%v", ns), simplePodName, fmt.Sprintf(":%d", simplePodPort))
@@ -791,3 +816,20 @@ func getUDData(jpgExpected string, ns string) func(*client.Client, string) error
 		}
 	}
 }
+
+// newBlockingReader returns a reader that allows reading the given string,
+// then blocks until Close() is called on the returned closer.
+//
+// We're explicitly returning the reader and closer separately, because
+// the closer needs to be the *os.File we get from os.Pipe(). This is required
+// so the exec of kubectl can pass the underlying file descriptor to the exec
+// syscall, instead of creating another os.Pipe and blocking on the io.Copy
+// between the source (e.g. stdin) and the write half of the pipe.
+func newBlockingReader(s string) (io.Reader, io.Closer, error) {
+	r, w, err := os.Pipe()
+	if err != nil {
+		return nil, nil, err
+	}
+	w.Write([]byte(s))
+	return r, w, nil
+}

test/e2e/util.go

@@ -977,6 +977,11 @@ func (b kubectlBuilder) withStdinData(data string) *kubectlBuilder {
 	return &b
 }
 
+func (b kubectlBuilder) withStdinReader(reader io.Reader) *kubectlBuilder {
+	b.cmd.Stdin = reader
+	return &b
+}
+
 func (b kubectlBuilder) exec() string {
 	var stdout, stderr bytes.Buffer
 	cmd := b.cmd