[querier] s3: add getObject retry

CHANGELOG.md

@@ -1,6 +1,7 @@
 ## Main
 * [4400](https://github.com/grafana/loki/pull/4400) **trevorwhitney**: Config: automatically apply memberlist config too all rings when provided
 * [4443](https://github.com/grafana/loki/pull/4443) **DylanGuedes**: Loki: Change how push API checks for contentType
+* [4453](https://github.com/grafana/loki/pull/4453) **liguozhong**: Loki: Implement retry to s3 chunk storage
 
 # 2.3.0 (2021/08/06)
 

docs/sources/configuration/_index.md

@@ -1215,6 +1215,19 @@ aws:
     # endpoint.
     # CLI flag: -s3.http.ca-file
     [ca_file: <string> | default = ""]
+  # Configures backoff when s3 get Object.
+  backoff_config:
+    # Minimum delay when backing off.
+    # CLI flag: -s3.backoff-min-period
+    [min_period: <duration> | default = 100ms]
+
+    # The maximum delay when backing off.
+    # CLI flag: -s3.backoff-max-period
+    [max_period: <duration> | default = 3s]
+
+    # Number of times to backoff and retry before failing.
+    # CLI flag: -s3.backoff-retries
+    [max_retries: <int> | default = 3]
 
   # Configure the DynamoDB connection
   dynamodb:

pkg/storage/chunk/aws/s3_storage_client.go

@@ -30,6 +30,7 @@ import (
 
 	cortex_s3 "github.com/cortexproject/cortex/pkg/storage/bucket/s3"
 	"github.com/cortexproject/cortex/pkg/util"
+	"github.com/grafana/dskit/backoff"
 	"github.com/grafana/dskit/flagext"
 
 	"github.com/grafana/loki/pkg/storage/chunk"
@@ -75,6 +76,7 @@ type S3Config struct {
 	HTTPConfig       HTTPConfig          `yaml:"http_config"`
 	SignatureVersion string              `yaml:"signature_version"`
 	SSEConfig        cortex_s3.SSEConfig `yaml:"sse"`
+	BackoffConfig    backoff.Config      `yaml:"backoff_config"`
 
 	Inject InjectRequestMiddleware `yaml:"-"`
 }
@@ -115,6 +117,9 @@ func (cfg *S3Config) RegisterFlagsWithPrefix(prefix string, f *flag.FlagSet) {
 	f.BoolVar(&cfg.HTTPConfig.InsecureSkipVerify, prefix+"s3.http.insecure-skip-verify", false, "Set to true to skip verifying the certificate chain and hostname.")
 	f.StringVar(&cfg.HTTPConfig.CAFile, prefix+"s3.http.ca-file", "", "Path to the trusted CA file that signed the SSL certificate of the S3 endpoint.")
 	f.StringVar(&cfg.SignatureVersion, prefix+"s3.signature-version", SignatureVersionV4, fmt.Sprintf("The signature version to use for authenticating against S3. Supported values are: %s.", strings.Join(supportedSignatureVersions, ", ")))
+	f.DurationVar(&cfg.BackoffConfig.MinBackoff, "s3.min-backoff", 100*time.Millisecond, "Minimum backoff time when s3 get Object")
+	f.DurationVar(&cfg.BackoffConfig.MaxBackoff, "s3.max-backoff", 3*time.Second, "Maximum backoff time when s3 get Object")
+	f.IntVar(&cfg.BackoffConfig.MaxRetries, "s3.max-retries", 5, "Maximum number of times to retry when s3 get Object")
 }
 
 // Validate config and returns error on failure
@@ -126,6 +131,7 @@ func (cfg *S3Config) Validate() error {
 }
 
 type S3ObjectClient struct {
+	cfg         S3Config
 	bucketNames []string
 	S3          s3iface.S3API
 	sseConfig   *SSEParsedConfig
@@ -155,6 +161,7 @@ func NewS3ObjectClient(cfg S3Config) (*S3ObjectClient, error) {
 	}
 
 	client := S3ObjectClient{
+		cfg:         cfg,
 		S3:          s3Client,
 		bucketNames: bucketNames,
 		sseConfig:   sseCfg,
@@ -333,19 +340,29 @@ func (a *S3ObjectClient) GetObject(ctx context.Context, objectKey string) (io.Re
 	// Map the key into a bucket
 	bucket := a.bucketFromKey(objectKey)
 
-	err := instrument.CollectedRequest(ctx, "S3.GetObject", s3RequestDuration, instrument.ErrorCode, func(ctx context.Context) error {
-		var err error
-		resp, err = a.S3.GetObjectWithContext(ctx, &s3.GetObjectInput{
-			Bucket: aws.String(bucket),
-			Key:    aws.String(objectKey),
-		})
-		return err
-	})
+	retries := backoff.New(ctx, a.cfg.BackoffConfig)
+	err := ctx.Err()
 	if err != nil {
-		return nil, err
+		return nil, errors.Wrap(err, "ctx related error during s3 getObject")
 	}
-
-	return resp.Body, nil
+	for retries.Ongoing() {
+		if ctx.Err() != nil {
+			return nil, errors.Wrap(ctx.Err(), "ctx related error during s3 getObject")
+		}
+		err = instrument.CollectedRequest(ctx, "S3.GetObject", s3RequestDuration, instrument.ErrorCode, func(ctx context.Context) error {
+			var requestErr error
+			resp, requestErr = a.S3.GetObjectWithContext(ctx, &s3.GetObjectInput{
+				Bucket: aws.String(bucket),
+				Key:    aws.String(objectKey),
+			})
+			return requestErr
+		})
+		if err == nil {
+			return resp.Body, nil
+		}
+		retries.Wait()
+	}
+	return nil, errors.Wrap(err, "failed to get s3 object")
 }
 
 // PutObject into the store