feat(origin): setup http transport config and connection pool

origin/blobclient/client.go

@@ -74,6 +74,7 @@ type HTTPClient struct {
 	addr      string
 	chunkSize uint64
 	tls       *tls.Config
+	transport http.RoundTripper
 	tracer    trace.Tracer
 }
 
@@ -90,6 +91,33 @@ func WithTLS(tls *tls.Config) Option {
 	return func(c *HTTPClient) { c.tls = tls }
 }
 
+// WithTransport configures an HTTPClient to use a shared HTTP RoundTripper
+// (typically built via httputil.NewClientTransport).
+func WithTransport(t http.RoundTripper) Option {
+	return func(c *HTTPClient) { c.transport = t }
+}
+
+// sendOpt returns the appropriate httputil.SendOption based on whether a
+// shared transport and TLS are configured.
+//
+// When transport is non-nil it is injected directly (connection pooling
+// active). tlsCfg is NOT used to configure TLS on the connection, the
+// transport already has a *tls.Config baked in from httputil.NewClientTransport.
+// Callers must therefore always pair WithTransport with WithTLS when connecting
+// to an HTTPS origin.
+//
+// When transport is nil httputil.SendTLS creates a
+// fresh *http.Transport per request (no pooling) and also sets the scheme.
+func sendOpt(tlsCfg *tls.Config, transport http.RoundTripper) httputil.SendOption {
+	if transport != nil {
+		if tlsCfg != nil {
+			return httputil.SendTLSTransport(transport)
+		}
+		return httputil.SendTransport(transport)
+	}
+	return httputil.SendTLS(tlsCfg)
+}
+
 // New returns a new HTTPClient scoped to addr.
 func New(addr string, opts ...Option) *HTTPClient {
 	c := &HTTPClient{
@@ -112,7 +140,7 @@ func (c *HTTPClient) CheckReadiness() error {
 	_, err := httputil.Get(
 		fmt.Sprintf("http://%s/readiness", c.addr),
 		httputil.SendTimeout(5*time.Second),
-		httputil.SendTLS(c.tls))
+		sendOpt(c.tls, c.transport))
 	if err != nil {
 		return fmt.Errorf("origin not ready: %v", err)
 	}
@@ -124,7 +152,7 @@ func (c *HTTPClient) Locations(d core.Digest) ([]string, error) {
 	r, err := httputil.Get(
 		fmt.Sprintf("http://%s/blobs/%s/locations", c.addr, d),
 		httputil.SendTimeout(5*time.Second),
-		httputil.SendTLS(c.tls))
+		sendOpt(c.tls, c.transport))
 	if err != nil {
 		return nil, err
 	}
@@ -163,7 +191,7 @@ func (c *HTTPClient) stat(namespace string, d core.Digest, local bool) (*core.Bl
 	r, err := httputil.Head(
 		u,
 		httputil.SendTimeout(15*time.Second),
-		httputil.SendTLS(c.tls))
+		sendOpt(c.tls, c.transport))
 	if err != nil {
 		if httputil.IsNotFound(err) {
 			return nil, ErrBlobNotFound
@@ -186,14 +214,14 @@ func (c *HTTPClient) DeleteBlob(d core.Digest) error {
 	_, err := httputil.Delete(
 		fmt.Sprintf("http://%s/internal/blobs/%s", c.addr, d),
 		httputil.SendAcceptedCodes(http.StatusAccepted),
-		httputil.SendTLS(c.tls))
+		sendOpt(c.tls, c.transport))
 	return err
 }
 
 // TransferBlob uploads a blob to a single origin server. Unlike its cousin UploadBlob,
 // TransferBlob is an internal API which does not replicate the blob.
 func (c *HTTPClient) TransferBlob(d core.Digest, blob io.Reader) error {
-	tc := newTransferClient(c.addr, c.tls)
+	tc := newTransferClient(c.addr, c.tls, c.transport)
 	return runChunkedUpload(tc, d, blob, int64(c.chunkSize))
 }
 
@@ -211,7 +239,7 @@ func (c *HTTPClient) UploadBlob(ctx context.Context, namespace string, d core.Di
 	)
 	defer span.End()
 
-	uc := newUploadClientWithContext(ctx, c.addr, namespace, _publicUpload, 0, c.tls)
+	uc := newUploadClientWithContext(ctx, c.addr, namespace, _publicUpload, 0, c.tls, c.transport)
 	if err := runChunkedUpload(uc, d, blob, int64(c.chunkSize)); err != nil {
 		span.RecordError(err)
 		span.SetStatus(codes.Error, "upload failed")
@@ -227,7 +255,7 @@ func (c *HTTPClient) UploadBlob(ctx context.Context, namespace string, d core.Di
 func (c *HTTPClient) DuplicateUploadBlob(
 	namespace string, d core.Digest, blob io.Reader, delay time.Duration,
 ) error {
-	uc := newUploadClient(c.addr, namespace, _duplicateUpload, delay, c.tls)
+	uc := newUploadClient(c.addr, namespace, _duplicateUpload, delay, c.tls, c.transport)
 	return runChunkedUpload(uc, d, blob, int64(c.chunkSize))
 }
 
@@ -251,7 +279,7 @@ func (c *HTTPClient) DownloadBlob(ctx context.Context, namespace string, d core.
 	r, err := httputil.Get(
 		fmt.Sprintf("http://%s/namespace/%s/blobs/%s", c.addr, url.PathEscape(namespace), d),
 		httputil.SendContext(ctx),
-		httputil.SendTLS(c.tls))
+		sendOpt(c.tls, c.transport))
 	if err != nil {
 		span.RecordError(err)
 		span.SetStatus(codes.Error, "download request failed")
@@ -277,7 +305,7 @@ func (c *HTTPClient) PrefetchBlob(namespace string, d core.Digest) error {
 	r, err := httputil.Post(
 		fmt.Sprintf("http://%s/namespace/%s/blobs/%s/prefetch", c.addr, url.PathEscape(namespace), d),
 		httputil.SendAcceptedCodes(http.StatusOK, http.StatusAccepted),
-		httputil.SendTLS(c.tls))
+		sendOpt(c.tls, c.transport))
 	if err != nil {
 		return err
 	}
@@ -292,7 +320,7 @@ func (c *HTTPClient) ReplicateToRemote(namespace string, d core.Digest, remoteDN
 	_, err := httputil.Post(
 		fmt.Sprintf("http://%s/namespace/%s/blobs/%s/remote/%s",
 			c.addr, url.PathEscape(namespace), d, remoteDNS),
-		httputil.SendTLS(c.tls))
+		sendOpt(c.tls, c.transport))
 	return err
 }
 
@@ -305,7 +333,7 @@ func (c *HTTPClient) GetMetaInfo(namespace string, d core.Digest) (*core.MetaInf
 		fmt.Sprintf("http://%s/internal/namespace/%s/blobs/%s/metainfo",
 			c.addr, url.PathEscape(namespace), d),
 		httputil.SendTimeout(15*time.Second),
-		httputil.SendTLS(c.tls))
+		sendOpt(c.tls, c.transport))
 	if err != nil {
 		return nil, err
 	}
@@ -326,7 +354,7 @@ func (c *HTTPClient) GetMetaInfo(namespace string, d core.Digest) (*core.MetaInf
 func (c *HTTPClient) OverwriteMetaInfo(d core.Digest, pieceLength int64) error {
 	_, err := httputil.Post(
 		fmt.Sprintf("http://%s/internal/blobs/%s/metainfo?piece_length=%d", c.addr, d, pieceLength),
-		httputil.SendTLS(c.tls))
+		sendOpt(c.tls, c.transport))
 	return err
 }
 
@@ -336,7 +364,7 @@ func (c *HTTPClient) GetPeerContext() (core.PeerContext, error) {
 	r, err := httputil.Get(
 		fmt.Sprintf("http://%s/internal/peercontext", c.addr),
 		httputil.SendTimeout(5*time.Second),
-		httputil.SendTLS(c.tls))
+		sendOpt(c.tls, c.transport))
 	if err != nil {
 		return pctx, err
 	}
@@ -354,6 +382,6 @@ func (c *HTTPClient) ForceCleanup(ttl time.Duration) error {
 	_, err := httputil.Post(
 		fmt.Sprintf("http://%s/forcecleanup?%s", c.addr, v.Encode()),
 		httputil.SendTimeout(2*time.Minute),
-		httputil.SendTLS(c.tls))
+		sendOpt(c.tls, c.transport))
 	return err
 }

origin/blobclient/uploader.go

@@ -21,6 +21,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"net/http"
 	"net/url"
 	"time"
 
@@ -69,18 +70,19 @@ func runChunkedUploadHelper(u uploader, d core.Digest, blob io.Reader, chunkSize
 
 // transferClient executes chunked uploads for internal blob transfers.
 type transferClient struct {
-	addr string
-	tls  *tls.Config
+	addr      string
+	tls       *tls.Config
+	transport http.RoundTripper
 }
 
-func newTransferClient(addr string, tls *tls.Config) *transferClient {
-	return &transferClient{addr, tls}
+func newTransferClient(addr string, tls *tls.Config, transport http.RoundTripper) *transferClient {
+	return &transferClient{addr, tls, transport}
 }
 
 func (c *transferClient) start(d core.Digest) (uid string, err error) {
 	r, err := httputil.Post(
 		fmt.Sprintf("http://%s/internal/blobs/%s/uploads", c.addr, d),
-		httputil.SendTLS(c.tls))
+		sendOpt(c.tls, c.transport))
 	if err != nil {
 		return "", err
 	}
@@ -100,15 +102,15 @@ func (c *transferClient) patch(
 		httputil.SendHeaders(map[string]string{
 			"Content-Range": fmt.Sprintf("%d-%d", start, stop),
 		}),
-		httputil.SendTLS(c.tls))
+		sendOpt(c.tls, c.transport))
 	return err
 }
 
 func (c *transferClient) commit(d core.Digest, uid string) error {
 	_, err := httputil.Put(
 		fmt.Sprintf("http://%s/internal/blobs/%s/uploads/%s", c.addr, d, uid),
 		httputil.SendTimeout(15*time.Minute),
-		httputil.SendTLS(c.tls))
+		sendOpt(c.tls, c.transport))
 	return err
 }
 
@@ -127,16 +129,17 @@ type uploadClient struct {
 	uploadType uploadType
 	delay      time.Duration
 	tls        *tls.Config
+	transport  http.RoundTripper
 }
 
 func newUploadClient(
-	addr string, namespace string, t uploadType, delay time.Duration, tls *tls.Config,
+	addr string, namespace string, t uploadType, delay time.Duration, tls *tls.Config, transport http.RoundTripper,
 ) *uploadClient {
-	return newUploadClientWithContext(context.Background(), addr, namespace, t, delay, tls)
+	return newUploadClientWithContext(context.Background(), addr, namespace, t, delay, tls, transport)
 }
 
 func newUploadClientWithContext(
-	ctx context.Context, addr string, namespace string, t uploadType, delay time.Duration, tls *tls.Config,
+	ctx context.Context, addr string, namespace string, t uploadType, delay time.Duration, tls *tls.Config, transport http.RoundTripper,
 ) *uploadClient {
 	return &uploadClient{
 		ctx:        ctx,
@@ -145,6 +148,7 @@ func newUploadClientWithContext(
 		uploadType: t,
 		delay:      delay,
 		tls:        tls,
+		transport:  transport,
 	}
 }
 
@@ -153,7 +157,7 @@ func (c *uploadClient) start(d core.Digest) (uid string, err error) {
 		fmt.Sprintf("http://%s/namespace/%s/blobs/%s/uploads",
 			c.addr, url.PathEscape(c.namespace), d),
 		httputil.SendContext(c.ctx),
-		httputil.SendTLS(c.tls))
+		sendOpt(c.tls, c.transport))
 	if err != nil {
 		return "", err
 	}
@@ -175,7 +179,7 @@ func (c *uploadClient) patch(
 		httputil.SendHeaders(map[string]string{
 			"Content-Range": fmt.Sprintf("%d-%d", start, stop),
 		}),
-		httputil.SendTLS(c.tls))
+		sendOpt(c.tls, c.transport))
 	return err
 }
 
@@ -206,6 +210,6 @@ func (c *uploadClient) commit(d core.Digest, uid string) error {
 		httputil.SendContext(c.ctx),
 		httputil.SendTimeout(15*time.Minute),
 		httputil.SendBody(body),
-		httputil.SendTLS(c.tls))
+		sendOpt(c.tls, c.transport))
 	return err
 }

utils/httputil/transport.go

@@ -0,0 +1,48 @@
+// Copyright (c) 2016-2019 Uber Technologies, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//	http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+package httputil
+
+import (
+	"crypto/tls"
+	"net/http"
+)
+
+// ConnPoolConfig tunes a shared *http.Transport. Zero values are replaced with
+// safe defaults by NewClientTransport. All fields are YAML-deserializable so
+// callers can expose pool sizing through their service config.
+type ConnPoolConfig struct {
+	MaxIdleConns        int `yaml:"max_idle_conns"`
+	MaxIdleConnsPerHost int `yaml:"max_idle_conns_per_host"`
+}
+
+func (c *ConnPoolConfig) applyDefaults() {
+	if c.MaxIdleConns == 0 {
+		c.MaxIdleConns = 256
+	}
+	if c.MaxIdleConnsPerHost == 0 {
+		c.MaxIdleConnsPerHost = 32
+	}
+}
+
+// NewClientTransport returns a single *http.Transport suitable for sharing
+// across goroutines. If tlsCfg is nil the transport runs in cleartext.
+// Callers should construct this once at startup and never mutate it.
+func NewClientTransport(tlsCfg *tls.Config, cfg ConnPoolConfig) *http.Transport {
+	cfg.applyDefaults()
+	return &http.Transport{
+		MaxIdleConns:        cfg.MaxIdleConns,
+		MaxIdleConnsPerHost: cfg.MaxIdleConnsPerHost,
+		TLSClientConfig:     tlsCfg,
+	}
+}