Merge pull request ClickHouse#80035 from vitlibar/s3-slow-down-after-network-error

Slow down S3 client after network error

Author: vitlibar

src/Backups/BackupIO_S3.cpp

@@ -33,6 +33,7 @@ namespace Setting
     extern const SettingsBool s3_disable_checksum;
     extern const SettingsUInt64 s3_max_connections;
     extern const SettingsUInt64 s3_max_redirects;
+    extern const SettingsBool s3_slow_all_threads_after_network_error;
 }
 
 namespace S3AuthSetting
@@ -88,6 +89,7 @@ namespace
             context->getRemoteHostFilter(),
             static_cast<unsigned>(local_settings[Setting::s3_max_redirects]),
             static_cast<unsigned>(local_settings[Setting::backup_restore_s3_retry_attempts]),
+            local_settings[Setting::s3_slow_all_threads_after_network_error],
             local_settings[Setting::enable_s3_requests_logging],
             /* for_disk_s3 = */ false,
             request_settings.get_request_throttler,

src/Coordination/KeeperSnapshotManagerS3.cpp

@@ -104,6 +104,7 @@ void KeeperSnapshotManagerS3::updateS3Configuration(const Poco::Util::AbstractCo
 
         static constexpr size_t s3_max_redirects = 10;
         static constexpr size_t s3_retry_attempts = 10;
+        static constexpr bool s3_slow_all_threads_after_network_error = true;
         static constexpr bool enable_s3_requests_logging = false;
 
         if (!new_uri.key.empty())
@@ -114,7 +115,7 @@ void KeeperSnapshotManagerS3::updateS3Configuration(const Poco::Util::AbstractCo
 
         S3::PocoHTTPClientConfiguration client_configuration = S3::ClientFactory::instance().createClientConfiguration(
             auth_settings[S3AuthSetting::region],
-            RemoteHostFilter(), s3_max_redirects, s3_retry_attempts,
+            RemoteHostFilter(), s3_max_redirects, s3_retry_attempts, s3_slow_all_threads_after_network_error,
             enable_s3_requests_logging,
             /* for_disk_s3 = */ false, /* get_request_throttler = */ {}, /* put_request_throttler = */ {},
             new_uri.uri.getScheme());

src/Core/Settings.cpp

@@ -430,6 +430,11 @@ Maximum number of files that could be returned in batch by ListObject request
     DECLARE(Bool, s3_use_adaptive_timeouts, S3::DEFAULT_USE_ADAPTIVE_TIMEOUTS, R"(
 When set to `true` than for all s3 requests first two attempts are made with low send and receive timeouts.
 When set to `false` than all attempts are made with identical timeouts.
+)", 0) \
+    DECLARE(Bool, s3_slow_all_threads_after_network_error, true, R"(
+When set to `true` than all threads executing s3 requests to the same endpoint get slow down for a while
+after one s3 request fails with a retryable network error.
+When set to `false` than each thread executing s3 request uses an independent set of backoffs on network errors.
 )", 0) \
     DECLARE(UInt64, azure_list_object_keys_size, 1000, R"(
 Maximum number of files that could be returned in batch by ListObject request

src/Core/SettingsChangesHistory.cpp

@@ -105,6 +105,7 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory()
             {"compile_expressions", false, true, "We believe that the LLVM infrastructure behind the JIT compiler is stable enough to enable this setting by default."},
             {"use_legacy_to_time", false, false, "New setting. Allows for user to use the old function logic for toTime, which works as toTimeWithFixedDate."},
             {"input_format_parquet_allow_geoparquet_parser", false, true, "A new setting to use geo columns in parquet file"},
+            {"s3_slow_all_threads_after_network_error", false, true, "New setting"},
         });
         addSettingsChanges(settings_changes_history, "25.4",
         {

src/Databases/DataLake/GlueCatalog.cpp

@@ -43,6 +43,7 @@ namespace DB::Setting
     extern const SettingsUInt64 s3_max_connections;
     extern const SettingsUInt64 s3_max_redirects;
     extern const SettingsUInt64 s3_retry_attempts;
+    extern const SettingsBool s3_slow_all_threads_after_network_error;
     extern const SettingsBool enable_s3_requests_logging;
     extern const SettingsUInt64 s3_connect_timeout_ms;
     extern const SettingsUInt64 s3_request_timeout_ms;
@@ -75,12 +76,15 @@ GlueCatalog::GlueCatalog(
 
     int s3_max_redirects = static_cast<int>(global_settings[DB::Setting::s3_max_redirects]);
     int s3_retry_attempts = static_cast<int>(global_settings[DB::Setting::s3_retry_attempts]);
+    bool s3_slow_all_threads_after_network_error = global_settings[DB::Setting::s3_slow_all_threads_after_network_error];
     bool enable_s3_requests_logging = global_settings[DB::Setting::enable_s3_requests_logging];
+
     DB::S3::PocoHTTPClientConfiguration poco_config = DB::S3::ClientFactory::instance().createClientConfiguration(
         region,
         getContext()->getRemoteHostFilter(),
         s3_max_redirects,
         s3_retry_attempts,
+        s3_slow_all_threads_after_network_error,
         enable_s3_requests_logging,
         false,
         nullptr,

src/Disks/ObjectStorages/S3/diskSettings.cpp

@@ -31,6 +31,7 @@ namespace Setting
     extern const SettingsBool enable_s3_requests_logging;
     extern const SettingsUInt64 s3_max_redirects;
     extern const SettingsUInt64 s3_retry_attempts;
+    extern const SettingsBool s3_slow_all_threads_after_network_error;
 }
 
 namespace S3AuthSetting
@@ -123,6 +124,10 @@ std::unique_ptr<S3::Client> getClient(
     if (!for_disk_s3 && local_settings.isChanged("s3_retry_attempts"))
         s3_retry_attempts = static_cast<int>(local_settings[Setting::s3_retry_attempts]);
 
+    bool s3_slow_all_threads_after_network_error = static_cast<int>(global_settings[Setting::s3_slow_all_threads_after_network_error]);
+    if (!for_disk_s3 && local_settings.isChanged("s3_slow_all_threads_after_network_error"))
+        s3_slow_all_threads_after_network_error = static_cast<int>(local_settings[Setting::s3_slow_all_threads_after_network_error]);
+
     bool enable_s3_requests_logging = global_settings[Setting::enable_s3_requests_logging];
     if (!for_disk_s3 && local_settings.isChanged("enable_s3_requests_logging"))
         enable_s3_requests_logging = local_settings[Setting::enable_s3_requests_logging];
@@ -132,6 +137,7 @@ std::unique_ptr<S3::Client> getClient(
         context->getRemoteHostFilter(),
         s3_max_redirects,
         s3_retry_attempts,
+        s3_slow_all_threads_after_network_error,
         enable_s3_requests_logging,
         for_disk_s3,
         request_settings.get_request_throttler,

src/IO/S3/Client.cpp

@@ -653,6 +653,9 @@ Client::doRequestWithRetryNetworkErrors(RequestType & request, RequestFn request
         std::exception_ptr last_exception = nullptr;
         for (Int64 attempt_no = 0; attempt_no < max_attempts; ++attempt_no)
         {
+            /// Sometimes we need to slow down because other requests failed with network errors to free the S3 server a bit.
+            slowDownAfterNetworkError();
+
             try
             {
                 /// S3 does retries network errors actually.
@@ -695,10 +698,7 @@ Client::doRequestWithRetryNetworkErrors(RequestType & request, RequestFn request
                 if (!client_configuration.retryStrategy->ShouldRetry(error, attempt_no))
                     break;
 
-                auto sleep_ms = client_configuration.retryStrategy->CalculateDelayBeforeNextRetry(error, attempt_no);
-                LOG_WARNING(log, "Request failed, now waiting {} ms before attempting again", sleep_ms);
-                sleepForMilliseconds(sleep_ms);
-
+                sleepAfterNetworkError(error, attempt_no);
                 continue;
             }
         }
@@ -730,6 +730,44 @@ RequestResult Client::processRequestResult(RequestResult && outcome) const
     return RequestResult(error);
 }
 
+void Client::sleepAfterNetworkError(Aws::Client::AWSError<Aws::Client::CoreErrors> error, Int64 attempt_no) const
+{
+    auto sleep_ms = client_configuration.retryStrategy->CalculateDelayBeforeNextRetry(error, attempt_no);
+    if (!client_configuration.s3_slow_all_threads_after_network_error)
+    {
+        LOG_WARNING(log, "Request failed, now waiting {} ms before attempting again", sleep_ms);
+        sleepForMilliseconds(sleep_ms);
+        return;
+    }
+
+    /// Set the time other s3 requests must wait until.
+    UInt64 current_time_ms = std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now().time_since_epoch()).count();
+    UInt64 next_time_ms = current_time_ms + sleep_ms;
+    /// next_time_to_retry_after_network_error = std::max(next_time_to_retry_after_network_error, next_time_ms)
+    for (UInt64 stored_next_time = next_time_to_retry_after_network_error;
+         (stored_next_time < next_time_ms) && !next_time_to_retry_after_network_error.compare_exchange_weak(stored_next_time, next_time_ms);)
+    {
+    }
+}
+
+void Client::slowDownAfterNetworkError() const
+{
+    if (!client_configuration.s3_slow_all_threads_after_network_error)
+        return;
+
+    /// Wait until `next_time_to_retry_after_network_error`.
+    for (;;)
+    {
+        UInt64 current_time_ms = std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now().time_since_epoch()).count();
+        UInt64 next_time_ms = next_time_to_retry_after_network_error.load();
+        if (current_time_ms >= next_time_ms)
+            break;
+        UInt64 sleep_ms = next_time_ms - current_time_ms;
+        LOG_WARNING(log, "Some request failed, now waiting {} ms before executing a request", sleep_ms);
+        sleepForMilliseconds(sleep_ms);
+    }
+}
+
 bool Client::supportsMultiPartCopy() const
 {
     return provider_type != ProviderType::GCS;
@@ -990,6 +1028,7 @@ PocoHTTPClientConfiguration ClientFactory::createClientConfiguration( // NOLINT
     const RemoteHostFilter & remote_host_filter,
     unsigned int s3_max_redirects,
     unsigned int s3_retry_attempts,
+    bool s3_slow_all_threads_after_network_error,
     bool enable_s3_requests_logging,
     bool for_disk_s3,
     const ThrottlerPtr & get_request_throttler,
@@ -1009,6 +1048,7 @@ PocoHTTPClientConfiguration ClientFactory::createClientConfiguration( // NOLINT
         remote_host_filter,
         s3_max_redirects,
         s3_retry_attempts,
+        s3_slow_all_threads_after_network_error,
         enable_s3_requests_logging,
         for_disk_s3,
         context->getGlobalContext()->getSettingsRef()[Setting::s3_use_adaptive_timeouts],

src/IO/S3/Client.h

@@ -279,6 +279,9 @@ class Client : private Aws::S3::S3Client
     template <typename RequestResult>
     RequestResult processRequestResult(RequestResult && outcome) const;
 
+    void sleepAfterNetworkError(Aws::Client::AWSError<Aws::Client::CoreErrors> error, Int64 attempt_no) const;
+    void slowDownAfterNetworkError() const;
+
     String initial_endpoint;
     std::shared_ptr<Aws::Auth::AWSCredentialsProvider> credentials_provider;
     PocoHTTPClientConfiguration client_configuration;
@@ -299,6 +302,9 @@ class Client : private Aws::S3::S3Client
 
     const size_t max_redirects;
 
+    /// S3 requests must wait until this time because some s3 request fails with a retryable network error.
+    mutable std::atomic<UInt64> next_time_to_retry_after_network_error = 0;
+
     const ServerSideEncryptionKMSConfig sse_kms_config;
 
     LoggerPtr log;
@@ -327,6 +333,7 @@ class ClientFactory
         const RemoteHostFilter & remote_host_filter,
         unsigned int s3_max_redirects,
         unsigned int s3_retry_attempts,
+        bool s3_slow_all_threads_after_network_error,
         bool enable_s3_requests_logging,
         bool for_disk_s3,
         const ThrottlerPtr & get_request_throttler,

src/IO/S3/Credentials.cpp

@@ -712,6 +712,7 @@ S3CredentialsProviderChain::S3CredentialsProviderChain(
                 configuration.remote_host_filter,
                 configuration.s3_max_redirects,
                 configuration.s3_retry_attempts,
+                configuration.s3_slow_all_threads_after_network_error,
                 configuration.enable_s3_requests_logging,
                 configuration.for_disk_s3,
                 configuration.get_request_throttler,
@@ -727,6 +728,7 @@ S3CredentialsProviderChain::S3CredentialsProviderChain(
                 configuration.remote_host_filter,
                 configuration.s3_max_redirects,
                 configuration.s3_retry_attempts,
+                configuration.s3_slow_all_threads_after_network_error,
                 configuration.enable_s3_requests_logging,
                 configuration.for_disk_s3,
                 configuration.get_request_throttler,
@@ -785,6 +787,7 @@ S3CredentialsProviderChain::S3CredentialsProviderChain(
                 configuration.remote_host_filter,
                 configuration.s3_max_redirects,
                 configuration.s3_retry_attempts,
+                configuration.s3_slow_all_threads_after_network_error,
                 configuration.enable_s3_requests_logging,
                 configuration.for_disk_s3,
                 configuration.get_request_throttler,

src/IO/S3/PocoHTTPClient.cpp

@@ -117,6 +117,7 @@ PocoHTTPClientConfiguration::PocoHTTPClientConfiguration(
         const RemoteHostFilter & remote_host_filter_,
         unsigned int s3_max_redirects_,
         unsigned int s3_retry_attempts_,
+        bool s3_slow_all_threads_after_network_error_,
         bool enable_s3_requests_logging_,
         bool for_disk_s3_,
         bool s3_use_adaptive_timeouts_,
@@ -128,6 +129,7 @@ PocoHTTPClientConfiguration::PocoHTTPClientConfiguration(
     , remote_host_filter(remote_host_filter_)
     , s3_max_redirects(s3_max_redirects_)
     , s3_retry_attempts(s3_retry_attempts_)
+    , s3_slow_all_threads_after_network_error(s3_slow_all_threads_after_network_error_)
     , enable_s3_requests_logging(enable_s3_requests_logging_)
     , for_disk_s3(for_disk_s3_)
     , get_request_throttler(get_request_throttler_)