github-actions[bot] commented on code in PR #66350: URL: https://github.com/apache/doris/pull/66350#discussion_r3717613226
########## common/cpp/client/azure_obj_storage_backend.cpp: ########## @@ -0,0 +1,498 @@ +// 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) { Review Comment: [P1] Restore the standard-exception boundary for Azure SDK calls Both this handler and `head_object` now catch only `RequestFailedException`, but the removed implementations ran these calls through `do_azure_client_call`, which also converted `std::exception`. With the pinned Azure SDK, successful `StageBlock`/`GetProperties` response parsing uses `map::at` (and `std::stoll` for `Content-Length`), so missing or malformed provider/proxy headers can throw `std::out_of_range`/`std::invalid_argument`. Those exceptions escape the facade; multipart upload runs on the S3 upload pool whose worker has no catch, so it can terminate instead of setting the buffer's error status, while HEAD callers likewise cannot return failure. Please restore an equivalent catch for both methods and return `INTERNAL_ERROR` with path context. -- 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]
