Replace use of httputil in client hijack
- addresses part of https://github.com/moby/moby/issues/39926


Simplify the hijack process by just performing the http request/response on the connection and returning the raw conn after success. The client conn from httputil is deprecated and easily replaced.

Just a draft to see if the tests are ok with it.
client/hijack.go
package client // import "github.com/docker/docker/client"

import (
	"bufio"
	"context"
	"fmt"
	"net"
	"net/http"
	"net/http/httputil"
	"net/url"
	"time"

	"github.com/docker/docker/api/types"
	"github.com/docker/docker/api/types/versions"
	"github.com/pkg/errors"
	"go.opentelemetry.io/otel"
	"go.opentelemetry.io/otel/codes"
	"go.opentelemetry.io/otel/propagation"
	"go.opentelemetry.io/otel/trace"
)

// postHijacked sends a POST request and hijacks the connection.
func (cli *Client) postHijacked(ctx context.Context, path string, query url.Values, body interface{}, headers map[string][]string) (types.HijackedResponse, error) {
	bodyEncoded, err := encodeData(body)
	if err != nil {
		return types.HijackedResponse{}, err
	}
	req, err := cli.buildRequest(ctx, http.MethodPost, cli.getAPIPath(ctx, path, query), bodyEncoded, headers)
	if err != nil {
		return types.HijackedResponse{}, err
	}
	conn, mediaType, err := cli.setupHijackConn(req, "tcp")
	if err != nil {
		return types.HijackedResponse{}, err
	}

	return types.NewHijackedResponse(conn, mediaType), err
}

// DialHijack returns a hijacked connection with negotiated protocol proto.
func (cli *Client) DialHijack(ctx context.Context, url, proto string, meta map[string][]string) (net.Conn, error) {
	req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, nil)
	if err != nil {
		return nil, err
	}
	req = cli.addHeaders(req, meta)

	conn, _, err := cli.setupHijackConn(req, proto)
	return conn, err
}

func (cli *Client) setupHijackConn(req *http.Request, proto string) (_ net.Conn, _ string, retErr error) {
	ctx := req.Context()
	req.Header.Set("Connection", "Upgrade")
	req.Header.Set("Upgrade", proto)

	// We aren't using the configured RoundTripper here so manually inject the trace context
	tp := cli.tp
	if tp == nil {
		if span := trace.SpanFromContext(ctx); span.SpanContext().IsValid() {
			tp = span.TracerProvider()
		} else {
			tp = otel.GetTracerProvider()
		}
	}

	ctx, span := tp.Tracer("").Start(ctx, req.Method+" "+req.URL.Path, trace.WithSpanKind(trace.SpanKindClient))
	// FIXME(thaJeztah): httpconv.ClientRequest is now an internal package; replace this with alternative for semconv v1.21
	// span.SetAttributes(httpconv.ClientRequest(req)...)
	defer func() {
		if retErr != nil {
			span.RecordError(retErr)
			span.SetStatus(codes.Error, retErr.Error())
		}
		span.End()
	}()
	otel.GetTextMapPropagator().Inject(ctx, propagation.HeaderCarrier(req.Header))

	dialer := cli.Dialer()
	conn, err := dialer(ctx)
	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 resp != nil {
		// This is a simplified variant of "httpconv.ClientStatus(resp.StatusCode))";
		//
		// The main purpose of httpconv.ClientStatus() is to detect whether the
		// status was successful (1xx, 2xx, 3xx) or non-successful (4xx/5xx).
		//
		// It also provides complex logic to *validate* status-codes against
		// a hard-coded list meant to exclude "bogus" status codes in "success"
		// ranges (1xx, 2xx) and convert them into an error status. That code
		// seemed over-reaching (and not accounting for potential future valid
		// status codes). We assume we only get valid status codes, and only
		// look at status-code ranges.
		//
		// For reference, see:
		// https://github.com/open-telemetry/opentelemetry-go/blob/v1.21.0/semconv/v1.17.0/httpconv/http.go#L85-L89
		// https://github.com/open-telemetry/opentelemetry-go/blob/v1.21.0/semconv/internal/v2/http.go#L322-L330
		// https://github.com/open-telemetry/opentelemetry-go/blob/v1.21.0/semconv/internal/v2/http.go#L356-L404
		code := codes.Unset
		if resp.StatusCode >= http.StatusBadRequest {
			code = codes.Error
		}
		span.SetStatus(code, "")
	}

	//nolint:staticcheck // ignore SA1019 for connecting to old (pre go1.8) daemons
	if err != httputil.ErrPersistEOF {
		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.  We return an
		// object that implements CloseWrite if the underlying connection
		// implements it.
		if _, ok := c.(types.CloseWriter); ok {
			c = &hijackedConnCloseWriter{&hijackedConn{c, br}}
		} else {
			c = &hijackedConn{c, br}
		}
	} else {
		br.Reset(nil)
	}

	var mediaType string
	if versions.GreaterThanOrEqualTo(cli.ClientVersion(), "1.42") {
		// Prior to 1.42, Content-Type is always set to raw-stream and not relevant
		mediaType = resp.Header.Get("Content-Type")
	}

	return c, mediaType, nil
}

// hijackedConn wraps a net.Conn and is returned by setupHijackConn in the case
// that a) there was already buffered data in the http layer when Hijack() was
// called, and b) the underlying net.Conn does *not* implement CloseWrite().
// hijackedConn does not implement CloseWrite() either.
type hijackedConn struct {
	net.Conn
	r *bufio.Reader
}

func (c *hijackedConn) Read(b []byte) (int, error) {
	return c.r.Read(b)
}

// hijackedConnCloseWriter is a hijackedConn which additionally implements
// CloseWrite().  It is returned by setupHijackConn in the case that a) there
// was already buffered data in the http layer when Hijack() was called, and b)
// the underlying net.Conn *does* implement CloseWrite().
type hijackedConnCloseWriter struct {
	*hijackedConn
}

var _ types.CloseWriter = &hijackedConnCloseWriter{}

func (c *hijackedConnCloseWriter) CloseWrite() error {
	conn := c.Conn.(types.CloseWriter)
	return conn.CloseWrite()
}

package client // import "github.com/docker/docker/client"

import (
	"bufio"
	"context"
	"fmt"
	"net"
	"net/http"
	"net/url"
	"time"

	"github.com/docker/docker/api/types"
	"github.com/docker/docker/api/types/versions"
	"github.com/pkg/errors"
	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
)

// postHijacked sends a POST request and hijacks the connection.
func (cli *Client) postHijacked(ctx context.Context, path string, query url.Values, body interface{}, headers map[string][]string) (types.HijackedResponse, error) {
	bodyEncoded, err := encodeData(body)
	if err != nil {
		return types.HijackedResponse{}, err
	}
	req, err := cli.buildRequest(ctx, http.MethodPost, cli.getAPIPath(ctx, path, query), bodyEncoded, headers)
	if err != nil {
		return types.HijackedResponse{}, err
	}
	conn, mediaType, err := cli.setupHijackConn(req, "tcp")
	if err != nil {
		return types.HijackedResponse{}, err
	}

	return types.NewHijackedResponse(conn, mediaType), err
}

// DialHijack returns a hijacked connection with negotiated protocol proto.
func (cli *Client) DialHijack(ctx context.Context, url, proto string, meta map[string][]string) (net.Conn, error) {
	req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, nil)
	if err != nil {
		return nil, err
	}
	req = cli.addHeaders(req, meta)

	conn, _, err := cli.setupHijackConn(req, proto)
	return conn, err
}

func (cli *Client) setupHijackConn(req *http.Request, proto string) (_ net.Conn, _ string, retErr error) {
	ctx := req.Context()
	req.Header.Set("Connection", "Upgrade")
	req.Header.Set("Upgrade", proto)

	dialer := cli.Dialer()
	conn, err := dialer(ctx)
	if err != nil {
		return nil, "", errors.Wrap(err, "cannot connect to the Docker daemon. Is 'docker daemon' running on this host?")
	}
	defer func() {
		if retErr != nil {
			conn.Close()
		}
	}()

	// 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)
	}

	hc := &hijackedConn{conn, bufio.NewReader(conn)}

	// Server hijacks the connection, error 'connection closed' expected
	resp, err := otelhttp.NewTransport(hc).RoundTrip(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)
	}

	if hc.r.Buffered() > 0 {
		// If there is buffered content, wrap the connection.  We return an
		// object that implements CloseWrite if the underlying connection
		// implements it.
		if _, ok := hc.Conn.(types.CloseWriter); ok {
			conn = &hijackedConnCloseWriter{hc}
		} else {
			conn = hc
		}
	} else {
		hc.r.Reset(nil)
	}

	var mediaType string
	if versions.GreaterThanOrEqualTo(cli.ClientVersion(), "1.42") {
		// Prior to 1.42, Content-Type is always set to raw-stream and not relevant
		mediaType = resp.Header.Get("Content-Type")
	}

	return conn, mediaType, nil
}

// hijackedConn wraps a net.Conn and is returned by setupHijackConn in the case
// that a) there was already buffered data in the http layer when Hijack() was
// called, and b) the underlying net.Conn does *not* implement CloseWrite().
// hijackedConn does not implement CloseWrite() either.
type hijackedConn struct {
	net.Conn
	r *bufio.Reader
}

func (c *hijackedConn) RoundTrip(req *http.Request) (*http.Response, error) {
	if err := req.Write(c.Conn); err != nil {
		return nil, err
	}
	return http.ReadResponse(c.r, req)
}

func (c *hijackedConn) Read(b []byte) (int, error) {
	return c.r.Read(b)
}

// hijackedConnCloseWriter is a hijackedConn which additionally implements
// CloseWrite().  It is returned by setupHijackConn in the case that a) there
// was already buffered data in the http layer when Hijack() was called, and b)
// the underlying net.Conn *does* implement CloseWrite().
type hijackedConnCloseWriter struct {
	*hijackedConn
}

var _ types.CloseWriter = &hijackedConnCloseWriter{}

func (c *hijackedConnCloseWriter) CloseWrite() error {
	conn := c.Conn.(types.CloseWriter)
	return conn.CloseWrite()
}

The workflow with `httputil.ClientConn` was (simplified): construct a `ClientConn`, round-trip one HTTP request, then take its `net.Conn`. Looking at the implementation, [`(*httputil.ClientConn).Do()`](https://cs.opensource.google/go/go/+/refs/tags/go1.21.5:src/net/http/httputil/persist.go;l=424-431) looks uncannily similar to `(*hijackedConn).RoundTrip()`. And when you ignore the pipelining cruft in the httputil code (waiting until the last request has been fully written before writing the next one to the conn) which does not come into play when only a single round-trip is made with the instance, they are functionally indistinguishable.

`http.Client` does a lot of things above and beyond `httputil.ClientConn`. For instance, it handles 3xx redirect responses by internally making a follow-up request to the new location. For this to work as expected, the `http.RoundTripper` assigned to the client's `Transport` field has to adhere to the interface contract, particularly the ability to make independent requests to different servers depending on the request URL. I think it's safe to say that things would not go well if the HTTP client tried to follow a redirect using `hijackedConn` as the RoundTripper.

```suggestion
	resp, err := hc.RoundTrip(req)
```
As a bonus, it would then be possible to wrap `hc` in an `otelhttp.Transport` for injecting the span context into the request headers.
