Add long-running client session endpoint

Signed-off-by: Tonis Tiigi <[email protected]>

Author: tonistiigi

api/server/backend/build/backend.go

@@ -27,9 +27,9 @@ type Backend struct {
 }
 
 // NewBackend creates a new build backend from components
-func NewBackend(components ImageComponent, builderBackend builder.Backend, idMappings *idtools.IDMappings) *Backend {
-	manager := dockerfile.NewBuildManager(builderBackend, idMappings)
-	return &Backend{imageComponent: components, manager: manager}
+func NewBackend(components ImageComponent, builderBackend builder.Backend, sg dockerfile.SessionGetter, idMappings *idtools.IDMappings) (*Backend, error) {
+	manager := dockerfile.NewBuildManager(builderBackend, sg, idMappings)
+	return &Backend{imageComponent: components, manager: manager}, nil
 }
 
 // Build builds an image from a Source

api/server/router/build/build_routes.go

@@ -127,6 +127,7 @@ func newImageBuildOptions(ctx context.Context, r *http.Request) (*types.ImageBui
 		}
 		options.CacheFrom = cacheFrom
 	}
+	options.SessionID = r.FormValue("session")
 
 	return options, nil
 }

api/server/router/session/backend.go

@@ -0,0 +1,12 @@
+package session
+
+import (
+	"net/http"
+
+	"golang.org/x/net/context"
+)
+
+// Backend abstracts an session receiver from an http request.
+type Backend interface {
+	HandleHTTPRequest(ctx context.Context, w http.ResponseWriter, r *http.Request) error
+}

api/server/router/session/session.go

@@ -0,0 +1,29 @@
+package session
+
+import "github.com/docker/docker/api/server/router"
+
+// sessionRouter is a router to talk with the session controller
+type sessionRouter struct {
+	backend Backend
+	routes  []router.Route
+}
+
+// NewRouter initializes a new session router
+func NewRouter(b Backend) router.Router {
+	r := &sessionRouter{
+		backend: b,
+	}
+	r.initRoutes()
+	return r
+}
+
+// Routes returns the available routers to the session controller
+func (r *sessionRouter) Routes() []router.Route {
+	return r.routes
+}
+
+func (r *sessionRouter) initRoutes() {
+	r.routes = []router.Route{
+		router.Experimental(router.NewPostRoute("/session", r.startSession)),
+	}
+}

api/server/router/session/session_routes.go

@@ -0,0 +1,16 @@
+package session
+
+import (
+	"net/http"
+
+	apierrors "github.com/docker/docker/api/errors"
+	"golang.org/x/net/context"
+)
+
+func (sr *sessionRouter) startSession(ctx context.Context, w http.ResponseWriter, r *http.Request, vars map[string]string) error {
+	err := sr.backend.HandleHTTPRequest(ctx, w, r)
+	if err != nil {
+		return apierrors.NewBadRequestError(err)
+	}
+	return nil
+}

api/types/client.go

@@ -7,7 +7,7 @@ import (
 
 	"github.com/docker/docker/api/types/container"
 	"github.com/docker/docker/api/types/filters"
-	"github.com/docker/go-units"
+	units "github.com/docker/go-units"
 )
 
 // CheckpointCreateOptions holds parameters to create a checkpoint from a container
@@ -178,6 +178,7 @@ type ImageBuildOptions struct {
 	SecurityOpt []string
 	ExtraHosts  []string // List of extra hosts
 	Target      string
+	SessionID   string
 
 	// TODO @jhowardmsft LCOW Support: This will require extending to include
 	// `Platform string`, but is ommited for now as it's hard-coded temporarily

builder/dockerfile/builder.go

@@ -16,6 +16,7 @@ import (
 	"github.com/docker/docker/builder/dockerfile/command"
 	"github.com/docker/docker/builder/dockerfile/parser"
 	"github.com/docker/docker/builder/remotecontext"
+	"github.com/docker/docker/client/session"
 	"github.com/docker/docker/pkg/archive"
 	"github.com/docker/docker/pkg/chrootarchive"
 	"github.com/docker/docker/pkg/idtools"
@@ -40,18 +41,25 @@ var validCommitCommands = map[string]bool{
 	"workdir":     true,
 }
 
+// SessionGetter is object used to get access to a session by uuid
+type SessionGetter interface {
+	Get(ctx context.Context, uuid string) (session.Caller, error)
+}
+
 // BuildManager is shared across all Builder objects
 type BuildManager struct {
 	archiver  *archive.Archiver
 	backend   builder.Backend
 	pathCache pathCache // TODO: make this persistent
+	sg        SessionGetter
 }
 
 // NewBuildManager creates a BuildManager
-func NewBuildManager(b builder.Backend, idMappings *idtools.IDMappings) *BuildManager {
+func NewBuildManager(b builder.Backend, sg SessionGetter, idMappings *idtools.IDMappings) *BuildManager {
 	return &BuildManager{
 		backend:   b,
 		pathCache: &syncmap.Map{},
+		sg:        sg,
 		archiver:  chrootarchive.NewArchiver(idMappings),
 	}
 }
@@ -84,6 +92,13 @@ func (bm *BuildManager) Build(ctx context.Context, config backend.BuildConfig) (
 		}
 	}
 
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
+	if err := bm.initializeClientSession(ctx, cancel, config.Options); err != nil {
+		return nil, err
+	}
+
 	builderOptions := builderOptions{
 		Options:        config.Options,
 		ProgressWriter: config.ProgressWriter,
@@ -96,6 +111,22 @@ func (bm *BuildManager) Build(ctx context.Context, config backend.BuildConfig) (
 	return newBuilder(ctx, builderOptions).build(source, dockerfile)
 }
 
+func (bm *BuildManager) initializeClientSession(ctx context.Context, cancel func(), options *types.ImageBuildOptions) error {
+	if options.SessionID == "" || bm.sg == nil {
+		return nil
+	}
+	logrus.Debug("client is session enabled")
+	c, err := bm.sg.Get(ctx, options.SessionID)
+	if err != nil {
+		return err
+	}
+	go func() {
+		<-c.Context().Done()
+		cancel()
+	}()
+	return nil
+}
+
 // builderOptions are the dependencies required by the builder
 type builderOptions struct {
 	Options        *types.ImageBuildOptions

client/hijack.go

@@ -1,11 +1,9 @@
 package client
 
 import (
-	"bytes"
+	"bufio"
 	"crypto/tls"
-	"errors"
 	"fmt"
-	"io/ioutil"
 	"net"
 	"net/http"
 	"net/http/httputil"
@@ -16,6 +14,7 @@ import (
 	"github.com/docker/docker/api/types"
 	"github.com/docker/docker/pkg/tlsconfig"
 	"github.com/docker/go-connections/sockets"
+	"github.com/pkg/errors"
 	"golang.org/x/net/context"
 )
 
@@ -48,49 +47,12 @@ func (cli *Client) postHijacked(ctx context.Context, path string, query url.Valu
 	}
 	req = cli.addHeaders(req, headers)
 
-	req.Host = cli.addr
-	req.Header.Set("Connection", "Upgrade")
-	req.Header.Set("Upgrade", "tcp")
-
-	conn, err := dial(cli.proto, cli.addr, resolveTLSConfig(cli.client.Transport))
+	conn, err := cli.setupHijackConn(req, "tcp")
 	if err != nil {
-		if strings.Contains(err.Error(), "connection refused") {
-			return types.HijackedResponse{}, fmt.Errorf("Cannot connect to the Docker daemon. Is 'docker daemon' running on this host?")
-		}
 		return types.HijackedResponse{}, err
 	}
 
-	// When we set up a TCP connection for hijack, there could be long periods
-	// of inactivity (a long running command with no output) that in certain
-	// network setups may cause ECONNTIMEOUT, leaving the client in an unknown
-	// state. Setting TCP KeepAlive on the socket connection will prohibit
-	// ECONNTIMEOUT unless the socket connection truly is broken
-	if tcpConn, ok := conn.(*net.TCPConn); ok {
-		tcpConn.SetKeepAlive(true)
-		tcpConn.SetKeepAlivePeriod(30 * time.Second)
-	}
-
-	clientconn := httputil.NewClientConn(conn, nil)
-	defer clientconn.Close()
-
-	// Server hijacks the connection, error 'connection closed' expected
-	resp, err := clientconn.Do(req)
-	if err != nil {
-		return types.HijackedResponse{}, err
-	}
-
-	defer resp.Body.Close()
-	switch resp.StatusCode {
-	case http.StatusOK, http.StatusSwitchingProtocols:
-		rwc, br := clientconn.Hijack()
-		return types.HijackedResponse{Conn: rwc, Reader: br}, err
-	}
-
-	errbody, err := ioutil.ReadAll(resp.Body)
-	if err != nil {
-		return types.HijackedResponse{}, err
-	}
-	return types.HijackedResponse{}, fmt.Errorf("Error response from daemon: %s", bytes.TrimSpace(errbody))
+	return types.HijackedResponse{Conn: conn, Reader: bufio.NewReader(conn)}, err
 }
 
 func tlsDial(network, addr string, config *tls.Config) (net.Conn, error) {
@@ -189,3 +151,56 @@ func dial(proto, addr string, tlsConfig *tls.Config) (net.Conn, error) {
 	}
 	return net.Dial(proto, addr)
 }
+
+func (cli *Client) setupHijackConn(req *http.Request, proto string) (net.Conn, error) {
+	req.Host = cli.addr
+	req.Header.Set("Connection", "Upgrade")
+	req.Header.Set("Upgrade", proto)
+
+	conn, err := dial(cli.proto, cli.addr, resolveTLSConfig(cli.client.Transport))
+	if err != nil {
+		return nil, errors.Wrap(err, "cannot connect to the Docker daemon. Is 'docker daemon' running on this host?")
+	}
+
+	// When we set up a TCP connection for hijack, there could be long periods
+	// of inactivity (a long running command with no output) that in certain
+	// network setups may cause ECONNTIMEOUT, leaving the client in an unknown
+	// state. Setting TCP KeepAlive on the socket connection will prohibit
+	// ECONNTIMEOUT unless the socket connection truly is broken
+	if tcpConn, ok := conn.(*net.TCPConn); ok {
+		tcpConn.SetKeepAlive(true)
+		tcpConn.SetKeepAlivePeriod(30 * time.Second)
+	}
+
+	clientconn := httputil.NewClientConn(conn, nil)
+	defer clientconn.Close()
+
+	// Server hijacks the connection, error 'connection closed' expected
+	resp, err := clientconn.Do(req)
+	if err != nil {
+		return nil, err
+	}
+	if resp.StatusCode != http.StatusSwitchingProtocols {
+		resp.Body.Close()
+		return nil, fmt.Errorf("unable to upgrade to %s, received %d", proto, resp.StatusCode)
+	}
+
+	c, br := clientconn.Hijack()
+	if br.Buffered() > 0 {
+		// If there is buffered content, wrap the connection
+		c = &hijackedConn{c, br}
+	} else {
+		br.Reset(nil)
+	}
+
+	return c, nil
+}
+
+type hijackedConn struct {
+	net.Conn
+	r *bufio.Reader
+}
+
+func (c *hijackedConn) Read(b []byte) (int, error) {
+	return c.r.Read(b)
+}

client/image_build.go

@@ -120,6 +120,9 @@ func (cli *Client) imageBuildOptionsToQuery(options types.ImageBuildOptions) (ur
 		return query, err
 	}
 	query.Set("cachefrom", string(cacheFromJSON))
+	if options.SessionID != "" {
+		query.Set("session", options.SessionID)
+	}
 
 	return query, nil
 }

client/interface.go

@@ -2,6 +2,7 @@ package client
 
 import (
 	"io"
+	"net"
 	"time"
 
 	"github.com/docker/docker/api/types"
@@ -35,6 +36,7 @@ type CommonAPIClient interface {
 	ServerVersion(ctx context.Context) (types.Version, error)
 	NegotiateAPIVersion(ctx context.Context)
 	NegotiateAPIVersionPing(types.Ping)
+	DialSession(ctx context.Context, proto string, meta map[string][]string) (net.Conn, error)
 }
 
 // ContainerAPIClient defines API client methods for the containers

client/session.go

@@ -0,0 +1,19 @@
+package client
+
+import (
+	"net"
+	"net/http"
+
+	"golang.org/x/net/context"
+)
+
+// DialSession returns a connection that can be used communication with daemon
+func (cli *Client) DialSession(ctx context.Context, proto string, meta map[string][]string) (net.Conn, error) {
+	req, err := http.NewRequest("POST", "/session", nil)
+	if err != nil {
+		return nil, err
+	}
+	req = cli.addHeaders(req, meta)
+
+	return cli.setupHijackConn(req, proto)
+}