[feat] curvefs: s3Adaptor retry after failed

Signed-off-by: montaguelhz <[email protected]>

conf/s3.conf

@@ -29,4 +29,11 @@ s3.throttle.bpsTotalMB=1280
 s3.throttle.bpsReadMB=1280
 s3.throttle.bpsWriteMB=1280
 s3.useVirtualAddressing=false
+# The based version of aws SDK suppot 2 retry strategies now, default and standard.
+# The default depends on maxRetries and scaleFactor, while the standard depends on maxRetries.
+# Both maxRetries and scaleFactor should not be negative, and these arguments are optional.
+# default standard disable
+s3.retryMode=default
+s3.maxRetries=10
+s3.scaleFactor=25
 

curvefs/conf/client.conf

@@ -251,6 +251,10 @@ s3.useVirtualAddressing=false
 s3.maxReadRetryIntervalMs = 1000
 # retry interval
 s3.readRetryIntervalMs = 100
+# default standard disable
+s3.retryMode=default
+s3.maxRetries=10
+s3.scaleFactor=25
 # TODO(hongsong): limit bytes、iops/bps
 #### disk cache options
 # 0:not enable disk cache

curvefs/conf/mds.conf

@@ -119,7 +119,7 @@ mds.fsmanager.reloadSpaceConcurrency=10
 mds.fsmanager.client.timeoutSec=20
 
 #### s3
-# TODO(huyao): use more meaningfull name
+# TODO(huyao): use more meaningful name
 # http = 0, https = 1
 s3.http_scheme=0
 s3.verify_SSL=False
@@ -142,6 +142,10 @@ s3.throttle.bpsTotalMB=0
 s3.throttle.bpsReadMB=0
 s3.throttle.bpsWriteMB=0
 s3.useVirtualAddressing=false
+# default standard disable
+s3.retryMode=default
+s3.maxRetries=10
+s3.scaleFactor=25
 
 # TTL(millisecond) for distributed lock
 dlock.ttl_ms=5000

curvefs/conf/metaserver.conf

@@ -29,6 +29,10 @@ s3.throttle.bpsTotalMB=0
 s3.throttle.bpsReadMB=0
 s3.throttle.bpsWriteMB=0
 s3.useVirtualAddressing=false
+# default standard disable
+s3.retryMode=default
+s3.maxRetries=10
+s3.scaleFactor=25
 # s3 workqueue
 s3compactwq.enable=True
 s3compactwq.thread_num=2

curvefs/test/client/client_s3_test.cpp

@@ -67,6 +67,34 @@ TEST_F(ClientS3Test, init_s3Adapter_userAgent) {
     ASSERT_STREQ(userAgent.c_str(), s3Adapter.GetConfig()->userAgent.c_str());
 }
 
+TEST_F(ClientS3Test, init_s3Adapter_retry_strategy) {
+    curve::common::S3Adapter s3Adapter;
+    curve::common::S3AdapterOption option;
+
+    std::string retryMode = "default";
+    int maxRetries = 10;
+    int scaleFactors = 25;
+    option.retryMode = retryMode;
+    option.maxRetries = maxRetries;
+    option.scaleFactor = scaleFactors;
+    s3Adapter.Init(option);
+    ASSERT_EQ(maxRetries + 1,
+              (int)s3Adapter.GetConfig()->retryStrategy->GetMaxAttempts());
+
+    retryMode = "standard";
+    maxRetries = 3;
+    option.retryMode = retryMode;
+    option.maxRetries = maxRetries;
+    s3Adapter.Init(option);
+    ASSERT_EQ(maxRetries + 1,
+              (int)s3Adapter.GetConfig()->retryStrategy->GetMaxAttempts());
+
+    retryMode = "disable";
+    option.retryMode = retryMode;
+    s3Adapter.Init(option);
+    ASSERT_EQ(1, (int)s3Adapter.GetConfig()->retryStrategy->GetMaxAttempts());
+}
+
 TEST_F(ClientS3Test, upload) {
     const std::string obj("test");
     uint64_t len = 1024;
@@ -122,12 +150,10 @@ TEST_F(ClientS3Test, downloadAsync) {
 
     auto context = std::make_shared<GetObjectAsyncContext>("name", buf, 1, 10);
 
-    EXPECT_CALL(*s3Client_, GetObjectAsync(_))
-        .WillOnce(Return());
+    EXPECT_CALL(*s3Client_, GetObjectAsync(_)).WillOnce(Return());
     client_->DownloadAsync(context);
     delete[] buf;
 }
 
-
 }  // namespace client
 }  // namespace curvefs

src/common/s3_adapter.cpp

@@ -107,6 +107,26 @@ void InitS3AdaptorOptionExceptS3InfoOption(Configuration* conf,
         &s3Opt->useVirtualAddressing));
     LOG_IF(FATAL, !conf->GetStringValue("s3.region", &s3Opt->region));
 
+    if (!conf->GetStringValue("s3.retryMode", &s3Opt->retryMode)) {
+        LOG(WARNING) << "Not found s3.retryMode in conf";
+        s3Opt->retryMode = DEFAULT_MODE;
+    }
+    if (!conf->GetIntValue("s3.maxRetries", &s3Opt->maxRetries) ||
+        s3Opt->maxRetries < 0) {
+        LOG(WARNING) << "Not found s3.maxRetries in conf";
+        if (s3Opt->retryMode == DEFAULT_MODE) {
+            s3Opt->maxRetries = 10;
+        } else if (s3Opt->retryMode == STANDARD_MODE) {
+            s3Opt->maxRetries = 3;
+        }
+    }
+    if (!conf->GetIntValue("s3.scaleFactor", &s3Opt->scaleFactor) ||
+        s3Opt->scaleFactor < 0) {
+        LOG(WARNING) << "Not found s3.scaleFactor in conf";
+        if (s3Opt->retryMode == DEFAULT_MODE) {
+            s3Opt->scaleFactor = 25;
+        }
+    }
     if (!conf->GetUInt64Value("s3.maxAsyncRequestInflightBytes",
                               &s3Opt->maxAsyncRequestInflightBytes)) {
         LOG(WARNING) << "Not found s3.maxAsyncRequestInflightBytes in conf";
@@ -156,6 +176,22 @@ void S3Adapter::Init(const S3AdapterOption &option) {
     clientCfg_->connectTimeoutMs = option.connectTimeout;
     clientCfg_->requestTimeoutMs = option.requestTimeout;
     clientCfg_->endpointOverride = s3Address_;
+
+    if (option.retryMode == STANDARD_MODE) {
+        clientCfg_->retryStrategy =
+            Aws::MakeShared<Aws::Client::StandardRetryStrategy>(
+                AWS_ALLOCATE_TAG, option.maxRetries);
+    } else if (option.retryMode == DEFAULT_MODE) {
+        clientCfg_->retryStrategy =
+            Aws::MakeShared<Aws::Client::DefaultRetryStrategy>(
+                AWS_ALLOCATE_TAG, option.maxRetries,
+                option.scaleFactor);
+    } else if (option.retryMode == DISABLE_MODE) {
+        clientCfg_->retryStrategy =
+            Aws::MakeShared<Aws::Client::DefaultRetryStrategy>(
+                AWS_ALLOCATE_TAG, 0);
+    }
+
     auto asyncThreadNum = option.asyncThreadNum;
     LOG(INFO) << "S3Adapter init thread num = " << asyncThreadNum << std::endl;
     clientCfg_->executor =

src/common/s3_adapter.h

@@ -25,6 +25,7 @@
 #include <aws/core/Aws.h>                                 //NOLINT
 #include <aws/core/auth/AWSCredentialsProvider.h>         //NOLINT
 #include <aws/core/client/ClientConfiguration.h>          //NOLINT
+#include <aws/core/client/DefaultRetryStrategy.h>         //NOLINT
 #include <aws/core/http/HttpRequest.h>                    //NOLINT
 #include <aws/core/http/Scheme.h>                         //NOLINT
 #include <aws/core/utils/memory/AWSMemory.h>              //NOLINT
@@ -69,6 +70,9 @@ struct GetObjectAsyncContext;
 struct PutObjectAsyncContext;
 class S3Adapter;
 
+const char STANDARD_MODE[] = "standard";
+const char DEFAULT_MODE[] = "default";
+const char DISABLE_MODE[] = "disable";
 struct S3AdapterOption {
     std::string ak;
     std::string sk;
@@ -92,6 +96,9 @@ struct S3AdapterOption {
     uint64_t bpsReadMB;
     uint64_t bpsWriteMB;
     bool useVirtualAddressing;
+    std::string retryMode;
+    int maxRetries;
+    int scaleFactor;
 };
 
 struct S3InfoOption {