Merge pull request ClickHouse#88023 from ClickHouse/backport/25.8/88001

Backport ClickHouse#88001 to 25.8: Add logs to S3 Client

Author: nikitamikhaylov

src/Backups/BackupIO_S3.cpp

@@ -117,6 +117,7 @@ namespace
             local_settings[Setting::s3_slow_all_threads_after_retryable_error],
             local_settings[Setting::enable_s3_requests_logging],
             /* for_disk_s3 = */ false,
+            /* opt_disk_name = */ {},
             request_settings.get_request_throttler,
             request_settings.put_request_throttler,
             s3_uri.uri.getScheme());

src/Coordination/KeeperSnapshotManagerS3.cpp

@@ -123,6 +123,7 @@ void KeeperSnapshotManagerS3::updateS3Configuration(const Poco::Util::AbstractCo
             s3_slow_all_threads_after_retryable_error,
             enable_s3_requests_logging,
             /* for_disk_s3 = */ false,
+            /* opt_disk_name = */ {},
             /* get_request_throttler = */ {},
             /* put_request_throttler = */ {},
             new_uri.uri.getScheme());

src/Databases/DataLake/GlueCatalog.cpp

@@ -122,9 +122,10 @@ GlueCatalog::GlueCatalog(
         s3_slow_all_threads_after_network_error,
         s3_slow_all_threads_after_retryable_error,
         enable_s3_requests_logging,
-        false,
-        nullptr,
-        nullptr);
+        /* for_disk_s3 = */ false,
+        /* opt_disk_name = */ {},
+        /* get_request_throttler = */ nullptr,
+        /* put_request_throttler = */ nullptr);
 
     Aws::Glue::GlueClientConfiguration client_configuration;
     client_configuration.maxConnections = static_cast<unsigned>(global_settings[DB::Setting::s3_max_connections]);

src/Disks/ObjectStorages/ObjectStorageFactory.cpp

@@ -189,7 +189,7 @@ void registerS3ObjectStorage(ObjectStorageFactory & factory)
         auto endpoint = getEndpoint(config, config_prefix, context);
         auto settings = std::make_unique<S3Settings>();
         settings->loadFromConfigForObjectStorage(config, config_prefix, context->getSettingsRef(), uri.uri.getScheme(), true);
-        auto client = getClient(endpoint, *settings, context, /* for_disk_s3 */true);
+        auto client = getClient(endpoint, *settings, context, /* for_disk_s3 */ true, name);
         auto key_generator = getKeyGenerator(uri, config, config_prefix);
 
         auto object_storage = createObjectStorage<S3ObjectStorage>(
@@ -217,7 +217,7 @@ void registerS3PlainObjectStorage(ObjectStorageFactory & factory)
         auto endpoint = getEndpoint(config, config_prefix, context);
         auto settings = std::make_unique<S3Settings>();
         settings->loadFromConfigForObjectStorage(config, config_prefix, context->getSettingsRef(), uri.uri.getScheme(), true);
-        auto client = getClient(endpoint, *settings, context, /* for_disk_s3 */true);
+        auto client = getClient(endpoint, *settings, context, /* for_disk_s3 */ true, name);
         auto key_generator = getKeyGenerator(uri, config, config_prefix);
 
         auto object_storage = std::make_shared<PlainObjectStorage<S3ObjectStorage>>(
@@ -244,7 +244,7 @@ void registerS3PlainRewritableObjectStorage(ObjectStorageFactory & factory)
             auto endpoint = getEndpoint(config, config_prefix, context);
             auto settings = std::make_unique<S3Settings>();
             settings->loadFromConfigForObjectStorage(config, config_prefix, context->getSettingsRef(), uri.uri.getScheme(), true);
-            auto client = getClient(endpoint, *settings, context, /* for_disk_s3 */true);
+            auto client = getClient(endpoint, *settings, context, /* for_disk_s3 */ true, name);
             auto key_generator = getKeyGenerator(uri, config, config_prefix);
 
             auto metadata_storage_metrics = DB::MetadataStorageMetrics::create<S3ObjectStorage, MetadataStorageType::PlainRewritable>();

src/Disks/ObjectStorages/S3/S3ObjectStorage.cpp

@@ -508,7 +508,7 @@ void S3ObjectStorage::applyNewSettings(
     if (options.allow_client_change
         && (current_settings->auth_settings.hasUpdates(modified_settings->auth_settings) || for_disk_s3))
     {
-        auto new_client = getClient(uri, *modified_settings, context, for_disk_s3);
+        auto new_client = getClient(uri, *modified_settings, context, for_disk_s3, disk_name);
         client.set(std::move(new_client));
     }
     s3_settings.set(std::move(modified_settings));

src/Disks/ObjectStorages/S3/diskSettings.cpp

@@ -72,19 +72,18 @@ std::unique_ptr<S3::Client> getClient(
     const std::string & endpoint,
     const S3Settings & settings,
     ContextPtr context,
-    bool for_disk_s3)
+    bool for_disk_s3,
+    std::optional<std::string> opt_disk_name)
+
 {
     auto url = S3::URI(endpoint);
     if (!url.key.ends_with('/'))
         url.key.push_back('/');
-    return getClient(url, settings, context, for_disk_s3);
+    return getClient(url, settings, context, for_disk_s3, opt_disk_name);
 }
 
-std::unique_ptr<S3::Client> getClient(
-    const S3::URI & url,
-    const S3Settings & settings,
-    ContextPtr context,
-    bool for_disk_s3)
+std::unique_ptr<S3::Client>
+getClient(const S3::URI & url, const S3Settings & settings, ContextPtr context, bool for_disk_s3, std::optional<std::string> opt_disk_name)
 {
     const Settings & global_settings = context->getGlobalContext()->getSettingsRef();
     const auto & auth_settings = settings.auth_settings;
@@ -129,6 +128,7 @@ std::unique_ptr<S3::Client> getClient(
         s3_slow_all_threads_after_retryable_error,
         enable_s3_requests_logging,
         for_disk_s3,
+        opt_disk_name,
         request_settings.get_request_throttler,
         request_settings.put_request_throttler,
         url.uri.getScheme());

src/Disks/ObjectStorages/S3/diskSettings.h

@@ -17,14 +17,11 @@ std::unique_ptr<S3::Client> getClient(
     const std::string & endpoint,
     const S3Settings & settings,
     ContextPtr context,
-    bool for_disk_s3);
+    bool for_disk_s3,
+    std::optional<std::string> opt_disk_name = {});
 
 std::unique_ptr<S3::Client> getClient(
-    const S3::URI & url_,
-    const S3Settings & settings,
-    ContextPtr context,
-    bool for_disk_s3);
-
+    const S3::URI & url_, const S3Settings & settings, ContextPtr context, bool for_disk_s3, std::optional<std::string> opt_disk_name = {});
 }
 
 #endif

src/IO/S3/Client.cpp

@@ -84,6 +84,7 @@ namespace S3
 
 Client::RetryStrategy::RetryStrategy(const PocoHTTPClientConfiguration::RetryStrategy & config_)
     : config(config_)
+    , log(getLogger("S3ClientRetryStrategy"))
 {
     chassert(config.max_delay_ms <= (1.0 + config.jitter_factor) * config.initial_delay_ms * (1ul << 31l));
     chassert(config.jitter_factor >= 0 && config.jitter_factor <= 1);
@@ -133,7 +134,7 @@ long Client::RetryStrategy::CalculateDelayBeforeNextRetry(const Aws::Client::AWS
     else
         res = std::min<uint64_t>(config.initial_delay_ms * backoffLimitedPow, config.max_delay_ms);
 
-    LOG_TEST(getLogger("RetryStrategy"), "Next retry in {} ms", res);
+    LOG_TEST(log, "Next retry in {} ms", res);
     return res;
 }
 
@@ -143,6 +144,37 @@ long Client::RetryStrategy::GetMaxAttempts() const
     return config.max_retries + 1;
 }
 
+void Client::RetryStrategy::RequestBookkeeping(const Aws::Client::HttpResponseOutcome & httpResponseOutcome)
+{
+    if (!httpResponseOutcome.IsSuccess())
+    {
+        const auto & error = httpResponseOutcome.GetError();
+        if (error.ShouldRetry())
+            LOG_TRACE(
+                log,
+                "Attempt {}/{} failed with retryable error: {}, {}",
+                httpResponseOutcome.GetRetryCount() + 1,
+                GetMaxAttempts(),
+                static_cast<size_t>(error.GetResponseCode()),
+                error.GetMessage());
+    }
+}
+
+void Client::RetryStrategy::RequestBookkeeping(
+    const Aws::Client::HttpResponseOutcome & httpResponseOutcome, const Aws::Client::AWSError<Aws::Client::CoreErrors> & lastError)
+{
+    if (httpResponseOutcome.IsSuccess())
+        LOG_TRACE(
+            log,
+            "Attempt {}/{} succeeded with response code {}, last error: {}, {}",
+            httpResponseOutcome.GetRetryCount() + 1,
+            GetMaxAttempts(),
+            static_cast<size_t>(httpResponseOutcome.GetResult()->GetResponseCode()),
+            static_cast<size_t>(lastError.GetResponseCode()),
+            lastError.GetMessage());
+    RequestBookkeeping(httpResponseOutcome);
+}
+
 namespace
 {
 
@@ -254,15 +286,27 @@ Client::Client(
 
     LOG_TRACE(log, "API mode of the S3 client: {}", api_mode);
 
-    LOG_TRACE(
-        log,
-        "Slowing down threads on retryable errors is {}",
-        client_configuration.s3_slow_all_threads_after_retryable_error ? "enabled" : "disabled");
-
-    LOG_TRACE(
-        log,
-        "Slowing down threads on network errors is {}",
-        client_configuration.s3_slow_all_threads_after_network_error ? "enabled" : "disabled");
+    if (client_configuration.for_disk_s3)
+    {
+        LOG_TRACE(
+            log,
+            "S3 client for disk '{}' initialized with s3_retry_attempts: {}",
+            client_configuration.opt_disk_name.value_or(""),
+            client_configuration.retry_strategy.max_retries);
+        LOG_TRACE(
+            log,
+            "S3 client for disk '{}': slowing down threads on retryable errors is {}",
+            client_configuration.opt_disk_name.value_or(""),
+            client_configuration.s3_slow_all_threads_after_retryable_error ? "enabled" : "disabled");
+    }
+    else
+    {
+        LOG_TRACE(log, "S3 client initialized with s3_retry_attempts: {}", client_configuration.retry_strategy.max_retries);
+        LOG_TRACE(
+            log,
+            "S3 client: slowing down threads on retryable errors is {}",
+            client_configuration.s3_slow_all_threads_after_retryable_error ? "enabled" : "disabled");
+    }
 
     detect_region = provider_type == ProviderType::AWS && explicit_region == Aws::Region::AWS_GLOBAL;
 
@@ -1143,6 +1187,7 @@ PocoHTTPClientConfiguration ClientFactory::createClientConfiguration( // NOLINT
     bool s3_slow_all_threads_after_retryable_error,
     bool enable_s3_requests_logging,
     bool for_disk_s3,
+    std::optional<std::string> opt_disk_name,
     const ThrottlerPtr & get_request_throttler,
     const ThrottlerPtr & put_request_throttler,
     const String & protocol)
@@ -1164,6 +1209,7 @@ PocoHTTPClientConfiguration ClientFactory::createClientConfiguration( // NOLINT
         s3_slow_all_threads_after_retryable_error,
         enable_s3_requests_logging,
         for_disk_s3,
+        opt_disk_name,
         context->getGlobalContext()->getSettingsRef()[Setting::s3_use_adaptive_timeouts],
         get_request_throttler,
         put_request_throttler,

src/IO/S3/Client.h

@@ -165,6 +165,11 @@ class Client : private Aws::S3::S3Client
         /// NOLINTNEXTLINE(google-runtime-int)
         long GetMaxAttempts() const override;
 
+        void RequestBookkeeping(const Aws::Client::HttpResponseOutcome & httpResponseOutcome) override;
+        void RequestBookkeeping(
+            const Aws::Client::HttpResponseOutcome & httpResponseOutcome,
+            const Aws::Client::AWSError<Aws::Client::CoreErrors> & lastError) override;
+
         /// Sometimes [1] GCS may suggest to use Rewrite over CopyObject, i.e.:
         ///
         ///     AWSError 'InternalError': Copy spanning locations and/or storage classes could not complete within 30 seconds. Please use the Rewrite method in the JSON API (https://cloud.google.com/storage/docs/json_api/v1/objects/rewrite) instead.
@@ -177,6 +182,7 @@ class Client : private Aws::S3::S3Client
 
     private:
         PocoHTTPClientConfiguration::RetryStrategy config;
+        LoggerPtr log;
     };
 
     /// SSE-KMS headers MUST be signed, so they need to be added before the SDK signs the message
@@ -350,6 +356,7 @@ class ClientFactory
         bool s3_slow_all_threads_after_retryable_error,
         bool enable_s3_requests_logging,
         bool for_disk_s3,
+        std::optional<std::string> opt_disk_name,
         const ThrottlerPtr & get_request_throttler,
         const ThrottlerPtr & put_request_throttler,
         const String & protocol = "https");

src/IO/S3/Credentials.cpp

@@ -742,6 +742,7 @@ S3CredentialsProviderChain::S3CredentialsProviderChain(
                 configuration.s3_slow_all_threads_after_retryable_error,
                 configuration.enable_s3_requests_logging,
                 configuration.for_disk_s3,
+                configuration.opt_disk_name,
                 configuration.get_request_throttler,
                 configuration.put_request_throttler);
             AddProvider(std::make_shared<AwsAuthSTSAssumeRoleWebIdentityCredentialsProvider>(aws_client_configuration, credentials_configuration.expiration_window_seconds, credentials_configuration.kms_role_arn));
@@ -759,6 +760,7 @@ S3CredentialsProviderChain::S3CredentialsProviderChain(
                 configuration.s3_slow_all_threads_after_retryable_error,
                 configuration.enable_s3_requests_logging,
                 configuration.for_disk_s3,
+                configuration.opt_disk_name,
                 configuration.get_request_throttler,
                 configuration.put_request_throttler);
             AddProvider(std::make_shared<SSOCredentialsProvider>(
@@ -811,6 +813,7 @@ S3CredentialsProviderChain::S3CredentialsProviderChain(
                 configuration.s3_slow_all_threads_after_retryable_error,
                 configuration.enable_s3_requests_logging,
                 configuration.for_disk_s3,
+                configuration.opt_disk_name,
                 configuration.get_request_throttler,
                 configuration.put_request_throttler,
                 Aws::Http::SchemeMapper::ToString(Aws::Http::Scheme::HTTP));