github-actions[bot] commented on code in PR #66350:
URL: https://github.com/apache/doris/pull/66350#discussion_r3714530278
##########
be/test/io/s3_client_factory_test.cpp:
##########
@@ -80,23 +61,7 @@ S3ClientConf make_hash_collision_conf(std::string endpoint,
bool is_internal_buc
} // namespace
-TEST_F(S3ClientFactoryTest, WrapsAllClientsInNonCloudMode) {
- CloudModeConfigGuard guard(false);
- auto& factory = S3ClientFactory::instance();
-
- auto external_client =
-
factory.create(make_factory_conf("non-cloud-external-rate-limit.example.com",
false));
- auto internal_client =
-
factory.create(make_factory_conf("non-cloud-internal-rate-limit.example.com",
true));
-
- ASSERT_NE(external_client, nullptr);
- ASSERT_NE(internal_client, nullptr);
-
EXPECT_NE(std::dynamic_pointer_cast<io::RateLimitedObjStorageClient>(external_client),
nullptr);
-
EXPECT_NE(std::dynamic_pointer_cast<io::RateLimitedObjStorageClient>(internal_client),
nullptr);
-}
-
-TEST_F(S3ClientFactoryTest,
WrapsOnlyInternalClientsInCloudModeAndDistinguishesHashCollisions) {
- CloudModeConfigGuard guard(true);
+TEST_F(S3ClientFactoryTest, DistinguishesHashCollisions) {
Review Comment:
[P2] Keep behavioral coverage for cloud limiter selection
The deleted factory tests asserted that non-cloud clients are limited and
that cloud mode limits only internal storage-vault clients. This replacement
now checks only hash-collision/cache identity, and no remaining test switches
cloud mode and observes the `is_internal_bucket` policy branch in
`S3ClientFactory::create`. Please retain a behavior-level test (for example
with an observable counting policy/backend) for non-cloud, cloud-internal, and
cloud-external clients so this production admission boundary cannot silently
flip.
##########
common/cpp/client/azure_obj_storage_backend.cpp:
##########
@@ -0,0 +1,497 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "azure_obj_storage_backend.h"
+
+#include <cctype>
+#include <string_view>
+
+#include "cpp/obj_retry_strategy.h"
+
+using namespace Azure::Storage::Blobs;
+
+namespace {
+std::string wrap_object_storage_path_msg(const
doris::ObjectStoragePathOptions& opts) {
+ return fmt::format("bucket {}, key {}, prefix {}, path {}", opts.bucket,
opts.key, opts.prefix,
+ opts.path.native());
+}
+
+std::string to_lower_ascii(std::string_view input) {
+ std::string lowered(input);
+ std::transform(lowered.begin(), lowered.end(), lowered.begin(),
+ [](unsigned char ch) { return
static_cast<char>(std::tolower(ch)); });
+ return lowered;
+}
+
+template <std::endian target, typename T>
+T to_endian(T value) {
+ if constexpr (std::endian::native == target) {
+ return value; // No swap needed
+ } else {
+ static_assert(std::endian::native == std::endian::big ||
+ std::endian::native == std::endian::little,
+ "Unsupported endianness");
+ return byte_swap(value);
+ }
+}
+
+inline void encode_fixed32_le(uint8_t* buf, uint32_t val) {
+ val = to_endian<std::endian::little>(val);
+ memcpy(buf, &val, sizeof(val));
+}
+
+auto base64_encode_part_num(int part_num) {
+ uint8_t buf[4];
+ encode_fixed32_le(buf, static_cast<uint32_t>(part_num));
+ return Aws::Utils::HashingUtils::Base64Encode({buf, sizeof(buf)});
+}
+
+constexpr char SAS_TOKEN_URL_TEMPLATE[] = "{}/{}/{}{}";
+constexpr char BlobNotFound[] = "BlobNotFound";
+} // namespace
+
+namespace doris {
+
+// As Azure's doc said, the batch size is 256
+// You can find out the num in
https://learn.microsoft.com/en-us/rest/api/storageservices/blob-batch?tabs=microsoft-entra-id
+// > Each batch request supports a maximum of 256 subrequests.
+constexpr size_t BlobBatchMaxOperations = 256;
+
+bool is_azure_tls_ca_error_message(std::string_view message) {
+ std::string lower = to_lower_ascii(message);
+ return lower.find("ssl ca cert") != std::string::npos ||
+ lower.find("peer failed verification") != std::string::npos ||
+ lower.find("unable to get local issuer certificate") !=
std::string::npos ||
+ lower.find("problem with the ssl ca cert") != std::string::npos;
+}
+
+std::string build_azure_tls_debug_suffix(std::string_view error_message,
+ std::string_view tls_debug_context) {
+ if (tls_debug_context.empty() ||
!is_azure_tls_ca_error_message(error_message)) {
+ return "";
+ }
+ return fmt::format(", {}", tls_debug_context);
+}
+
+template <typename Func>
+ObjectStorageResponse do_azure_client_call(Func f, const
ObjectStoragePathOptions& opts,
+ std::string_view tls_debug_context)
{
+ try {
+ f();
+ } catch (Azure::Core::RequestFailedException& e) {
+ doris::record_object_request_failed(static_cast<int>(e.StatusCode));
+ auto msg = fmt::format(
+ "Azure request failed because {}, error msg {}, http code {},
path msg {}{}",
+ e.what(), e.Message, static_cast<int>(e.StatusCode),
+ wrap_object_storage_path_msg(opts),
+ build_azure_tls_debug_suffix(fmt::format("{} {}", e.what(),
e.Message),
+ tls_debug_context));
+ LOG(WARNING) << msg;
+ return {.status = ObjectStorageStatus {TStatusCode::INTERNAL_ERROR,
std::move(msg)},
+ .http_code = static_cast<int>(e.StatusCode),
+ .request_id = std::move(e.RequestId)};
+ } catch (std::exception& e) {
+ auto msg = fmt::format("Azure request failed because {}, path msg
{}{}", e.what(),
+ wrap_object_storage_path_msg(opts),
+ build_azure_tls_debug_suffix(e.what(),
tls_debug_context));
+ LOG(WARNING) << msg;
+ return {.status = ObjectStorageStatus {TStatusCode::INTERNAL_ERROR,
std::move(msg)},
+ .http_code = 0,
+ .request_id = ""};
+ }
+ return ObjectStorageResponse::OK();
+}
+
+struct AzureBatchDeleter {
+ AzureBatchDeleter(BlobContainerClient* client, const
ObjectStoragePathOptions& opts,
+ std::string_view tls_debug_context)
+ : _client(client),
+ _batch(client->CreateBatch()),
+ _opts(opts),
+ _tls_debug_context(tls_debug_context) {}
+ // Submit one blob to be deleted in `AzureBatchDeleter::execute`
+ void delete_blob(const std::string& blob_name) {
+ deferred_resps.emplace_back(_batch.DeleteBlob(blob_name));
+ }
+ ObjectStorageResponse execute() {
+ if (deferred_resps.empty()) {
+ return ObjectStorageResponse::OK();
+ }
+ auto resp = do_azure_client_call(
+ [&]() {
+ client_bvar::ScopedLatency scoped_latency(
+ client_bvar::s3_delete_objects_latency);
+ _client->SubmitBatch(_batch);
+ },
+ _opts, _tls_debug_context);
+ if (resp.status.code != TStatusCode::OK) {
+ return resp;
+ }
+
+ for (auto&& defer_response : deferred_resps) {
+ try {
+ auto r = defer_response.GetResponse();
+ if (!r.Value.Deleted) {
+ auto msg = fmt::format("Azure batch delete failed, path
msg {}",
+
wrap_object_storage_path_msg(_opts));
+ LOG(WARNING) << msg;
+ return {.status = ObjectStorageStatus
{TStatusCode::INTERNAL_ERROR,
+ std::move(msg)},
+ .http_code = 0,
+ .request_id = ""};
+ }
+ } catch (Azure::Core::RequestFailedException& e) {
+ if (Azure::Core::Http::HttpStatusCode::NotFound ==
e.StatusCode &&
+ 0 == strcmp(e.ErrorCode.c_str(), BlobNotFound)) {
+ continue;
+ }
+
doris::record_object_request_failed(static_cast<int>(e.StatusCode));
+ auto msg = fmt::format(
+ "Azure request failed because {}, error msg {}, http
code {}, path msg "
+ "{}{}",
+ e.what(), e.Message, static_cast<int>(e.StatusCode),
+ wrap_object_storage_path_msg(_opts),
+ build_azure_tls_debug_suffix(fmt::format("{} {}",
e.what(), e.Message),
+ _tls_debug_context));
+ LOG(WARNING) << msg;
+ return {.status = ObjectStorageStatus
{TStatusCode::INTERNAL_ERROR, std::move(msg)},
+ .http_code = static_cast<int>(e.StatusCode),
+ .request_id = std::move(e.RequestId)};
+ }
+ }
+
+ return ObjectStorageResponse::OK();
+ }
+
+private:
+ BlobContainerClient* _client;
+ BlobContainerBatch _batch;
+ const ObjectStoragePathOptions& _opts;
+ std::string_view _tls_debug_context;
+ std::vector<Azure::Storage::DeferredResponse<Models::DeleteBlobResult>>
deferred_resps;
+};
+
+// Azure would do nothing
+ObjectStorageUploadResponse AzureObjStorageBackend::create_multipart_upload(
+ const ObjectStoragePathOptions& opts) {
+ return ObjectStorageUploadResponse {
+ .resp = ObjectStorageResponse::OK(),
+ };
+}
+
+ObjectStorageResponse AzureObjStorageBackend::put_object(const
ObjectStoragePathOptions& opts,
+ std::string_view
stream) {
+ auto client = _client->GetBlockBlobClient(opts.key);
+ return do_azure_client_call(
+ [&]() {
+ client_bvar::ScopedLatency
scoped_latency(client_bvar::s3_put_latency);
+ client.UploadFrom(reinterpret_cast<const
uint8_t*>(stream.data()), stream.size());
+ },
+ opts, _config.tls_debug_context);
+}
+
+ObjectStorageUploadResponse AzureObjStorageBackend::upload_part(
+ const ObjectStoragePathOptions& opts, std::string_view stream, int
part_num) {
+ auto client = _client->GetBlockBlobClient(opts.key);
+ try {
+ Azure::Core::IO::MemoryBodyStream memory_body(
+ reinterpret_cast<const uint8_t*>(stream.data()),
stream.size());
+ // The blockId must be base64 encoded
+ client_bvar::ScopedLatency
scoped_latency(client_bvar::s3_multi_part_upload_latency);
+ client.StageBlock(base64_encode_part_num(part_num), memory_body);
+ } catch (Azure::Core::RequestFailedException& e) {
+ record_object_request_failed(static_cast<int>(e.StatusCode));
+ auto tls_debug_suffix = build_azure_tls_debug_suffix(
+ fmt::format("{} {}", e.what(), e.Message),
_config.tls_debug_context);
+ auto msg = fmt::format(
+ "Azure request failed because {}, error msg {}, http code {},
path msg {}{}",
+ e.what(), e.Message, static_cast<int>(e.StatusCode),
+ wrap_object_storage_path_msg(opts), tls_debug_suffix);
+ LOG(WARNING) << msg;
+ // clang-format off
+ return {
+ .resp = {
+ .status = ObjectStorageStatus {TStatusCode::INTERNAL_ERROR,
std::move(msg)},
+ .http_code = static_cast<int>(e.StatusCode),
+ .request_id = std::move(e.RequestId),
+ },
+ };
+ // clang-format on
+ }
+ return ObjectStorageUploadResponse {.resp = ObjectStorageResponse::OK()};
+}
+
+ObjectStorageResponse AzureObjStorageBackend::complete_multipart_upload(
+ const ObjectStoragePathOptions& opts,
+ const std::vector<ObjectCompleteMultiPart>& completed_parts) {
+ auto client = _client->GetBlockBlobClient(opts.key);
+ std::vector<std::string> string_block_ids;
+ std::ranges::transform(
+ completed_parts, std::back_inserter(string_block_ids),
+ [](const ObjectCompleteMultiPart& i) { return
base64_encode_part_num(i.part_num); });
+ return do_azure_client_call(
+ [&]() {
+ client_bvar::ScopedLatency scoped_latency(
+ client_bvar::s3_multi_part_upload_latency);
+ client.CommitBlockList(string_block_ids);
+ },
+ opts, _config.tls_debug_context);
+}
+
+ObjectStorageHeadResponse AzureObjStorageBackend::head_object(
+ const ObjectStoragePathOptions& opts) {
+ try {
+ Models::BlobProperties properties = [&]() {
+ client_bvar::ScopedLatency
scoped_latency(client_bvar::s3_head_latency);
+ return _client->GetBlockBlobClient(opts.key).GetProperties().Value;
+ }();
+ return {.resp = ObjectStorageResponse::OK(), .file_size =
properties.BlobSize};
+ } catch (Azure::Core::RequestFailedException& e) {
+ if (e.StatusCode == Azure::Core::Http::HttpStatusCode::NotFound) {
+ return ObjectStorageHeadResponse {
+ .resp = {.status = ObjectStorageStatus
{TStatusCode::NOT_FOUND, ""},
+ .http_code = static_cast<int>(e.StatusCode),
+ .request_id = std::move(e.RequestId)},
+ };
+ }
+ record_object_request_failed(static_cast<int>(e.StatusCode));
+ auto tls_debug_suffix = build_azure_tls_debug_suffix(
+ fmt::format("{} {}", e.what(), e.Message),
_config.tls_debug_context);
+ auto msg = fmt::format(
+ "Azure request failed because {}, error msg {}, http code {},
path msg {}{}",
+ e.what(), e.Message, static_cast<int>(e.StatusCode),
+ wrap_object_storage_path_msg(opts), tls_debug_suffix);
+ return ObjectStorageHeadResponse {
Review Comment:
[P2] Log Azure head failures before Recycler discards the response
Unlike the general Azure call helper and the parallel S3 head path, this
catch returns the detailed message, HTTP code, and request ID without logging
them. Recycler's `S3Accessor::exists` converts every non-404 response to `-1`,
and its production checker can then report only the object path and `ret=-1`;
the Azure failure and request ID are lost. Please emit the constructed warning
here (or preserve the full response at the adapter boundary) before returning.
##########
common/cpp/client/s3_obj_storage_backend.cpp:
##########
@@ -0,0 +1,582 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "s3_obj_storage_backend.h"
+
+#include <cpp/client/obj_storage_client.h>
+#include <gen_cpp/Status_types.h>
+
+#include <algorithm>
+#include <chrono>
+
+#include "client_bvar.h"
+#include "cpp/obj_retry_strategy.h"
+
+namespace Aws::S3::Model {
+class DeleteObjectRequest;
+} // namespace Aws::S3::Model
+
+using Aws::S3::Model::CompletedPart;
+using Aws::S3::Model::CompletedMultipartUpload;
+using Aws::S3::Model::CompleteMultipartUploadRequest;
+using Aws::S3::Model::CreateMultipartUploadRequest;
+using Aws::S3::Model::UploadPartRequest;
+using Aws::S3::Model::UploadPartOutcome;
+
+namespace doris {
+using namespace Aws::S3::Model;
+namespace {
+
+constexpr int64_t S3_REQUEST_THRESHOLD_MS = 5000;
+
+int64_t elapsed_time_milliseconds(std::chrono::steady_clock::time_point start)
{
+ return
std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now()
-
+ start)
+ .count();
+}
+
+void record_s3_request_failed(const Aws::S3::S3Error& error) {
+ record_object_request_failed(static_cast<int>(error.GetResponseCode()));
+}
+
+} // namespace
+
+ObjectStorageStatus s3fs_error(const Aws::S3::S3Error& err, std::string_view
msg) {
+ using namespace Aws::Http;
+ switch (err.GetResponseCode()) {
+ case HttpResponseCode::NOT_FOUND:
+ return {TStatusCode::NOT_FOUND,
+ fmt::format("{}: {} {}", msg, err.GetExceptionName(),
err.GetMessage())};
+ case HttpResponseCode::FORBIDDEN:
+ // TODO: no permission and other 4xx errors should be handled
separately
+ return {TStatusCode::NOT_AUTHORIZED,
+ fmt::format("{}: {} {}", msg, err.GetExceptionName(),
err.GetMessage())};
+ case HttpResponseCode::REQUEST_NOT_MADE:
+ return {-1, fmt::format("{}: {} {}", msg, err.GetExceptionName(),
err.GetMessage())};
+ default:
+ return {TStatusCode::INTERNAL_ERROR,
+ fmt::format("{}: {} {}", msg, err.GetExceptionName(),
err.GetMessage())};
+ }
+}
+
+ObjectStorageUploadResponse S3ObjStorageBackend::create_multipart_upload(
+ const ObjectStoragePathOptions& opts) {
+ CreateMultipartUploadRequest request;
+ request.WithBucket(opts.bucket).WithKey(opts.key);
+ request.SetContentType("application/octet-stream");
+
+ const auto start = std::chrono::steady_clock::now();
+ auto outcome = SYNC_POINT_HOOK_RETURN_VALUE(
+ [&]() {
+ client_bvar::ScopedLatency scoped_latency(
+ client_bvar::s3_multi_part_upload_latency);
+ return _client->CreateMultipartUpload(request);
+ }(),
+ "s3_file_writer::create_multi_part_upload",
std::cref(request).get());
+ SYNC_POINT_CALLBACK("s3_file_writer::_open", &outcome);
+ const auto elapsed_ms = elapsed_time_milliseconds(start);
+
+ const auto& request_id = outcome.IsSuccess() ?
outcome.GetResult().GetRequestId()
+ :
outcome.GetError().GetRequestId();
+
+ LOG_IF(INFO, elapsed_ms > S3_REQUEST_THRESHOLD_MS)
+ << "CreateMultipartUpload cost=" << elapsed_ms << "ms"
+ << ", request_id=" << request_id << ", bucket=" << opts.bucket <<
", key=" << opts.key;
+
+ if (!outcome.IsSuccess()) {
+ record_s3_request_failed(outcome.GetError());
+ auto st = s3fs_error(outcome.GetError(), fmt::format("failed to
CreateMultipartUpload: {} ",
+
opts.path.native()));
+ LOG(WARNING) << st.code << " request_id=" << request_id;
+ return ObjectStorageUploadResponse {
+ .resp = {.status = st,
+ .http_code =
static_cast<int>(outcome.GetError().GetResponseCode()),
+ .request_id = outcome.GetError().GetRequestId()},
+ };
+ }
+
+ return ObjectStorageUploadResponse {.resp = ObjectStorageResponse::OK(),
+ .upload_id
{outcome.GetResult().GetUploadId()}};
+}
+
+ObjectStorageResponse S3ObjStorageBackend::put_object(const
ObjectStoragePathOptions& opts,
+ std::string_view stream)
{
+ Aws::S3::Model::PutObjectRequest request;
+ request.WithBucket(opts.bucket).WithKey(opts.key);
+ auto string_view_stream =
std::make_shared<StringViewStream>(stream.data(), stream.size());
+ 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");
+
+ const auto start = std::chrono::steady_clock::now();
+ auto outcome = SYNC_POINT_HOOK_RETURN_VALUE(
+ [&]() {
+ client_bvar::ScopedLatency
scoped_latency(client_bvar::s3_put_latency);
+ return _client->PutObject(request);
+ }(),
+ "s3_file_writer::put_object", std::cref(request).get(), &stream);
+ const auto elapsed_ms = elapsed_time_milliseconds(start);
+
+ const auto& request_id = outcome.IsSuccess() ?
outcome.GetResult().GetRequestId()
+ :
outcome.GetError().GetRequestId();
+
+ if (!outcome.IsSuccess()) {
+ record_s3_request_failed(outcome.GetError());
+ auto st = s3fs_error(outcome.GetError(),
+ fmt::format("failed to put object: {}",
opts.path.native()));
+ LOG(WARNING) << st.code << ", request_id=" << request_id;
+ return ObjectStorageResponse {
+ .status = st,
+ .http_code =
static_cast<int>(outcome.GetError().GetResponseCode()),
+ .request_id = outcome.GetError().GetRequestId()};
+ }
+
+ LOG_IF(INFO, elapsed_ms > S3_REQUEST_THRESHOLD_MS)
+ << "PutObject cost=" << elapsed_ms << "ms"
+ << ", request_id=" << request_id << ", bucket=" << opts.bucket <<
", key=" << opts.key;
+ return ObjectStorageResponse::OK();
+}
+
+ObjectStorageUploadResponse S3ObjStorageBackend::upload_part(const
ObjectStoragePathOptions& opts,
+ std::string_view
stream,
+ int part_num) {
+ UploadPartRequest request;
+ request.WithBucket(opts.bucket)
+ .WithKey(opts.key)
+ .WithPartNumber(part_num)
+ .WithUploadId(*opts.upload_id);
+ auto string_view_stream =
std::make_shared<StringViewStream>(stream.data(), stream.size());
+
+ 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));
+
+ request.SetContentLength(stream.size());
+ request.SetContentType("application/octet-stream");
+
+ const auto start = std::chrono::steady_clock::now();
+ auto outcome = SYNC_POINT_HOOK_RETURN_VALUE(
+ [&]() {
+ client_bvar::ScopedLatency scoped_latency(
+ client_bvar::s3_multi_part_upload_latency);
+
+ return _client->UploadPart(request);
+ }(),
+ "s3_file_writer::upload_part", std::cref(request).get(), &stream);
+ const auto elapsed_ms = elapsed_time_milliseconds(start);
+
+ const auto& request_id = outcome.IsSuccess() ?
outcome.GetResult().GetRequestId()
+ :
outcome.GetError().GetRequestId();
+
+ TEST_SYNC_POINT_CALLBACK("S3FileWriter::_upload_one_part", &outcome);
+ if (!outcome.IsSuccess()) {
+ record_s3_request_failed(outcome.GetError());
+ auto st = s3fs_error(outcome.GetError(),
+ fmt::format("failed to UploadPart: {}, part_num
{}, upload_id={}",
+ opts.path.native(), part_num,
*opts.upload_id));
+
+ LOG(WARNING) << st.code << ", request_id=" << request_id;
+ return ObjectStorageUploadResponse {
+ .resp = {.status = st,
+ .http_code =
static_cast<int>(outcome.GetError().GetResponseCode()),
+ .request_id = outcome.GetError().GetRequestId()}};
+ }
+ LOG_IF(INFO, elapsed_ms > S3_REQUEST_THRESHOLD_MS)
+ << "UploadPart cost=" << elapsed_ms << "ms"
+ << ", request_id=" << request_id << ", bucket=" << opts.bucket <<
", key=" << opts.key
+ << ", part_num=" << part_num << ", upload_id=" << *opts.upload_id;
+ return ObjectStorageUploadResponse {.resp = ObjectStorageResponse::OK(),
+ .etag = outcome.GetResult().GetETag()};
+}
+
+ObjectStorageResponse S3ObjStorageBackend::complete_multipart_upload(
+ const ObjectStoragePathOptions& opts,
+ const std::vector<ObjectCompleteMultiPart>& completed_parts) {
+ CompleteMultipartUploadRequest request;
+
request.WithBucket(opts.bucket).WithKey(opts.key).WithUploadId(*opts.upload_id);
+
+ CompletedMultipartUpload completed_upload;
+ std::vector<CompletedPart> complete_parts;
+ std::ranges::transform(completed_parts, std::back_inserter(complete_parts),
+ [](const ObjectCompleteMultiPart& part_ptr) {
+ CompletedPart part;
+ part.SetPartNumber(part_ptr.part_num);
+ part.SetETag(part_ptr.etag);
+ return part;
+ });
+ completed_upload.SetParts(std::move(complete_parts));
+ request.WithMultipartUpload(completed_upload);
+
+ TEST_SYNC_POINT_RETURN_WITH_VALUE("S3FileWriter::_complete:3",
ObjectStorageResponse(), this);
+
+ const auto start = std::chrono::steady_clock::now();
+ auto outcome = SYNC_POINT_HOOK_RETURN_VALUE(
+ [&]() {
+ client_bvar::ScopedLatency scoped_latency(
+ client_bvar::s3_multi_part_upload_latency);
+ return _client->CompleteMultipartUpload(request);
+ }(),
+ "s3_file_writer::complete_multi_part", std::cref(request).get());
+ const auto elapsed_ms = elapsed_time_milliseconds(start);
+
+ const auto& request_id = outcome.IsSuccess() ?
outcome.GetResult().GetRequestId()
+ :
outcome.GetError().GetRequestId();
+
+ if (!outcome.IsSuccess()) {
+ record_s3_request_failed(outcome.GetError());
+ auto st = s3fs_error(outcome.GetError(),
+ fmt::format("failed to CompleteMultipartUpload:
{}, upload_id={}",
+ opts.path.native(), *opts.upload_id));
+ LOG(WARNING) << st.code << ", request_id=" << request_id;
+ return {.status = st,
+ .http_code =
static_cast<int>(outcome.GetError().GetResponseCode()),
+ .request_id = outcome.GetError().GetRequestId()};
+ }
+
+ LOG_IF(INFO, elapsed_ms > S3_REQUEST_THRESHOLD_MS)
+ << "CompleteMultipartUpload cost=" << elapsed_ms << "ms"
+ << ", request_id=" << request_id << ", bucket=" << opts.bucket <<
", key=" << opts.key
+ << ", upload_id=" << *opts.upload_id;
+ return ObjectStorageResponse::OK();
+}
+
+ObjectStorageHeadResponse S3ObjStorageBackend::head_object(const
ObjectStoragePathOptions& opts) {
+ Aws::S3::Model::HeadObjectRequest request;
+ request.WithBucket(opts.bucket).WithKey(opts.key);
+
+ auto outcome = SYNC_POINT_HOOK_RETURN_VALUE(
+ [&]() {
+ client_bvar::ScopedLatency
scoped_latency(client_bvar::s3_head_latency);
+ return _client->HeadObject(request);
+ }(),
+ "s3_file_system::head_object", std::ref(request).get());
+
+ if (outcome.IsSuccess()) {
+ return {.resp = ObjectStorageResponse::OK(),
+ .file_size = outcome.GetResult().GetContentLength()};
+ } else if (outcome.GetError().GetResponseCode() ==
Aws::Http::HttpResponseCode::NOT_FOUND) {
+ return {.resp = {.status = TStatusCode::NOT_FOUND}, .file_size = 0};
+ } else {
+ record_s3_request_failed(outcome.GetError());
+ LOG(WARNING) << "failed to head object"
+ << "bucket " << opts.bucket << " key " << opts.key << "
responseCode "
+ << outcome.GetError() << " error " <<
outcome.GetError().GetMessage()
+ << " request_id " << outcome.GetError().GetRequestId();
+ return {.resp = {.status = s3fs_error(
+ outcome.GetError(),
+ fmt::format("failed to head object: {}",
opts.path.native())),
+ .http_code =
static_cast<int>(outcome.GetError().GetResponseCode()),
+ .request_id = outcome.GetError().GetRequestId()},
+ .file_size = -1};
+ }
+}
+
+ObjectStorageResponse S3ObjStorageBackend::get_object(const
ObjectStoragePathOptions& opts,
+ void* buffer, size_t
offset,
+ size_t bytes_read,
size_t* size_return) {
+ Aws::S3::Model::GetObjectRequest request;
+ request.WithBucket(opts.bucket).WithKey(opts.key);
+ request.SetRange(fmt::format("bytes={}-{}", offset, offset + bytes_read -
1));
+ request.SetResponseStreamFactory(AwsWriteableStreamFactory(buffer,
bytes_read));
+
+ auto outcome = [&]() {
+ client_bvar::ScopedLatency scoped_latency(client_bvar::s3_get_latency);
+ return _client->GetObject(request);
+ }();
+ if (!outcome.IsSuccess()) {
+ record_s3_request_failed(outcome.GetError());
+ return ObjectStorageResponse {
+ .status = s3fs_error(outcome.GetError(),
Review Comment:
[P2] Keep object and request context in GetObject errors
Both production callers populate only `bucket` and `key`, so `opts.path` is
empty here. They then rebuild a BE `Status` from only `status.code/msg`,
discarding this response's HTTP code and request ID; the deleted path used
`opts.key` and embedded provider code/type/request ID in the returned error. A
failed read therefore reaches operators without the object identity or request
context needed to diagnose it. Please use the key (or a path/key fallback) and
retain/log the provider classification and request ID; the incomplete-read
branch below should retain its successful request ID as well.
##########
common/cpp/client/s3_obj_storage_backend.cpp:
##########
@@ -0,0 +1,582 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "s3_obj_storage_backend.h"
+
+#include <cpp/client/obj_storage_client.h>
+#include <gen_cpp/Status_types.h>
+
+#include <algorithm>
+#include <chrono>
+
+#include "client_bvar.h"
+#include "cpp/obj_retry_strategy.h"
+
+namespace Aws::S3::Model {
+class DeleteObjectRequest;
+} // namespace Aws::S3::Model
+
+using Aws::S3::Model::CompletedPart;
+using Aws::S3::Model::CompletedMultipartUpload;
+using Aws::S3::Model::CompleteMultipartUploadRequest;
+using Aws::S3::Model::CreateMultipartUploadRequest;
+using Aws::S3::Model::UploadPartRequest;
+using Aws::S3::Model::UploadPartOutcome;
+
+namespace doris {
+using namespace Aws::S3::Model;
+namespace {
+
+constexpr int64_t S3_REQUEST_THRESHOLD_MS = 5000;
+
+int64_t elapsed_time_milliseconds(std::chrono::steady_clock::time_point start)
{
+ return
std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now()
-
+ start)
+ .count();
+}
+
+void record_s3_request_failed(const Aws::S3::S3Error& error) {
+ record_object_request_failed(static_cast<int>(error.GetResponseCode()));
+}
+
+} // namespace
+
+ObjectStorageStatus s3fs_error(const Aws::S3::S3Error& err, std::string_view
msg) {
+ using namespace Aws::Http;
+ switch (err.GetResponseCode()) {
+ case HttpResponseCode::NOT_FOUND:
+ return {TStatusCode::NOT_FOUND,
+ fmt::format("{}: {} {}", msg, err.GetExceptionName(),
err.GetMessage())};
+ case HttpResponseCode::FORBIDDEN:
+ // TODO: no permission and other 4xx errors should be handled
separately
+ return {TStatusCode::NOT_AUTHORIZED,
+ fmt::format("{}: {} {}", msg, err.GetExceptionName(),
err.GetMessage())};
+ case HttpResponseCode::REQUEST_NOT_MADE:
+ return {-1, fmt::format("{}: {} {}", msg, err.GetExceptionName(),
err.GetMessage())};
+ default:
+ return {TStatusCode::INTERNAL_ERROR,
+ fmt::format("{}: {} {}", msg, err.GetExceptionName(),
err.GetMessage())};
+ }
+}
+
+ObjectStorageUploadResponse S3ObjStorageBackend::create_multipart_upload(
+ const ObjectStoragePathOptions& opts) {
+ CreateMultipartUploadRequest request;
+ request.WithBucket(opts.bucket).WithKey(opts.key);
+ request.SetContentType("application/octet-stream");
+
+ const auto start = std::chrono::steady_clock::now();
+ auto outcome = SYNC_POINT_HOOK_RETURN_VALUE(
+ [&]() {
+ client_bvar::ScopedLatency scoped_latency(
+ client_bvar::s3_multi_part_upload_latency);
+ return _client->CreateMultipartUpload(request);
+ }(),
+ "s3_file_writer::create_multi_part_upload",
std::cref(request).get());
+ SYNC_POINT_CALLBACK("s3_file_writer::_open", &outcome);
+ const auto elapsed_ms = elapsed_time_milliseconds(start);
+
+ const auto& request_id = outcome.IsSuccess() ?
outcome.GetResult().GetRequestId()
+ :
outcome.GetError().GetRequestId();
+
+ LOG_IF(INFO, elapsed_ms > S3_REQUEST_THRESHOLD_MS)
+ << "CreateMultipartUpload cost=" << elapsed_ms << "ms"
+ << ", request_id=" << request_id << ", bucket=" << opts.bucket <<
", key=" << opts.key;
+
+ if (!outcome.IsSuccess()) {
+ record_s3_request_failed(outcome.GetError());
+ auto st = s3fs_error(outcome.GetError(), fmt::format("failed to
CreateMultipartUpload: {} ",
+
opts.path.native()));
+ LOG(WARNING) << st.code << " request_id=" << request_id;
+ return ObjectStorageUploadResponse {
+ .resp = {.status = st,
+ .http_code =
static_cast<int>(outcome.GetError().GetResponseCode()),
+ .request_id = outcome.GetError().GetRequestId()},
+ };
+ }
+
+ return ObjectStorageUploadResponse {.resp = ObjectStorageResponse::OK(),
+ .upload_id
{outcome.GetResult().GetUploadId()}};
+}
+
+ObjectStorageResponse S3ObjStorageBackend::put_object(const
ObjectStoragePathOptions& opts,
+ std::string_view stream)
{
+ Aws::S3::Model::PutObjectRequest request;
+ request.WithBucket(opts.bucket).WithKey(opts.key);
+ auto string_view_stream =
std::make_shared<StringViewStream>(stream.data(), stream.size());
+ 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");
+
+ const auto start = std::chrono::steady_clock::now();
+ auto outcome = SYNC_POINT_HOOK_RETURN_VALUE(
+ [&]() {
+ client_bvar::ScopedLatency
scoped_latency(client_bvar::s3_put_latency);
+ return _client->PutObject(request);
+ }(),
+ "s3_file_writer::put_object", std::cref(request).get(), &stream);
+ const auto elapsed_ms = elapsed_time_milliseconds(start);
+
+ const auto& request_id = outcome.IsSuccess() ?
outcome.GetResult().GetRequestId()
+ :
outcome.GetError().GetRequestId();
+
+ if (!outcome.IsSuccess()) {
+ record_s3_request_failed(outcome.GetError());
+ auto st = s3fs_error(outcome.GetError(),
+ fmt::format("failed to put object: {}",
opts.path.native()));
+ LOG(WARNING) << st.code << ", request_id=" << request_id;
+ return ObjectStorageResponse {
+ .status = st,
+ .http_code =
static_cast<int>(outcome.GetError().GetResponseCode()),
+ .request_id = outcome.GetError().GetRequestId()};
+ }
+
+ LOG_IF(INFO, elapsed_ms > S3_REQUEST_THRESHOLD_MS)
+ << "PutObject cost=" << elapsed_ms << "ms"
+ << ", request_id=" << request_id << ", bucket=" << opts.bucket <<
", key=" << opts.key;
+ return ObjectStorageResponse::OK();
+}
+
+ObjectStorageUploadResponse S3ObjStorageBackend::upload_part(const
ObjectStoragePathOptions& opts,
+ std::string_view
stream,
+ int part_num) {
+ UploadPartRequest request;
+ request.WithBucket(opts.bucket)
+ .WithKey(opts.key)
+ .WithPartNumber(part_num)
+ .WithUploadId(*opts.upload_id);
+ auto string_view_stream =
std::make_shared<StringViewStream>(stream.data(), stream.size());
+
+ 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));
+
+ request.SetContentLength(stream.size());
+ request.SetContentType("application/octet-stream");
+
+ const auto start = std::chrono::steady_clock::now();
+ auto outcome = SYNC_POINT_HOOK_RETURN_VALUE(
+ [&]() {
+ client_bvar::ScopedLatency scoped_latency(
+ client_bvar::s3_multi_part_upload_latency);
+
+ return _client->UploadPart(request);
+ }(),
+ "s3_file_writer::upload_part", std::cref(request).get(), &stream);
+ const auto elapsed_ms = elapsed_time_milliseconds(start);
+
+ const auto& request_id = outcome.IsSuccess() ?
outcome.GetResult().GetRequestId()
+ :
outcome.GetError().GetRequestId();
+
+ TEST_SYNC_POINT_CALLBACK("S3FileWriter::_upload_one_part", &outcome);
+ if (!outcome.IsSuccess()) {
+ record_s3_request_failed(outcome.GetError());
+ auto st = s3fs_error(outcome.GetError(),
+ fmt::format("failed to UploadPart: {}, part_num
{}, upload_id={}",
+ opts.path.native(), part_num,
*opts.upload_id));
+
+ LOG(WARNING) << st.code << ", request_id=" << request_id;
+ return ObjectStorageUploadResponse {
+ .resp = {.status = st,
+ .http_code =
static_cast<int>(outcome.GetError().GetResponseCode()),
+ .request_id = outcome.GetError().GetRequestId()}};
+ }
+ LOG_IF(INFO, elapsed_ms > S3_REQUEST_THRESHOLD_MS)
+ << "UploadPart cost=" << elapsed_ms << "ms"
+ << ", request_id=" << request_id << ", bucket=" << opts.bucket <<
", key=" << opts.key
+ << ", part_num=" << part_num << ", upload_id=" << *opts.upload_id;
+ return ObjectStorageUploadResponse {.resp = ObjectStorageResponse::OK(),
+ .etag = outcome.GetResult().GetETag()};
+}
+
+ObjectStorageResponse S3ObjStorageBackend::complete_multipart_upload(
+ const ObjectStoragePathOptions& opts,
+ const std::vector<ObjectCompleteMultiPart>& completed_parts) {
+ CompleteMultipartUploadRequest request;
+
request.WithBucket(opts.bucket).WithKey(opts.key).WithUploadId(*opts.upload_id);
+
+ CompletedMultipartUpload completed_upload;
+ std::vector<CompletedPart> complete_parts;
+ std::ranges::transform(completed_parts, std::back_inserter(complete_parts),
+ [](const ObjectCompleteMultiPart& part_ptr) {
+ CompletedPart part;
+ part.SetPartNumber(part_ptr.part_num);
+ part.SetETag(part_ptr.etag);
+ return part;
+ });
+ completed_upload.SetParts(std::move(complete_parts));
+ request.WithMultipartUpload(completed_upload);
+
+ TEST_SYNC_POINT_RETURN_WITH_VALUE("S3FileWriter::_complete:3",
ObjectStorageResponse(), this);
+
+ const auto start = std::chrono::steady_clock::now();
+ auto outcome = SYNC_POINT_HOOK_RETURN_VALUE(
+ [&]() {
+ client_bvar::ScopedLatency scoped_latency(
+ client_bvar::s3_multi_part_upload_latency);
+ return _client->CompleteMultipartUpload(request);
+ }(),
+ "s3_file_writer::complete_multi_part", std::cref(request).get());
+ const auto elapsed_ms = elapsed_time_milliseconds(start);
+
+ const auto& request_id = outcome.IsSuccess() ?
outcome.GetResult().GetRequestId()
+ :
outcome.GetError().GetRequestId();
+
+ if (!outcome.IsSuccess()) {
+ record_s3_request_failed(outcome.GetError());
+ auto st = s3fs_error(outcome.GetError(),
+ fmt::format("failed to CompleteMultipartUpload:
{}, upload_id={}",
+ opts.path.native(), *opts.upload_id));
+ LOG(WARNING) << st.code << ", request_id=" << request_id;
+ return {.status = st,
+ .http_code =
static_cast<int>(outcome.GetError().GetResponseCode()),
+ .request_id = outcome.GetError().GetRequestId()};
+ }
+
+ LOG_IF(INFO, elapsed_ms > S3_REQUEST_THRESHOLD_MS)
+ << "CompleteMultipartUpload cost=" << elapsed_ms << "ms"
+ << ", request_id=" << request_id << ", bucket=" << opts.bucket <<
", key=" << opts.key
+ << ", upload_id=" << *opts.upload_id;
+ return ObjectStorageResponse::OK();
+}
+
+ObjectStorageHeadResponse S3ObjStorageBackend::head_object(const
ObjectStoragePathOptions& opts) {
+ Aws::S3::Model::HeadObjectRequest request;
+ request.WithBucket(opts.bucket).WithKey(opts.key);
+
+ auto outcome = SYNC_POINT_HOOK_RETURN_VALUE(
+ [&]() {
+ client_bvar::ScopedLatency
scoped_latency(client_bvar::s3_head_latency);
+ return _client->HeadObject(request);
+ }(),
+ "s3_file_system::head_object", std::ref(request).get());
+
+ if (outcome.IsSuccess()) {
+ return {.resp = ObjectStorageResponse::OK(),
+ .file_size = outcome.GetResult().GetContentLength()};
+ } else if (outcome.GetError().GetResponseCode() ==
Aws::Http::HttpResponseCode::NOT_FOUND) {
+ return {.resp = {.status = TStatusCode::NOT_FOUND}, .file_size = 0};
+ } else {
+ record_s3_request_failed(outcome.GetError());
+ LOG(WARNING) << "failed to head object"
+ << "bucket " << opts.bucket << " key " << opts.key << "
responseCode "
+ << outcome.GetError() << " error " <<
outcome.GetError().GetMessage()
+ << " request_id " << outcome.GetError().GetRequestId();
+ return {.resp = {.status = s3fs_error(
+ outcome.GetError(),
+ fmt::format("failed to head object: {}",
opts.path.native())),
+ .http_code =
static_cast<int>(outcome.GetError().GetResponseCode()),
+ .request_id = outcome.GetError().GetRequestId()},
+ .file_size = -1};
+ }
+}
+
+ObjectStorageResponse S3ObjStorageBackend::get_object(const
ObjectStoragePathOptions& opts,
+ void* buffer, size_t
offset,
+ size_t bytes_read,
size_t* size_return) {
+ Aws::S3::Model::GetObjectRequest request;
+ request.WithBucket(opts.bucket).WithKey(opts.key);
+ request.SetRange(fmt::format("bytes={}-{}", offset, offset + bytes_read -
1));
+ request.SetResponseStreamFactory(AwsWriteableStreamFactory(buffer,
bytes_read));
+
+ auto outcome = [&]() {
+ client_bvar::ScopedLatency scoped_latency(client_bvar::s3_get_latency);
+ return _client->GetObject(request);
+ }();
+ if (!outcome.IsSuccess()) {
+ record_s3_request_failed(outcome.GetError());
+ return ObjectStorageResponse {
+ .status = s3fs_error(outcome.GetError(),
+ fmt::format("failed to get object: {}",
opts.path.native())),
+ .http_code =
static_cast<int>(outcome.GetError().GetResponseCode()),
+ .request_id = outcome.GetError().GetRequestId(),
+ };
+ }
+ *size_return = outcome.GetResult().GetContentLength();
+ SYNC_POINT_CALLBACK("s3_obj_storage_client::get_object", size_return);
+ if (*size_return != bytes_read) {
+ return ObjectStorageResponse {
+ .status = {TStatusCode::INTERNAL_ERROR,
+ fmt::format("incomplete read from {}, expect {},
got {}",
+ opts.path.native(), bytes_read,
*size_return)}};
+ }
+ return ObjectStorageResponse::OK();
+}
+
+ObjectStorageListPage S3ObjStorageBackend::list_objects(const
ObjectStoragePathOptions& opts,
+ std::string_view
continuation_token) {
+ const auto& prefix = opts.prefix.empty() ? opts.key : opts.prefix;
+ Aws::S3::Model::ListObjectsV2Request request;
+
request.WithBucket(opts.bucket).WithPrefix(prefix).WithMaxKeys(OBJECT_LIST_PAGE_SIZE);
+ if (!continuation_token.empty()) {
+ request.SetContinuationToken(std::string(continuation_token));
+ }
+ TEST_SYNC_POINT_CALLBACK("S3ObjStorageBackend::list_objects", &request);
+
+ auto outcome = [&]() {
+ client_bvar::ScopedLatency
scoped_latency(client_bvar::s3_list_latency);
+ return _client->ListObjectsV2(request);
+ }();
+
+ const auto& request_id = outcome.IsSuccess() ?
outcome.GetResult().GetRequestId()
+ :
outcome.GetError().GetRequestId();
+ if (!outcome.IsSuccess()) {
+ // Some S3-compatible providers (for example TOS) return NoSuchKey
instead of an empty page
+ // when a prefix does not exist.
+ if (outcome.GetError().GetErrorType() ==
Aws::S3::S3Errors::NO_SUCH_KEY) {
+ LOG(INFO) << fmt::format(
+ "NoSuchKey when listing objects, treat as empty response,
endpoint: {}, "
+ "bucket: {}, prefix: {}, request_id: {}",
+ _config.endpoint, request.GetBucket(),
request.GetPrefix(), request_id);
+ return {.resp = ObjectStorageResponse::OK()};
+ }
+
record_object_request_failed(static_cast<int>(outcome.GetError().GetResponseCode()));
+ const auto status = s3fs_error(outcome.GetError(),
+ fmt::format("failed to list objects:
{}, prefix: {}",
+ request.GetBucket(),
request.GetPrefix()));
+ LOG(WARNING) << fmt::format(
+ "failed to list objects, endpoint: {}, bucket: {}, prefix: {},
responseCode: {}, "
+ "error: {}, request_id: {}",
+ _config.endpoint, request.GetBucket(), request.GetPrefix(),
+ static_cast<int>(outcome.GetError().GetResponseCode()),
+ outcome.GetError().GetMessage(), request_id);
+ return {
+ .resp = {.status = status,
+ .http_code =
static_cast<int>(outcome.GetError().GetResponseCode()),
+ .request_id = request_id},
+ };
+ }
+
+ const auto& result = outcome.GetResult();
+ if (result.GetIsTruncated() && result.GetNextContinuationToken().empty()) {
+ LOG(WARNING) << fmt::format(
+ "failed to list objects, isTruncated but no continuation
token, endpoint: {}, "
+ "bucket: {}, prefix: {}, request_id: {}",
+ _config.endpoint, request.GetBucket(), request.GetPrefix(),
request_id);
+ return {
+ .resp = {.status = {TStatusCode::INTERNAL_ERROR,
+ fmt::format("failed to list objects: {},
prefix: {}",
+ request.GetBucket(),
request.GetPrefix())},
+ .http_code = 0,
+ .request_id = request_id},
+ };
+ }
+
+ ObjectStorageListPage page {
+ .resp = ObjectStorageResponse::OK(),
+ .continuation_token = result.GetNextContinuationToken(),
+ .has_more = result.GetIsTruncated(),
+ };
+ const auto& content = result.GetContents();
+ page.objects.reserve(content.size());
+ for (const auto& obj : content) {
+ DCHECK(obj.GetKey().starts_with(request.GetPrefix()))
+ << obj.GetKey() << ' ' << request.GetPrefix();
+ page.objects.emplace_back(ObjectMeta {.file_path = obj.GetKey(),
+ .size = obj.GetSize(),
+ .mtime_s =
obj.GetLastModified().Seconds()});
+ }
+ return page;
+}
+
+ObjectStorageResponse S3ObjStorageBackend::delete_objects(const
ObjectStoragePathOptions& opts,
+
std::vector<std::string> objs) {
+ size_t max_delete_batch = 1000;
+ TEST_SYNC_POINT_CALLBACK("S3ObjClient::delete_objects", &max_delete_batch);
+ TEST_SYNC_POINT_CALLBACK("S3ObjStorageClient::delete_objects",
&max_delete_batch);
+ max_delete_batch = std::max<size_t>(1, max_delete_batch);
+ for (size_t begin = 0; begin < objs.size(); begin += max_delete_batch) {
+ const size_t end = std::min(begin + max_delete_batch, objs.size());
+ if (end - begin == 1) {
+ auto single_opts = opts;
+ single_opts.key = std::move(objs[begin]);
+ auto resp = delete_object(single_opts);
+ if (!resp.ok()) {
+ return resp;
+ }
+ continue;
+ }
+
+ Aws::S3::Model::DeleteObjectsRequest delete_request;
+ delete_request.SetBucket(opts.bucket);
+ Aws::S3::Model::Delete del;
+ Aws::Vector<Aws::S3::Model::ObjectIdentifier> objects;
+ objects.reserve(end - begin);
+ for (size_t i = begin; i < end; ++i) {
+ Aws::S3::Model::ObjectIdentifier object;
+ object.SetKey(std::move(objs[i]));
+ objects.emplace_back(std::move(object));
+ }
+ del.WithObjects(std::move(objects)).SetQuiet(true);
+ delete_request.SetDelete(std::move(del));
+
+ auto delete_outcome = [&]() {
+ client_bvar::ScopedLatency
scoped_latency(client_bvar::s3_delete_objects_latency);
+ return _client->DeleteObjects(delete_request);
+ }();
+ SYNC_POINT_CALLBACK("s3_obj_storage_client::delete_objects",
&delete_outcome);
+
SYNC_POINT_CALLBACK("s3_obj_storage_client::delete_objects_recursively",
&delete_outcome);
+ if (!delete_outcome.IsSuccess()) {
+ record_s3_request_failed(delete_outcome.GetError());
+ return ObjectStorageResponse {
Review Comment:
[P2] Preserve S3 delete failure diagnostics for Recycler
This return carries the provider message, HTTP code, and request ID, but the
Recycler adapters for prefix, batch, and single deletion immediately collapse
`ObjectStorageResponse` to `status.code`; recursive execution can additionally
replace it with a generic cancellation error. The operation URI is pre-logged,
but because this backend no longer logs failed responses, production recycling
loses the provider error, HTTP code, and request ID needed to correlate that
URI with the failed request. Please log those response fields before returning
(or preserve/log the full response at the adapter boundary); this is separate
from the already-fixed noisy success logging.
##########
common/cpp/client/s3_obj_storage_backend.cpp:
##########
@@ -0,0 +1,582 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "s3_obj_storage_backend.h"
+
+#include <cpp/client/obj_storage_client.h>
+#include <gen_cpp/Status_types.h>
+
+#include <algorithm>
+#include <chrono>
+
+#include "client_bvar.h"
+#include "cpp/obj_retry_strategy.h"
+
+namespace Aws::S3::Model {
+class DeleteObjectRequest;
+} // namespace Aws::S3::Model
+
+using Aws::S3::Model::CompletedPart;
+using Aws::S3::Model::CompletedMultipartUpload;
+using Aws::S3::Model::CompleteMultipartUploadRequest;
+using Aws::S3::Model::CreateMultipartUploadRequest;
+using Aws::S3::Model::UploadPartRequest;
+using Aws::S3::Model::UploadPartOutcome;
+
+namespace doris {
+using namespace Aws::S3::Model;
+namespace {
+
+constexpr int64_t S3_REQUEST_THRESHOLD_MS = 5000;
+
+int64_t elapsed_time_milliseconds(std::chrono::steady_clock::time_point start)
{
+ return
std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now()
-
+ start)
+ .count();
+}
+
+void record_s3_request_failed(const Aws::S3::S3Error& error) {
+ record_object_request_failed(static_cast<int>(error.GetResponseCode()));
+}
+
+} // namespace
+
+ObjectStorageStatus s3fs_error(const Aws::S3::S3Error& err, std::string_view
msg) {
+ using namespace Aws::Http;
+ switch (err.GetResponseCode()) {
+ case HttpResponseCode::NOT_FOUND:
+ return {TStatusCode::NOT_FOUND,
+ fmt::format("{}: {} {}", msg, err.GetExceptionName(),
err.GetMessage())};
+ case HttpResponseCode::FORBIDDEN:
+ // TODO: no permission and other 4xx errors should be handled
separately
+ return {TStatusCode::NOT_AUTHORIZED,
+ fmt::format("{}: {} {}", msg, err.GetExceptionName(),
err.GetMessage())};
+ case HttpResponseCode::REQUEST_NOT_MADE:
+ return {-1, fmt::format("{}: {} {}", msg, err.GetExceptionName(),
err.GetMessage())};
+ default:
+ return {TStatusCode::INTERNAL_ERROR,
+ fmt::format("{}: {} {}", msg, err.GetExceptionName(),
err.GetMessage())};
+ }
+}
+
+ObjectStorageUploadResponse S3ObjStorageBackend::create_multipart_upload(
+ const ObjectStoragePathOptions& opts) {
+ CreateMultipartUploadRequest request;
+ request.WithBucket(opts.bucket).WithKey(opts.key);
+ request.SetContentType("application/octet-stream");
+
+ const auto start = std::chrono::steady_clock::now();
+ auto outcome = SYNC_POINT_HOOK_RETURN_VALUE(
+ [&]() {
+ client_bvar::ScopedLatency scoped_latency(
+ client_bvar::s3_multi_part_upload_latency);
+ return _client->CreateMultipartUpload(request);
+ }(),
+ "s3_file_writer::create_multi_part_upload",
std::cref(request).get());
+ SYNC_POINT_CALLBACK("s3_file_writer::_open", &outcome);
+ const auto elapsed_ms = elapsed_time_milliseconds(start);
+
+ const auto& request_id = outcome.IsSuccess() ?
outcome.GetResult().GetRequestId()
+ :
outcome.GetError().GetRequestId();
+
+ LOG_IF(INFO, elapsed_ms > S3_REQUEST_THRESHOLD_MS)
+ << "CreateMultipartUpload cost=" << elapsed_ms << "ms"
+ << ", request_id=" << request_id << ", bucket=" << opts.bucket <<
", key=" << opts.key;
+
+ if (!outcome.IsSuccess()) {
+ record_s3_request_failed(outcome.GetError());
+ auto st = s3fs_error(outcome.GetError(), fmt::format("failed to
CreateMultipartUpload: {} ",
+
opts.path.native()));
+ LOG(WARNING) << st.code << " request_id=" << request_id;
+ return ObjectStorageUploadResponse {
+ .resp = {.status = st,
+ .http_code =
static_cast<int>(outcome.GetError().GetResponseCode()),
+ .request_id = outcome.GetError().GetRequestId()},
+ };
+ }
+
+ return ObjectStorageUploadResponse {.resp = ObjectStorageResponse::OK(),
+ .upload_id
{outcome.GetResult().GetUploadId()}};
+}
+
+ObjectStorageResponse S3ObjStorageBackend::put_object(const
ObjectStoragePathOptions& opts,
+ std::string_view stream)
{
+ Aws::S3::Model::PutObjectRequest request;
+ request.WithBucket(opts.bucket).WithKey(opts.key);
+ auto string_view_stream =
std::make_shared<StringViewStream>(stream.data(), stream.size());
+ 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");
+
+ const auto start = std::chrono::steady_clock::now();
+ auto outcome = SYNC_POINT_HOOK_RETURN_VALUE(
+ [&]() {
+ client_bvar::ScopedLatency
scoped_latency(client_bvar::s3_put_latency);
+ return _client->PutObject(request);
+ }(),
+ "s3_file_writer::put_object", std::cref(request).get(), &stream);
+ const auto elapsed_ms = elapsed_time_milliseconds(start);
+
+ const auto& request_id = outcome.IsSuccess() ?
outcome.GetResult().GetRequestId()
+ :
outcome.GetError().GetRequestId();
+
+ if (!outcome.IsSuccess()) {
+ record_s3_request_failed(outcome.GetError());
+ auto st = s3fs_error(outcome.GetError(),
+ fmt::format("failed to put object: {}",
opts.path.native()));
+ LOG(WARNING) << st.code << ", request_id=" << request_id;
+ return ObjectStorageResponse {
+ .status = st,
+ .http_code =
static_cast<int>(outcome.GetError().GetResponseCode()),
+ .request_id = outcome.GetError().GetRequestId()};
+ }
+
+ LOG_IF(INFO, elapsed_ms > S3_REQUEST_THRESHOLD_MS)
+ << "PutObject cost=" << elapsed_ms << "ms"
+ << ", request_id=" << request_id << ", bucket=" << opts.bucket <<
", key=" << opts.key;
+ return ObjectStorageResponse::OK();
+}
+
+ObjectStorageUploadResponse S3ObjStorageBackend::upload_part(const
ObjectStoragePathOptions& opts,
+ std::string_view
stream,
+ int part_num) {
+ UploadPartRequest request;
+ request.WithBucket(opts.bucket)
+ .WithKey(opts.key)
+ .WithPartNumber(part_num)
+ .WithUploadId(*opts.upload_id);
+ auto string_view_stream =
std::make_shared<StringViewStream>(stream.data(), stream.size());
+
+ 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));
+
+ request.SetContentLength(stream.size());
+ request.SetContentType("application/octet-stream");
+
+ const auto start = std::chrono::steady_clock::now();
+ auto outcome = SYNC_POINT_HOOK_RETURN_VALUE(
+ [&]() {
+ client_bvar::ScopedLatency scoped_latency(
+ client_bvar::s3_multi_part_upload_latency);
+
+ return _client->UploadPart(request);
+ }(),
+ "s3_file_writer::upload_part", std::cref(request).get(), &stream);
+ const auto elapsed_ms = elapsed_time_milliseconds(start);
+
+ const auto& request_id = outcome.IsSuccess() ?
outcome.GetResult().GetRequestId()
+ :
outcome.GetError().GetRequestId();
+
+ TEST_SYNC_POINT_CALLBACK("S3FileWriter::_upload_one_part", &outcome);
+ if (!outcome.IsSuccess()) {
+ record_s3_request_failed(outcome.GetError());
+ auto st = s3fs_error(outcome.GetError(),
+ fmt::format("failed to UploadPart: {}, part_num
{}, upload_id={}",
+ opts.path.native(), part_num,
*opts.upload_id));
+
+ LOG(WARNING) << st.code << ", request_id=" << request_id;
+ return ObjectStorageUploadResponse {
+ .resp = {.status = st,
+ .http_code =
static_cast<int>(outcome.GetError().GetResponseCode()),
+ .request_id = outcome.GetError().GetRequestId()}};
+ }
+ LOG_IF(INFO, elapsed_ms > S3_REQUEST_THRESHOLD_MS)
+ << "UploadPart cost=" << elapsed_ms << "ms"
+ << ", request_id=" << request_id << ", bucket=" << opts.bucket <<
", key=" << opts.key
+ << ", part_num=" << part_num << ", upload_id=" << *opts.upload_id;
+ return ObjectStorageUploadResponse {.resp = ObjectStorageResponse::OK(),
+ .etag = outcome.GetResult().GetETag()};
+}
+
+ObjectStorageResponse S3ObjStorageBackend::complete_multipart_upload(
+ const ObjectStoragePathOptions& opts,
+ const std::vector<ObjectCompleteMultiPart>& completed_parts) {
+ CompleteMultipartUploadRequest request;
+
request.WithBucket(opts.bucket).WithKey(opts.key).WithUploadId(*opts.upload_id);
+
+ CompletedMultipartUpload completed_upload;
+ std::vector<CompletedPart> complete_parts;
+ std::ranges::transform(completed_parts, std::back_inserter(complete_parts),
+ [](const ObjectCompleteMultiPart& part_ptr) {
+ CompletedPart part;
+ part.SetPartNumber(part_ptr.part_num);
+ part.SetETag(part_ptr.etag);
+ return part;
+ });
+ completed_upload.SetParts(std::move(complete_parts));
+ request.WithMultipartUpload(completed_upload);
+
+ TEST_SYNC_POINT_RETURN_WITH_VALUE("S3FileWriter::_complete:3",
ObjectStorageResponse(), this);
+
+ const auto start = std::chrono::steady_clock::now();
+ auto outcome = SYNC_POINT_HOOK_RETURN_VALUE(
+ [&]() {
+ client_bvar::ScopedLatency scoped_latency(
+ client_bvar::s3_multi_part_upload_latency);
+ return _client->CompleteMultipartUpload(request);
+ }(),
+ "s3_file_writer::complete_multi_part", std::cref(request).get());
+ const auto elapsed_ms = elapsed_time_milliseconds(start);
+
+ const auto& request_id = outcome.IsSuccess() ?
outcome.GetResult().GetRequestId()
+ :
outcome.GetError().GetRequestId();
+
+ if (!outcome.IsSuccess()) {
+ record_s3_request_failed(outcome.GetError());
+ auto st = s3fs_error(outcome.GetError(),
+ fmt::format("failed to CompleteMultipartUpload:
{}, upload_id={}",
+ opts.path.native(), *opts.upload_id));
+ LOG(WARNING) << st.code << ", request_id=" << request_id;
+ return {.status = st,
+ .http_code =
static_cast<int>(outcome.GetError().GetResponseCode()),
+ .request_id = outcome.GetError().GetRequestId()};
+ }
+
+ LOG_IF(INFO, elapsed_ms > S3_REQUEST_THRESHOLD_MS)
+ << "CompleteMultipartUpload cost=" << elapsed_ms << "ms"
+ << ", request_id=" << request_id << ", bucket=" << opts.bucket <<
", key=" << opts.key
+ << ", upload_id=" << *opts.upload_id;
+ return ObjectStorageResponse::OK();
+}
+
+ObjectStorageHeadResponse S3ObjStorageBackend::head_object(const
ObjectStoragePathOptions& opts) {
+ Aws::S3::Model::HeadObjectRequest request;
+ request.WithBucket(opts.bucket).WithKey(opts.key);
+
+ auto outcome = SYNC_POINT_HOOK_RETURN_VALUE(
+ [&]() {
+ client_bvar::ScopedLatency
scoped_latency(client_bvar::s3_head_latency);
+ return _client->HeadObject(request);
+ }(),
+ "s3_file_system::head_object", std::ref(request).get());
+
+ if (outcome.IsSuccess()) {
+ return {.resp = ObjectStorageResponse::OK(),
+ .file_size = outcome.GetResult().GetContentLength()};
+ } else if (outcome.GetError().GetResponseCode() ==
Aws::Http::HttpResponseCode::NOT_FOUND) {
+ return {.resp = {.status = TStatusCode::NOT_FOUND}, .file_size = 0};
+ } else {
+ record_s3_request_failed(outcome.GetError());
+ LOG(WARNING) << "failed to head object"
+ << "bucket " << opts.bucket << " key " << opts.key << "
responseCode "
+ << outcome.GetError() << " error " <<
outcome.GetError().GetMessage()
+ << " request_id " << outcome.GetError().GetRequestId();
+ return {.resp = {.status = s3fs_error(
+ outcome.GetError(),
+ fmt::format("failed to head object: {}",
opts.path.native())),
+ .http_code =
static_cast<int>(outcome.GetError().GetResponseCode()),
+ .request_id = outcome.GetError().GetRequestId()},
+ .file_size = -1};
+ }
+}
+
+ObjectStorageResponse S3ObjStorageBackend::get_object(const
ObjectStoragePathOptions& opts,
+ void* buffer, size_t
offset,
+ size_t bytes_read,
size_t* size_return) {
+ Aws::S3::Model::GetObjectRequest request;
+ request.WithBucket(opts.bucket).WithKey(opts.key);
+ request.SetRange(fmt::format("bytes={}-{}", offset, offset + bytes_read -
1));
+ request.SetResponseStreamFactory(AwsWriteableStreamFactory(buffer,
bytes_read));
+
+ auto outcome = [&]() {
+ client_bvar::ScopedLatency scoped_latency(client_bvar::s3_get_latency);
+ return _client->GetObject(request);
+ }();
+ if (!outcome.IsSuccess()) {
+ record_s3_request_failed(outcome.GetError());
+ return ObjectStorageResponse {
+ .status = s3fs_error(outcome.GetError(),
+ fmt::format("failed to get object: {}",
opts.path.native())),
+ .http_code =
static_cast<int>(outcome.GetError().GetResponseCode()),
+ .request_id = outcome.GetError().GetRequestId(),
+ };
+ }
+ *size_return = outcome.GetResult().GetContentLength();
+ SYNC_POINT_CALLBACK("s3_obj_storage_client::get_object", size_return);
+ if (*size_return != bytes_read) {
+ return ObjectStorageResponse {
+ .status = {TStatusCode::INTERNAL_ERROR,
+ fmt::format("incomplete read from {}, expect {},
got {}",
+ opts.path.native(), bytes_read,
*size_return)}};
+ }
+ return ObjectStorageResponse::OK();
+}
+
+ObjectStorageListPage S3ObjStorageBackend::list_objects(const
ObjectStoragePathOptions& opts,
+ std::string_view
continuation_token) {
+ const auto& prefix = opts.prefix.empty() ? opts.key : opts.prefix;
+ Aws::S3::Model::ListObjectsV2Request request;
+
request.WithBucket(opts.bucket).WithPrefix(prefix).WithMaxKeys(OBJECT_LIST_PAGE_SIZE);
+ if (!continuation_token.empty()) {
+ request.SetContinuationToken(std::string(continuation_token));
+ }
+ TEST_SYNC_POINT_CALLBACK("S3ObjStorageBackend::list_objects", &request);
+
+ auto outcome = [&]() {
+ client_bvar::ScopedLatency
scoped_latency(client_bvar::s3_list_latency);
+ return _client->ListObjectsV2(request);
+ }();
+
+ const auto& request_id = outcome.IsSuccess() ?
outcome.GetResult().GetRequestId()
+ :
outcome.GetError().GetRequestId();
+ if (!outcome.IsSuccess()) {
+ // Some S3-compatible providers (for example TOS) return NoSuchKey
instead of an empty page
+ // when a prefix does not exist.
+ if (outcome.GetError().GetErrorType() ==
Aws::S3::S3Errors::NO_SUCH_KEY) {
+ LOG(INFO) << fmt::format(
+ "NoSuchKey when listing objects, treat as empty response,
endpoint: {}, "
+ "bucket: {}, prefix: {}, request_id: {}",
+ _config.endpoint, request.GetBucket(),
request.GetPrefix(), request_id);
+ return {.resp = ObjectStorageResponse::OK()};
+ }
+
record_object_request_failed(static_cast<int>(outcome.GetError().GetResponseCode()));
+ const auto status = s3fs_error(outcome.GetError(),
+ fmt::format("failed to list objects:
{}, prefix: {}",
+ request.GetBucket(),
request.GetPrefix()));
+ LOG(WARNING) << fmt::format(
+ "failed to list objects, endpoint: {}, bucket: {}, prefix: {},
responseCode: {}, "
+ "error: {}, request_id: {}",
+ _config.endpoint, request.GetBucket(), request.GetPrefix(),
+ static_cast<int>(outcome.GetError().GetResponseCode()),
+ outcome.GetError().GetMessage(), request_id);
+ return {
+ .resp = {.status = status,
+ .http_code =
static_cast<int>(outcome.GetError().GetResponseCode()),
+ .request_id = request_id},
+ };
+ }
+
+ const auto& result = outcome.GetResult();
+ if (result.GetIsTruncated() && result.GetNextContinuationToken().empty()) {
+ LOG(WARNING) << fmt::format(
+ "failed to list objects, isTruncated but no continuation
token, endpoint: {}, "
+ "bucket: {}, prefix: {}, request_id: {}",
+ _config.endpoint, request.GetBucket(), request.GetPrefix(),
request_id);
+ return {
+ .resp = {.status = {TStatusCode::INTERNAL_ERROR,
+ fmt::format("failed to list objects: {},
prefix: {}",
+ request.GetBucket(),
request.GetPrefix())},
+ .http_code = 0,
+ .request_id = request_id},
+ };
+ }
+
+ ObjectStorageListPage page {
+ .resp = ObjectStorageResponse::OK(),
+ .continuation_token = result.GetNextContinuationToken(),
+ .has_more = result.GetIsTruncated(),
+ };
+ const auto& content = result.GetContents();
+ page.objects.reserve(content.size());
+ for (const auto& obj : content) {
+ DCHECK(obj.GetKey().starts_with(request.GetPrefix()))
+ << obj.GetKey() << ' ' << request.GetPrefix();
+ page.objects.emplace_back(ObjectMeta {.file_path = obj.GetKey(),
+ .size = obj.GetSize(),
+ .mtime_s =
obj.GetLastModified().Seconds()});
+ }
+ return page;
+}
+
+ObjectStorageResponse S3ObjStorageBackend::delete_objects(const
ObjectStoragePathOptions& opts,
+
std::vector<std::string> objs) {
+ size_t max_delete_batch = 1000;
+ TEST_SYNC_POINT_CALLBACK("S3ObjClient::delete_objects", &max_delete_batch);
+ TEST_SYNC_POINT_CALLBACK("S3ObjStorageClient::delete_objects",
&max_delete_batch);
+ max_delete_batch = std::max<size_t>(1, max_delete_batch);
+ for (size_t begin = 0; begin < objs.size(); begin += max_delete_batch) {
+ const size_t end = std::min(begin + max_delete_batch, objs.size());
+ if (end - begin == 1) {
+ auto single_opts = opts;
+ single_opts.key = std::move(objs[begin]);
+ auto resp = delete_object(single_opts);
+ if (!resp.ok()) {
+ return resp;
+ }
+ continue;
+ }
+
+ Aws::S3::Model::DeleteObjectsRequest delete_request;
+ delete_request.SetBucket(opts.bucket);
+ Aws::S3::Model::Delete del;
+ Aws::Vector<Aws::S3::Model::ObjectIdentifier> objects;
+ objects.reserve(end - begin);
+ for (size_t i = begin; i < end; ++i) {
+ Aws::S3::Model::ObjectIdentifier object;
+ object.SetKey(std::move(objs[i]));
+ objects.emplace_back(std::move(object));
+ }
+ del.WithObjects(std::move(objects)).SetQuiet(true);
+ delete_request.SetDelete(std::move(del));
+
+ auto delete_outcome = [&]() {
+ client_bvar::ScopedLatency
scoped_latency(client_bvar::s3_delete_objects_latency);
+ return _client->DeleteObjects(delete_request);
+ }();
+ SYNC_POINT_CALLBACK("s3_obj_storage_client::delete_objects",
&delete_outcome);
+
SYNC_POINT_CALLBACK("s3_obj_storage_client::delete_objects_recursively",
&delete_outcome);
+ if (!delete_outcome.IsSuccess()) {
+ record_s3_request_failed(delete_outcome.GetError());
+ return ObjectStorageResponse {
+ .status = s3fs_error(delete_outcome.GetError(),
+ fmt::format("failed to delete dir
{}", opts.key)),
+ .http_code =
static_cast<int>(delete_outcome.GetError().GetResponseCode()),
+ .request_id = delete_outcome.GetError().GetRequestId()};
+ }
+ if (!delete_outcome.GetResult().GetErrors().empty()) {
+ const auto& error = delete_outcome.GetResult().GetErrors().front();
+ return ObjectStorageResponse {
+ .status = {TStatusCode::INTERNAL_ERROR,
+ fmt::format("failed to delete object {}: {},
request_id={}",
+ error.GetKey(), error.GetMessage(),
+
delete_outcome.GetResult().GetRequestId())}};
+ }
+ }
+ return ObjectStorageResponse::OK();
+}
+
+ObjectStorageResponse S3ObjStorageBackend::delete_object(const
ObjectStoragePathOptions& opts) {
+ Aws::S3::Model::DeleteObjectRequest request;
+ request.WithBucket(opts.bucket).WithKey(opts.key);
+
+ auto outcome = [&]() {
+ client_bvar::ScopedLatency
scoped_latency(client_bvar::s3_delete_object_latency);
+
+ return _client->DeleteObject(request);
+ }();
+ TEST_SYNC_POINT_CALLBACK("S3ObjClient::delete_object", &outcome);
+ TEST_SYNC_POINT_CALLBACK("S3ObjStorageClient::delete_object", &outcome);
+ if (outcome.IsSuccess() ||
+ outcome.GetError().GetResponseCode() ==
Aws::Http::HttpResponseCode::NOT_FOUND) {
+ return ObjectStorageResponse::OK();
+ }
+ record_s3_request_failed(outcome.GetError());
+ return ObjectStorageResponse {
+ .status = {TStatusCode::INTERNAL_ERROR,
Review Comment:
[P2] Preserve the permission status for single-object deletes
For a 403 `DeleteObject` response this branch returns `INTERNAL_ERROR`,
while the new multi-object failure path uses `s3fs_error` and the deleted BE
implementation also used its permission-aware mapper.
`S3FileSystem::delete_file_impl` forwards this code directly, and a one-key
`delete_objects` batch delegates here, so credentials without delete permission
now surface as a generic internal failure and classification changes with batch
cardinality. Please return `s3fs_error(outcome.GetError(), ...)` here while
retaining the HTTP code and request ID.
##########
common/cpp/client/auth/aws_credential_factory.cpp:
##########
@@ -0,0 +1,109 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "aws_credential_factory.h"
+
+#include <aws/core/auth/AWSCredentials.h>
+#include <aws/core/auth/AWSCredentialsProvider.h>
+#include <aws/core/auth/AWSCredentialsProviderChain.h>
+#include <aws/core/auth/STSCredentialsProvider.h>
+#include <aws/core/platform/Environment.h>
+#include <aws/identity-management/auth/STSAssumeRoleCredentialsProvider.h>
+#include <aws/sts/STSClient.h>
+
+#include "cpp/custom_aws_credentials_provider_chain.h"
+
+namespace doris {
+namespace {
+
+using Provider = Aws::Auth::AWSCredentialsProvider;
+
+std::shared_ptr<Provider> create_v2_base_provider(CredProviderType type) {
+ switch (type) {
+ case CredProviderType::Env:
+ return
std::make_shared<Aws::Auth::EnvironmentAWSCredentialsProvider>();
+ case CredProviderType::SystemProperties:
+ return
std::make_shared<Aws::Auth::ProfileConfigFileAWSCredentialsProvider>();
+ case CredProviderType::WebIdentity:
+ return
std::make_shared<Aws::Auth::STSAssumeRoleWebIdentityCredentialsProvider>();
+ case CredProviderType::Container:
+ return std::make_shared<Aws::Auth::TaskRoleCredentialsProvider>(
+
Aws::Environment::GetEnv("AWS_CONTAINER_CREDENTIALS_RELATIVE_URI").c_str());
+ case CredProviderType::Anonymous:
+ return std::make_shared<Aws::Auth::AnonymousAWSCredentialsProvider>();
+ case CredProviderType::Default:
+ case CredProviderType::Simple:
+ return std::make_shared<CustomAwsCredentialsProviderChain>();
+ case CredProviderType::InstanceProfile:
+ return
std::make_shared<Aws::Auth::InstanceProfileCredentialsProvider>();
+ }
+ return nullptr;
+}
+
+AwsCredentialResult assume_role(const AwsCredentialOptions& options,
+ std::shared_ptr<Provider> base_provider) {
+ auto sts_client =
+ std::make_shared<Aws::STS::STSClient>(base_provider,
options.sts_client_config);
+ return {
+ .provider =
std::make_shared<Aws::Auth::STSAssumeRoleCredentialsProvider>(
+ options.role_arn, Aws::String(), options.external_id,
+ Aws::Auth::DEFAULT_CREDS_LOAD_FREQ_SECONDS,
std::move(sts_client)),
+ };
+}
+
+} // namespace
+
+AwsCredentialResult AwsCredentialFactory::create(const AwsCredentialOptions&
options) {
+ const bool has_access_key = !options.access_key.empty();
+ const bool has_secret_key = !options.secret_key.empty();
+
+ if (has_access_key && has_secret_key) {
+ Aws::Auth::AWSCredentials credentials(options.access_key,
options.secret_key);
+ if (!options.session_token.empty()) {
+ credentials.SetSessionToken(options.session_token);
+ }
+ return {
+ .provider =
std::make_shared<Aws::Auth::SimpleAWSCredentialsProvider>(
+ std::move(credentials)),
+ };
+ }
+
+ if (options.version == AwsCredentialProviderVersion::V1) {
+ if (options.provider_type == CredProviderType::InstanceProfile) {
+ auto base =
std::make_shared<Aws::Auth::InstanceProfileCredentialsProvider>();
+ return options.role_arn.empty() ? AwsCredentialResult {.provider =
std::move(base)}
+ : assume_role(options,
std::move(base));
+ }
+ if (options.empty_credentials == EmptyCredentialsBehavior::ANONYMOUS) {
Review Comment:
[P2] Preserve the V1 fallback for accepted partial credentials
BE passes `ANONYMOUS` here, but this condition does not verify that both
static fields are empty. Current validation and tests accept AK-only or SK-only
configurations when a role ARN is present; on the
non-`InstanceProfile`/default-provider path the deleted V1 logic fell back to
`DefaultAWSCredentialsProviderChain`, while this branch now sends anonymous
requests. Because `v1` remains supported, those accepted configurations can
lose working environment/default credentials. Please select anonymous only when
both fields are empty (or reject partial pairs consistently) and add
provider-selection tests for both variants.
##########
common/cpp/client/obj_storage_client.cpp:
##########
@@ -0,0 +1,298 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "obj_storage_client.h"
+
+#include <cpp/sync_point.h>
+
+#include <algorithm>
+#include <iterator>
+
+namespace doris {
+namespace {
+
+ObjStorageRateLimitToken acquire_rate_limit(
+ const std::shared_ptr<const ObjStorageRateLimitPolicy>& policy,
ObjStorageRequestType type,
+ size_t estimated_bytes = 0) {
+ if (!policy) {
+ return {};
+ }
+ return policy->acquire(type, estimated_bytes);
+}
+
+} // namespace
+
+ObjStorageRateLimitToken ObjStorageClient::acquire(ObjStorageRequestType type,
+ size_t estimated_bytes)
const {
+ return acquire_rate_limit(rate_limit_policy_, type, estimated_bytes);
+}
+
+ObjectStorageUploadResponse ObjStorageClient::create_multipart_upload(
+ const ObjectStoragePathOptions& opts) {
+ auto rate_limit = acquire(ObjStorageRequestType::PUT);
+ if (!rate_limit.resp.ok()) {
+ return {.resp = std::move(rate_limit.resp)};
+ }
+ return backend_->create_multipart_upload(opts);
+}
+
+ObjectStorageResponse ObjStorageClient::put_object(const
ObjectStoragePathOptions& opts,
+ std::string_view stream) {
+ auto rate_limit = acquire(ObjStorageRequestType::PUT, stream.size());
+ if (!rate_limit.resp.ok()) {
+ return rate_limit.resp;
+ }
+ return backend_->put_object(opts, stream);
+}
+
+ObjectStorageUploadResponse ObjStorageClient::upload_part(const
ObjectStoragePathOptions& opts,
+ std::string_view
stream, int part_num) {
+ auto rate_limit = acquire(ObjStorageRequestType::PUT, stream.size());
+ if (!rate_limit.resp.ok()) {
+ return {.resp = std::move(rate_limit.resp)};
+ }
+ return backend_->upload_part(opts, stream, part_num);
+}
+
+ObjectStorageResponse ObjStorageClient::complete_multipart_upload(
+ const ObjectStoragePathOptions& opts,
+ const std::vector<ObjectCompleteMultiPart>& completed_parts) {
+ auto rate_limit = acquire(ObjStorageRequestType::PUT);
+ if (!rate_limit.resp.ok()) {
+ return rate_limit.resp;
+ }
+ return backend_->complete_multipart_upload(opts, completed_parts);
+}
+
+ObjectStorageHeadResponse ObjStorageClient::head_object(const
ObjectStoragePathOptions& opts) {
+ auto rate_limit = acquire(ObjStorageRequestType::GET);
+ if (!rate_limit.resp.ok()) {
+ return {.resp = std::move(rate_limit.resp)};
+ }
+ return backend_->head_object(opts);
+}
+
+ObjectStorageResponse ObjStorageClient::get_object(const
ObjectStoragePathOptions& opts,
+ void* buffer, size_t
offset, size_t bytes_read,
+ size_t* size_return) {
+ auto rate_limit = acquire(ObjStorageRequestType::GET, bytes_read);
+ if (!rate_limit.resp.ok()) {
+ return rate_limit.resp;
+ }
+ auto response = backend_->get_object(opts, buffer, offset, bytes_read,
size_return);
+ if (response.ok()) {
+ rate_limit.settle_bytes(*size_return);
Review Comment:
[P2] Port the direct admission-policy tests to the shared facade
This facade is now the sole mapping from each object API to GET/PUT
admission, estimated bytes, and short-read settlement, but the PR deletes the
direct suite that asserted those contracts without adding an equivalent
`ObjStorageClient` suite. The remaining changed tests count list/recursive
requests or exercise integration paths; they do not pin rejection propagation,
per-API request type, payload charging, or this `settle_bytes` call. Please
port those direct cases to the shared facade so a future mapping or bypass
regression is caught.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
