s3 backend: optional HTTP client settings for the cloud tier (#11897)

* s3 backend: optional HTTP client settings for the cloud tier

The S3 tier backend builds its client from a bare aws.Config, so it always
uses Go's default transport: HTTP/2 wherever the endpoint offers it over TLS,
no cap on connections per host, and a 90s idle timeout. Against some
S3-compatible services that is a poor fit. Over HTTP/2 every upload and
range read of a tiered volume shares one connection and its flow-control
window; with a small window, tier uploads and concurrent reads of tiered
volumes become throughput-bound on that one connection, while HTTP/1.1 opens
a connection per concurrent request. Some services also limit parallel
connections per client address, or drop idle connections before 90s.

Add three optional settings to [storage.backend.s3.<id>], all off by
default so existing deployments keep the SDK's default client:
- disable_http2: keep the client on HTTP/1.1 (ForceAttemptHTTP2=false and an
  empty TLSNextProto, the documented opt-out);
- max_connections: MaxConnsPerHost, 0 = no cap;
- idle_connection_timeout_seconds: IdleConnTimeout, 0 = Go's default.

The session cache key includes the options when any is set, so two backends
on the same endpoint with different settings do not share a client.
ToProperties reports the settings; the master.toml scaffold documents them.

Tests: TestS3BackendHTTPOptionsReachTheClient fails without the change (the
settings never reach the client); TestS3BackendDefaultHTTPClientUnchanged,
TestS3BackendSessionsDifferByHTTPOptions, TestParseNonNegative.

* s3 backend: reject an idle timeout too large for time.Duration

An idle_connection_timeout_seconds above math.MaxInt64/1e9 overflowed when
multiplied by time.Second. The negative result made the transport keep its
default timeout and ToProperties export a negative value. Warn and use the
default instead.

* s3 backend: parse the idle timeout as 64-bit seconds

parseIdleConnTimeout went through strconv.Atoi, so on 32-bit platforms
any value above 2147483647 seconds was rejected while the same value was
accepted on 64-bit, and the overflow test failed under GOARCH=386. Parse
with ParseInt(…, 64) and export the effective value with FormatInt so the
accepted range is the same on every platform.

---------

Co-authored-by: Javier de la Torre <jatorre@lakehousebox.com>

weed/command/scaffold/master.toml

@@ -28,6 +28,10 @@ region = "us-east-2"
 bucket = "your_bucket_name"    # an existing bucket
 endpoint = ""
 storage_class = "STANDARD_IA"
+# optional HTTP client tuning for S3-compatible services:
+# disable_http2 = false                 # true keeps the client on HTTP/1.1, one connection per concurrent request
+# max_connections = 0                   # cap on connections to the endpoint, 0 = no cap
+# idle_connection_timeout_seconds = 0   # close idle connections sooner, 0 = Go default (90s)
 
 # create this number of logical volumes if no more writable volumes
 # count_x means how many copies of data.

weed/storage/backend/s3_backend/s3_backend.go

@@ -3,6 +3,7 @@ package s3_backend
 import (
 	"fmt"
 	"io"
+	"math"
 	"os"
 	"strconv"
 	"strings"
@@ -44,6 +45,7 @@ type S3BackendStorage struct {
 	forcePathStyle        bool
 	uploadConcurrency     int
 	downloadConcurrency   int
+	httpOptions           httpOptions
 	conn                  s3iface.S3API
 }
 
@@ -59,11 +61,16 @@ func newS3BackendStorage(configuration backend.StringProperties, configPrefix st
 	s.forcePathStyle = util.ParseBool(configuration.GetString(configPrefix+"force_path_style"), true)
 	s.uploadConcurrency = parseConcurrency(configuration.GetString(configPrefix+"upload_concurrency"), defaultUploadConcurrency)
 	s.downloadConcurrency = parseConcurrency(configuration.GetString(configPrefix+"download_concurrency"), defaultDownloadConcurrency)
+	s.httpOptions = httpOptions{
+		disableHTTP2:    util.ParseBool(configuration.GetString(configPrefix+"disable_http2"), false),
+		maxConnsPerHost: parseNonNegative(configuration.GetString(configPrefix+"max_connections"), "max_connections"),
+		idleConnTimeout: parseIdleConnTimeout(configuration.GetString(configPrefix + "idle_connection_timeout_seconds")),
+	}
 	if s.storageClass == "" {
 		s.storageClass = "STANDARD_IA"
 	}
 
-	s.conn, err = createSession(s.aws_access_key_id, s.aws_secret_access_key, s.region, s.endpoint, s.forcePathStyle)
+	s.conn, err = createSession(s.aws_access_key_id, s.aws_secret_access_key, s.region, s.endpoint, s.forcePathStyle, s.httpOptions)
 
 	glog.V(0).Infof("created backend storage s3.%s for region %s bucket %s", s.id, s.region, s.bucket)
 	return
@@ -80,6 +87,9 @@ func (s *S3BackendStorage) ToProperties() map[string]string {
 	m["force_path_style"] = util.BoolToString(s.forcePathStyle)
 	m["upload_concurrency"] = strconv.Itoa(s.uploadConcurrency)
 	m["download_concurrency"] = strconv.Itoa(s.downloadConcurrency)
+	m["disable_http2"] = util.BoolToString(s.httpOptions.disableHTTP2)
+	m["max_connections"] = strconv.Itoa(s.httpOptions.maxConnsPerHost)
+	m["idle_connection_timeout_seconds"] = strconv.FormatInt(int64(s.httpOptions.idleConnTimeout/time.Second), 10)
 	return m
 }
 
@@ -99,6 +109,42 @@ func parseConcurrency(value string, def int) int {
 	return def
 }
 
+// parseNonNegative reads an optional non-negative integer setting; empty or
+// invalid means 0 (the default).
+func parseNonNegative(value, name string) int {
+	if value == "" {
+		return 0
+	}
+	if n, err := strconv.Atoi(value); err == nil && n >= 0 {
+		return n
+	}
+	glog.Warningf("invalid %s value %q, using the default", name, value)
+	return 0
+}
+
+// maxIdleConnTimeoutSeconds is the largest number of seconds a time.Duration
+// can hold.
+const maxIdleConnTimeoutSeconds = math.MaxInt64 / int64(time.Second)
+
+// parseIdleConnTimeout reads idle_connection_timeout_seconds; empty, invalid
+// or too large to represent as a time.Duration means 0 (the default). It
+// parses 64-bit seconds so the range is the same on 32-bit platforms.
+func parseIdleConnTimeout(value string) time.Duration {
+	if value == "" {
+		return 0
+	}
+	seconds, err := strconv.ParseInt(value, 10, 64)
+	if err != nil || seconds < 0 {
+		glog.Warningf("invalid idle_connection_timeout_seconds value %q, using the default", value)
+		return 0
+	}
+	if seconds > maxIdleConnTimeoutSeconds {
+		glog.Warningf("idle_connection_timeout_seconds %q is out of range, using the default", value)
+		return 0
+	}
+	return time.Duration(seconds) * time.Second
+}
+
 func (s *S3BackendStorage) NewStorageFile(key string, tierInfo *volume_server_pb.VolumeInfo) backend.BackendStorageFile {
 	if strings.HasPrefix(key, "/") {
 		key = key[1:]

weed/storage/backend/s3_backend/s3_sessions.go

@@ -1,8 +1,11 @@
 package s3_backend
 
 import (
+	"crypto/tls"
 	"fmt"
+	"net/http"
 	"sync"
+	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/credentials"
@@ -18,12 +21,49 @@ var (
 	sessionsLock sync.Mutex
 )
 
-func createSession(awsAccessKeyId, awsSecretAccessKey, region, endpoint string, forcePathStyle bool) (s3iface.S3API, error) {
+// httpOptions tunes the HTTP transport of the tier backend's S3 client. The
+// zero value keeps the SDK's default client.
+type httpOptions struct {
+	// disableHTTP2 keeps the client on HTTP/1.1. Over HTTP/2 every upload and
+	// range read to the endpoint shares one connection and its flow-control
+	// window, which some S3-compatible services keep small.
+	disableHTTP2 bool
+	// maxConnsPerHost caps the connections to the endpoint (0 = no cap), for
+	// services that limit parallel connections per client address.
+	maxConnsPerHost int
+	// idleConnTimeout closes idle connections sooner than Go's default 90s,
+	// for services that drop idle connections earlier (0 = default).
+	idleConnTimeout time.Duration
+}
+
+// newTransport returns the transport for the options, or nil for the default.
+func (o httpOptions) newTransport() *http.Transport {
+	if o == (httpOptions{}) {
+		return nil
+	}
+	t := http.DefaultTransport.(*http.Transport).Clone()
+	if o.disableHTTP2 {
+		t.ForceAttemptHTTP2 = false
+		t.TLSNextProto = map[string]func(string, *tls.Conn) http.RoundTripper{}
+	}
+	if o.maxConnsPerHost > 0 {
+		t.MaxConnsPerHost = o.maxConnsPerHost
+	}
+	if o.idleConnTimeout > 0 {
+		t.IdleConnTimeout = o.idleConnTimeout
+	}
+	return t
+}
+
+func createSession(awsAccessKeyId, awsSecretAccessKey, region, endpoint string, forcePathStyle bool, httpOpts httpOptions) (s3iface.S3API, error) {
 
 	sessionsLock.Lock()
 	defer sessionsLock.Unlock()
 
 	cacheKey := fmt.Sprintf("%s|%s", region, endpoint)
+	if httpOpts != (httpOptions{}) {
+		cacheKey = fmt.Sprintf("%s|%+v", cacheKey, httpOpts)
+	}
 	if t, found := s3Sessions[cacheKey]; found {
 		return t, nil
 	}
@@ -34,6 +74,9 @@ func createSession(awsAccessKeyId, awsSecretAccessKey, region, endpoint string,
 		S3ForcePathStyle:              aws.Bool(forcePathStyle),
 		S3DisableContentMD5Validation: aws.Bool(true),
 	}
+	if t := httpOpts.newTransport(); t != nil {
+		config.HTTPClient = &http.Client{Transport: t}
+	}
 	if awsAccessKeyId != "" && awsSecretAccessKey != "" {
 		config.Credentials = credentials.NewStaticCredentials(awsAccessKeyId, awsSecretAccessKey, "")
 	}

weed/storage/backend/s3_backend/s3_sessions_test.go

@@ -0,0 +1,129 @@
+package s3_backend
+
+import (
+	"math"
+	"net/http"
+	"strconv"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go/service/s3"
+)
+
+func backendTransport(t *testing.T, s *S3BackendStorage) *http.Transport {
+	t.Helper()
+	client, ok := s.conn.(*s3.S3)
+	if !ok {
+		t.Fatalf("unexpected client type %T", s.conn)
+	}
+	httpClient := client.Client.Config.HTTPClient
+	if httpClient == nil {
+		return nil
+	}
+	transport, _ := httpClient.Transport.(*http.Transport)
+	return transport
+}
+
+func TestS3BackendHTTPOptionsReachTheClient(t *testing.T) {
+	s, err := newS3BackendStorage(testProperties{
+		"endpoint":                        "http://127.0.0.1:1",
+		"disable_http2":                   "true",
+		"max_connections":                 "64",
+		"idle_connection_timeout_seconds": "50",
+	}, "", "http-options")
+	if err != nil {
+		t.Fatal(err)
+	}
+	transport := backendTransport(t, s)
+	if transport == nil {
+		t.Fatal("configured HTTP options did not reach the S3 client")
+	}
+	if transport.ForceAttemptHTTP2 || transport.TLSNextProto == nil || len(transport.TLSNextProto) != 0 {
+		t.Errorf("HTTP/2 not disabled: ForceAttemptHTTP2=%v TLSNextProto=%v", transport.ForceAttemptHTTP2, transport.TLSNextProto)
+	}
+	if transport.MaxConnsPerHost != 64 {
+		t.Errorf("MaxConnsPerHost = %d, want 64", transport.MaxConnsPerHost)
+	}
+	if transport.IdleConnTimeout != 50*time.Second {
+		t.Errorf("IdleConnTimeout = %v, want 50s", transport.IdleConnTimeout)
+	}
+
+	props := s.ToProperties()
+	if props["disable_http2"] != "true" || props["max_connections"] != "64" || props["idle_connection_timeout_seconds"] != "50" {
+		t.Errorf("ToProperties: %v", props)
+	}
+}
+
+func TestS3BackendDefaultHTTPClientUnchanged(t *testing.T) {
+	s, err := newS3BackendStorage(testProperties{"endpoint": "http://127.0.0.1:2"}, "", "http-defaults")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if transport := backendTransport(t, s); transport != nil && transport != http.DefaultTransport {
+		t.Fatalf("no options set, yet the client got a custom transport %+v", transport)
+	}
+	if (httpOptions{}).newTransport() != nil {
+		t.Fatal("zero options must keep the default client")
+	}
+}
+
+func TestS3BackendSessionsDifferByHTTPOptions(t *testing.T) {
+	a, err := createSession("", "", "us-east-1", "http://127.0.0.1:3", true, httpOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	b, err := createSession("", "", "us-east-1", "http://127.0.0.1:3", true, httpOptions{disableHTTP2: true})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if a == b {
+		t.Fatal("backends with different HTTP options share one client")
+	}
+	c, err := createSession("", "", "us-east-1", "http://127.0.0.1:3", true, httpOptions{disableHTTP2: true})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if b != c {
+		t.Fatal("the same endpoint and options must reuse the cached client")
+	}
+}
+
+func TestParseNonNegative(t *testing.T) {
+	for _, c := range []struct {
+		in   string
+		want int
+	}{{"", 0}, {"0", 0}, {"12", 12}, {"-1", 0}, {"x", 0}} {
+		if got := parseNonNegative(c.in, "test"); got != c.want {
+			t.Errorf("parseNonNegative(%q) = %d, want %d", c.in, got, c.want)
+		}
+	}
+}
+
+func TestS3BackendIdleTimeoutOverflowFallsBackToDefault(t *testing.T) {
+	maxSeconds := int64(math.MaxInt64 / int64(time.Second))
+	for i, c := range []struct {
+		in   string
+		want time.Duration
+	}{
+		{strconv.FormatInt(maxSeconds, 10), time.Duration(maxSeconds) * time.Second},
+		{strconv.FormatInt(maxSeconds+1, 10), 0},
+		{"10000000000", 0},
+		{"-1", 0},
+		{"x", 0},
+	} {
+		s, err := newS3BackendStorage(testProperties{
+			"endpoint":                        "http://127.0.0.1:4",
+			"idle_connection_timeout_seconds": c.in,
+		}, "", "idle-overflow-"+strconv.Itoa(i))
+		if err != nil {
+			t.Fatal(err)
+		}
+		if s.httpOptions.idleConnTimeout != c.want {
+			t.Errorf("idle_connection_timeout_seconds=%s: timeout %v, want %v", c.in, s.httpOptions.idleConnTimeout, c.want)
+		}
+		want := strconv.FormatInt(int64(c.want/time.Second), 10)
+		if got := s.ToProperties()["idle_connection_timeout_seconds"]; got != want {
+			t.Errorf("idle_connection_timeout_seconds=%s: ToProperties exports %s, want %s", c.in, got, want)
+		}
+	}
+}