diff --git a/be/src/common/config.cpp b/be/src/common/config.cpp index 9ec587bd100ec5..5fa1fa6091973d 100644 --- a/be/src/common/config.cpp +++ b/be/src/common/config.cpp @@ -1512,6 +1512,8 @@ DEFINE_mInt32(s3_read_max_wait_time_ms, "800"); DEFINE_mBool(enable_s3_object_check_after_upload, "true"); DEFINE_mInt32(aws_client_request_timeout_ms, "30000"); +DEFINE_mBool(s3_disable_content_md5, "false"); + DEFINE_mBool(enable_s3_rate_limiter, "false"); DEFINE_mInt64(s3_get_bucket_tokens, "1000000000000000000"); DEFINE_Validator(s3_get_bucket_tokens, [](int64_t config) -> bool { return config > 0; }); diff --git a/be/src/common/config.h b/be/src/common/config.h index 012f09a735c0cd..c9a6241b45882f 100644 --- a/be/src/common/config.h +++ b/be/src/common/config.h @@ -1621,6 +1621,12 @@ DECLARE_mInt32(s3_read_max_wait_time_ms); DECLARE_mBool(enable_s3_object_check_after_upload); DECLARE_mInt32(aws_client_request_timeout_ms); +// When true, omit the Content-MD5 header on S3 PutObject / UploadPart and send a +// CRC32C checksum instead. Required for S3 Express One Zone, which returns 501 +// NotImplemented for Content-MD5. Endpoints containing "s3express" auto-enable +// this regardless of the flag. +DECLARE_mBool(s3_disable_content_md5); + // write as inverted index tmp directory DECLARE_String(tmp_file_dir); diff --git a/be/src/io/fs/s3_obj_storage_client.cpp b/be/src/io/fs/s3_obj_storage_client.cpp index 1b8bf7b473c076..1d1f146cd6244f 100644 --- a/be/src/io/fs/s3_obj_storage_client.cpp +++ b/be/src/io/fs/s3_obj_storage_client.cpp @@ -34,6 +34,7 @@ #include #include #include +#include #include #include #include @@ -60,10 +61,12 @@ #include #include #include +#include #include #include +#include "common/config.h" #include "common/logging.h" #include "common/status.h" #include "cpp/sync_point.h" @@ -116,6 +119,22 @@ using namespace Aws::S3::Model; static constexpr int S3_REQUEST_THRESHOLD_MS = 5000; +namespace { +// AWS expects the CRC32C value as the big-endian 4-byte representation, base64-encoded. +Aws::String compute_crc32c_b64(std::string_view data) { + uint32_t crc = crc32c::Crc32c(reinterpret_cast(data.data()), data.size()); + uint8_t bytes[4] = {static_cast((crc >> 24) & 0xff), + static_cast((crc >> 16) & 0xff), + static_cast((crc >> 8) & 0xff), static_cast(crc & 0xff)}; + return Aws::Utils::HashingUtils::Base64Encode(Aws::Utils::ByteBuffer(bytes, sizeof(bytes))); +} +} // namespace + +S3ObjStorageClient::S3ObjStorageClient(std::shared_ptr client, + bool is_s3_express) + : _client(std::move(client)), + _disable_content_md5(config::s3_disable_content_md5 || is_s3_express) {} + ObjectStorageUploadResponse S3ObjStorageClient::create_multipart_upload( const ObjectStoragePathOptions& opts) { CreateMultipartUploadRequest request; @@ -158,8 +177,15 @@ ObjectStorageResponse S3ObjStorageClient::put_object(const ObjectStoragePathOpti Aws::S3::Model::PutObjectRequest request; request.WithBucket(opts.bucket).WithKey(opts.key); auto string_view_stream = std::make_shared(stream.data(), stream.size()); - Aws::Utils::ByteBuffer part_md5(Aws::Utils::HashingUtils::CalculateMD5(*string_view_stream)); - request.SetContentMD5(Aws::Utils::HashingUtils::Base64Encode(part_md5)); + if (_disable_content_md5) { + // S3 Express One Zone rejects Content-MD5; use CRC32C instead. + request.SetChecksumAlgorithm(ChecksumAlgorithm::CRC32C); + request.SetChecksumCRC32C(compute_crc32c_b64(stream)); + } else { + Aws::Utils::ByteBuffer part_md5( + Aws::Utils::HashingUtils::CalculateMD5(*string_view_stream)); + request.SetContentMD5(Aws::Utils::HashingUtils::Base64Encode(part_md5)); + } request.SetBody(string_view_stream); request.SetContentLength(stream.size()); request.SetContentType("application/octet-stream"); @@ -202,8 +228,15 @@ ObjectStorageUploadResponse S3ObjStorageClient::upload_part(const ObjectStorageP request.SetBody(string_view_stream); - Aws::Utils::ByteBuffer part_md5(Aws::Utils::HashingUtils::CalculateMD5(*string_view_stream)); - request.SetContentMD5(Aws::Utils::HashingUtils::Base64Encode(part_md5)); + if (_disable_content_md5) { + // S3 Express One Zone rejects Content-MD5; use CRC32C instead. + request.SetChecksumAlgorithm(ChecksumAlgorithm::CRC32C); + request.SetChecksumCRC32C(compute_crc32c_b64(stream)); + } else { + Aws::Utils::ByteBuffer part_md5( + Aws::Utils::HashingUtils::CalculateMD5(*string_view_stream)); + request.SetContentMD5(Aws::Utils::HashingUtils::Base64Encode(part_md5)); + } request.SetContentLength(stream.size()); request.SetContentType("application/octet-stream"); @@ -502,4 +535,4 @@ std::string S3ObjStorageClient::generate_presigned_url(const ObjectStoragePathOp expiration_secs); } -} // namespace doris::io \ No newline at end of file +} // namespace doris::io diff --git a/be/src/io/fs/s3_obj_storage_client.h b/be/src/io/fs/s3_obj_storage_client.h index 45294226594d81..0216318936e4f4 100644 --- a/be/src/io/fs/s3_obj_storage_client.h +++ b/be/src/io/fs/s3_obj_storage_client.h @@ -32,7 +32,8 @@ class ObjClientHolder; class S3ObjStorageClient final : public ObjStorageClient { public: - S3ObjStorageClient(std::shared_ptr client) : _client(std::move(client)) {} + explicit S3ObjStorageClient(std::shared_ptr client, + bool is_s3_express = false); ~S3ObjStorageClient() override = default; ObjectStorageUploadResponse create_multipart_upload( const ObjectStoragePathOptions& opts) override; @@ -58,6 +59,10 @@ class S3ObjStorageClient final : public ObjStorageClient { private: std::shared_ptr _client; + // True for S3 Express One Zone endpoints (or when config::s3_disable_content_md5 + // is on). When set, uploads send a CRC32C checksum instead of Content-MD5, + // since S3 Express returns 501 NotImplemented for the latter. + bool _disable_content_md5 = false; }; } // namespace doris::io diff --git a/be/src/util/s3_util.cpp b/be/src/util/s3_util.cpp index fc4c81d4bd96f2..028613f5533dc4 100644 --- a/be/src/util/s3_util.cpp +++ b/be/src/util/s3_util.cpp @@ -491,11 +491,13 @@ std::shared_ptr S3ClientFactory::_create_s3_client( TEST_SYNC_POINT_RETURN_WITH_VALUE( "s3_client_factory::create", std::make_shared(std::make_shared())); - Aws::Client::ClientConfiguration aws_config = S3ClientFactory::getClientConfiguration(); - if (s3_conf.need_override_endpoint) { - aws_config.endpointOverride = s3_conf.endpoint; - } + // Use S3ClientConfiguration (not the legacy ClientConfiguration) so the SDK's + // endpoint-rules resolver is active. This is required for S3 Express One Zone: + // the resolver detects the --x-s3 bucket suffix, calls CreateSession, and signs + // requests with service name "s3express" instead of "s3". + Aws::S3::S3ClientConfiguration aws_config; aws_config.region = s3_conf.region; + aws_config.useVirtualAddressing = s3_conf.use_virtual_addressing; if (_ca_cert_file_path.empty()) { _ca_cert_file_path = get_valid_ca_cert_path(doris::split(config::ca_cert_file_paths, ";")); @@ -527,12 +529,20 @@ std::shared_ptr S3ClientFactory::_create_s3_client( aws_config.retryStrategy = std::make_shared( config::max_s3_client_retry /*scaleFactor = 25*/, /*retry_slow_down=*/false); - std::shared_ptr new_client = std::make_shared( - get_aws_credentials_provider(s3_conf), std::move(aws_config), - Aws::Client::AWSAuthV4Signer::PayloadSigningPolicy::Never, - s3_conf.use_virtual_addressing); + // S3 Express buckets are identified by the --x-s3 suffix or an s3express endpoint. + // Skip endpointOverride for these so the SDK resolves the bucket-specific endpoint + // and manages CreateSession automatically. For all other buckets, keep the override. + const bool express = is_s3_express(s3_conf.bucket, s3_conf.endpoint); + if (s3_conf.need_override_endpoint && !express) { + aws_config.endpointOverride = s3_conf.endpoint; + } + aws_config.disableS3ExpressAuth = !express; + + auto new_client = std::make_shared( + get_aws_credentials_provider(s3_conf), + Aws::MakeShared("S3Client"), aws_config); - auto obj_client = std::make_shared(std::move(new_client)); + auto obj_client = std::make_shared(std::move(new_client), express); LOG_INFO("create one s3 client with {}", s3_conf.to_string()); return obj_client; } @@ -794,4 +804,19 @@ std::string hide_access_key(const std::string& ak) { return key; } +bool is_s3_express(std::string_view bucket, std::string_view endpoint) { + constexpr std::string_view bucket_suffix = "--x-s3"; + if (bucket.ends_with(bucket_suffix)) { + return true; + } + + const auto scheme_pos = endpoint.find("://"); + if (scheme_pos != std::string_view::npos) { + endpoint.remove_prefix(scheme_pos + 3); + } + endpoint = endpoint.substr(0, endpoint.find('/')); + return endpoint.starts_with("s3express-") || + endpoint.find(".s3express-") != std::string_view::npos; +} + } // end namespace doris diff --git a/be/src/util/s3_util.h b/be/src/util/s3_util.h index 5546db857c91b1..950600fe8a534f 100644 --- a/be/src/util/s3_util.h +++ b/be/src/util/s3_util.h @@ -63,6 +63,7 @@ extern bvar::LatencyRecorder s3_copy_object_latency; }; // namespace s3_bvar std::string hide_access_key(const std::string& ak); +bool is_s3_express(std::string_view bucket, std::string_view endpoint); int reset_s3_rate_limiter(S3RateLimitType type, size_t max_speed, size_t max_burst, size_t limit); // Rebuild the S3 GET/PUT rate limiters if the related configs have changed. // Safe to call periodically; it is a no-op when nothing changed. diff --git a/be/test/util/s3_util_test.cpp b/be/test/util/s3_util_test.cpp index a87d0e9a2d2980..cba03bec576773 100644 --- a/be/test/util/s3_util_test.cpp +++ b/be/test/util/s3_util_test.cpp @@ -77,6 +77,14 @@ TEST_F(S3UTILTest, hide_access_key_typical_aws_key) { EXPECT_EQ("xxxxxxxFODNN7xxxxxxx", result); } +TEST_F(S3UTILTest, is_s3_express_context) { + EXPECT_TRUE(is_s3_express("bucket--use1-az4--x-s3", "")); + EXPECT_TRUE(is_s3_express("", "bucket.s3express-use1-az4.us-east-1.amazonaws.com")); + EXPECT_FALSE(is_s3_express("bucket--x-s3-suffix", "")); + EXPECT_FALSE(is_s3_express("bucket", "https://example.com/s3express/path")); + EXPECT_FALSE(is_s3_express("bucket", "s3.us-east-1.amazonaws.com")); +} + // Verifies that check_s3_rate_limiter_config_changed() rebuilds the global GET rate // limiter when the related configs change. This is the behavior the cloud vault refresh // thread relies on to apply dynamically modified s3_get_* rate limiter configs without diff --git a/common/cpp/aws_logger.h b/common/cpp/aws_logger.h index 85734d13e1411c..9f06cec80fac0f 100644 --- a/common/cpp/aws_logger.h +++ b/common/cpp/aws_logger.h @@ -20,6 +20,8 @@ #include #include #include +#include +#include class DorisAWSLogger final : public Aws::Utils::Logging::LogSystemInterface { public: @@ -36,6 +38,13 @@ class DorisAWSLogger final : public Aws::Utils::Logging::LogSystemInterface { _log_impl(log_level, tag, message_stream.str().c_str()); } + void vaLog(Aws::Utils::Logging::LogLevel log_level, const char* tag, const char* format_str, + va_list args) final { + char buf[4096]; + vsnprintf(buf, sizeof(buf), format_str, args); + _log_impl(log_level, tag, buf); + } + void Flush() final {} private: diff --git a/thirdparty/vars.sh b/thirdparty/vars.sh index af46e566b8a30f..1b4ca87919f81b 100644 --- a/thirdparty/vars.sh +++ b/thirdparty/vars.sh @@ -356,10 +356,10 @@ BOOTSTRAP_TABLE_CSS_FILE="bootstrap-table.min.css" BOOTSTRAP_TABLE_CSS_MD5SUM="23389d4456da412e36bae30c469a766a" # aws sdk -AWS_SDK_DOWNLOAD="https://github.com/aws/aws-sdk-cpp/archive/refs/tags/1.11.219.tar.gz" -AWS_SDK_NAME="aws-sdk-cpp-1.11.219.tar.gz" -AWS_SDK_SOURCE="aws-sdk-cpp-1.11.219" -AWS_SDK_MD5SUM="80aa616efe1a3e7a9bf0dfbc44a97864" +AWS_SDK_DOWNLOAD="https://github.com/aws/aws-sdk-cpp/archive/refs/tags/1.11.400.tar.gz" +AWS_SDK_NAME="aws-sdk-cpp-1.11.400.tar.gz" +AWS_SDK_SOURCE="aws-sdk-cpp-1.11.400" +AWS_SDK_MD5SUM="329c46612df55ba6715d9d2ccd61dc03" # tsan_header TSAN_HEADER_DOWNLOAD="https://gcc.gnu.org/git/?p=gcc.git;a=blob_plain;f=libsanitizer/include/sanitizer/tsan_interface_atomic.h;hb=refs/heads/releases/gcc-7"