Configurable http response header timeouts (alioss/gcs/s3)

alioss/README.md

@@ -16,7 +16,8 @@ The AliOSS client requires a JSON configuration file with the following structur
   "access_key_secret":         "<string> (required)",
   "endpoint":                  "<string> (required)",
   "bucket_name":               "<string> (required)",
-  "http_request_timeout":      "<string duration> (optional)"
+  "http_request_timeout":      "<string duration> (optional)",
+  "http_response_header_timeout": "<string duration> (optional)"
 }
 ```
 

alioss/client/storage_client.go

@@ -94,11 +94,22 @@ func NewStorageClient(storageConfig config.AliStorageConfig) (StorageClient, err
 	}, nil
 }
 
-func newOSSClient(endpoint, accessKeyID, accessKeySecret string, httpRequestTimeoutSeconds int64) (*oss.Client, error) {
+func setOSSResponseHeaderTimeout(timeout time.Duration) oss.ClientOption {
+	return func(client *oss.Client) {
+		if client.Config != nil {
+			client.Config.HTTPTimeout.HeaderTimeout = timeout
+		}
+	}
+}
+
+func newOSSClient(endpoint, accessKeyID, accessKeySecret string, httpRequestTimeoutSeconds int64, httpResponseHeaderTimeout time.Duration) (*oss.Client, error) {
 	options := make([]oss.ClientOption, 0, 3)
 	if httpRequestTimeoutSeconds > 0 {
 		options = append(options, oss.Timeout(httpRequestTimeoutSeconds, httpRequestTimeoutSeconds))
 	}
+	if httpResponseHeaderTimeout > 0 {
+		options = append(options, setOSSResponseHeaderTimeout(httpResponseHeaderTimeout))
+	}
 
 	if common.IsDebug() {
 		slogLogger := slog.Default()
@@ -115,11 +126,17 @@ func (dsc DefaultStorageClient) newOSSClient() (*oss.Client, error) {
 		return nil, err
 	}
 
+	httpResponseHeaderTimeout, err := dsc.storageConfig.HTTPResponseHeaderTimeoutDuration()
+	if err != nil {
+		return nil, err
+	}
+
 	return newOSSClient(
 		dsc.storageConfig.Endpoint,
 		dsc.storageConfig.AccessKeyID,
 		dsc.storageConfig.AccessKeySecret,
 		httpRequestTimeoutSeconds,
+		httpResponseHeaderTimeout,
 	)
 }
 

alioss/config/config.go

@@ -9,14 +9,16 @@ import (
 )
 
 type AliStorageConfig struct {
-	AccessKeyID        string `json:"access_key_id"`
-	AccessKeySecret    string `json:"access_key_secret"`
-	Endpoint           string `json:"endpoint"`
-	BucketName         string `json:"bucket_name"`
-	HTTPRequestTimeout string `json:"http_request_timeout"`
+	AccessKeyID               string `json:"access_key_id"`
+	AccessKeySecret           string `json:"access_key_secret"`
+	Endpoint                  string `json:"endpoint"`
+	BucketName                string `json:"bucket_name"`
+	HTTPRequestTimeout        string `json:"http_request_timeout"`
+	HTTPResponseHeaderTimeout string `json:"http_response_header_timeout"`
 }
 
 var errorNonPositiveHTTPRequestTimeout = errors.New("http_request_timeout must be greater than 0")
+var errorNonPositiveHTTPResponseHeaderTimeout = errors.New("http_response_header_timeout must be greater than 0")
 
 // NewFromReader returns a new ali-storage-cli configuration struct from the contents of reader.
 // reader.Read() is expected to return valid JSON
@@ -35,6 +37,9 @@ func NewFromReader(reader io.Reader) (AliStorageConfig, error) {
 	if _, err := config.HTTPRequestTimeoutSeconds(); err != nil {
 		return AliStorageConfig{}, err
 	}
+	if _, err := config.HTTPResponseHeaderTimeoutDuration(); err != nil {
+		return AliStorageConfig{}, err
+	}
 
 	return config, nil
 }
@@ -61,3 +66,20 @@ func (c AliStorageConfig) HTTPRequestTimeoutSeconds() (int64, error) {
 
 	return timeoutSeconds, nil
 }
+
+func (c AliStorageConfig) HTTPResponseHeaderTimeoutDuration() (time.Duration, error) {
+	if c.HTTPResponseHeaderTimeout == "" {
+		return 0, nil
+	}
+
+	httpResponseHeaderTimeout, err := time.ParseDuration(c.HTTPResponseHeaderTimeout)
+	if err != nil {
+		return 0, fmt.Errorf("invalid http_response_header_timeout: %w", err)
+	}
+
+	if httpResponseHeaderTimeout <= 0 {
+		return 0, errorNonPositiveHTTPResponseHeaderTimeout
+	}
+
+	return httpResponseHeaderTimeout, nil
+}

alioss/config/config_test.go

@@ -16,7 +16,8 @@ var _ = Describe("Config", func() {
 								"access_key_secret": "foo_access_key_secret",
                                 "endpoint": "foo_endpoint",
 								"bucket_name": "foo_bucket_name",
-								"http_request_timeout": "30s"}`)
+								"http_request_timeout": "30s",
+								"http_response_header_timeout": "5s"}`)
 		configReader := bytes.NewReader(configJson)
 
 		config, err := config.NewFromReader(configReader)
@@ -27,10 +28,14 @@ var _ = Describe("Config", func() {
 		Expect(config.Endpoint).To(Equal("foo_endpoint"))
 		Expect(config.BucketName).To(Equal("foo_bucket_name"))
 		Expect(config.HTTPRequestTimeout).To(Equal("30s"))
+		Expect(config.HTTPResponseHeaderTimeout).To(Equal("5s"))
 
 		timeoutSeconds, err := config.HTTPRequestTimeoutSeconds()
 		Expect(err).ToNot(HaveOccurred())
 		Expect(timeoutSeconds).To(Equal(int64(30)))
+		headerTimeout, err := config.HTTPResponseHeaderTimeoutDuration()
+		Expect(err).ToNot(HaveOccurred())
+		Expect(headerTimeout.Seconds()).To(Equal(5.0))
 	})
 
 	It("rounds up sub-second timeout in HTTPRequestTimeoutSeconds getter", func() {
@@ -65,6 +70,22 @@ var _ = Describe("Config", func() {
 		Expect(timeoutSeconds).To(BeZero())
 	})
 
+	It("leaves response header timeout unset when http_response_header_timeout is not provided", func() {
+		configJson := []byte(`{"access_key_id": "foo_access_key_id",
+								"access_key_secret": "foo_access_key_secret",
+								"endpoint": "foo_endpoint",
+								"bucket_name": "foo_bucket_name"}`)
+		configReader := bytes.NewReader(configJson)
+
+		config, err := config.NewFromReader(configReader)
+
+		Expect(err).ToNot(HaveOccurred())
+		Expect(config.HTTPResponseHeaderTimeout).To(BeEmpty())
+		headerTimeout, err := config.HTTPResponseHeaderTimeoutDuration()
+		Expect(err).ToNot(HaveOccurred())
+		Expect(headerTimeout).To(BeZero())
+	})
+
 	It("returns an error when http_request_timeout has invalid format", func() {
 		configJson := []byte(`{"access_key_id": "foo_access_key_id",
 								"access_key_secret": "foo_access_key_secret",
@@ -92,6 +113,33 @@ var _ = Describe("Config", func() {
 		Expect(err).To(MatchError("http_request_timeout must be greater than 0"))
 	})
 
+	It("returns an error when http_response_header_timeout has invalid format", func() {
+		configJson := []byte(`{"access_key_id": "foo_access_key_id",
+								"access_key_secret": "foo_access_key_secret",
+								"endpoint": "foo_endpoint",
+								"bucket_name": "foo_bucket_name",
+								"http_response_header_timeout": "bananas"}`)
+		configReader := bytes.NewReader(configJson)
+
+		_, err := config.NewFromReader(configReader)
+
+		Expect(err).To(HaveOccurred())
+		Expect(err.Error()).To(ContainSubstring("invalid http_response_header_timeout"))
+	})
+
+	It("returns an error when http_response_header_timeout is non-positive", func() {
+		configJson := []byte(`{"access_key_id": "foo_access_key_id",
+								"access_key_secret": "foo_access_key_secret",
+								"endpoint": "foo_endpoint",
+								"bucket_name": "foo_bucket_name",
+								"http_response_header_timeout": "0s"}`)
+		configReader := bytes.NewReader(configJson)
+
+		_, err := config.NewFromReader(configReader)
+
+		Expect(err).To(MatchError("http_response_header_timeout must be greater than 0"))
+	})
+
 	It("is empty if config cannot be parsed", func() {
 		configJson := []byte(`~`)
 		configReader := bytes.NewReader(configJson)

gcs/README.md

@@ -18,6 +18,7 @@ The GCS client requires a JSON configuration file.
   "json_key":               "<string> (required if credentials_source = 'static')",
   "storage_class":          "<string> (optional - default: 'STANDARD', check for more options=https://docs.cloud.google.com/storage/docs/storage-classes)",
   "http_request_timeout":   "<string duration> (optional)",
+  "http_response_header_timeout": "<string duration> (optional)",
   "encryption_key":         "<string> (optional)",
   "uniform_bucket_level_access": "<boolean> (optional)"
 }

gcs/client/sdk.go

@@ -21,15 +21,15 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"net/http"
+	"time"
 
 	"golang.org/x/oauth2"
 	"golang.org/x/oauth2/google"
 	"golang.org/x/oauth2/jwt"
 
 	"google.golang.org/api/option"
 
-	"net/http"
-
 	"cloud.google.com/go/storage"
 	"github.com/cloudfoundry/storage-cli/common"
 	"github.com/cloudfoundry/storage-cli/gcs/client/middleware"
@@ -38,15 +38,40 @@ import (
 
 const uaString = "storage-cli-gcs"
 
+func withResponseHeaderTimeout(base http.RoundTripper, timeout time.Duration) http.RoundTripper {
+	if timeout == 0 {
+		return base
+	}
+
+	if base == nil {
+		base = http.DefaultTransport
+	}
+
+	transport, ok := base.(*http.Transport)
+	if !ok {
+		return base
+	}
+
+	cloned := transport.Clone()
+	cloned.ResponseHeaderTimeout = timeout
+	return cloned
+}
+
 func newStorageClients(ctx context.Context, cfg *config.GCSCli) (*storage.Client, *storage.Client, error) {
 	requestTimeout, err := cfg.HTTPRequestTimeoutValue()
 	if err != nil {
 		return nil, nil, err
 	}
 
+	responseHeaderTimeout, err := cfg.HTTPResponseHeaderTimeoutDuration()
+	if err != nil {
+		return nil, nil, err
+	}
+
 	publicHTTPClient := &http.Client{Timeout: requestTimeout}
+	publicHTTPClient.Transport = withResponseHeaderTimeout(publicHTTPClient.Transport, responseHeaderTimeout)
 	if common.IsDebug() {
-		publicHTTPClient.Transport = middleware.NewLoggingTransport(http.DefaultTransport)
+		publicHTTPClient.Transport = middleware.NewLoggingTransport(publicHTTPClient.Transport)
 	}
 
 	publicClient, err := storage.NewClient(ctx, option.WithUserAgent(uaString), option.WithHTTPClient(publicHTTPClient))
@@ -60,6 +85,7 @@ func newStorageClients(ctx context.Context, cfg *config.GCSCli) (*storage.Client
 	case config.DefaultCredentialsSource:
 		if tokenSource, err = google.DefaultTokenSource(ctx, storage.ScopeFullControl); err == nil {
 			baseClient := oauth2.NewClient(ctx, tokenSource)
+			baseClient.Transport = withResponseHeaderTimeout(baseClient.Transport, responseHeaderTimeout)
 			if common.IsDebug() {
 				baseClient.Transport = middleware.NewLoggingTransport(baseClient.Transport)
 			}
@@ -70,6 +96,7 @@ func newStorageClients(ctx context.Context, cfg *config.GCSCli) (*storage.Client
 		if token, err = google.JWTConfigFromJSON([]byte(cfg.ServiceAccountFile), storage.ScopeFullControl); err == nil {
 			tokenSource := token.TokenSource(ctx)
 			baseClient := oauth2.NewClient(ctx, tokenSource)
+			baseClient.Transport = withResponseHeaderTimeout(baseClient.Transport, responseHeaderTimeout)
 			if common.IsDebug() {
 				baseClient.Transport = middleware.NewLoggingTransport(baseClient.Transport)
 			}

gcs/config/config.go

@@ -55,6 +55,9 @@ type GCSCli struct {
 	// HTTPRequestTimeout specifies the maximum duration for each GCS HTTP request.
 	// If empty, requests have no client-side timeout.
 	HTTPRequestTimeout string `json:"http_request_timeout"`
+	// HTTPResponseHeaderTimeout specifies how long to wait for response headers.
+	// If empty, no explicit response header timeout is configured.
+	HTTPResponseHeaderTimeout string `json:"http_response_header_timeout"`
 
 	EncryptionKeyEncoded string
 	EncryptionKeySha256  string
@@ -87,12 +90,14 @@ var ErrWrongLengthEncryptionKey = errors.New("encryption_key not 32 bytes")
 // ErrNonPositiveHTTPRequestTimeout is returned when http_request_timeout is <= 0.
 var ErrNonPositiveHTTPRequestTimeout = errors.New("http_request_timeout must be greater than 0")
 
+// ErrNonPositiveHTTPResponseHeaderTimeout is returned when http_response_header_timeout is <= 0.
+var ErrNonPositiveHTTPResponseHeaderTimeout = errors.New("http_response_header_timeout must be greater than 0")
+
 // NewFromReader returns the new gcscli configuration struct from the
 // contents of the reader.
 //
 // reader.Read() is expected to return valid JSON.
 func NewFromReader(reader io.Reader) (GCSCli, error) {
-
 	dec := json.NewDecoder(reader)
 	var c GCSCli
 	if err := dec.Decode(&c); err != nil {
@@ -103,8 +108,7 @@ func NewFromReader(reader io.Reader) (GCSCli, error) {
 		return GCSCli{}, ErrEmptyBucketName
 	}
 
-	if c.CredentialsSource == ServiceAccountFileCredentialsSource &&
-		c.ServiceAccountFile == "" {
+	if c.CredentialsSource == ServiceAccountFileCredentialsSource && c.ServiceAccountFile == "" {
 		return GCSCli{}, ErrEmptyServiceAccountFile
 	}
 
@@ -124,22 +128,34 @@ func NewFromReader(reader io.Reader) (GCSCli, error) {
 		return GCSCli{}, err
 	}
 
+	if _, err := c.HTTPResponseHeaderTimeoutDuration(); err != nil {
+		return GCSCli{}, err
+	}
+
 	return c, nil
 }
 
 func (c *GCSCli) HTTPRequestTimeoutValue() (time.Duration, error) {
-	if c.HTTPRequestTimeout == "" {
+	return parseOptionalPositiveDuration("http_request_timeout", c.HTTPRequestTimeout, ErrNonPositiveHTTPRequestTimeout)
+}
+
+func (c *GCSCli) HTTPResponseHeaderTimeoutDuration() (time.Duration, error) {
+	return parseOptionalPositiveDuration("http_response_header_timeout", c.HTTPResponseHeaderTimeout, ErrNonPositiveHTTPResponseHeaderTimeout)
+}
+
+func parseOptionalPositiveDuration(fieldName, value string, nonPositiveErr error) (time.Duration, error) {
+	if value == "" {
 		return 0, nil
 	}
 
-	requestTimeout, err := time.ParseDuration(c.HTTPRequestTimeout)
+	parsedDuration, err := time.ParseDuration(value)
 	if err != nil {
-		return 0, fmt.Errorf("invalid http_request_timeout: %w", err)
+		return 0, fmt.Errorf("invalid %s: %w", fieldName, err)
 	}
 
-	if requestTimeout <= 0 {
-		return 0, ErrNonPositiveHTTPRequestTimeout
+	if parsedDuration <= 0 {
+		return 0, nonPositiveErr
 	}
 
-	return requestTimeout, nil
+	return parsedDuration, nil
 }