Merge pull request ClickHouse#84011 from ClickHouse/sync-disks

Implement AWS S3 authentication with an explicitly provided IAM role; implement OAuth for GCS.

Author: alexey-milovidov

src/Backups/BackupCoordinationReplicatedTables.cpp

@@ -1,6 +1,9 @@
 #include <Backups/BackupCoordinationReplicatedTables.h>
 #include <Storages/MergeTree/MergeTreePartInfo.h>
 #include <Storages/MergeTree/ReplicatedMergeTreeMutationEntry.h>
+#if CLICKHOUSE_CLOUD
+    #include <Storages/SharedMergeTree/SharedMergeTreeMutationEntry.h>
+#endif
 #include <Common/Exception.h>
 
 #include <boost/range/adaptor/map.hpp>
@@ -292,13 +295,6 @@ void BackupCoordinationReplicatedTables::prepare() const
                 auto part_info = MergeTreePartInfo::fromPartName(part_name, MERGE_TREE_DATA_MIN_FORMAT_VERSION_WITH_CUSTOM_PARTITIONING);
                 [[maybe_unused]] const auto & partition_id = part_info.getPartitionId();
 
-                auto & min_data_versions_by_partition = table_info.min_data_versions_by_partition;
-                auto it2 = min_data_versions_by_partition.find(partition_id);
-                if (it2 == min_data_versions_by_partition.end())
-                    min_data_versions_by_partition[partition_id] = part_info.getDataVersion();
-                else
-                    it2->second = std::min(it2->second, part_info.getDataVersion());
-
                 table_info.covered_parts_finder->addPartInfo(std::move(part_info), part_replicas.replica_names[0]);
             }
 
@@ -310,24 +306,6 @@ void BackupCoordinationReplicatedTables::prepare() const
                 const auto & chosen_replica_name = *part_replicas.replica_names[chosen_index];
                 table_info.part_names_by_replica_name[chosen_replica_name].push_back(part_name);
             }
-
-            /// Remove finished or unrelated mutations.
-            std::unordered_map<String, String> unfinished_mutations;
-            for (const auto & [mutation_id, mutation_entry_str] : table_info.mutations)
-            {
-                auto mutation_entry = ReplicatedMergeTreeMutationEntry::parse(mutation_entry_str, mutation_id);
-                std::map<String, Int64> new_block_numbers;
-                for (const auto & [partition_id, block_number] : mutation_entry.block_numbers)
-                {
-                    auto it = table_info.min_data_versions_by_partition.find(partition_id);
-                    if ((it != table_info.min_data_versions_by_partition.end()) && (it->second < block_number))
-                        new_block_numbers[partition_id] = block_number;
-                }
-                mutation_entry.block_numbers = std::move(new_block_numbers);
-                if (!mutation_entry.block_numbers.empty())
-                    unfinished_mutations[mutation_id] = mutation_entry.toString();
-            }
-            table_info.mutations = unfinished_mutations;
         }
         catch (Exception & e)
         {

src/Backups/BackupCoordinationReplicatedTables.h

@@ -104,7 +104,6 @@ class BackupCoordinationReplicatedTables
         std::map<String /* part_name */, PartReplicas> replicas_by_part_name; /// Should be ordered because we need this map to be in the same order on every replica.
         mutable std::unordered_map<String /* replica_name> */, Strings> part_names_by_replica_name;
         std::unique_ptr<CoveredPartsFinder> covered_parts_finder;
-        mutable std::unordered_map<String, Int64> min_data_versions_by_partition;
         mutable std::unordered_map<String, String> mutations;
         String replica_name_to_store_mutations;
         std::unordered_set<String> data_paths;

src/Backups/BackupEntriesCollector.cpp

@@ -45,6 +45,9 @@ namespace Setting
     extern const SettingsUInt64 backup_restore_keeper_retry_max_backoff_ms;
     extern const SettingsUInt64 backup_restore_keeper_max_retries;
     extern const SettingsSeconds lock_acquire_timeout;
+
+    /// Cloud only
+    extern const SettingsBool cloud_mode;
 }
 
 namespace ErrorCodes
@@ -112,7 +115,10 @@ BackupEntriesCollector::BackupEntriesCollector(
           context->getConfigRef().getUInt64("backups.min_sleep_before_next_attempt_to_collect_metadata", 100))
     , max_sleep_before_next_attempt_to_collect_metadata(
           context->getConfigRef().getUInt64("backups.max_sleep_before_next_attempt_to_collect_metadata", 5000))
-    , compare_collected_metadata(context->getConfigRef().getBool("backups.compare_collected_metadata", true))
+    , compare_collected_metadata(
+        context->getConfigRef().getBool("backups.compare_collected_metadata",
+                                        !context->getSettingsRef()[Setting::cloud_mode])) /// Collected metadata shouldn't be compared by default in our Cloud
+                                                                                          /// (because in the Cloud only Replicated databases are used)
     , log(getLogger("BackupEntriesCollector"))
     , zookeeper_retries_info(
           context->getSettingsRef()[Setting::backup_restore_keeper_max_retries],

src/Backups/BackupEntryWrappedWith.h

@@ -23,6 +23,10 @@ class BackupEntryWrappedWith : public IBackupEntry
     bool isEncryptedByDisk() const override { return entry->isEncryptedByDisk(); }
     bool isFromFile() const override { return entry->isFromFile(); }
     bool isFromImmutableFile() const override { return entry->isFromImmutableFile(); }
+    bool isFromRemoteFile() const override { return entry->isFromRemoteFile(); }
+    String getEndpointURI() const override { return entry->getEndpointURI(); }
+    String getNamespace() const override { return entry->getNamespace(); }
+    String getRemotePath() const override { return entry->getRemotePath(); }
     String getFilePath() const override { return entry->getFilePath(); }
     DiskPtr getDisk() const override { return entry->getDisk(); }
 

src/Backups/BackupFileInfo.cpp

@@ -124,6 +124,12 @@ BackupFileInfo buildFileInfoForBackupEntry(
     info.size = backup_entry->getSize();
     info.encrypted_by_disk = backup_entry->isEncryptedByDisk();
 
+    if (backup_entry->isFromRemoteFile())
+    {
+        info.object_key = backup_entry->getRemotePath();
+        return info;
+    }
+
     /// We don't set `info.data_file_name` and `info.data_file_index` in this function because they're set during backup coordination
     /// (see the class BackupCoordinationFileInfos).
 

src/Backups/BackupIO_AzureBlobStorage.cpp

@@ -13,6 +13,7 @@
 
 #include <Poco/Util/AbstractConfiguration.h>
 #include <azure/storage/blobs/blob_options.hpp>
+#include <azure/core/context.hpp>
 
 #include <filesystem>
 
@@ -65,6 +66,16 @@ static bool compareAuthMethod (AzureBlobStorage::AuthMethod auth_method_a, Azure
             Azure::Core::Credentials::TokenRequestContext tokenRequestContext;
             return managed_identity_a->get()->GetToken(tokenRequestContext, {}).Token == managed_identity_b->get()->GetToken(tokenRequestContext, {}).Token;
         }
+
+        const auto * static_credential_a = std::get_if<std::shared_ptr<AzureBlobStorage::StaticCredential>>(&auth_method_a);
+        const auto * static_credential_b = std::get_if<std::shared_ptr<AzureBlobStorage::StaticCredential>>(&auth_method_b);
+
+        if (static_credential_a && static_credential_b)
+        {
+            Azure::Core::Credentials::TokenRequestContext tokenRequestContext;
+            auto az_context = Azure::Core::Context();
+            return static_credential_a->get()->GetToken(tokenRequestContext, az_context).Token == static_credential_b->get()->GetToken(tokenRequestContext, az_context).Token;
+        }
     }
     catch (const Azure::Core::Credentials::AuthenticationException & e)
     {
@@ -83,7 +94,7 @@ BackupReaderAzureBlobStorage::BackupReaderAzureBlobStorage(
     const WriteSettings & write_settings_,
     const ContextPtr & context_)
     : BackupReaderDefault(read_settings_, write_settings_, getLogger("BackupReaderAzureBlobStorage"))
-    , data_source_description{DataSourceType::ObjectStorage, ObjectStorageType::Azure, MetadataStorageType::None, connection_params_.getConnectionURL(), false, false}
+    , data_source_description{DataSourceType::ObjectStorage, ObjectStorageType::Azure, MetadataStorageType::None, connection_params_.getConnectionURL(), false, false, ""}
     , connection_params(connection_params_)
     , blob_path(blob_path_)
 {
@@ -95,8 +106,10 @@ BackupReaderAzureBlobStorage::BackupReaderAzureBlobStorage(
         connection_params.auth_method,
         std::move(client_ptr),
         std::move(settings_ptr),
+        connection_params,
         connection_params.getContainer(),
-        connection_params.getConnectionURL());
+        connection_params.getConnectionURL(),
+        /*common_key_prefix*/ "");
 
     client = object_storage->getAzureBlobStorageClient();
     settings = object_storage->getSettings();
@@ -177,7 +190,7 @@ BackupWriterAzureBlobStorage::BackupWriterAzureBlobStorage(
     const ContextPtr & context_,
     bool attempt_to_create_container)
     : BackupWriterDefault(read_settings_, write_settings_, getLogger("BackupWriterAzureBlobStorage"))
-    , data_source_description{DataSourceType::ObjectStorage, ObjectStorageType::Azure, MetadataStorageType::None, connection_params_.getConnectionURL(), false, false}
+    , data_source_description{DataSourceType::ObjectStorage, ObjectStorageType::Azure, MetadataStorageType::None, connection_params_.getConnectionURL(), false, false, ""}
     , connection_params(connection_params_)
     , blob_path(blob_path_)
 {
@@ -192,8 +205,10 @@ BackupWriterAzureBlobStorage::BackupWriterAzureBlobStorage(
         connection_params.auth_method,
         std::move(client_ptr),
         std::move(settings_ptr),
+        connection_params,
         connection_params.getContainer(),
-        connection_params.getConnectionURL());
+        connection_params.getConnectionURL(),
+        /*common_key_prefix*/ "");
 
 
     client = object_storage->getAzureBlobStorageClient();

src/Backups/BackupIO_S3.cpp

@@ -46,6 +46,13 @@ namespace S3AuthSetting
     extern const S3AuthSettingsString server_side_encryption_customer_key_base64;
     extern const S3AuthSettingsBool use_environment_credentials;
     extern const S3AuthSettingsBool use_insecure_imds_request;
+
+    extern const S3AuthSettingsString role_arn;
+    extern const S3AuthSettingsString role_session_name;
+    extern const S3AuthSettingsString http_client;
+    extern const S3AuthSettingsString service_account;
+    extern const S3AuthSettingsString metadata_service;
+    extern const S3AuthSettingsString request_token_path;
 }
 
 namespace S3RequestSetting
@@ -69,6 +76,8 @@ namespace
         const S3::URI & s3_uri,
         const String & access_key_id,
         const String & secret_access_key,
+        String role_arn,
+        String role_session_name,
         const S3Settings & settings,
         const ContextPtr & context)
     {
@@ -84,6 +93,12 @@ namespace
         const Settings & global_settings = context->getGlobalContext()->getSettingsRef();
         const Settings & local_settings = context->getSettingsRef();
 
+        if (role_arn.empty())
+        {
+            role_arn = settings.auth_settings[S3AuthSetting::role_arn];
+            role_session_name = settings.auth_settings[S3AuthSetting::role_session_name];
+        }
+
         S3::PocoHTTPClientConfiguration client_configuration = S3::ClientFactory::instance().createClientConfiguration(
             settings.auth_settings[S3AuthSetting::region],
             context->getRemoteHostFilter(),
@@ -108,6 +123,11 @@ namespace
         client_configuration.http_max_field_name_size = request_settings[S3RequestSetting::http_max_field_name_size];
         client_configuration.http_max_field_value_size = request_settings[S3RequestSetting::http_max_field_value_size];
 
+        client_configuration.http_client = settings.auth_settings[S3AuthSetting::http_client];
+        client_configuration.service_account = settings.auth_settings[S3AuthSetting::service_account];
+        client_configuration.metadata_service = settings.auth_settings[S3AuthSetting::metadata_service];
+        client_configuration.request_token_path = settings.auth_settings[S3AuthSetting::request_token_path];
+
         S3::ClientSettings client_settings{
             .use_virtual_addressing = s3_uri.is_virtual_hosted_style,
             .disable_checksum = local_settings[Setting::s3_disable_checksum],
@@ -128,7 +148,10 @@ namespace
                 settings.auth_settings[S3AuthSetting::use_environment_credentials],
                 settings.auth_settings[S3AuthSetting::use_insecure_imds_request],
                 settings.auth_settings[S3AuthSetting::expiration_window_seconds],
-                settings.auth_settings[S3AuthSetting::no_sign_request]
+                settings.auth_settings[S3AuthSetting::no_sign_request],
+                std::move(role_arn),
+                std::move(role_session_name),
+                /*sts_endpoint_override=*/""
             });
     }
 
@@ -150,14 +173,16 @@ BackupReaderS3::BackupReaderS3(
     const S3::URI & s3_uri_,
     const String & access_key_id_,
     const String & secret_access_key_,
+    const String & role_arn,
+    const String & role_session_name,
     bool allow_s3_native_copy,
     const ReadSettings & read_settings_,
     const WriteSettings & write_settings_,
     const ContextPtr & context_,
     bool is_internal_backup)
     : BackupReaderDefault(read_settings_, write_settings_, getLogger("BackupReaderS3"))
     , s3_uri(s3_uri_)
-    , data_source_description{DataSourceType::ObjectStorage, ObjectStorageType::S3, MetadataStorageType::None, s3_uri.endpoint, false, false}
+    , data_source_description{DataSourceType::ObjectStorage, ObjectStorageType::S3, MetadataStorageType::None, s3_uri.endpoint, false, false, ""}
 {
     s3_settings.loadFromConfig(context_->getConfigRef(), "s3", context_->getSettingsRef());
 
@@ -170,7 +195,7 @@ BackupReaderS3::BackupReaderS3(
     s3_settings.request_settings.updateFromSettings(context_->getSettingsRef(), /* if_changed */true);
     s3_settings.request_settings[S3RequestSetting::allow_native_copy] = allow_s3_native_copy;
 
-    client = makeS3Client(s3_uri_, access_key_id_, secret_access_key_, s3_settings, context_);
+    client = makeS3Client(s3_uri_, access_key_id_, secret_access_key_, role_arn, role_session_name, s3_settings, context_);
 
     if (auto blob_storage_system_log = context_->getBlobStorageLog())
         blob_storage_log = std::make_shared<BlobStorageLogWriter>(blob_storage_system_log);
@@ -247,6 +272,8 @@ BackupWriterS3::BackupWriterS3(
     const S3::URI & s3_uri_,
     const String & access_key_id_,
     const String & secret_access_key_,
+    const String & role_arn,
+    const String & role_session_name,
     bool allow_s3_native_copy,
     const String & storage_class_name,
     const ReadSettings & read_settings_,
@@ -255,7 +282,7 @@ BackupWriterS3::BackupWriterS3(
     bool is_internal_backup)
     : BackupWriterDefault(read_settings_, write_settings_, getLogger("BackupWriterS3"))
     , s3_uri(s3_uri_)
-    , data_source_description{DataSourceType::ObjectStorage, ObjectStorageType::S3, MetadataStorageType::None, s3_uri.endpoint, false, false}
+    , data_source_description{DataSourceType::ObjectStorage, ObjectStorageType::S3, MetadataStorageType::None, s3_uri.endpoint, false, false, ""}
     , s3_capabilities(getCapabilitiesFromConfig(context_->getConfigRef(), "s3"))
 {
     s3_settings.loadFromConfig(context_->getConfigRef(), "s3", context_->getSettingsRef());
@@ -270,7 +297,8 @@ BackupWriterS3::BackupWriterS3(
     s3_settings.request_settings[S3RequestSetting::allow_native_copy] = allow_s3_native_copy;
     s3_settings.request_settings[S3RequestSetting::storage_class_name] = storage_class_name;
 
-    client = makeS3Client(s3_uri_, access_key_id_, secret_access_key_, s3_settings, context_);
+    client = makeS3Client(s3_uri_, access_key_id_, secret_access_key_, role_arn, role_session_name, s3_settings, context_);
+
     if (auto blob_storage_system_log = context_->getBlobStorageLog())
     {
         blob_storage_log = std::make_shared<BlobStorageLogWriter>(blob_storage_system_log);

src/Backups/BackupIO_S3.h

@@ -23,6 +23,8 @@ class BackupReaderS3 : public BackupReaderDefault
         const S3::URI & s3_uri_,
         const String & access_key_id_,
         const String & secret_access_key_,
+        const String & role_arn,
+        const String & role_session_name,
         bool allow_s3_native_copy,
         const ReadSettings & read_settings_,
         const WriteSettings & write_settings_,
@@ -54,6 +56,8 @@ class BackupWriterS3 : public BackupWriterDefault
         const S3::URI & s3_uri_,
         const String & access_key_id_,
         const String & secret_access_key_,
+        const String & role_arn,
+        const String & role_session_name,
         bool allow_s3_native_copy,
         const String & storage_class_name,
         const ReadSettings & read_settings_,