Move blobstore retry logic into specific implementations (cadence-wor…

…kflow#1596)

common/blobstore/filestore/client.go

@@ -23,15 +23,17 @@ package filestore
 import (
 	"context"
 	"fmt"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
+
 	"github.com/uber-common/bark"
 	"github.com/uber/cadence/.gen/go/shared"
+	"github.com/uber/cadence/common/backoff"
 	"github.com/uber/cadence/common/blobstore"
 	"github.com/uber/cadence/common/blobstore/blob"
 	"github.com/uber/cadence/common/logging"
-	"os"
-	"path/filepath"
-	"strings"
-	"sync"
 )
 
 const (
@@ -325,6 +327,16 @@ func (c *client) BucketMetadata(_ context.Context, bucket string) (*blobstore.Bu
 	}, nil
 }
 
+func (c *client) IsRetryableError(err error) bool {
+	return false
+}
+
+func (c *client) GetRetryPolicy() backoff.RetryPolicy {
+	policy := backoff.NewExponentialRetryPolicy(0)
+	policy.SetMaximumAttempts(1)
+	return policy
+}
+
 func setupDirectories(cfg *Config) error {
 	if err := mkdirAll(cfg.StoreDirectory); err != nil {
 		return err

common/blobstore/interface.go

@@ -22,7 +22,9 @@ package blobstore
 
 import (
 	"context"
+
 	"github.com/uber/cadence/.gen/go/shared"
+	"github.com/uber/cadence/common/backoff"
 	"github.com/uber/cadence/common/blobstore/blob"
 )
 
@@ -51,4 +53,7 @@ type Client interface {
 	Delete(ctx context.Context, bucket string, key blob.Key) (bool, error)
 	ListByPrefix(ctx context.Context, bucket string, prefix string) ([]blob.Key, error)
 	BucketMetadata(ctx context.Context, bucket string) (*BucketMetadataResponse, error)
+
+	IsRetryableError(err error) bool
+	GetRetryPolicy() backoff.RetryPolicy
 }

common/blobstore/metricClient.go

@@ -22,6 +22,7 @@ package blobstore
 
 import (
 	"context"
+	"github.com/uber/cadence/common/backoff"
 	"github.com/uber/cadence/common/blobstore/blob"
 	"github.com/uber/cadence/common/metrics"
 )
@@ -118,3 +119,11 @@ func (c *metricClient) BucketMetadata(ctx context.Context, bucket string) (*Buck
 	}
 	return resp, err
 }
+
+func (c *metricClient) IsRetryableError(err error) bool {
+	return c.client.IsRetryableError(err)
+}
+
+func (c *metricClient) GetRetryPolicy() backoff.RetryPolicy {
+	return c.client.GetRetryPolicy()
+}

common/blobstore/retryableClient.go

@@ -104,3 +104,11 @@ func (c *retryableClient) BucketMetadata(ctx context.Context, bucket string) (*B
 	err := backoff.Retry(op, c.policy, c.isRetryable)
 	return resp, err
 }
+
+func (c *retryableClient) IsRetryableError(err error) bool {
+	return c.client.IsRetryableError(err)
+}
+
+func (c *retryableClient) GetRetryPolicy() backoff.RetryPolicy {
+	return c.client.GetRetryPolicy()
+}

common/util.go

@@ -71,10 +71,6 @@ const (
 	retryKafkaOperationMaxInterval        = 10 * time.Second
 	retryKafkaOperationExpirationInterval = 30 * time.Second
 
-	retryBlobstoreClientInitialInterval    = time.Second
-	retryBlobstoreClientMaxInterval        = 10 * time.Second
-	retryBlobstoreClientExpirationInterval = time.Minute
-
 	// FailureReasonCompleteResultExceedsLimit is failureReason for complete result exceeds limit
 	FailureReasonCompleteResultExceedsLimit = "COMPLETE_RESULT_EXCEEDS_LIMIT"
 	// FailureReasonFailureDetailsExceedsLimit is failureReason for failure details exceeds limit
@@ -187,15 +183,6 @@ func CreateKafkaOperationRetryPolicy() backoff.RetryPolicy {
 	return policy
 }
 
-// CreateBlobstoreClientRetryPolicy creates a retry policy for blobstore client
-func CreateBlobstoreClientRetryPolicy() backoff.RetryPolicy {
-	policy := backoff.NewExponentialRetryPolicy(retryBlobstoreClientInitialInterval)
-	policy.SetMaximumInterval(retryBlobstoreClientMaxInterval)
-	policy.SetExpirationInterval(retryBlobstoreClientExpirationInterval)
-
-	return policy
-}
-
 // IsPersistenceTransientError checks if the error is a transient persistence error
 func IsPersistenceTransientError(err error) bool {
 	switch err.(type) {
@@ -211,22 +198,6 @@ func IsKafkaTransientError(err error) bool {
 	return true
 }
 
-// IsBlobstoreTransientError checks if the error is a retryable error.
-func IsBlobstoreTransientError(err error) bool {
-	return !IsBlobstoreNonRetryableError(err)
-}
-
-// IsBlobstoreNonRetryableError checks if the error is a non retryable error.
-func IsBlobstoreNonRetryableError(err error) bool {
-	switch err.(type) {
-	case *workflow.BadRequestError:
-		return true
-	case *workflow.EntityNotExistsError:
-		return true
-	}
-	return false
-}
-
 // IsServiceTransientError checks if the error is a retryable error.
 func IsServiceTransientError(err error) bool {
 	return !IsServiceNonRetryableError(err)

host/onebox.go

@@ -507,8 +507,8 @@ func (c *cadenceImpl) startWorkerClientWorker(params *service.BootstrapParams, s
 	)
 	blobstoreClient := blobstore.NewRetryableClient(
 		blobstore.NewMetricClient(c.blobstoreClient, service.GetMetricsClient()),
-		common.CreateBlobstoreClientRetryPolicy(),
-		common.IsBlobstoreTransientError)
+		c.blobstoreClient.GetRetryPolicy(),
+		c.blobstoreClient.IsRetryableError)
 	workerConfig := worker.NewConfig(dynamicconfig.NewNopCollection())
 	workerConfig.ArchiverConfig.ArchiverConcurrency = dynamicconfig.GetIntPropertyFn(10)
 	bc := &archiver.BootstrapContainer{

service/worker/archiver/activities.go

@@ -212,7 +212,7 @@ func blobExists(ctx context.Context, blobstoreClient blobstore.Client, bucket st
 	cancel()
 	for err != nil {
 		activity.RecordHeartbeat(ctx)
-		if !common.IsBlobstoreTransientError(err) {
+		if !blobstoreClient.IsRetryableError(err) {
 			return false, cadence.NewCustomError(errBlobExists)
 		}
 		if contextExpired(ctx) {
@@ -231,7 +231,7 @@ func uploadBlob(ctx context.Context, blobstoreClient blobstore.Client, bucket st
 	cancel()
 	for err != nil {
 		activity.RecordHeartbeat(ctx)
-		if !common.IsBlobstoreTransientError(err) {
+		if !blobstoreClient.IsRetryableError(err) {
 			return cadence.NewCustomError(errUploadBlob)
 		}
 		if contextExpired(ctx) {

service/worker/archiver/activities_test.go

@@ -48,8 +48,6 @@ const (
 var (
 	errPersistenceNonRetryable = errors.New("persistence non-retryable error")
 	errPersistenceRetryable    = &shared.InternalServiceError{}
-	errBlobstoreNonRetryable   = &shared.BadRequestError{}
-	errBlobstoreRetryable      = errors.New("blobstore retryable error")
 )
 
 type activitiesSuite struct {
@@ -303,7 +301,8 @@ func (s *activitiesSuite) TestUploadHistoryActivity_Fail_BlobExistsNonRetryableE
 	}
 	mockHistoryBlobIterator.On("Next").Return(historyBlob, nil)
 	mockBlobstore := &mocks.BlobstoreClient{}
-	mockBlobstore.On("Exists", mock.Anything, mock.Anything, mock.Anything).Return(false, errBlobstoreNonRetryable)
+	mockBlobstore.On("Exists", mock.Anything, mock.Anything, mock.Anything).Return(false, errors.New("some error"))
+	mockBlobstore.On("IsRetryableError", mock.Anything).Return(false)
 	container := &BootstrapContainer{
 		Logger:              s.logger,
 		MetricsClient:       s.metricsClient,
@@ -340,7 +339,8 @@ func (s *activitiesSuite) TestUploadHistoryActivity_Fail_TimeoutOnBlobExists() {
 	}
 	mockHistoryBlobIterator.On("Next").Return(historyBlob, nil)
 	mockBlobstore := &mocks.BlobstoreClient{}
-	mockBlobstore.On("Exists", mock.Anything, mock.Anything, mock.Anything).Return(false, errBlobstoreRetryable)
+	mockBlobstore.On("Exists", mock.Anything, mock.Anything, mock.Anything).Return(false, errors.New("some error"))
+	mockBlobstore.On("IsRetryableError", mock.Anything).Return(true)
 	container := &BootstrapContainer{
 		Logger:              s.logger,
 		MetricsClient:       s.metricsClient,
@@ -418,7 +418,8 @@ func (s *activitiesSuite) TestUploadHistoryActivity_Fail_UploadBlobNonRetryableE
 	mockHistoryBlobIterator.On("Next").Return(historyBlob, nil)
 	mockBlobstore := &mocks.BlobstoreClient{}
 	mockBlobstore.On("Exists", mock.Anything, mock.Anything, mock.Anything).Return(false, nil)
-	mockBlobstore.On("Upload", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(errBlobstoreNonRetryable)
+	mockBlobstore.On("Upload", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(errors.New("some error"))
+	mockBlobstore.On("IsRetryableError", mock.Anything).Return(false)
 	container := &BootstrapContainer{
 		Logger:              s.logger,
 		MetricsClient:       s.metricsClient,
@@ -458,7 +459,8 @@ func (s *activitiesSuite) TestUploadHistoryActivity_Fail_TimeoutOnUploadBlob() {
 	mockHistoryBlobIterator.On("Next").Return(historyBlob, nil)
 	mockBlobstore := &mocks.BlobstoreClient{}
 	mockBlobstore.On("Exists", mock.Anything, mock.Anything, mock.Anything).Return(false, nil)
-	mockBlobstore.On("Upload", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(errBlobstoreRetryable)
+	mockBlobstore.On("Upload", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(errors.New("some error"))
+	mockBlobstore.On("IsRetryableError", mock.Anything).Return(true)
 	container := &BootstrapContainer{
 		Logger:              s.logger,
 		MetricsClient:       s.metricsClient,

service/worker/service.go

@@ -207,8 +207,8 @@ func (s *Service) startArchiver(base service.Service, pFactory persistencefactor
 
 	blobstoreClient := blobstore.NewRetryableClient(
 		blobstore.NewMetricClient(s.params.BlobstoreClient, s.metricsClient),
-		common.CreateBlobstoreClientRetryPolicy(),
-		common.IsBlobstoreTransientError)
+		s.params.BlobstoreClient.GetRetryPolicy(),
+		s.params.BlobstoreClient.IsRetryableError)
 
 	bc := &archiver.BootstrapContainer{
 		PublicClient:     publicClient,