Add Support for retries when using Azure Blob Storage

CHANGELOG.md

@@ -15,6 +15,8 @@ We use *breaking* word for marking changes that are not backward compatible (rel
 
 - [#1097](https://github.com/improbable-eng/thanos/pull/1097) Added `thanos check rules` linter for Thanos rule rules files.
 
+- [#1253](https://github.com/improbable-eng/thanos/pull/1253) Add support for specifying a maximum amount of retries when using Azure Blob storage (default: no retries).
+
 ## [v0.5.0](https://github.com/improbable-eng/thanos/releases/tag/v0.5.0) - 2019.06.05
 
 TL;DR: Store LRU cache is no longer leaking, Upgraded Thanos UI to Prometheus 2.9, Fixed auto-downsampling, Moved to Go 1.12.5 and more.

docs/storage.md

@@ -255,6 +255,7 @@ config:
   storage_account_key: ""
   container: ""
   endpoint: ""
+  max_retries: 0
 ```
 
 ### OpenStack Swift Configuration

pkg/objstore/azure/azure.go

@@ -27,6 +27,7 @@ type Config struct {
 	StorageAccountKey  string `yaml:"storage_account_key"`
 	ContainerName      string `yaml:"container"`
 	Endpoint           string `yaml:"endpoint"`
+	MaxRetries         int    `yaml:"max_retries"`
 }
 
 // Bucket implements the store.Bucket interface against Azure APIs.
@@ -43,17 +44,20 @@ func (conf *Config) validate() error {
 		return errors.New("invalid Azure storage configuration")
 	}
 	if conf.StorageAccountName == "" && conf.StorageAccountKey != "" {
-		return errors.New("no Azure storage_account specified while storage_account_key is present in config file; both should be present.")
+		return errors.New("no Azure storage_account specified while storage_account_key is present in config file; both should be present")
 	}
 	if conf.StorageAccountName != "" && conf.StorageAccountKey == "" {
-		return errors.New("no Azure storage_account_key specified while storage_account is present in config file; both should be present.")
+		return errors.New("no Azure storage_account_key specified while storage_account is present in config file; both should be present")
 	}
 	if conf.ContainerName == "" {
 		return errors.New("no Azure container specified")
 	}
 	if conf.Endpoint == "" {
 		conf.Endpoint = azureDefaultEndpoint
 	}
+	if conf.MaxRetries < 0 {
+		return errors.New("the value of maxretries must be greater than or equal to 0 in the config file")
+	}
 	return nil
 }
 
@@ -199,6 +203,9 @@ func (b *Bucket) getBlobReader(ctx context.Context, name string, offset, length
 			BlockSize:   blob.BlobDefaultDownloadBlockSize,
 			Parallelism: uint16(3),
 			Progress:    nil,
+			RetryReaderOptionsPerBlock: blob.RetryReaderOptions{
+				MaxRetryRequests: b.config.MaxRetries,
+			},
 		},
 	); err != nil {
 		return nil, errors.Wrapf(err, "cannot download blob, address: %s", blobURL.BlobURL)

pkg/objstore/azure/azure_test.go

@@ -12,6 +12,7 @@ func TestConfig_validate(t *testing.T) {
 		StorageAccountKey  string
 		ContainerName      string
 		Endpoint           string
+		MaxRetries         int
 	}
 	tests := []struct {
 		name         string
@@ -25,6 +26,7 @@ func TestConfig_validate(t *testing.T) {
 				StorageAccountName: "foo",
 				StorageAccountKey:  "bar",
 				ContainerName:      "roo",
+				MaxRetries:         3,
 			},
 			wantErr:      false,
 			wantEndpoint: azureDefaultEndpoint,
@@ -67,6 +69,17 @@ func TestConfig_validate(t *testing.T) {
 			},
 			wantErr: true,
 		},
+		{
+			name: "invalid max retries (negative)",
+			fields: fields{
+				StorageAccountName: "foo",
+				StorageAccountKey:  "bar",
+				ContainerName:      "roo",
+				MaxRetries:         -3,
+			},
+			wantErr:      true,
+			wantEndpoint: azureDefaultEndpoint,
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
@@ -75,6 +88,7 @@ func TestConfig_validate(t *testing.T) {
 				StorageAccountKey:  tt.fields.StorageAccountKey,
 				ContainerName:      tt.fields.ContainerName,
 				Endpoint:           tt.fields.Endpoint,
+				MaxRetries:         tt.fields.MaxRetries,
 			}
 			err := conf.validate()
 			if (err != nil) != tt.wantErr {

pkg/objstore/azure/helpers.go

@@ -21,7 +21,9 @@ func getContainerURL(ctx context.Context, conf Config) (blob.ContainerURL, error
 		return blob.ContainerURL{}, err
 	}
 
-	retryOptions := blob.RetryOptions{}
+	retryOptions := blob.RetryOptions{
+		MaxTries: int32(conf.MaxRetries),
+	}
 	if deadline, ok := ctx.Deadline(); ok {
 		retryOptions.TryTimeout = deadline.Sub(time.Now())
 	}