github-actions[bot] commented on code in PR #66350:
URL: https://github.com/apache/doris/pull/66350#discussion_r3706695258
##########
cloud/src/recycler/s3_accessor.cpp:
##########
@@ -503,29 +553,30 @@ int S3Accessor::delete_files(const
std::vector<std::string>& paths) {
keys.emplace_back(get_key(path));
}
- return obj_client_->delete_objects(conf_.bucket, std::move(keys),
{.executor = worker_pool})
- .ret;
+ return obj_client_->delete_objects({.bucket = conf_.bucket},
std::move(keys)).status.code;
}
int S3Accessor::delete_file(const std::string& path) {
LOG_INFO("delete file").tag("uri", to_uri(path));
- int ret = obj_client_->delete_object({.bucket = conf_.bucket, .key =
get_key(path)}).ret;
- static_assert(ObjectStorageResponse::OK == 0);
- if (ret == ObjectStorageResponse::OK || ret ==
ObjectStorageResponse::NOT_FOUND) {
+ int ret =
+ obj_client_->delete_object({.bucket = conf_.bucket, .key =
get_key(path)}).status.code;
+ static_assert(ObjectStorageStatus::OK == 0);
+ if (ret == ObjectStorageStatus::OK || ret ==
ObjectStorageStatus::NOT_FOUND) {
return 0;
}
return ret;
}
int S3Accessor::put_file(const std::string& path, const std::string& content) {
- return obj_client_->put_object({.bucket = conf_.bucket, .key =
get_key(path)}, content).ret;
+ return obj_client_->put_object({.bucket = conf_.bucket, .key =
get_key(path)}, content)
+ .status.code;
}
int S3Accessor::list_prefix(const std::string& path_prefix,
std::unique_ptr<ListIterator>* res) {
- size_t prefix_length = conf_.prefix.empty() ? 0 : conf_.prefix.length() +
1;
*res = std::make_unique<S3ListIterator>(
- obj_client_->list_objects({.bucket = conf_.bucket, .key =
get_key(path_prefix)}),
- prefix_length);
+ obj_client_,
Review Comment:
[P1] Preserve Recycler's exists return contract
The common response uses positive Thrift codes (`NOT_FOUND` is 31, while
internal/auth errors are also positive), but `StorageVaultAccessor::exists`
promises 0 for present, 1 for absent, and a negative value for errors.
`checker.cpp` consequently treats an S3 permission or server failure as a
missing file, and the existing accessor tests still require 1 for a 404. Map OK
to 0, NOT_FOUND to 1, and every other status to a negative error at this
adapter boundary.
##########
gensrc/proto/cloud.proto:
##########
@@ -309,6 +309,7 @@ message ObjectStoreInfoPB {
optional CredProviderTypePB cred_provider_type = 17;
optional string role_arn = 18; // aws assumed role's arn
optional string external_id = 19; // aws assumed role's external_id if
configure
+ optional string token = 20; // optional session token paired with ak/sk
Review Comment:
[P1] Give the token the same secret lifecycle as SK
This field is currently left plaintext by `create_object_info_with_encrypt`
and external-stage creation while SK is encrypted. It can then be emitted by
full-instance INFO logs that do not call `hide_token`, returned by the list-all
external-stage display path, and serialized by `SHOW STORAGE VAULT` for any
`USAGE` grantee because that converter masks only SK. Please encrypt/decrypt
the token like other credential secrets and explicitly strip or mask it from
every log and display response.
##########
common/cpp/client/s3_obj_storage_provider.cpp:
##########
@@ -0,0 +1,558 @@
+// 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_provider.h"
+
+#include <cpp/client/obj_storage_client.h>
+#include <gen_cpp/Status_types.h>
+
+#include <algorithm>
+
+#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 {
+
+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 S3ObjStorageProvider::create_multipart_upload(
+ const ObjectStoragePathOptions& opts) {
+ CreateMultipartUploadRequest request;
+ request.WithBucket(opts.bucket).WithKey(opts.key);
+ request.SetContentType("application/octet-stream");
+
+ 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& request_id = outcome.IsSuccess() ?
outcome.GetResult().GetRequestId()
+ :
outcome.GetError().GetRequestId();
+
+ LOG(INFO) << "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 S3ObjStorageProvider::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");
+
+ 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& 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(INFO) << "request_id = " << request_id << ", bucket = " << opts.bucket
Review Comment:
[P2] Avoid INFO logging every successful object write
This unconditional success log turns normal PutObject traffic into one INFO
record per object; create and complete multipart have the same pattern. The
deleted BE implementation logged these only when they exceeded the 5-second
threshold, and Recycler logged PutObject failures only, so ingest/file-cache
traffic can now generate log I/O proportional to storage QPS. Restore
slow-request-only logging or lower these success messages to VLOG while keeping
latency bvars for aggregate observability.
##########
cloud/src/recycler/s3_accessor.cpp:
##########
@@ -567,11 +618,11 @@ int S3Accessor::abort_multipart_upload(const std::string&
path, const std::strin
}
int S3Accessor::get_life_cycle(int64_t* expiration_days) {
- return obj_client_->get_life_cycle(conf_.bucket, expiration_days).ret;
Review Comment:
[P1] Finish converting the GCS iterator to value semantics
This declaration changes `iter` from a pointer to a stack
`ObjectListIterator`, and the new loop correctly calls `iter.next()`, but the
post-loop check at line 678 still calls `iter->is_valid()`. The common iterator
defines no `operator->`, so the Cloud target cannot compile regardless of which
provider is selected at runtime. Change that remaining check to
`iter.is_valid()`.
##########
common/cpp/client/obj_storage_client.cpp:
##########
@@ -0,0 +1,299 @@
+// 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 {
+
+ObjStorageRateLimitToken ObjStorageClient::acquire(ObjStorageRequestType type,
+ size_t estimated_bytes)
const {
+ if (!rate_limit_policy_) {
+ return {};
+ }
+ return rate_limit_policy_->acquire(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 provider_->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 provider_->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 provider_->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 provider_->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 provider_->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 = provider_->get_object(opts, buffer, offset, bytes_read,
size_return);
+ if (response.ok()) {
+ rate_limit.settle_bytes(*size_return);
+ }
+ return response;
+}
+
+ObjectStorageListPage ObjStorageClient::list_objects(const
ObjectStoragePathOptions& opts,
+ std::string_view
continuation_token) {
+ auto rate_limit = acquire(ObjStorageRequestType::GET);
+ if (!rate_limit.resp.ok()) {
+ return {.resp = std::move(rate_limit.resp)};
+ }
+ return provider_->list_objects(opts, continuation_token);
+}
+
+ObjectStorageResponse ObjectListIterator::has_next() {
+ if (!is_valid_) {
+ return {
+ .status = {TStatusCode::INTERNAL_ERROR, "Iterator is invalid"},
+ .http_code = 0,
+ };
+ }
+ while (next_index_ == objects_.size()) {
+ if (!has_more_) {
+ return {
+ .status = {TStatusCode::END_OF_FILE, "No more results"},
+ .http_code = 200,
+ };
+ }
+ auto page = client_->list_objects(opts_, continuation_token_);
+ if (!page.resp.ok()) {
+ is_valid_ = false;
+ return page.resp;
+ }
+ objects_ = std::move(page.objects);
+ next_index_ = 0;
+ continuation_token_ = std::move(page.continuation_token);
+ has_more_ = page.has_more;
+ }
+ return ObjectStorageResponse::OK();
+}
+
+ObjectStorageListResponse ObjectListIterator::next() {
+ auto response = has_next();
+ if (response.status.code == ObjectStorageStatus::END_OF_FILE) {
+ return {.resp = ObjectStorageResponse::OK(), .results_ = {}};
+ }
+ if (!response.ok()) {
+ return {.resp = std::move(response), .results_ = {}};
+ }
+ return {
+ .resp = ObjectStorageResponse::OK(),
+ .results_ = std::move(objects_[next_index_++]),
+ };
+}
+
+ObjectStorageResponse ObjStorageClient::delete_objects(const
ObjectStoragePathOptions& opts,
+
std::vector<std::string> objs) {
+ // TODO: One public delete_objects call may issue multiple SDK requests.
Move batching to an
+ // upper-layer helper and make the client API handle exactly one
provider-sized batch, so one
+ // client call maps to one rate-limit admission and one SDK request.
+ const auto max_batch_size = std::max<size_t>(1,
provider_->capabilities().max_delete_batch);
+ for (size_t begin = 0; begin < objs.size(); begin += max_batch_size) {
+ const auto end = std::min(begin + max_batch_size, objs.size());
+ auto rate_limit = acquire(ObjStorageRequestType::PUT);
+ if (!rate_limit.resp.ok()) {
+ return rate_limit.resp;
+ }
+ std::vector<std::string> batch(std::make_move_iterator(objs.begin() +
begin),
+ std::make_move_iterator(objs.begin() +
end));
+ auto response = provider_->delete_objects(opts, std::move(batch));
+ if (!response.ok()) {
+ return response;
+ }
+ }
+ return ObjectStorageResponse::OK();
+}
+
+ObjectStorageResponse ObjStorageClient::delete_object(const
ObjectStoragePathOptions& opts) {
+ auto rate_limit = acquire(ObjStorageRequestType::PUT);
+ if (!rate_limit.resp.ok()) {
+ return rate_limit.resp;
+ }
+ return provider_->delete_object(opts);
+}
+
+ObjectStorageResponse ObjStorageClient::delete_objects_recursively(
+ const ObjectStoragePathOptions& opts, const RecursiveDeleteOptions&
options) {
+ // TODO: A recursive delete may issue multiple list and delete SDK
requests. Keep the previous
+ // BE behavior of charging it as one logical PUT call in this refactor.
Move recursive
+ // orchestration above ObjStorageClient in a follow-up, using the one-page
list API and a
+ // one-batch delete API so every client call maps to one admission and one
SDK request.
+ auto rate_limit = acquire(ObjStorageRequestType::PUT);
Review Comment:
[P1] Keep Recycler recursive deletes under request-level admission
After this single PUT admission, every list page and delete batch calls the
provider directly, so `RecyclerObjStorageRateLimitPolicy` is never invoked for
the actual SDK requests. The deleted Recycler implementations admitted every
GET page and PUT batch; a large prefix can now issue unbounded pages and
concurrent deletes after one token and one fault-injection decision. Route each
page/batch through the policy-bearing facade, or make BE's
one-logical-operation behavior an explicit BE-only mode.
##########
cloud/src/recycler/s3_accessor.cpp:
##########
@@ -207,18 +220,28 @@ std::optional<S3Conf> S3Conf::from_obj_store_info(const
ObjectStoreInfoPB& obj_i
s3_conf.provider = S3Conf::AZURE;
break;
default:
- LOG_WARNING("unknown provider type {}").tag("obj_info",
proto_to_json(obj_info));
+ LOG_WARNING("unknown object storage provider")
+ .tag("provider", obj_info.provider())
+ .tag("bucket", obj_info.bucket());
return std::nullopt;
}
if (!skip_aksk) {
+ if (obj_info.ak().empty() != obj_info.sk().empty()) {
+ LOG_WARNING("access key and secret key must be provided together")
+ .tag("provider", obj_info.provider())
+ .tag("bucket", obj_info.bucket());
+ return std::nullopt;
+ }
if (!obj_info.ak().empty() && !obj_info.sk().empty()) {
if (obj_info.has_encryption_info()) {
AkSkPair plain_ak_sk_pair;
int ret = decrypt_ak_sk_helper(obj_info.ak(), obj_info.sk(),
obj_info.encryption_info(),
&plain_ak_sk_pair);
if (ret != 0) {
- LOG_WARNING("fail to decrypt ak sk").tag("obj_info",
proto_to_json(obj_info));
+ LOG_WARNING("fail to decrypt ak sk")
+ .tag("provider", obj_info.provider())
+ .tag("bucket", obj_info.bucket());
return std::nullopt;
} else {
Review Comment:
[P1] Preserve token across the complete DDL and metadata round trip
FE creation accepts and validates `s3.session_token`, but
`CloudObjectStoreAdapter` never puts it in `ObjectStoreInfoPB`, and the S3
vault ALTER allow-list rejects token rotation. Even for direct protobuf paths,
`object_info_pb_factory` and internal-stage reconstruction omit the token,
while meta-service alteration neither updates nor clears it with AK/SK.
Temporary credentials can therefore validate and still reach the new
BE/Recycler/FE consumers with an empty or stale token. Thread token through the
FE builder/allow-list and every meta create, alter, and reconstruction path as
part of the AK/SK tuple, with end-to-end tests.
--
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]
