thevar1able/ClickHouse
1
0
Fork 0
mirror of https://github.com/ClickHouse/ClickHouse.git synced 2024-11-28 02:21:59 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #59145 from ClickHouse/capability-check-s3-plain

Browse Source 
Capability check for `s3_plain`
This commit is contained in:
Antonio Andelic 2024-01-25 09:28:36 +01:00 committed by GitHub
commit 0386830b98
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 31 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

50
src/Disks/ObjectStorages/ObjectStorageFactory.cpp
View File

@ -80,9 +80,10 @@ ObjectStoragePtr ObjectStorageFactory::create(
} }
#if USE_AWS_S3 #if USE_AWS_S3
static S3::URI getS3URI(const Poco::Util::AbstractConfiguration & config, namespace
const std::string & config_prefix, {
const ContextPtr & context)
S3::URI getS3URI(const Poco::Util::AbstractConfiguration & config, const std::string & config_prefix, const ContextPtr & context)
{ {
String endpoint = context->getMacros()->expand(config.getString(config_prefix + ".endpoint")); String endpoint = context->getMacros()->expand(config.getString(config_prefix + ".endpoint"));
S3::URI uri(endpoint); S3::URI uri(endpoint);
@ -94,6 +95,23 @@ static S3::URI getS3URI(const Poco::Util::AbstractConfiguration & config,
return uri; return uri;
} }
void checkS3Capabilities(
S3ObjectStorage & storage, const S3Capabilities s3_capabilities, const String & name, const String & key_with_trailing_slash)
{
/// If `support_batch_delete` is turned on (default), check and possibly switch it off.
if (s3_capabilities.support_batch_delete && !checkBatchRemove(storage, key_with_trailing_slash))
{
LOG_WARNING(
&Poco::Logger::get("S3ObjectStorage"),
"Storage for disk {} does not support batch delete operations, "
"so `s3_capabilities.support_batch_delete` was automatically turned off during the access check. "
"To remove this message set `s3_capabilities.support_batch_delete` for the disk to `false`.",
name);
storage.setCapabilitiesSupportBatchDelete(false);
}
}
}
void registerS3ObjectStorage(ObjectStorageFactory & factory) void registerS3ObjectStorage(ObjectStorageFactory & factory)
{ {
static constexpr auto disk_type = "s3"; static constexpr auto disk_type = "s3";
@ -116,20 +134,8 @@ void registerS3ObjectStorage(ObjectStorageFactory & factory)
/// NOTE: should we still perform this check for clickhouse-disks? /// NOTE: should we still perform this check for clickhouse-disks?
if (!skip_access_check) if (!skip_access_check)
{ checkS3Capabilities(*object_storage, s3_capabilities, name, uri.key);
/// If `support_batch_delete` is turned on (default), check and possibly switch it off.
if (s3_capabilities.support_batch_delete && !checkBatchRemove(*object_storage, uri.key))
{
LOG_WARNING(
&Poco::Logger::get("S3ObjectStorage"),
"Storage for disk {} does not support batch delete operations, "
"so `s3_capabilities.support_batch_delete` was automatically turned off during the access check. "
"To remove this message set `s3_capabilities.support_batch_delete` for the disk to `false`.",
name
);
object_storage->setCapabilitiesSupportBatchDelete(false);
}
}
return object_storage; return object_storage;
}); });
} }
@ -143,7 +149,7 @@ void registerS3PlainObjectStorage(ObjectStorageFactory & factory)
const Poco::Util::AbstractConfiguration & config, const Poco::Util::AbstractConfiguration & config,
const std::string & config_prefix, const std::string & config_prefix,
const ContextPtr & context, const ContextPtr & context,
bool /* skip_access_check */) -> ObjectStoragePtr bool skip_access_check) -> ObjectStoragePtr
{ {
/// send_metadata changes the filenames (includes revision), while /// send_metadata changes the filenames (includes revision), while
/// s3_plain do not care about this, and expect that the file name /// s3_plain do not care about this, and expect that the file name
@ -159,8 +165,14 @@ void registerS3PlainObjectStorage(ObjectStorageFactory & factory)
auto client = getClient(config, config_prefix, context, *settings); auto client = getClient(config, config_prefix, context, *settings);
auto key_generator = getKeyGenerator(disk_type, uri, config, config_prefix); auto key_generator = getKeyGenerator(disk_type, uri, config, config_prefix);
return std::make_shared<S3PlainObjectStorage>( auto object_storage = std::make_shared<S3PlainObjectStorage>(
std::move(client), std::move(settings), uri, s3_capabilities, key_generator, name); std::move(client), std::move(settings), uri, s3_capabilities, key_generator, name);
/// NOTE: should we still perform this check for clickhouse-disks?
if (!skip_access_check)
checkS3Capabilities(*object_storage, s3_capabilities, name, uri.key);
return object_storage;
}); });
} }
#endif #endif