diff --git a/weed/command/scaffold/master.toml b/weed/command/scaffold/master.toml index 34a7168c6a..b7a344e231 100644 --- a/weed/command/scaffold/master.toml +++ b/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. diff --git a/weed/storage/backend/s3_backend/s3_backend.go b/weed/storage/backend/s3_backend/s3_backend.go index 882f21e99a..08651dedea 100644 --- a/weed/storage/backend/s3_backend/s3_backend.go +++ b/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:] diff --git a/weed/storage/backend/s3_backend/s3_sessions.go b/weed/storage/backend/s3_backend/s3_sessions.go index 999dc6fe51..bdadb56118 100644 --- a/weed/storage/backend/s3_backend/s3_sessions.go +++ b/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, "") } diff --git a/weed/storage/backend/s3_backend/s3_sessions_test.go b/weed/storage/backend/s3_backend/s3_sessions_test.go new file mode 100644 index 0000000000..eb9f621c4d --- /dev/null +++ b/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) + } + } +}