From 76e6583679031f45262d302d69228d41e3280f52 Mon Sep 17 00:00:00 2001 From: Sambhav Jain Date: Mon, 18 May 2026 12:00:19 +0000 Subject: [PATCH] feat(origin): add http transport config --- origin/blobclient/client.go | 56 ++++++++++++++++++++++++++--------- origin/blobclient/uploader.go | 30 +++++++++++-------- utils/httputil/transport.go | 48 ++++++++++++++++++++++++++++++ 3 files changed, 107 insertions(+), 27 deletions(-) create mode 100644 utils/httputil/transport.go diff --git a/origin/blobclient/client.go b/origin/blobclient/client.go index f6387f209..569463c74 100644 --- a/origin/blobclient/client.go +++ b/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 } diff --git a/origin/blobclient/uploader.go b/origin/blobclient/uploader.go index 504779525..988e0915f 100644 --- a/origin/blobclient/uploader.go +++ b/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,7 +102,7 @@ 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 } @@ -108,7 +110,7 @@ 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 } diff --git a/utils/httputil/transport.go b/utils/httputil/transport.go new file mode 100644 index 000000000..1b155c6fa --- /dev/null +++ b/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, + } +}