diff --git a/README.md b/README.md index 05c249b7..8e63fb64 100644 --- a/README.md +++ b/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) diff --git a/alioss/README.md b/alioss/README.md index 595b68bd..a33f39ff 100644 --- a/alioss/README.md +++ b/alioss/README.md @@ -15,7 +15,8 @@ The AliOSS client requires a JSON configuration file with the following structur "access_key_id": " (required)", "access_key_secret": " (required)", "endpoint": " (required)", - "bucket_name": " (required)" + "bucket_name": " (required)", + "http_request_timeout": " (optional)" } ``` diff --git a/alioss/client/storage_client.go b/alioss/client/storage_client.go index 6592bdfb..9a942847 100644 --- a/alioss/client/storage_client.go +++ b/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 } diff --git a/alioss/config/config.go b/alioss/config/config.go index 72fb15b4..773a08ca 100644 --- a/alioss/config/config.go +++ b/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 +} diff --git a/alioss/config/config_test.go b/alioss/config/config_test.go index 6212d37d..166410aa 100644 --- a/alioss/config/config_test.go +++ b/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() { diff --git a/alioss/integration/general_ali_test.go b/alioss/integration/general_ali_test.go index b380a686..070a9d44 100644 --- a/alioss/integration/general_ali_test.go +++ b/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() { diff --git a/gcs/README.md b/gcs/README.md index 7917f940..7746fe4b 100644 --- a/gcs/README.md +++ b/gcs/README.md @@ -17,6 +17,7 @@ The GCS client requires a JSON configuration file. "credentials_source": " ['static'|'none'|""]", "json_key": " (required if credentials_source = 'static')", "storage_class": " (optional - default: 'STANDARD', check for more options=https://docs.cloud.google.com/storage/docs/storage-classes)", + "http_request_timeout": " (optional)", "encryption_key": " (optional)", "uniform_bucket_level_access": " (optional)" } diff --git a/gcs/client/sdk.go b/gcs/client/sdk.go index e1e155b2..dd14c8cf 100644 --- a/gcs/client/sdk.go +++ b/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") diff --git a/gcs/config/config.go b/gcs/config/config.go index 28dcf090..0594780e 100644 --- a/gcs/config/config.go +++ b/gcs/config/config.go @@ -21,7 +21,9 @@ import ( "encoding/base64" "encoding/json" "errors" + "fmt" "io" + "time" ) // GCSCli represents the configuration for the gcscli @@ -50,6 +52,9 @@ type GCSCli struct { // GCS transparently encrypts data using server-side encryption keys. // https://cloud.google.com/storage/docs/encryption EncryptionKey []byte `json:"encryption_key"` + // HTTPRequestTimeout specifies the maximum duration for each GCS HTTP request. + // If empty, requests have no client-side timeout. + HTTPRequestTimeout string `json:"http_request_timeout"` EncryptionKeyEncoded string EncryptionKeySha256 string @@ -79,6 +84,9 @@ var ErrEmptyServiceAccountFile = errors.New("json_key must be set") // in the config is not exactly 32 bytes. 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") + // NewFromReader returns the new gcscli configuration struct from the // contents of the reader. // @@ -112,5 +120,26 @@ func NewFromReader(reader io.Reader) (GCSCli, error) { c.EncryptionKeySha256 = base64.StdEncoding.EncodeToString(encryptionKeySha.Sum(nil)) } + if _, err := c.HTTPRequestTimeoutValue(); err != nil { + return GCSCli{}, err + } + return c, nil } + +func (c *GCSCli) HTTPRequestTimeoutValue() (time.Duration, error) { + if c.HTTPRequestTimeout == "" { + return 0, nil + } + + requestTimeout, err := time.ParseDuration(c.HTTPRequestTimeout) + if err != nil { + return 0, fmt.Errorf("invalid http_request_timeout: %w", err) + } + + if requestTimeout <= 0 { + return 0, ErrNonPositiveHTTPRequestTimeout + } + + return requestTimeout, nil +} diff --git a/gcs/config/config_suite_test.go b/gcs/config/config_suite_test.go index 513cc3c9..71da09b8 100644 --- a/gcs/config/config_suite_test.go +++ b/gcs/config/config_suite_test.go @@ -25,5 +25,5 @@ import ( func TestConfig(t *testing.T) { RegisterFailHandler(Fail) - RunSpecs(t, "Gcs Config Suite") + RunSpecs(t, "GCS Config Suite") } diff --git a/gcs/config/config_test.go b/gcs/config/config_test.go index 5031fd33..372106f5 100644 --- a/gcs/config/config_test.go +++ b/gcs/config/config_test.go @@ -169,4 +169,53 @@ var _ = Describe("BlobstoreClient configuration", func() { }) }) + Describe("when http_request_timeout is set", func() { + dummyJSONBytes := []byte(`{"bucket_name": "some-bucket", "http_request_timeout":"30s"}`) + dummyJSONReader := bytes.NewReader(dummyJSONBytes) + + It("parses and stores timeout", func() { + c, err := NewFromReader(dummyJSONReader) + Expect(err).To(BeNil()) + Expect(c.HTTPRequestTimeout).To(Equal("30s")) + requestTimeoutValue, err := c.HTTPRequestTimeoutValue() + Expect(err).To(BeNil()) + Expect(requestTimeoutValue.Seconds()).To(Equal(30.0)) + }) + }) + + Describe("when http_request_timeout is not set", func() { + dummyJSONBytes := []byte(`{"bucket_name": "some-bucket"}`) + dummyJSONReader := bytes.NewReader(dummyJSONBytes) + + It("leaves timeout unset", func() { + c, err := NewFromReader(dummyJSONReader) + Expect(err).To(BeNil()) + Expect(c.HTTPRequestTimeout).To(BeEmpty()) + requestTimeoutValue, err := c.HTTPRequestTimeoutValue() + Expect(err).To(BeNil()) + Expect(requestTimeoutValue).To(BeZero()) + }) + }) + + Describe("when http_request_timeout has invalid format", func() { + dummyJSONBytes := []byte(`{"bucket_name": "some-bucket", "http_request_timeout":"bananas"}`) + dummyJSONReader := bytes.NewReader(dummyJSONBytes) + + It("returns an error", func() { + _, err := NewFromReader(dummyJSONReader) + Expect(err).To(HaveOccurred()) + Expect(err.Error()).To(ContainSubstring("invalid http_request_timeout")) + }) + }) + + Describe("when http_request_timeout is non-positive", func() { + dummyJSONBytes := []byte(`{"bucket_name": "some-bucket", "http_request_timeout":"0s"}`) + dummyJSONReader := bytes.NewReader(dummyJSONBytes) + + It("returns an error", func() { + _, err := NewFromReader(dummyJSONReader) + Expect(err).To(MatchError(ErrNonPositiveHTTPRequestTimeout)) + }) + }) + }) diff --git a/gcs/integration/gcs_general_test.go b/gcs/integration/gcs_general_test.go index 49c4fa8f..e01b9e12 100644 --- a/gcs/integration/gcs_general_test.go +++ b/gcs/integration/gcs_general_test.go @@ -50,6 +50,15 @@ var _ = Describe("Integration", func() { }, configurations) + DescribeTable("Blobstore lifecycle works with http_request_timeout set", + func(cfg *config.GCSCli) { + cfgCopy := *cfg + cfgCopy.HTTPRequestTimeout = "30s" + env.AddConfig(&cfgCopy) + AssertLifecycleWorks(gcsCLIPath, env) + }, + configurations) + DescribeTable("Delete silently ignores that the file doesn't exist", func(config *config.GCSCli) { env.AddConfig(config) diff --git a/s3/README.md b/s3/README.md index 886084a0..c006ebc9 100644 --- a/s3/README.md +++ b/s3/README.md @@ -21,6 +21,7 @@ The S3 client requires a JSON configuration file with the following structure: "port": (optional), "ssl_verify_peer": (optional - default: true), "use_ssl": (optional - default: true), + "http_request_timeout": " (optional)", "signature_version": " (optional)", "server_side_encryption": " (optional)", "sse_kms_key_id": " (optional)", @@ -37,6 +38,8 @@ The S3 client requires a JSON configuration file with the following structure: } ``` +If `http_request_timeout` is omitted, the HTTP client timeout is left unset. + **Usage examples:** ```shell # Upload a file to S3 diff --git a/s3/client/sdk.go b/s3/client/sdk.go index 11f654e1..c2b97324 100644 --- a/s3/client/sdk.go +++ b/s3/client/sdk.go @@ -46,6 +46,12 @@ func NewAwsS3ClientWithApiOptions( httpClient.Transport = s3middleware.NewS3LoggingTransport(httpClient.Transport) } + httpRequestTimeout, err := c.HTTPRequestTimeoutValue() + if err != nil { + return nil, err + } + httpClient.Timeout = httpRequestTimeout + options := []func(*config.LoadOptions) error{ config.WithHTTPClient(httpClient), } diff --git a/s3/config/config.go b/s3/config/config.go index 9fb08d71..0a74c38c 100644 --- a/s3/config/config.go +++ b/s3/config/config.go @@ -6,10 +6,12 @@ import ( "fmt" "io" "math" + "strconv" "strings" + "time" ) -// The S3Cli represents configuration for the s3cli +// The S3Cli represents configuration for the s3 cli type S3Cli struct { AccessKeyID string `json:"access_key_id"` SecretAccessKey string `json:"secret_access_key"` @@ -21,6 +23,7 @@ type S3Cli struct { Region string `json:"region"` SSLVerifyPeer bool `json:"ssl_verify_peer"` UseSSL bool `json:"use_ssl"` + HTTPRequestTimeout string `json:"http_request_timeout"` ServerSideEncryption string `json:"server_side_encryption"` SSEKMSKeyID string `json:"sse_kms_key_id"` AssumeRoleArn string `json:"assume_role_arn"` @@ -70,6 +73,7 @@ const credentialsSourceEnvOrProfile = "env_or_profile" const noCredentialsSourceProvided = "" var errorStaticCredentialsMissing = errors.New("access_key_id and secret_access_key must be provided") +var errorNonPositiveHTTPRequestTimeout = errors.New("http_request_timeout must be greater than 0") type errorStaticCredentialsPresent struct { credentialsSource string @@ -132,6 +136,10 @@ func NewFromReader(reader io.Reader) (S3Cli, error) { return S3Cli{}, fmt.Errorf("multipart_copy_part_size must be at least %d bytes (5MB - AWS minimum)", multipartCopyMinPartSize) } + if _, err := c.HTTPRequestTimeoutValue(); err != nil { + return S3Cli{}, err + } + switch c.CredentialsSource { case StaticCredentialsSource: if c.AccessKeyID == "" || c.SecretAccessKey == "" { @@ -254,3 +262,24 @@ func (c *S3Cli) ShouldDisableResponseChecksumCalculation() bool { func (c *S3Cli) ShouldDisableUploaderRequestChecksumCalculation() bool { return !c.UploaderRequestChecksumCalculationEnabled } + +func (c *S3Cli) HTTPRequestTimeoutValue() (time.Duration, error) { + if c.HTTPRequestTimeout == "" { + return 0, nil + } + + if _, err := strconv.ParseFloat(c.HTTPRequestTimeout, 64); err == nil { + return 0, fmt.Errorf("invalid http_request_timeout: missing duration unit") + } + + 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 + } + + return httpRequestTimeout, nil +} diff --git a/s3/config/config_test.go b/s3/config/config_test.go index ad7923de..f1765c29 100644 --- a/s3/config/config_test.go +++ b/s3/config/config_test.go @@ -396,6 +396,66 @@ var _ = Describe("BlobstoreClient configuration", func() { }) }) + Describe("http_request_timeout", func() { + It("leaves timeout unset when not set", func() { + dummyJSONBytes := []byte(`{"access_key_id":"id","secret_access_key":"key","bucket_name":"some-bucket"}`) + dummyJSONReader := bytes.NewReader(dummyJSONBytes) + + c, err := config.NewFromReader(dummyJSONReader) + Expect(err).ToNot(HaveOccurred()) + Expect(c.HTTPRequestTimeout).To(BeEmpty()) + timeout, err := c.HTTPRequestTimeoutValue() + Expect(err).ToNot(HaveOccurred()) + Expect(timeout).To(BeZero()) + }) + + It("parses a valid duration", func() { + dummyJSONBytes := []byte(`{"access_key_id":"id","secret_access_key":"key","bucket_name":"some-bucket","http_request_timeout":"45s"}`) + dummyJSONReader := bytes.NewReader(dummyJSONBytes) + + c, err := config.NewFromReader(dummyJSONReader) + Expect(err).ToNot(HaveOccurred()) + Expect(c.HTTPRequestTimeout).To(Equal("45s")) + timeout, err := c.HTTPRequestTimeoutValue() + Expect(err).ToNot(HaveOccurred()) + Expect(timeout.Seconds()).To(Equal(45.0)) + }) + + It("rejects numeric timeout values", func() { + dummyJSONBytes := []byte(`{"access_key_id":"id","secret_access_key":"key","bucket_name":"some-bucket","http_request_timeout":45}`) + dummyJSONReader := bytes.NewReader(dummyJSONBytes) + + _, err := config.NewFromReader(dummyJSONReader) + Expect(err).To(HaveOccurred()) + Expect(err.Error()).To(ContainSubstring("cannot unmarshal number into Go struct field")) + }) + + It("rejects invalid duration formats", func() { + dummyJSONBytes := []byte(`{"access_key_id":"id","secret_access_key":"key","bucket_name":"some-bucket","http_request_timeout":"bananas"}`) + dummyJSONReader := bytes.NewReader(dummyJSONBytes) + + _, err := config.NewFromReader(dummyJSONReader) + Expect(err).To(HaveOccurred()) + Expect(err.Error()).To(ContainSubstring("invalid http_request_timeout")) + }) + + It("rejects negative durations", func() { + dummyJSONBytes := []byte(`{"access_key_id":"id","secret_access_key":"key","bucket_name":"some-bucket","http_request_timeout":"-1s"}`) + dummyJSONReader := bytes.NewReader(dummyJSONBytes) + + _, err := config.NewFromReader(dummyJSONReader) + Expect(err).To(MatchError("http_request_timeout must be greater than 0")) + }) + + It("rejects zero durations", func() { + dummyJSONBytes := []byte(`{"access_key_id":"id","secret_access_key":"key","bucket_name":"some-bucket","http_request_timeout":"0s"}`) + dummyJSONReader := bytes.NewReader(dummyJSONBytes) + + _, err := config.NewFromReader(dummyJSONReader) + Expect(err).To(MatchError("http_request_timeout must be greater than 0")) + }) + }) + Describe("returning the S3 endpoint", func() { Context("when port is provided", func() { It("returns a URI in the form `host:port`", func() { diff --git a/s3/integration/general_aws_test.go b/s3/integration/general_aws_test.go index 3d52a7db..d808baf7 100644 --- a/s3/integration/general_aws_test.go +++ b/s3/integration/general_aws_test.go @@ -59,6 +59,14 @@ var _ = Describe("General testing for all AWS regions", Label("aws", "static", " func(cfg *config.S3Cli) { integration.AssertLifecycleWorks(s3CLIPath, cfg) }, configurations, ) + DescribeTable("Blobstore lifecycle works with http_request_timeout set", + func(cfg *config.S3Cli) { + cfgCopy := *cfg + cfgCopy.HTTPRequestTimeout = "30s" + integration.AssertLifecycleWorks(s3CLIPath, &cfgCopy) + }, + configurations, + ) DescribeTable("Invoking `ensure-storage-exists` works", func(cfg *config.S3Cli) { integration.AssertOnStorageExists(s3CLIPath, cfg) },