Configurable http client timeouts for alioss/gcs/s3

README.md

@@ -27,7 +27,7 @@ Key points
 - [Azurebs](./azurebs/README.md)
 - [Dav](./dav/README.md)
   - additional endpoints needed by CAPI still missing
-- [Gcs](./gcs/README.md)
+- [GCS](./gcs/README.md)
 - [S3](./s3/README.md)
 
 

alioss/README.md

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

alioss/client/storage_client.go

@@ -94,20 +94,39 @@ func NewStorageClient(storageConfig config.AliStorageConfig) (StorageClient, err
 	}, nil
 }
 
-func newOSSClient(endpoint, accesKeyID, accessKeySecret string) (*oss.Client, error) {
+func newOSSClient(endpoint, accessKeyID, accessKeySecret string, httpRequestTimeoutSeconds int64) (*oss.Client, error) {
+	options := make([]oss.ClientOption, 0, 3)
+	if httpRequestTimeoutSeconds > 0 {
+		options = append(options, oss.Timeout(httpRequestTimeoutSeconds, httpRequestTimeoutSeconds))
+	}
+
 	if common.IsDebug() {
 		slogLogger := slog.Default()
 		ossLogger := slog.NewLogLogger(slogLogger.Handler(), slog.LevelDebug)
-		return oss.New(endpoint, accesKeyID, accessKeySecret, oss.SetLogLevel(oss.Debug), oss.SetLogger(ossLogger))
-	} else {
-		return oss.New(endpoint, accesKeyID, accessKeySecret)
+		options = append(options, oss.SetLogLevel(oss.Debug), oss.SetLogger(ossLogger))
+	}
+
+	return oss.New(endpoint, accessKeyID, accessKeySecret, options...)
+}
+
+func (dsc DefaultStorageClient) newOSSClient() (*oss.Client, error) {
+	httpRequestTimeoutSeconds, err := dsc.storageConfig.HTTPRequestTimeoutSeconds()
+	if err != nil {
+		return nil, err
 	}
+
+	return newOSSClient(
+		dsc.storageConfig.Endpoint,
+		dsc.storageConfig.AccessKeyID,
+		dsc.storageConfig.AccessKeySecret,
+		httpRequestTimeoutSeconds,
+	)
 }
 
 func (dsc DefaultStorageClient) Upload(sourceFilePath string, sourceFileMD5 string, destinationObject string) error {
 	slog.Info("Uploading object to OSS bucket", "bucket", dsc.storageConfig.BucketName, "object_key", destinationObject, "file_path", sourceFilePath)
 
-	client, err := newOSSClient(dsc.storageConfig.Endpoint, dsc.storageConfig.AccessKeyID, dsc.storageConfig.AccessKeySecret)
+	client, err := dsc.newOSSClient()
 	if err != nil {
 		return err
 	}
@@ -132,7 +151,7 @@ func (dsc DefaultStorageClient) Upload(sourceFilePath string, sourceFileMD5 stri
 func (dsc DefaultStorageClient) Download(sourceObject string, destinationFilePath string) error {
 	slog.Info("Downloading object from OSS bucket", "bucket", dsc.storageConfig.BucketName, "object_key", sourceObject, "file_path", destinationFilePath)
 
-	client, err := newOSSClient(dsc.storageConfig.Endpoint, dsc.storageConfig.AccessKeyID, dsc.storageConfig.AccessKeySecret)
+	client, err := dsc.newOSSClient()
 	if err != nil {
 		return err
 	}
@@ -150,7 +169,7 @@ func (dsc DefaultStorageClient) Copy(sourceObject string, destinationObject stri
 	srcOut := fmt.Sprintf("%s/%s", dsc.storageConfig.BucketName, sourceObject)
 	destOut := fmt.Sprintf("%s/%s", dsc.storageConfig.BucketName, destinationObject)
 
-	client, err := newOSSClient(dsc.storageConfig.Endpoint, dsc.storageConfig.AccessKeyID, dsc.storageConfig.AccessKeySecret)
+	client, err := dsc.newOSSClient()
 	if err != nil {
 		return err
 	}
@@ -170,7 +189,7 @@ func (dsc DefaultStorageClient) Copy(sourceObject string, destinationObject stri
 func (dsc DefaultStorageClient) Delete(object string) error {
 	slog.Info("Deleting object from OSS bucket", "bucket", dsc.storageConfig.BucketName, "object_key", object)
 
-	client, err := newOSSClient(dsc.storageConfig.Endpoint, dsc.storageConfig.AccessKeyID, dsc.storageConfig.AccessKeySecret)
+	client, err := dsc.newOSSClient()
 	if err != nil {
 		return err
 	}
@@ -190,7 +209,7 @@ func (dsc DefaultStorageClient) DeleteRecursive(prefix string) error {
 		slog.Info("Deleting all objects from OSS bucket", "bucket", dsc.storageConfig.BucketName)
 	}
 
-	client, err := newOSSClient(dsc.storageConfig.Endpoint, dsc.storageConfig.AccessKeyID, dsc.storageConfig.AccessKeySecret)
+	client, err := dsc.newOSSClient()
 	if err != nil {
 		return err
 	}
@@ -247,7 +266,7 @@ func (dsc DefaultStorageClient) DeleteRecursive(prefix string) error {
 func (dsc DefaultStorageClient) Exists(object string) (bool, error) {
 	slog.Info("Checking if object exists in OSS bucket", "bucket", dsc.storageConfig.BucketName, "object_key", object)
 
-	client, err := newOSSClient(dsc.storageConfig.Endpoint, dsc.storageConfig.AccessKeyID, dsc.storageConfig.AccessKeySecret)
+	client, err := dsc.newOSSClient()
 	if err != nil {
 		return false, err
 	}
@@ -274,7 +293,7 @@ func (dsc DefaultStorageClient) Exists(object string) (bool, error) {
 func (dsc DefaultStorageClient) SignedUrlPut(object string, expiredInSec int64) (string, error) {
 	slog.Info("Generating signed PUT URL for OSS object", "bucket", dsc.storageConfig.BucketName, "object_key", object, "expiration_seconds", expiredInSec)
 
-	client, err := newOSSClient(dsc.storageConfig.Endpoint, dsc.storageConfig.AccessKeyID, dsc.storageConfig.AccessKeySecret)
+	client, err := dsc.newOSSClient()
 	if err != nil {
 		return "", err
 	}
@@ -290,7 +309,7 @@ func (dsc DefaultStorageClient) SignedUrlPut(object string, expiredInSec int64)
 func (dsc DefaultStorageClient) SignedUrlGet(object string, expiredInSec int64) (string, error) {
 	slog.Info("Generating signed GET URL for OSS object", "bucket", dsc.storageConfig.BucketName, "object_key", object, "expiration_seconds", expiredInSec)
 
-	client, err := newOSSClient(dsc.storageConfig.Endpoint, dsc.storageConfig.AccessKeyID, dsc.storageConfig.AccessKeySecret)
+	client, err := dsc.newOSSClient()
 	if err != nil {
 		return "", err
 	}
@@ -324,7 +343,7 @@ func (dsc DefaultStorageClient) List(prefix string) ([]string, error) {
 			opts = append(opts, oss.Marker(marker))
 		}
 
-		client, err := newOSSClient(dsc.storageConfig.Endpoint, dsc.storageConfig.AccessKeyID, dsc.storageConfig.AccessKeySecret)
+		client, err := dsc.newOSSClient()
 		if err != nil {
 			return nil, err
 		}
@@ -361,7 +380,7 @@ type BlobProperties struct {
 func (dsc DefaultStorageClient) Properties(object string) error {
 	slog.Info("Getting object properties from OSS bucket", "bucket", dsc.storageConfig.BucketName, "object_key", object)
 
-	client, err := newOSSClient(dsc.storageConfig.Endpoint, dsc.storageConfig.AccessKeyID, dsc.storageConfig.AccessKeySecret)
+	client, err := dsc.newOSSClient()
 	if err != nil {
 		return err
 	}
@@ -423,7 +442,7 @@ func (dsc DefaultStorageClient) Properties(object string) error {
 func (dsc DefaultStorageClient) EnsureBucketExists() error {
 	slog.Info("Ensuring OSS bucket exists", "bucket", dsc.storageConfig.BucketName)
 
-	client, err := newOSSClient(dsc.storageConfig.Endpoint, dsc.storageConfig.AccessKeyID, dsc.storageConfig.AccessKeySecret)
+	client, err := dsc.newOSSClient()
 	if err != nil {
 		return err
 	}

alioss/config/config.go

@@ -2,16 +2,22 @@ package config
 
 import (
 	"encoding/json"
+	"errors"
+	"fmt"
 	"io"
+	"time"
 )
 
 type AliStorageConfig struct {
-	AccessKeyID     string `json:"access_key_id"`
-	AccessKeySecret string `json:"access_key_secret"`
-	Endpoint        string `json:"endpoint"`
-	BucketName      string `json:"bucket_name"`
+	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"`
 }
 
+var errorNonPositiveHTTPRequestTimeout = errors.New("http_request_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
 func NewFromReader(reader io.Reader) (AliStorageConfig, error) {
@@ -26,5 +32,32 @@ func NewFromReader(reader io.Reader) (AliStorageConfig, error) {
 		return AliStorageConfig{}, err
 	}
 
+	if _, err := config.HTTPRequestTimeoutSeconds(); err != nil {
+		return AliStorageConfig{}, err
+	}
+
 	return config, nil
 }
+
+func (c AliStorageConfig) HTTPRequestTimeoutSeconds() (int64, error) {
+	if c.HTTPRequestTimeout == "" {
+		return 0, nil
+	}
+
+	httpRequestTimeout, err := time.ParseDuration(c.HTTPRequestTimeout)
+	if err != nil {
+		return 0, fmt.Errorf("invalid http_request_timeout: %w", err)
+	}
+
+	if httpRequestTimeout <= 0 {
+		return 0, errorNonPositiveHTTPRequestTimeout
+	}
+
+	// round up if necessary
+	timeoutSeconds := int64(httpRequestTimeout / time.Second)
+	if httpRequestTimeout%time.Second != 0 {
+		timeoutSeconds++
+	}
+
+	return timeoutSeconds, nil
+}

alioss/config/config_test.go

@@ -15,7 +15,8 @@ var _ = Describe("Config", 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"}`)
+								"bucket_name": "foo_bucket_name",
+								"http_request_timeout": "30s"}`)
 		configReader := bytes.NewReader(configJson)
 
 		config, err := config.NewFromReader(configReader)
@@ -25,6 +26,70 @@ var _ = Describe("Config", func() {
 		Expect(config.AccessKeySecret).To(Equal("foo_access_key_secret"))
 		Expect(config.Endpoint).To(Equal("foo_endpoint"))
 		Expect(config.BucketName).To(Equal("foo_bucket_name"))
+		Expect(config.HTTPRequestTimeout).To(Equal("30s"))
+
+		timeoutSeconds, err := config.HTTPRequestTimeoutSeconds()
+		Expect(err).ToNot(HaveOccurred())
+		Expect(timeoutSeconds).To(Equal(int64(30)))
+	})
+
+	It("rounds up sub-second timeout in HTTPRequestTimeoutSeconds getter", 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_request_timeout": "1500ms"}`)
+		configReader := bytes.NewReader(configJson)
+
+		config, err := config.NewFromReader(configReader)
+
+		Expect(err).ToNot(HaveOccurred())
+		timeoutSeconds, err := config.HTTPRequestTimeoutSeconds()
+		Expect(err).ToNot(HaveOccurred())
+		Expect(timeoutSeconds).To(Equal(int64(2)))
+	})
+
+	It("leaves timeout unset when http_request_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.HTTPRequestTimeout).To(BeEmpty())
+		timeoutSeconds, err := config.HTTPRequestTimeoutSeconds()
+		Expect(err).ToNot(HaveOccurred())
+		Expect(timeoutSeconds).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",
+								"endpoint": "foo_endpoint",
+								"bucket_name": "foo_bucket_name",
+								"http_request_timeout": "bananas"}`)
+		configReader := bytes.NewReader(configJson)
+
+		_, err := config.NewFromReader(configReader)
+
+		Expect(err).To(HaveOccurred())
+		Expect(err.Error()).To(ContainSubstring("invalid http_request_timeout"))
+	})
+
+	It("returns an error when http_request_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_request_timeout": "0s"}`)
+		configReader := bytes.NewReader(configJson)
+
+		_, err := config.NewFromReader(configReader)
+
+		Expect(err).To(MatchError("http_request_timeout must be greater than 0"))
 	})
 
 	It("is empty if config cannot be parsed", func() {

alioss/integration/general_ali_test.go

@@ -194,6 +194,33 @@ var _ = Describe("General testing for all Ali regions", func() {
 			fileContent, _ := os.ReadFile(outputFilePath) //nolint:errcheck
 			Expect(string(fileContent)).To(Equal("foo"))
 		})
+
+		It("downloads a file when http_request_timeout is set", func() {
+			outputFilePath := "/tmp/" + integration.GenerateRandomString()
+			cfg := defaultConfig
+			cfg.HTTPRequestTimeout = "30s"
+			timeoutConfigPath := integration.MakeConfigFile(&cfg)
+
+			defer func() {
+				cliSession, err := integration.RunCli(cliPath, timeoutConfigPath, storageType, "delete", blobName)
+				Expect(err).ToNot(HaveOccurred())
+				Expect(cliSession.ExitCode()).To(BeZero())
+
+				_ = os.Remove(outputFilePath)    //nolint:errcheck
+				_ = os.Remove(timeoutConfigPath) //nolint:errcheck
+			}()
+
+			cliSession, err := integration.RunCli(cliPath, timeoutConfigPath, storageType, "put", contentFile, blobName)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(cliSession.ExitCode()).To(BeZero())
+
+			cliSession, err = integration.RunCli(cliPath, timeoutConfigPath, storageType, "get", blobName, outputFilePath)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(cliSession.ExitCode()).To(BeZero())
+
+			fileContent, _ := os.ReadFile(outputFilePath) //nolint:errcheck
+			Expect(string(fileContent)).To(Equal("foo"))
+		})
 	})
 
 	Describe("Invoking `delete`", func() {

gcs/README.md

@@ -17,6 +17,7 @@ The GCS client requires a JSON configuration file.
   "credentials_source":     "<string> ['static'|'none'|""]",
   "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)",
   "encryption_key":         "<string> (optional)",
   "uniform_bucket_level_access": "<boolean> (optional)"
 }

gcs/client/sdk.go

@@ -39,40 +39,42 @@ import (
 const uaString = "storage-cli-gcs"
 
 func newStorageClients(ctx context.Context, cfg *config.GCSCli) (*storage.Client, *storage.Client, error) {
-	publicClient, err := storage.NewClient(ctx, option.WithUserAgent(uaString), option.WithHTTPClient(http.DefaultClient))
+	requestTimeout, err := cfg.HTTPRequestTimeoutValue()
+	if err != nil {
+		return nil, nil, err
+	}
+
+	publicHTTPClient := &http.Client{Timeout: requestTimeout}
+	if common.IsDebug() {
+		publicHTTPClient.Transport = middleware.NewLoggingTransport(http.DefaultTransport)
+	}
+
+	publicClient, err := storage.NewClient(ctx, option.WithUserAgent(uaString), option.WithHTTPClient(publicHTTPClient))
 	var authenticatedClient *storage.Client
 	var tokenSource oauth2.TokenSource
 	var token *jwt.Config
 
 	switch cfg.CredentialsSource {
 	case config.NoneCredentialsSource:
-		if common.IsDebug() {
-			httpClient := &http.Client{
-				Transport: middleware.NewLoggingTransport(http.DefaultTransport),
-			}
-			publicClient, err = storage.NewClient(ctx, option.WithUserAgent(uaString), option.WithHTTPClient(httpClient))
-		}
+		// Public client already initialized with the configured timeout.
 	case config.DefaultCredentialsSource:
 		if tokenSource, err = google.DefaultTokenSource(ctx, storage.ScopeFullControl); err == nil {
+			baseClient := oauth2.NewClient(ctx, tokenSource)
 			if common.IsDebug() {
-				baseClient := oauth2.NewClient(ctx, tokenSource)
 				baseClient.Transport = middleware.NewLoggingTransport(baseClient.Transport)
-				authenticatedClient, err = storage.NewClient(ctx, option.WithHTTPClient(baseClient), option.WithUserAgent(uaString))
-
-			} else {
-				authenticatedClient, err = storage.NewClient(ctx, option.WithUserAgent(uaString), option.WithTokenSource(tokenSource)) //nolint:ineffassign,staticcheck
 			}
+			baseClient.Timeout = requestTimeout
+			authenticatedClient, err = storage.NewClient(ctx, option.WithHTTPClient(baseClient), option.WithUserAgent(uaString))
 		}
 	case config.ServiceAccountFileCredentialsSource:
 		if token, err = google.JWTConfigFromJSON([]byte(cfg.ServiceAccountFile), storage.ScopeFullControl); err == nil {
+			tokenSource := token.TokenSource(ctx)
+			baseClient := oauth2.NewClient(ctx, tokenSource)
 			if common.IsDebug() {
-				tokenSource := token.TokenSource(ctx)
-				baseClient := oauth2.NewClient(ctx, tokenSource)
 				baseClient.Transport = middleware.NewLoggingTransport(baseClient.Transport)
-				authenticatedClient, err = storage.NewClient(ctx, option.WithHTTPClient(baseClient), option.WithUserAgent(uaString))
-			} else {
-				authenticatedClient, err = storage.NewClient(ctx, option.WithUserAgent(uaString), option.WithTokenSource(token.TokenSource(ctx))) //nolint:ineffassign,staticcheck
 			}
+			baseClient.Timeout = requestTimeout
+			authenticatedClient, err = storage.NewClient(ctx, option.WithHTTPClient(baseClient), option.WithUserAgent(uaString))
 		}
 	default:
 		return nil, nil, errors.New("unknown credentials_source in configuration")