add storage retry policy

sdk/storage/azure-storage-blobs/src/blob_client.cpp

@@ -18,6 +18,7 @@
 #include <azure/storage/common/internal/shared_key_policy.hpp>
 #include <azure/storage/common/internal/storage_bearer_token_auth.hpp>
 #include <azure/storage/common/internal/storage_per_retry_policy.hpp>
+#include <azure/storage/common/internal/storage_pipeline.hpp>
 #include <azure/storage/common/internal/storage_service_version_policy.hpp>
 #include <azure/storage/common/internal/storage_switch_to_secondary_policy.hpp>
 #include <azure/storage/common/storage_common.hpp>
@@ -58,23 +59,18 @@ namespace Azure { namespace Storage { namespace Blobs {
       const BlobClientOptions& options)
       : BlobClient(blobUrl, options)
   {
-    BlobClientOptions newOptions = options;
-    newOptions.PerRetryPolicies.emplace_back(
-        std::make_unique<_internal::SharedKeyPolicy>(credential));
-
     std::vector<std::unique_ptr<Azure::Core::Http::Policies::HttpPolicy>> perRetryPolicies;
-    std::vector<std::unique_ptr<Azure::Core::Http::Policies::HttpPolicy>> perOperationPolicies;
     perRetryPolicies.emplace_back(std::make_unique<_internal::StorageSwitchToSecondaryPolicy>(
-        m_blobUrl.GetHost(), newOptions.SecondaryHostForRetryReads));
-    perRetryPolicies.emplace_back(std::make_unique<_internal::StoragePerRetryPolicy>());
-    perOperationPolicies.emplace_back(
-        std::make_unique<_internal::StorageServiceVersionPolicy>(newOptions.ApiVersion));
-    m_pipeline = std::make_shared<Azure::Core::Http::_internal::HttpPipeline>(
-        newOptions,
+        m_blobUrl.GetHost(), options.SecondaryHostForRetryReads));
+
+    m_pipeline = _internal::ConstructStorageHttpPipeline(
+        options.ApiVersion,
         _internal::BlobServicePackageName,
         _detail::PackageVersion::ToString(),
+        std::move(credential),
+        {},
         std::move(perRetryPolicies),
-        std::move(perOperationPolicies));
+        options);
   }
 
   BlobClient::BlobClient(

sdk/storage/azure-storage-common/CMakeLists.txt

@@ -55,6 +55,8 @@ set(
     inc/azure/storage/common/internal/storage_bearer_token_auth.hpp
     inc/azure/storage/common/internal/storage_bearer_token_authentication_policy.hpp
     inc/azure/storage/common/internal/storage_per_retry_policy.hpp
+    inc/azure/storage/common/internal/storage_pipeline.hpp
+    inc/azure/storage/common/internal/storage_retry_policy.hpp
     inc/azure/storage/common/internal/storage_service_version_policy.hpp
     inc/azure/storage/common/internal/storage_switch_to_secondary_policy.hpp
     inc/azure/storage/common/internal/xml_wrapper.hpp
@@ -76,6 +78,8 @@ set(
     src/storage_credential.cpp
     src/storage_exception.cpp
     src/storage_per_retry_policy.cpp
+    src/storage_pipeline.cpp
+    src/storage_retry_policy.cpp
     src/storage_switch_to_secondary_policy.cpp
     src/xml_wrapper.cpp
 )

sdk/storage/azure-storage-common/inc/azure/storage/common/internal/storage_pipeline.hpp

@@ -0,0 +1,21 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+#pragma once
+
+#include "../storage_credential.hpp"
+
+#include <azure/core/internal/http/pipeline.hpp>
+
+namespace Azure { namespace Storage { namespace _internal {
+
+  std::shared_ptr<Azure::Core::Http::_internal::HttpPipeline> ConstructStorageHttpPipeline(
+      const std::string& apiVersion,
+      const std::string& telemetryPackageName,
+      const std::string& telemetryPackageVersion,
+      std::shared_ptr<StorageSharedKeyCredential> sharedKeyCredential,
+      std::vector<std::unique_ptr<Azure::Core::Http::Policies::HttpPolicy>> perCallPolicies,
+      std::vector<std::unique_ptr<Azure::Core::Http::Policies::HttpPolicy>> perRetryPolicies,
+      const Azure::Core::_internal::ClientOptions& clientOptions);
+
+}}} // namespace Azure::Storage::_internal

sdk/storage/azure-storage-common/inc/azure/storage/common/internal/storage_retry_policy.hpp

@@ -0,0 +1,24 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+#pragma once
+
+#include <azure/core/http/policies/policy.hpp>
+
+namespace Azure { namespace Storage { namespace _internal {
+
+  class StorageRetryPolicy final : public Core::Http::Policies::_internal::RetryPolicy {
+  public:
+    explicit StorageRetryPolicy(Core::Http::Policies::RetryOptions options)
+        : RetryPolicy(std::move(options))
+    {
+    }
+    ~StorageRetryPolicy() override {}
+
+  protected:
+    bool ShouldRetry(
+        const std::unique_ptr<Core::Http::RawResponse>& response,
+        const Core::Http::Policies::RetryOptions& retryOptions) const override;
+  };
+
+}}} // namespace Azure::Storage::_internal

sdk/storage/azure-storage-common/inc/azure/storage/common/internal/storage_service_version_policy.hpp

@@ -3,6 +3,8 @@
 
 #pragma once
 
+#include "constants.hpp"
+
 #include <azure/core/http/policies/policy.hpp>
 
 #include <memory>

sdk/storage/azure-storage-common/src/storage_pipeline.cpp

@@ -0,0 +1,86 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+#include "azure/storage/common/internal/storage_pipeline.hpp"
+
+#include "azure/storage/common/internal/shared_key_policy.hpp"
+#include "azure/storage/common/internal/storage_per_retry_policy.hpp"
+#include "azure/storage/common/internal/storage_retry_policy.hpp"
+#include "azure/storage/common/internal/storage_service_version_policy.hpp"
+
+namespace Azure { namespace Storage { namespace _internal {
+
+  std::shared_ptr<Azure::Core::Http::_internal::HttpPipeline> ConstructStorageHttpPipeline(
+      const std::string& apiVersion,
+      const std::string& telemetryPackageName,
+      const std::string& telemetryPackageVersion,
+      std::shared_ptr<StorageSharedKeyCredential> sharedKeyCredential,
+      std::vector<std::unique_ptr<Azure::Core::Http::Policies::HttpPolicy>> perCallPolicies,
+      std::vector<std::unique_ptr<Azure::Core::Http::Policies::HttpPolicy>> perRetryPolicies,
+      const Azure::Core::_internal::ClientOptions& clientOptions)
+  {
+    std::vector<std::unique_ptr<Azure::Core::Http::Policies::HttpPolicy>> policies;
+
+    // service-specific per call policies
+    for (auto& policy : perCallPolicies)
+    {
+      policies.push_back(std::move(policy));
+    }
+    policies.push_back(std::make_unique<_internal::StorageServiceVersionPolicy>(apiVersion));
+
+    // Request Id
+    policies.push_back(std::make_unique<Azure::Core::Http::Policies::_internal::RequestIdPolicy>());
+
+    // Telemetry (user-agent header)
+    policies.push_back(std::make_unique<Azure::Core::Http::Policies::_internal::TelemetryPolicy>(
+        telemetryPackageName, telemetryPackageVersion, clientOptions.Telemetry));
+
+    // client-options per call policies.
+    for (auto& policy : clientOptions.PerOperationPolicies)
+    {
+      policies.push_back(policy->Clone());
+    }
+
+    // Retry policy
+    policies.push_back(std::make_unique<StorageRetryPolicy>(clientOptions.Retry));
+
+    // service-specific per retry policies.
+    for (auto& policy : perRetryPolicies)
+    {
+      policies.push_back(std::move(policy));
+    }
+    policies.push_back(std::make_unique<StoragePerRetryPolicy>());
+
+    // client options per retry policies.
+    for (auto& policy : clientOptions.PerRetryPolicies)
+    {
+      policies.push_back(policy->Clone());
+    }
+
+    if (sharedKeyCredential)
+    {
+      auto policy = std::make_unique<SharedKeyPolicy>(sharedKeyCredential);
+      policies.push_back(policy->Clone());
+    }
+
+    // Policies after here cannot modify the request anymore
+
+    // Add a request activity policy which will generate distributed traces for the pipeline.
+    Azure::Core::Http::_internal::HttpSanitizer httpSanitizer(
+        clientOptions.Log.AllowedHttpQueryParameters, clientOptions.Log.AllowedHttpHeaders);
+    policies.push_back(
+        std::make_unique<Azure::Core::Http::Policies::_internal::RequestActivityPolicy>(
+            httpSanitizer));
+
+    // logging - won't update request
+    policies.push_back(
+        std::make_unique<Azure::Core::Http::Policies::_internal::LogPolicy>(clientOptions.Log));
+
+    // transport
+    policies.push_back(std::make_unique<Azure::Core::Http::Policies::_internal::TransportPolicy>(
+        clientOptions.Transport));
+
+    return std::make_shared<Azure::Core::Http::_internal::HttpPipeline>(policies);
+  }
+
+}}} // namespace Azure::Storage::_internal

sdk/storage/azure-storage-common/src/storage_retry_policy.cpp

@@ -0,0 +1,44 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+#include "azure/storage/common/internal/storage_retry_policy.hpp"
+
+#include "azure/core/internal/diagnostics/log.hpp"
+
+namespace Azure { namespace Storage { namespace _internal {
+
+  bool StorageRetryPolicy::ShouldRetry(
+      const std::unique_ptr<Core::Http::RawResponse>& response,
+      const Core::Http::Policies::RetryOptions& retryOptions) const
+  {
+    if (static_cast<std::underlying_type_t<Core::Http::HttpStatusCode>>(response->GetStatusCode())
+        >= 400)
+    {
+      const auto& headers = response->GetHeaders();
+      auto ite = headers.find("x-ms-copy-source-status-code");
+      if (ite != headers.end())
+      {
+        const auto innerStatusCodeInt = std::stoi(ite->second);
+        const auto innerStatusCode = static_cast<Core::Http::HttpStatusCode>(innerStatusCodeInt);
+
+        const bool shouldRetry
+            = retryOptions.StatusCodes.find(innerStatusCode) != retryOptions.StatusCodes.end();
+
+        if (Azure::Core::Diagnostics::_internal::Log::ShouldWrite(
+                Azure::Core::Diagnostics::Logger::Level::Informational))
+        {
+          Azure::Core::Diagnostics::_internal::Log::Write(
+              Azure::Core::Diagnostics::Logger::Level::Informational,
+              std::string("x-ms-copy-source-status-code ") + std::to_string(innerStatusCodeInt)
+                  + (shouldRetry ? " will be retried" : " won't be retried"));
+        }
+        if (shouldRetry)
+        {
+          return true;
+        }
+      }
+    }
+    return false;
+  }
+
+}}} // namespace Azure::Storage::_internal