diff --git a/be/src/common/config.cpp b/be/src/common/config.cpp index 9f697a2dcbed81..09c03967ef3829 100644 --- a/be/src/common/config.cpp +++ b/be/src/common/config.cpp @@ -1521,6 +1521,11 @@ DEFINE_mInt32(max_s3_client_retry, "10"); DEFINE_mInt32(s3_read_base_wait_time_ms, "100"); DEFINE_mInt32(s3_read_max_wait_time_ms, "800"); DEFINE_mBool(enable_s3_object_check_after_upload, "true"); + +DEFINE_Bool(enable_oss_native_sdk, "false"); +DEFINE_mInt32(max_oss_client_retry, "10"); +DEFINE_mInt32(oss_read_base_wait_time_ms, "100"); +DEFINE_mInt32(oss_read_max_wait_time_ms, "800"); DEFINE_mInt32(aws_client_request_timeout_ms, "30000"); DEFINE_mBool(enable_s3_rate_limiter, "false"); diff --git a/be/src/common/config.h b/be/src/common/config.h index f5c72eb776d5d3..0bfb1b63441646 100644 --- a/be/src/common/config.h +++ b/be/src/common/config.h @@ -1630,6 +1630,13 @@ DECLARE_mInt32(s3_read_max_wait_time_ms); DECLARE_mBool(enable_s3_object_check_after_upload); DECLARE_mInt32(aws_client_request_timeout_ms); +// native Alibaba Cloud OSS SDK (enable_oss_native_sdk=true routes ObjStorageType::OSS +// through OSSv2ObjStorageClient instead of the S3-compatible AWS SDK path) +DECLARE_Bool(enable_oss_native_sdk); +DECLARE_mInt32(max_oss_client_retry); +DECLARE_mInt32(oss_read_base_wait_time_ms); +DECLARE_mInt32(oss_read_max_wait_time_ms); + // write as inverted index tmp directory DECLARE_String(tmp_file_dir); diff --git a/be/src/io/CMakeLists.txt b/be/src/io/CMakeLists.txt index 56c2eeb94a3819..eb9cbd835be21f 100644 --- a/be/src/io/CMakeLists.txt +++ b/be/src/io/CMakeLists.txt @@ -26,6 +26,13 @@ if(BUILD_AZURE STREQUAL "OFF") list(REMOVE_ITEM IO_FILES "${CMAKE_CURRENT_SOURCE_DIR}/fs/azure_obj_storage_client.cpp") endif() +if(BUILD_OSS STREQUAL "OFF") + list(REMOVE_ITEM IO_FILES "${CMAKE_CURRENT_SOURCE_DIR}/fs/oss_v2_obj_storage_client.cpp") + list(REMOVE_ITEM IO_FILES "${CMAKE_CURRENT_SOURCE_DIR}/fs/oss_file_system.cpp") + list(REMOVE_ITEM IO_FILES "${CMAKE_CURRENT_SOURCE_DIR}/fs/oss_file_reader.cpp") + list(REMOVE_ITEM IO_FILES "${CMAKE_CURRENT_SOURCE_DIR}/fs/oss_file_writer.cpp") +endif() + if(ENABLE_TDE) list(REMOVE_ITEM IO_FILES "${CMAKE_CURRENT_SOURCE_DIR}/fs/encrypted_fs_factory.cpp") file(GLOB_RECURSE EXTRA_SOURCES CONFIGURE_DEPENDS diff --git a/be/src/io/fs/file_system.h b/be/src/io/fs/file_system.h index 2cf63b2ff008db..6840ff59fda7fb 100644 --- a/be/src/io/fs/file_system.h +++ b/be/src/io/fs/file_system.h @@ -48,7 +48,7 @@ namespace doris::io { } while (0); #endif -enum class FileSystemType : uint8_t { LOCAL, S3, HDFS, BROKER, HTTP }; +enum class FileSystemType : uint8_t { LOCAL, S3, HDFS, BROKER, HTTP, OSS }; inline std::ostream& operator<<(std::ostream& ostr, FileSystemType type) { switch (type) { @@ -67,6 +67,9 @@ inline std::ostream& operator<<(std::ostream& ostr, FileSystemType type) { case FileSystemType::HTTP: ostr << "HTTP"; return ostr; + case FileSystemType::OSS: + ostr << "OSS"; + return ostr; default: ostr << "UNKNOWN"; return ostr; diff --git a/be/src/io/fs/oss_file_reader.cpp b/be/src/io/fs/oss_file_reader.cpp new file mode 100644 index 00000000000000..a8f2a0a4eb7f64 --- /dev/null +++ b/be/src/io/fs/oss_file_reader.cpp @@ -0,0 +1,277 @@ +// 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 "io/fs/oss_file_reader.h" + +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include + +#include "common/compiler_util.h" +#include "common/config.h" +#include "common/metrics/doris_metrics.h" +#include "io/fs/err_utils.h" +#include "runtime/runtime_profile.h" +#include "runtime/thread_context.h" +#include "runtime/workload_management/io_throttle.h" +#include "util/bvar_helper.h" +#include "util/debug_points.h" + +namespace doris::io { + +namespace { +// Thread-local random number generator for jitter +// Using thread_local ensures each thread has its own seeded generator +thread_local std::mt19937 g_rng(std::random_device {}()); +} // namespace + +bvar::Adder oss_file_reader_read_counter("oss_file_reader", "read_at"); +bvar::Adder oss_file_reader_total("oss_file_reader", "total_num"); +bvar::Adder oss_bytes_read_total("oss_file_reader", "bytes_read"); +bvar::Adder oss_file_being_read("oss_file_reader", "file_being_read"); +bvar::LatencyRecorder oss_bytes_per_read("oss_file_reader", "bytes_per_read"); +bvar::PerSecond> oss_read_throughput("oss_file_reader", "oss_read_throughput", + &oss_bytes_read_total); +bvar::PerSecond> oss_get_request_qps("oss_file_reader", "oss_get_request", + &oss_file_reader_read_counter); +bvar::LatencyRecorder oss_file_reader_latency("oss_file_reader", "oss_latency"); + +Result OSSFileReader::create(std::shared_ptr client, + std::string bucket, std::string key, int64_t file_size, + RuntimeProfile* profile) { + if (file_size < 0) { + auto res = client->object_file_size(bucket, key); + if (!res.has_value()) { + return ResultError(std::move(res.error())); + } + file_size = res.value(); + } + + return std::make_shared(std::move(client), std::move(bucket), std::move(key), + file_size, profile); +} + +OSSFileReader::OSSFileReader(std::shared_ptr client, std::string bucket, + std::string key, size_t file_size, RuntimeProfile* profile) + : _path(fmt::format("oss://{}/{}", bucket, key)), + _file_size(file_size), + _bucket(std::move(bucket)), + _key(std::move(key)), + _client(std::move(client)), + _profile(profile) { + // TODO: Create separate oss_file_open_reading and oss_file_reader_total DorisMetrics + // Currently reusing s3 metrics for backward compatibility with dashboards + DorisMetrics::instance()->s3_file_open_reading->increment(1); + DorisMetrics::instance()->s3_file_reader_total->increment(1); + oss_file_reader_total << 1; + oss_file_being_read << 1; +} + +OSSFileReader::~OSSFileReader() { + static_cast(close()); + oss_file_being_read << -1; +} + +Status OSSFileReader::close() { + bool expected = false; + if (_closed.compare_exchange_strong(expected, true, std::memory_order_acq_rel)) { + // TODO: Should decrement oss_file_open_reading DorisMetric instead of s3 metric + DorisMetrics::instance()->s3_file_open_reading->increment(-1); + } + return Status::OK(); +} + +Status OSSFileReader::read_at_impl(size_t offset, Slice result, size_t* bytes_read, + const IOContext* /*io_ctx*/) { + DCHECK(!closed()); + if (offset > _file_size) { + return Status::InternalError( + "offset exceeds file size(offset: {}, file size: {}, path: {})", offset, _file_size, + _path.native()); + } + + size_t bytes_req = result.size; + char* to = result.data; + bytes_req = std::min(bytes_req, _file_size - offset); + + VLOG_DEBUG << "OSS read_at_impl: offset=" << offset << " bytes_req=" << bytes_req + << " result.size=" << result.size << " file_size=" << _file_size; + + if (UNLIKELY(bytes_req == 0)) { + *bytes_read = 0; + return Status::OK(); + } + + auto client = _client->get(); + if (!client) { + return Status::InternalError("OSS client not initialized"); + } + // TODO: add SCOPED_CONCURRENCY_COUNT for oss_file_reader_read like S3 + + int retry_count = 0; + const int base_wait_time = config::oss_read_base_wait_time_ms; + const int max_wait_time = config::oss_read_max_wait_time_ms; + const int max_retries = config::max_oss_client_retry; + + int64_t begin_ts = std::chrono::duration_cast( + std::chrono::system_clock::now().time_since_epoch()) + .count(); + LIMIT_REMOTE_SCAN_IO(bytes_read); + SCOPED_RAW_TIMER(&_oss_stats.total_get_request_time_ns); + Defer defer_latency {[&]() { + int64_t end_ts = std::chrono::duration_cast( + std::chrono::system_clock::now().time_since_epoch()) + .count(); + oss_file_reader_latency << (end_ts - begin_ts); + }}; + + int total_sleep_time = 0; + while (retry_count <= max_retries) { + *bytes_read = 0; + oss_file_reader_read_counter << 1; + + // Debug point: Simulate slow read + DBUG_EXECUTE_IF("OSSFileReader::read_at_impl.slow_read", { + auto sleep_ms = dp->param("sleep_ms", 1000); + LOG(INFO) << "Debug point: Simulating slow OSS read, sleeping " << sleep_ms << "ms"; + std::this_thread::sleep_for(std::chrono::milliseconds(sleep_ms)); + }); + + // Debug point: Simulate rate limiting error + DBUG_EXECUTE_IF("OSSFileReader::read_at_impl.rate_limit", { + LOG(WARNING) << "Debug point: Simulating OSS rate limit error"; + std::string msg = fmt::format("Debug OSS rate limit error for path {}", _path.native()); + return Status::IOError(msg); + }); + + AlibabaCloud::OSS::GetObjectRequest request(_bucket, _key); + request.setRange(static_cast(offset), + static_cast(offset + bytes_req - 1)); + + auto outcome = client->GetObject(request); + _oss_stats.total_get_request_counter++; + + // Debug point: Simulate read failure + DBUG_EXECUTE_IF("OSSFileReader::read_at_impl.read_error", { + LOG(WARNING) << "Debug point: Simulating OSS read error"; + std::string msg = fmt::format("Debug OSS read error for path {}", _path.native()); + return Status::IOError(msg); + }); + + if (!outcome.isSuccess()) { + std::string error_code = outcome.error().Code(); + + // Exponential backoff with jitter for rate limiting + if (error_code == "TooManyRequests" || error_code == "SlowDown") { + retry_count++; + if (retry_count > max_retries) { + std::string err_msg = fmt::format( + "OSS GetObject failed after {} retries for {}: {} - {}", max_retries, + _path.native(), error_code, outcome.error().Message()); + LOG(WARNING) << err_msg; + return Status::IOError(err_msg); + } + + // Exponential backoff with jitter + int wait_time = std::min(base_wait_time * (1 << retry_count), max_wait_time); + std::uniform_int_distribution dist(0, 99); + int jitter = dist(g_rng); + wait_time += jitter; + + std::this_thread::sleep_for(std::chrono::milliseconds(wait_time)); + _oss_stats.too_many_request_err_counter++; + _oss_stats.too_many_request_sleep_time_ms += wait_time; + total_sleep_time += wait_time; + + VLOG_DEBUG << "OSS rate limited, retry " << retry_count << "/" << max_retries + << " after " << wait_time << "ms (jitter: " << jitter + << "ms), path: " << _path.native(); + continue; + } + + std::string err_msg = + fmt::format("OSS GetObject failed for {}: {} - {}", _path.native(), error_code, + outcome.error().Message()); + LOG(WARNING) << err_msg; + return Status::IOError(err_msg); + } + + auto& content_stream = outcome.result().Content(); + content_stream->read(to, bytes_req); + *bytes_read = content_stream->gcount(); + + if (*bytes_read != bytes_req) { + std::string msg = fmt::format( + "OSS read size mismatch: path={} offset={} bytes_req={} bytes_read={} " + "file_size={}", + _path.native(), offset, bytes_req, *bytes_read, _file_size); + LOG(WARNING) << msg; + return Status::InternalError(msg); + } + + _oss_stats.total_bytes_read += bytes_req; + oss_bytes_read_total << bytes_req; + oss_bytes_per_read << bytes_req; + // TODO: Create separate oss_bytes_read_total DorisMetric to avoid confusing naming + // Currently reusing s3_bytes_read_total for backward compatibility with dashboards + DorisMetrics::instance()->s3_bytes_read_total->increment(bytes_req); + + if (retry_count > 0) { + LOG(INFO) << fmt::format("OSS read {} succeeded after {} retries with {} ms sleeping", + _path.native(), retry_count, total_sleep_time); + } + + return Status::OK(); + } + + return Status::IOError("OSS read failed: max retries exceeded"); +} + +void OSSFileReader::_collect_profile_before_close() { + if (_profile != nullptr) { + const char* oss_profile_name = "OSSProfile"; + ADD_TIMER(_profile, oss_profile_name); + RuntimeProfile::Counter* total_get_request_counter = + ADD_CHILD_COUNTER(_profile, "TotalGetRequest", TUnit::UNIT, oss_profile_name); + RuntimeProfile::Counter* too_many_request_err_counter = + ADD_CHILD_COUNTER(_profile, "TooManyRequestErr", TUnit::UNIT, oss_profile_name); + RuntimeProfile::Counter* too_many_request_sleep_time = ADD_CHILD_COUNTER( + _profile, "TooManyRequestSleepTime", TUnit::TIME_MS, oss_profile_name); + RuntimeProfile::Counter* total_bytes_read = + ADD_CHILD_COUNTER(_profile, "TotalBytesRead", TUnit::BYTES, oss_profile_name); + RuntimeProfile::Counter* total_get_request_time_ns = + ADD_CHILD_TIMER(_profile, "TotalGetRequestTime", oss_profile_name); + + COUNTER_UPDATE(total_get_request_counter, _oss_stats.total_get_request_counter); + COUNTER_UPDATE(too_many_request_err_counter, _oss_stats.too_many_request_err_counter); + COUNTER_UPDATE(too_many_request_sleep_time, _oss_stats.too_many_request_sleep_time_ms); + COUNTER_UPDATE(total_bytes_read, _oss_stats.total_bytes_read); + COUNTER_UPDATE(total_get_request_time_ns, _oss_stats.total_get_request_time_ns); + } +} + +} // namespace doris::io diff --git a/be/src/io/fs/oss_file_reader.h b/be/src/io/fs/oss_file_reader.h new file mode 100644 index 00000000000000..003d54b828c0c1 --- /dev/null +++ b/be/src/io/fs/oss_file_reader.h @@ -0,0 +1,81 @@ +// 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. + +#pragma once + +#include +#include +#include + +#include "io/fs/file_reader.h" +#include "io/fs/oss_file_system.h" +#include "io/fs/path.h" + +namespace doris { +class RuntimeProfile; + +namespace io { + +struct OSSFileReaderStats { + int64_t total_get_request_time_ns = 0; + int64_t total_get_request_counter = 0; + int64_t total_bytes_read = 0; + int64_t too_many_request_err_counter = 0; + int64_t too_many_request_sleep_time_ms = 0; +}; + +class OSSFileReader final : public FileReader { +public: + static Result create(std::shared_ptr client, + std::string bucket, std::string key, int64_t file_size, + RuntimeProfile* profile = nullptr); + + OSSFileReader(std::shared_ptr client, std::string bucket, std::string key, + size_t file_size, RuntimeProfile* profile); + + ~OSSFileReader() override; + + Status close() override; + + const Path& path() const override { return _path; } + + size_t size() const override { return _file_size; } + + bool closed() const override { return _closed.load(std::memory_order_acquire); } + + int64_t mtime() const override { return 0; } + +protected: + Status read_at_impl(size_t offset, Slice result, size_t* bytes_read, + const IOContext* io_ctx) override; + + void _collect_profile_before_close() override; + +private: + Path _path; + size_t _file_size; + std::string _bucket; + std::string _key; + std::shared_ptr _client; + std::atomic _closed {false}; + + OSSFileReaderStats _oss_stats; + RuntimeProfile* _profile = nullptr; +}; + +} // namespace io +} // namespace doris diff --git a/be/src/io/fs/oss_file_system.cpp b/be/src/io/fs/oss_file_system.cpp new file mode 100644 index 00000000000000..68559da1ddc37c --- /dev/null +++ b/be/src/io/fs/oss_file_system.cpp @@ -0,0 +1,631 @@ +// 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 "io/fs/oss_file_system.h" + +#include +#include + +#include +#include +#include +#include +#include + +#include "common/config.h" +#include "common/logging.h" +#include "common/status.h" +#include "io/fs/err_utils.h" +#include "io/fs/file_system.h" +#include "io/fs/file_writer.h" +#include "io/fs/local_file_system.h" +#include "io/fs/oss_file_reader.h" +#include "io/fs/oss_file_writer.h" +#include "io/fs/remote_file_system.h" + +namespace doris::io { +namespace { + +constexpr int64_t MULTIPART_COPY_THRESHOLD = 1073741824; // 1GB +constexpr int64_t MULTIPART_COPY_PART_SIZE = 104857600; // 100MB per part + +#ifndef CHECK_OSS_CLIENT +#define CHECK_OSS_CLIENT(client) \ + if (!client) { \ + return Status::InvalidArgument("init oss client error"); \ + } +#endif + +Result get_oss_key(const Path& full_path) { + std::string path_str = full_path.native(); + + const std::string oss_prefix = "oss://"; + if (path_str.find(oss_prefix) == 0) { + path_str = path_str.substr(oss_prefix.length()); + + size_t pos = path_str.find('/'); + if (pos == std::string::npos) { + return ""; // No key, just bucket + } + return path_str.substr(pos + 1); + } + + return path_str; +} + +} // namespace + +OSSClientHolder::OSSClientHolder(OSSClientConf conf) : _conf(std::move(conf)) {} + +OSSClientHolder::~OSSClientHolder() = default; + +Status OSSClientHolder::init() { + _client = OSSClientFactory::instance().create(_conf); + if (!_client) { + return Status::InvalidArgument("failed to init oss client with conf {}", _conf.to_string()); + } + + return Status::OK(); +} + +Status OSSClientHolder::reset(const OSSClientConf& conf) { + OSSClientConf reset_conf; + { + std::shared_lock lock(_mtx); + if (conf.get_hash() == _conf.get_hash()) { + return Status::OK(); + } + + reset_conf = _conf; + reset_conf.ak = conf.ak; + reset_conf.sk = conf.sk; + reset_conf.token = conf.token; + reset_conf.bucket = conf.bucket; + reset_conf.connect_timeout_ms = conf.connect_timeout_ms; + reset_conf.max_connections = conf.max_connections; + reset_conf.request_timeout_ms = conf.request_timeout_ms; + reset_conf.cred_provider_type = conf.cred_provider_type; + } + + auto client = OSSClientFactory::instance().create(reset_conf); + if (!client) { + return Status::InvalidArgument("failed to init oss client with conf {}", conf.to_string()); + } + + LOG(WARNING) << "reset oss client with new conf: " << conf.to_string(); + + { + std::lock_guard lock(_mtx); + _client = std::move(client); + _conf = std::move(reset_conf); + } + + return Status::OK(); +} + +Result OSSClientHolder::object_file_size(const std::string& bucket, + const std::string& key) const { + auto client = get(); + if (!client) { + return ResultError(Status::InvalidArgument("init oss client error")); + } + + auto outcome = client->HeadObject(bucket, key); + if (!outcome.isSuccess()) { + return ResultError(Status::IOError("failed to head oss file {}: {} - {}", + full_oss_path(bucket, key), outcome.error().Code(), + outcome.error().Message())); + } + + return outcome.result().ContentLength(); +} + +std::string OSSClientHolder::full_oss_path(std::string_view bucket, std::string_view key) const { + return fmt::format("{}/{}/{}", _conf.endpoint, bucket, key); +} + +std::string OSSFileSystem::full_oss_path(std::string_view key) const { + return _client->full_oss_path(_bucket, key); +} + +Result> OSSFileSystem::create(OSSConf oss_conf, std::string id) { + // Normalize prefix before construction so RemoteFileSystem and _prefix are consistent. + if (!oss_conf.prefix.empty()) { + size_t start = oss_conf.prefix.find_first_not_of('/'); + if (start == std::string::npos) { + oss_conf.prefix = ""; + } else { + size_t end = oss_conf.prefix.find_last_not_of('/'); + oss_conf.prefix = oss_conf.prefix.substr(start, end - start + 1); + } + } + std::shared_ptr fs(new OSSFileSystem(std::move(oss_conf), std::move(id))); + RETURN_IF_ERROR_RESULT(fs->init()); + return fs; +} + +OSSFileSystem::OSSFileSystem(OSSConf oss_conf, std::string id) + : RemoteFileSystem(oss_conf.prefix, std::move(id), FileSystemType::OSS), + _bucket(std::move(oss_conf.bucket)), + _prefix(std::move(oss_conf.prefix)), + _client(std::make_shared(std::move(oss_conf.client_conf))) {} + +Status OSSFileSystem::init() { + return _client->init(); +} + +OSSFileSystem::~OSSFileSystem() = default; + +Status OSSFileSystem::create_file_impl(const Path& file, FileWriterPtr* writer, + const FileWriterOptions* opts) { + auto key = DORIS_TRY(get_oss_key(file)); + *writer = std::make_unique(_client, _bucket, key, opts); + return Status::OK(); +} + +Status OSSFileSystem::open_file_internal(const Path& file, FileReaderSPtr* reader, + const FileReaderOptions& opts) { + auto key = DORIS_TRY(get_oss_key(file)); + int64_t fsize = opts.file_size; + + auto oss_reader = DORIS_TRY(OSSFileReader::create(_client, _bucket, key, fsize)); + *reader = oss_reader; + return Status::OK(); +} + +Status OSSFileSystem::create_directory_impl(const Path& dir, bool failed_if_exists) { + return Status::OK(); +} + +Status OSSFileSystem::delete_file_impl(const Path& file) { + auto client = _client->get(); + CHECK_OSS_CLIENT(client); + + auto key = DORIS_TRY(get_oss_key(file)); + + auto outcome = client->DeleteObject(_bucket, key); + + if (!outcome.isSuccess()) { + std::string error_code = outcome.error().Code(); + if (error_code != "NoSuchKey") { + return Status::IOError("failed to delete file {}: {} - {}", full_oss_path(key), + error_code, outcome.error().Message()); + } + } + + return Status::OK(); +} + +Status OSSFileSystem::delete_directory_impl(const Path& dir) { + auto client = _client->get(); + CHECK_OSS_CLIENT(client); + + auto prefix = DORIS_TRY(get_oss_key(dir)); + if (!prefix.empty() && prefix.back() != '/') { + prefix.push_back('/'); + } + + // Abort in-progress multipart uploads + { + bool is_truncated = true; + std::string key_marker; + std::string upload_id_marker; + int aborted_count = 0; + + while (is_truncated) { + AlibabaCloud::OSS::ListMultipartUploadsRequest list_uploads_request(_bucket); + list_uploads_request.setPrefix(prefix); + list_uploads_request.setMaxUploads(1000); + + if (!key_marker.empty()) { + list_uploads_request.setKeyMarker(key_marker); + } + if (!upload_id_marker.empty()) { + list_uploads_request.setUploadIdMarker(upload_id_marker); + } + + auto list_uploads_outcome = client->ListMultipartUploads(list_uploads_request); + if (!list_uploads_outcome.isSuccess()) { + LOG(WARNING) << "Failed to list multipart uploads for prefix " << prefix << ": " + << list_uploads_outcome.error().Code() << " - " + << list_uploads_outcome.error().Message() + << ". Continuing with object deletion."; + break; // Don't fail deletion if listing uploads fails + } + + const auto& uploads_result = list_uploads_outcome.result(); + const auto& uploads = uploads_result.MultipartUploadList(); + + for (const auto& upload : uploads) { + AlibabaCloud::OSS::AbortMultipartUploadRequest abort_request(_bucket, upload.Key, + upload.UploadId); + auto abort_outcome = client->AbortMultipartUpload(abort_request); + + if (!abort_outcome.isSuccess()) { + LOG(WARNING) << "Failed to abort multipart upload: key=" << upload.Key + << " upload_id=" << upload.UploadId + << " error=" << abort_outcome.error().Code() << " - " + << abort_outcome.error().Message(); + // Don't fail directory deletion if abort fails + } else { + aborted_count++; + VLOG(1) << "Aborted multipart upload: key=" << upload.Key + << " upload_id=" << upload.UploadId; + } + } + + is_truncated = uploads_result.IsTruncated(); + key_marker = uploads_result.NextKeyMarker(); + upload_id_marker = uploads_result.NextUploadIdMarker(); + } + + if (aborted_count > 0) { + LOG(INFO) << "Aborted " << aborted_count + << " in-progress multipart uploads under prefix: " << prefix; + } + } + + // Delete all objects with prefix + bool is_truncated = true; + std::string marker; + + while (is_truncated) { + AlibabaCloud::OSS::ListObjectsRequest list_request(_bucket); + list_request.setPrefix(prefix); + list_request.setMaxKeys(1000); + + if (!marker.empty()) { + list_request.setMarker(marker); + } + + auto list_outcome = client->ListObjects(list_request); + if (!list_outcome.isSuccess()) { + return Status::IOError("failed to list objects for delete directory {}: {} - {}", + full_oss_path(prefix), list_outcome.error().Code(), + list_outcome.error().Message()); + } + + const auto& result = list_outcome.result(); + const auto& objects = result.ObjectSummarys(); + + if (!objects.empty()) { + AlibabaCloud::OSS::DeletedKeyList keys; + for (const auto& obj : objects) { + keys.push_back(obj.Key()); + } + + AlibabaCloud::OSS::DeleteObjectsRequest delete_request(_bucket); + delete_request.setKeyList(keys); + delete_request.setQuiet(false); + auto delete_outcome = client->DeleteObjects(delete_request); + + if (!delete_outcome.isSuccess()) { + return Status::IOError("failed to batch delete objects: {} - {}", + delete_outcome.error().Code(), + delete_outcome.error().Message()); + } + } + + is_truncated = result.IsTruncated(); + marker = result.NextMarker(); + } + + return Status::OK(); +} + +Status OSSFileSystem::batch_delete_impl(const std::vector& remote_files) { + auto client = _client->get(); + CHECK_OSS_CLIENT(client); + + constexpr size_t max_delete_batch = 1000; + auto path_iter = remote_files.begin(); + + do { + AlibabaCloud::OSS::DeletedKeyList keys; + auto path_begin = path_iter; + + for (; path_iter != remote_files.end() && (path_iter - path_begin < max_delete_batch); + ++path_iter) { + auto key = DORIS_TRY(get_oss_key(*path_iter)); + keys.push_back(key); + } + + if (keys.empty()) { + break; + } + + AlibabaCloud::OSS::DeleteObjectsRequest request(_bucket); + request.setKeyList(keys); + request.setQuiet(false); + auto outcome = client->DeleteObjects(request); + + if (!outcome.isSuccess()) { + return Status::IOError("failed to batch delete objects: {} - {}", + outcome.error().Code(), outcome.error().Message()); + } + } while (path_iter != remote_files.end()); + + return Status::OK(); +} + +Status OSSFileSystem::exists_impl(const Path& path, bool* res) const { + auto client = _client->get(); + CHECK_OSS_CLIENT(client); + + auto key = DORIS_TRY(get_oss_key(path)); + + auto outcome = client->HeadObject(_bucket, key); + if (outcome.isSuccess()) { + *res = true; + } else if (outcome.error().Code() == "NoSuchKey") { + *res = false; + } else { + return Status::IOError("failed to check existence of {}: {} - {}", key, + outcome.error().Code(), outcome.error().Message()); + } + return Status::OK(); +} + +Status OSSFileSystem::file_size_impl(const Path& file, int64_t* file_size) const { + auto key = DORIS_TRY(get_oss_key(file)); + *file_size = DORIS_TRY(_client->object_file_size(_bucket, key)); + return Status::OK(); +} + +Status OSSFileSystem::list_impl(const Path& dir, bool only_file, std::vector* files, + bool* exists) { + auto client = _client->get(); + CHECK_OSS_CLIENT(client); + + auto prefix = DORIS_TRY(get_oss_key(dir)); + if (!prefix.empty() && prefix.back() != '/') { + prefix.push_back('/'); + } + + *exists = false; + + bool is_truncated = true; + std::string marker; + + while (is_truncated) { + AlibabaCloud::OSS::ListObjectsRequest request(_bucket); + request.setPrefix(prefix); + request.setMaxKeys(1000); + + // Use delimiter for non-recursive listing to prevent memory exhaustion + if (only_file) { + request.setDelimiter("/"); + } + + if (!marker.empty()) { + request.setMarker(marker); + } + + auto outcome = client->ListObjects(request); + if (!outcome.isSuccess()) { + return Status::IOError("failed to list objects: {} - {}", outcome.error().Code(), + outcome.error().Message()); + } + + const auto& result = outcome.result(); + const auto& objects = result.ObjectSummarys(); + + if (!objects.empty()) { + *exists = true; + } + + for (const auto& obj : objects) { + FileInfo file_info; + file_info.file_name = obj.Key(); + // Remove prefix from file name + if (file_info.file_name.find(prefix) == 0) { + file_info.file_name = file_info.file_name.substr(prefix.length()); + } + + file_info.file_size = obj.Size(); + file_info.is_file = true; + + files->push_back(std::move(file_info)); + } + + is_truncated = result.IsTruncated(); + marker = result.NextMarker(); + } + + return Status::OK(); +} + +Status OSSFileSystem::rename_impl(const Path& orig_name, const Path& new_name) { + auto client = _client->get(); + CHECK_OSS_CLIENT(client); + + auto src_key = DORIS_TRY(get_oss_key(orig_name)); + auto dst_key = DORIS_TRY(get_oss_key(new_name)); + + // Get source file size to determine copy strategy + int64_t file_size = DORIS_TRY(_client->object_file_size(_bucket, src_key)); + + if (file_size < MULTIPART_COPY_THRESHOLD) { + // Simple copy for files < 1GB + AlibabaCloud::OSS::CopyObjectRequest copy_request(_bucket, dst_key); + copy_request.setCopySource(_bucket, src_key); + + auto copy_outcome = client->CopyObject(copy_request); + if (!copy_outcome.isSuccess()) { + return Status::IOError("failed to copy object from {} to {}: {} - {}", + full_oss_path(src_key), full_oss_path(dst_key), + copy_outcome.error().Code(), copy_outcome.error().Message()); + } + } else { + // Multipart copy for large files to avoid timeouts + auto init_outcome = client->InitiateMultipartUpload( + AlibabaCloud::OSS::InitiateMultipartUploadRequest(_bucket, dst_key)); + if (!init_outcome.isSuccess()) { + return Status::IOError("failed to initiate multipart upload: {} - {}", + init_outcome.error().Code(), init_outcome.error().Message()); + } + + std::string upload_id = init_outcome.result().UploadId(); + AlibabaCloud::OSS::PartList part_etags; + + // Calculate number of parts + int64_t part_count = (file_size + MULTIPART_COPY_PART_SIZE - 1) / MULTIPART_COPY_PART_SIZE; + + // Copy parts + for (int64_t i = 0; i < part_count; ++i) { + int64_t start_offset = i * MULTIPART_COPY_PART_SIZE; + int64_t end_offset = + std::min(start_offset + MULTIPART_COPY_PART_SIZE - 1, file_size - 1); + + AlibabaCloud::OSS::UploadPartCopyRequest part_request(_bucket, dst_key, _bucket, + src_key); + part_request.setUploadId(upload_id); + part_request.setPartNumber(static_cast(i + 1)); + part_request.setCopySourceRange(start_offset, end_offset); + + auto part_outcome = client->UploadPartCopy(part_request); + if (!part_outcome.isSuccess()) { + // Abort multipart upload on failure + client->AbortMultipartUpload(AlibabaCloud::OSS::AbortMultipartUploadRequest( + _bucket, dst_key, upload_id)); + return Status::IOError("failed to copy part {}: {} - {}", i + 1, + part_outcome.error().Code(), part_outcome.error().Message()); + } + + part_etags.push_back(AlibabaCloud::OSS::Part(static_cast(i + 1), + part_outcome.result().ETag())); + } + + // Complete multipart upload + AlibabaCloud::OSS::CompleteMultipartUploadRequest complete_request(_bucket, dst_key, + part_etags, upload_id); + auto complete_outcome = client->CompleteMultipartUpload(complete_request); + if (!complete_outcome.isSuccess()) { + client->AbortMultipartUpload( + AlibabaCloud::OSS::AbortMultipartUploadRequest(_bucket, dst_key, upload_id)); + return Status::IOError("failed to complete multipart upload: {} - {}", + complete_outcome.error().Code(), + complete_outcome.error().Message()); + } + } + + // Delete source object + auto delete_outcome = client->DeleteObject(_bucket, src_key); + if (!delete_outcome.isSuccess()) { + LOG(WARNING) << "Failed to delete source object after copy: " << src_key << " - " + << delete_outcome.error().Code() << ": " << delete_outcome.error().Message(); + // Don't fail rename if delete fails, copy succeeded + } + + return Status::OK(); +} + +Status OSSFileSystem::upload_impl(const Path& local_file, const Path& remote_file) { + auto client = _client->get(); + CHECK_OSS_CLIENT(client); + + auto key = DORIS_TRY(get_oss_key(remote_file)); + + auto outcome = client->PutObject(_bucket, key, local_file.native()); + if (!outcome.isSuccess()) { + return Status::IOError("failed to upload file to {}: {} - {}", full_oss_path(key), + outcome.error().Code(), outcome.error().Message()); + } + + return Status::OK(); +} + +Status OSSFileSystem::batch_upload_impl(const std::vector& local_files, + const std::vector& remote_files) { + if (local_files.size() != remote_files.size()) { + return Status::InvalidArgument("local_files and remote_files size mismatch"); + } + + for (size_t i = 0; i < local_files.size(); ++i) { + RETURN_IF_ERROR(upload_impl(local_files[i], remote_files[i])); + } + + return Status::OK(); +} + +Status OSSFileSystem::download_impl(const Path& remote_file, const Path& local_file) { + auto client = _client->get(); + CHECK_OSS_CLIENT(client); + + auto key = DORIS_TRY(get_oss_key(remote_file)); + + AlibabaCloud::OSS::GetObjectRequest request(_bucket, key); + + auto outcome = client->GetObject(request); + if (!outcome.isSuccess()) { + return Status::IOError("failed to download file from {}: {} - {}", full_oss_path(key), + outcome.error().Code(), outcome.error().Message()); + } + + // Use temp file + rename pattern to ensure atomicity and proper cleanup on failure. + std::string temp_file_path = + local_file.native() + ".tmp." + std::to_string(getpid()) + "." + + std::to_string(std::hash {}(std::this_thread::get_id())); + + // RAII wrapper to ensure temp file cleanup on failure + struct TempFileGuard { + std::string path; + bool success = false; + + ~TempFileGuard() { + if (!success && !path.empty()) { + // Clean up temp file on failure + std::error_code ec; + std::filesystem::remove(path, ec); + if (ec) { + LOG(WARNING) << "Failed to remove temp file " << path << ": " << ec.message(); + } + } + } + } temp_guard {temp_file_path}; + + // Write to temp file + std::ofstream out(temp_file_path, std::ios::binary); + if (!out) { + return Status::IOError("failed to open temp file for writing: {}", temp_file_path); + } + + auto& content_stream = outcome.result().Content(); + out << content_stream->rdbuf(); + + if (!out.good()) { + return Status::IOError("failed to write to temp file: {}", temp_file_path); + } + + out.close(); + if (!out) { + return Status::IOError("failed to close temp file: {}", temp_file_path); + } + + // Atomically rename temp file to target file + std::error_code ec; + std::filesystem::rename(temp_file_path, local_file.native(), ec); + if (ec) { + return Status::IOError("failed to rename temp file {} to {}: {}", temp_file_path, + local_file.native(), ec.message()); + } + + temp_guard.success = true; // Prevent cleanup of successfully renamed file + return Status::OK(); +} + +} // namespace doris::io diff --git a/be/src/io/fs/oss_file_system.h b/be/src/io/fs/oss_file_system.h new file mode 100644 index 00000000000000..fc1c6256709c52 --- /dev/null +++ b/be/src/io/fs/oss_file_system.h @@ -0,0 +1,140 @@ +// 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. + +#pragma once + +#include +#include +#include +#include +#include +#include + +#include "common/status.h" +#include "io/fs/file_reader_writer_fwd.h" +#include "io/fs/path.h" +#include "io/fs/remote_file_system.h" +#include "util/oss_util.h" + +// Forward declare OSS SDK types +namespace AlibabaCloud { +namespace OSS { +class OssClient; +} // namespace OSS +} // namespace AlibabaCloud + +namespace doris::io { + +// OSSClientHolder holds a shared OSS client and provides thread-safe access +// In runtime, AK and SK may be modified via vault updates, and the original OssClient will be replaced. +// The OSSFileReader cached by the Segment must hold a shared OSSClientHolder in order to +// access OSS data with latest credentials. +class OSSClientHolder { +public: + explicit OSSClientHolder(OSSClientConf conf); + ~OSSClientHolder(); + + Status init(); + + // Update OSS conf and reset client if `conf` is different. This method is threadsafe. + Status reset(const OSSClientConf& conf); + + std::shared_ptr get() const { + std::shared_lock lock(_mtx); + return _client; + } + + Result object_file_size(const std::string& bucket, const std::string& key) const; + + // For error msg + std::string full_oss_path(std::string_view bucket, std::string_view key) const; + + const OSSClientConf& oss_client_conf() { return _conf; } + +private: + mutable std::shared_mutex _mtx; + std::shared_ptr _client; + OSSClientConf _conf; +}; + +// File system for Alibaba Cloud OSS (Object Storage Service) +// When creating OSSFileSystem, all required info should be set in OSSConf, +// such as ak, sk, region, endpoint, bucket. +// And the root_path of OSSFileSystem is oss_conf.prefix. +// When using OSSFileSystem, it accepts 2 kinds of path: +// 1. Full path: oss://bucket/path/to/file.txt +// In this case, the root_path is not used. +// 2. only key: path/to/file.txt +// In this case, the final key will be "prefix + path/to/file.txt" +class OSSFileSystem final : public RemoteFileSystem { +public: + static Result> create(OSSConf oss_conf, std::string id); + + ~OSSFileSystem() override; + + const std::shared_ptr& client_holder() const { return _client; } + + const std::string& bucket() const { return _bucket; } + const std::string& prefix() const { return _prefix; } + +protected: + Status create_file_impl(const Path& file, FileWriterPtr* writer, + const FileWriterOptions* opts) override; + Status open_file_internal(const Path& file, FileReaderSPtr* reader, + const FileReaderOptions& opts) override; + Status create_directory_impl(const Path& dir, bool failed_if_exists = false) override; + Status delete_file_impl(const Path& file) override; + Status delete_directory_impl(const Path& dir) override; + Status batch_delete_impl(const std::vector& files) override; + Status exists_impl(const Path& path, bool* res) const override; + Status file_size_impl(const Path& file, int64_t* file_size) const override; + Status list_impl(const Path& dir, bool only_file, std::vector* files, + bool* exists) override; + Status rename_impl(const Path& orig_name, const Path& new_name) override; + + Status upload_impl(const Path& local_file, const Path& remote_file) override; + Status batch_upload_impl(const std::vector& local_files, + const std::vector& remote_files) override; + Status download_impl(const Path& remote_file, const Path& local_file) override; + + Status absolute_path(const Path& path, Path& abs_path) const override { + if (path.string().find("://") != std::string::npos) { + // the path is with schema, which means this is a full path like: + // oss://bucket/path/to/file.txt + // so no need to concat with prefix + abs_path = path; + } else { + // path with no schema + abs_path = _prefix / path; + } + return Status::OK(); + } + +private: + OSSFileSystem(OSSConf oss_conf, std::string id); + + Status init(); + + // For error msg + std::string full_oss_path(std::string_view key) const; + + std::string _bucket; + std::string _prefix; + std::shared_ptr _client; +}; + +} // namespace doris::io diff --git a/be/src/io/fs/oss_file_writer.cpp b/be/src/io/fs/oss_file_writer.cpp new file mode 100644 index 00000000000000..45738c42f595eb --- /dev/null +++ b/be/src/io/fs/oss_file_writer.cpp @@ -0,0 +1,510 @@ +// 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 "io/fs/oss_file_writer.h" + +#include +#include +#include +#include + +#include +#include + +#include "common/config.h" +#include "common/status.h" +#include "io/cache/file_cache_common.h" +#include "io/fs/file_writer.h" +#include "io/fs/path.h" +#include "io/fs/s3_file_bufferpool.h" +#include "runtime/exec_env.h" +#include "util/debug_points.h" + +namespace doris::io { + +// Existing metrics +bvar::Adder oss_file_writer_total("oss_file_writer_total_num"); +bvar::Adder oss_bytes_written_total("oss_file_writer_bytes_written"); +bvar::Adder oss_file_created_total("oss_file_writer_file_created"); +bvar::Adder oss_file_being_written("oss_file_writer_file_being_written"); + +// New async metrics +bvar::Adder oss_file_writer_async_close_queuing("oss_file_writer_async_close_queuing"); +bvar::Adder oss_file_writer_async_close_processing( + "oss_file_writer_async_close_processing"); + +OSSFileWriter::OSSFileWriter(std::shared_ptr client, std::string bucket, + std::string key, const FileWriterOptions* opts) + : _path(fmt::format("oss://{}/{}", bucket, key)), + _bucket(std::move(bucket)), + _key(std::move(key)), + _client(std::move(client)), + _buffer_size(config::s3_write_buffer_size), + _used_by_oss_committer(opts ? opts->used_by_s3_committer : false) { + oss_file_writer_total << 1; + oss_file_being_written << 1; + + _completed_parts.reserve(100); + + init_cache_builder(opts, _path); +} + +OSSFileWriter::~OSSFileWriter() { + // Wait for async close task or pending part uploads to complete before destruction. + if (_async_close_pack != nullptr) { + std::ignore = _async_close_pack->future.get(); + _async_close_pack = nullptr; + } else { + _wait_until_finish("~OSSFileWriter"); + } + + // Abort multipart upload if not completed + if (state() != State::CLOSED && !_upload_id.empty()) { + LOG(WARNING) << "OSSFileWriter destroyed without close(), aborting multipart upload: " + << _path.native() << " upload_id: " << _upload_id; + static_cast(_abort_multipart_upload()); + } + + if (state() == State::OPENED && !_failed) { + oss_bytes_written_total << _bytes_appended; + } + oss_file_being_written << -1; +} + +Status OSSFileWriter::appendv(const Slice* data, size_t data_cnt) { + if (state() != State::OPENED) { + return Status::InternalError("append to closed file: {}", _path.native()); + } + + for (size_t i = 0; i < data_cnt; i++) { + size_t data_size = data[i].size; + const char* data_ptr = data[i].data; + size_t pos = 0; + + while (pos < data_size) { + if (_failed) { + return _st; + } + + // Create new buffer if needed + if (!_pending_buf) { + RETURN_IF_ERROR(_build_upload_buffer()); + } + + size_t remaining = data_size - pos; + size_t buffer_remaining = _buffer_size - _pending_buf->get_size(); + size_t to_append = std::min(remaining, buffer_remaining); + + Slice s(data_ptr + pos, to_append); + RETURN_IF_ERROR(_pending_buf->append_data(s)); + + pos += to_append; + _bytes_appended += to_append; + + if (_pending_buf->get_size() == _buffer_size) { + // Create multipart upload on first buffer flush + if (_cur_part_num == 1) { + RETURN_IF_ERROR(_create_multipart_upload()); + } + + _cur_part_num++; + _countdown_event.add_count(); + RETURN_IF_ERROR(FileBuffer::submit(std::move(_pending_buf))); + _pending_buf = nullptr; + } + } + } + + return Status::OK(); +} + +Status OSSFileWriter::_build_upload_buffer() { + auto builder = FileBufferBuilder(); + builder.set_type(BufferType::UPLOAD) + .set_upload_callback([part_num = _cur_part_num, this](UploadFileBuffer& buf) { + _upload_one_part(part_num, buf); + }) + .set_file_offset(_bytes_appended) + .set_sync_after_complete_task([this](auto&& s) { + return _complete_part_task_callback(std::forward(s)); + }) + .set_is_cancelled([this]() { return _failed.load(); }); + + if (cache_builder() != nullptr) { + int64_t tablet_id = get_tablet_id(_path.native()).value_or(0); + builder.set_allocate_file_blocks_holder([builder = *cache_builder(), + offset = _bytes_appended, + tablet_id]() -> FileBlocksHolderPtr { + return builder.allocate_cache_holder(offset, config::s3_write_buffer_size, tablet_id); + }); + } + + RETURN_IF_ERROR(builder.build(&_pending_buf)); + return Status::OK(); +} + +void OSSFileWriter::_upload_one_part(int64_t part_num, UploadFileBuffer& buf) { + if (buf.is_cancelled()) { + LOG(INFO) << "OSS file " << _path.native() << " skip part " << part_num + << " because previous failure"; + return; + } + + // Debug point: Simulate upload failure + DBUG_EXECUTE_IF("OSSFileWriter::_upload_one_part.upload_error", { + auto fail_part = dp->param("fail_part_num", 0); + if (fail_part == 0 || fail_part == part_num) { + LOG(WARNING) << "Debug point: Simulating OSS upload failure for part " << part_num; + buf.set_status(Status::IOError("Debug OSS upload error for part {}", part_num)); + return; + } + }); + + // Debug point: Simulate slow upload + DBUG_EXECUTE_IF("OSSFileWriter::_upload_one_part.slow_upload", { + auto sleep_ms = dp->param("sleep_ms", 1000); + LOG(INFO) << "Debug point: Simulating slow OSS upload, sleeping " << sleep_ms << "ms"; + std::this_thread::sleep_for(std::chrono::milliseconds(sleep_ms)); + }); + + auto client = _client->get(); + if (nullptr == client) { + buf.set_status(Status::InternalError("OSS client not initialized")); + return; + } + + auto stream = buf.get_stream(); + if (!stream) { + buf.set_status(Status::InternalError("Failed to get stream from upload buffer for part {}", + part_num)); + return; + } + + AlibabaCloud::OSS::UploadPartRequest request(_bucket, _key, stream); + request.setPartNumber(static_cast(part_num)); + request.setUploadId(_upload_id); + request.setContentLength(buf.get_size()); + + auto outcome = client->UploadPart(request); + if (!outcome.isSuccess()) { + std::string err = fmt::format("OSS UploadPart {} failed: {} - {}", part_num, + outcome.error().Code(), outcome.error().Message()); + LOG(WARNING) << err << ", path: " << _path.native(); + buf.set_status(Status::IOError(err)); + return; + } + + oss_bytes_written_total << buf.get_size(); + + AlibabaCloud::OSS::Part part(static_cast(part_num), outcome.result().ETag()); + + { + std::lock_guard lock(_completed_lock); + _completed_parts.push_back(part); + } + + VLOG_DEBUG << "OSS UploadPart " << part_num << " completed: " << _path.native() + << " size: " << buf.get_size(); +} + +bool OSSFileWriter::_complete_part_task_callback(Status s) { + if (!s.ok()) { + std::unique_lock lck {_completed_lock}; + _failed = true; + _st = std::move(s); + LOG(WARNING) << "OSS async upload failed: " << _path.native() << " error: " << _st; + } + _countdown_event.signal(); + return !s.ok(); +} + +void OSSFileWriter::_wait_until_finish(std::string_view task_name) { + auto timeout_duration = config::s3_file_writer_log_interval_second; + auto msg = fmt::format("OSS file {} still has unfinished async tasks: {}", _path.native(), + task_name); + timespec current_time; + auto current_time_second = time(nullptr); + current_time.tv_sec = current_time_second + timeout_duration; + current_time.tv_nsec = 0; + + while (0 != _countdown_event.timed_wait(current_time)) { + current_time.tv_sec += timeout_duration; + LOG(WARNING) << msg; + } +} + +Status OSSFileWriter::_create_multipart_upload() { + auto client = _client->get(); + if (!client) { + return Status::InternalError("OSS client not initialized"); + } + + AlibabaCloud::OSS::InitiateMultipartUploadRequest request(_bucket, _key); + auto outcome = client->InitiateMultipartUpload(request); + + if (!outcome.isSuccess()) { + std::string err = fmt::format("OSS InitiateMultipartUpload failed: {} - {}", + outcome.error().Code(), outcome.error().Message()); + LOG(WARNING) << err << ", path: " << _path.native(); + return Status::IOError(err); + } + + _upload_id = outcome.result().UploadId(); + LOG(INFO) << "OSS multipart upload initiated: " << _path.native() + << " upload_id: " << _upload_id; + + return Status::OK(); +} + +Status OSSFileWriter::_complete_multipart_upload() { + auto client = _client->get(); + if (!client) { + return Status::InternalError("OSS client not initialized"); + } + + // Debug point: Simulate complete multipart failure + DBUG_EXECUTE_IF("OSSFileWriter::_complete_multipart_upload.complete_error", { + LOG(WARNING) << "Debug point: Simulating OSS complete multipart upload failure"; + _failed = true; + return Status::IOError("Debug OSS complete multipart upload error for path {}", + _path.native()); + }); + + // CRITICAL: Sort parts by part number (OSS requires ascending order) + std::sort(_completed_parts.begin(), _completed_parts.end(), + [](const AlibabaCloud::OSS::Part& a, const AlibabaCloud::OSS::Part& b) { + return a.PartNumber() < b.PartNumber(); + }); + + AlibabaCloud::OSS::CompleteMultipartUploadRequest request(_bucket, _key, _completed_parts, + _upload_id); + auto outcome = client->CompleteMultipartUpload(request); + + if (!outcome.isSuccess()) { + _failed = true; + std::string err = fmt::format("OSS CompleteMultipartUpload failed: {} - {}", + outcome.error().Code(), outcome.error().Message()); + LOG(WARNING) << err << ", path: " << _path.native(); + static_cast(_abort_multipart_upload()); + return Status::IOError(err); + } + + LOG(INFO) << "OSS multipart upload completed: " << _path.native() + << " parts: " << _completed_parts.size() << " bytes: " << _bytes_appended; + + oss_file_created_total << 1; + return Status::OK(); +} + +Status OSSFileWriter::_abort_multipart_upload() { + if (_upload_id.empty()) { + return Status::OK(); + } + + auto client = _client->get(); + if (!client) { + return Status::InternalError("OSS client not initialized"); + } + + AlibabaCloud::OSS::AbortMultipartUploadRequest request(_bucket, _key, _upload_id); + auto outcome = client->AbortMultipartUpload(request); + + if (!outcome.isSuccess()) { + std::string err = fmt::format("OSS AbortMultipartUpload failed: {} - {}", + outcome.error().Code(), outcome.error().Message()); + LOG(WARNING) << err << ", path: " << _path.native() << " upload_id: " << _upload_id; + return Status::IOError(err); + } + + LOG(INFO) << "OSS multipart upload aborted: " << _path.native() << " upload_id: " << _upload_id; + _upload_id.clear(); + + return Status::OK(); +} + +Status OSSFileWriter::_put_object(UploadFileBuffer& buf) { + auto client = _client->get(); + if (!client) { + return Status::InternalError("OSS client not initialized"); + } + + auto stream = buf.get_stream(); + if (!stream) { + return Status::InternalError("Failed to get stream from upload buffer"); + } + + AlibabaCloud::OSS::PutObjectRequest request(_bucket, _key, stream); + auto outcome = client->PutObject(request); + + if (!outcome.isSuccess()) { + _failed = true; + std::string err = fmt::format("OSS PutObject failed: {} - {}", outcome.error().Code(), + outcome.error().Message()); + LOG(WARNING) << err << ", path: " << _path.native(); + return Status::IOError(err); + } + + LOG(INFO) << "OSS PutObject completed: " << _path.native() << " size: " << buf.get_size(); + + oss_file_created_total << 1; + oss_bytes_written_total << buf.get_size(); + + return Status::OK(); +} + +Status OSSFileWriter::_set_upload_to_remote_less_than_buffer_size() { + auto* buf = dynamic_cast(_pending_buf.get()); + if (buf == nullptr) { + return Status::InternalError( + "Invalid buffer type in _set_upload_to_remote_less_than_buffer_size: expected " + "UploadFileBuffer"); + } + + if (_used_by_oss_committer) { + // Committer mode: use multipart (FE needs upload_id and part ETags) + buf->set_upload_to_remote([part_num = _cur_part_num, this](UploadFileBuffer& buf) { + _upload_one_part(part_num, buf); + }); + DCHECK(_cur_part_num == 1); + RETURN_IF_ERROR(_create_multipart_upload()); + } else { + // Normal mode: use PutObject for small files + buf->set_upload_to_remote([this](UploadFileBuffer& buf) { + Status st = _put_object(buf); + if (!st.ok()) { + buf.set_status(st); + } + }); + } + + return Status::OK(); +} + +Status OSSFileWriter::close(bool non_block) { + std::lock_guard lock(_close_lock); + + if (state() == State::CLOSED) { + return _st; + } + + if (state() == State::ASYNC_CLOSING) { + if (non_block) { + return Status::InternalError("Don't submit async close multiple times"); + } + CHECK(_async_close_pack != nullptr); + _st = _async_close_pack->future.get(); + _async_close_pack = nullptr; + _state = State::CLOSED; + return _st; + } + + if (non_block) { + _state = State::ASYNC_CLOSING; + _async_close_pack = std::make_unique(); + _async_close_pack->future = _async_close_pack->promise.get_future(); + + oss_file_writer_async_close_queuing << 1; + + return ExecEnv::GetInstance()->non_block_close_thread_pool()->submit_func([this]() { + oss_file_writer_async_close_queuing << -1; + oss_file_writer_async_close_processing << 1; + _async_close_pack->promise.set_value(_close_impl()); + oss_file_writer_async_close_processing << -1; + }); + } + + _st = _close_impl(); + _state = State::CLOSED; + return _st; +} + +Status OSSFileWriter::_close_impl() { + // Wait for any pending async operations to complete + _wait_until_finish("_close_impl"); + + if (_failed) { + if (!_upload_id.empty()) { + static_cast(_abort_multipart_upload()); + } + return _st; + } + + if (_bytes_appended == 0) { + DCHECK_EQ(_cur_part_num, 1); + // Create an empty buffer for PutObject + RETURN_IF_ERROR(_build_upload_buffer()); + + if (!_used_by_oss_committer) { + auto* pending_buf = dynamic_cast(_pending_buf.get()); + DCHECK(pending_buf != nullptr); + pending_buf->set_upload_to_remote([this](UploadFileBuffer& buf) { + Status st = _put_object(buf); + if (!st.ok()) { + buf.set_status(st); + } + }); + } else { + RETURN_IF_ERROR(_create_multipart_upload()); + } + } + + if (_cur_part_num == 1 && _pending_buf) { + RETURN_IF_ERROR(_set_upload_to_remote_less_than_buffer_size()); + } + + if (_pending_buf && _pending_buf->get_size() > 0) { + _countdown_event.add_count(); + Status st = FileBuffer::submit(std::move(_pending_buf)); + _pending_buf = nullptr; + if (!st.ok()) { + if (!_upload_id.empty()) { + static_cast(_abort_multipart_upload()); + } + return st; + } + + _wait_until_finish("last_part"); + + if (_failed) { + if (!_upload_id.empty()) { + static_cast(_abort_multipart_upload()); + } + return _st; + } + } + + if (_cur_part_num == 1) { + _wait_until_finish("PutObject or single part"); + return _st; + } + + _wait_until_finish("Complete multipart"); + + if (_used_by_oss_committer) { + // OSS committer mode: FE will complete multipart upload + oss_file_created_total << 1; + LOG(INFO) << "OSS multipart upload parts completed (FE will finish): " << _path.native() + << " parts: " << _completed_parts.size() << " bytes: " << _bytes_appended + << " upload_id: " << _upload_id; + return Status::OK(); + } + + // TODO: add check_after_upload like S3 (off by default via config) + return _complete_multipart_upload(); +} + +} // namespace doris::io diff --git a/be/src/io/fs/oss_file_writer.h b/be/src/io/fs/oss_file_writer.h new file mode 100644 index 00000000000000..76119c1b06f370 --- /dev/null +++ b/be/src/io/fs/oss_file_writer.h @@ -0,0 +1,98 @@ +// 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. + +#pragma once + +#include +#include + +#include +#include +#include +#include +#include +#include + +#include "common/status.h" +#include "io/fs/file_writer.h" +#include "io/fs/oss_file_system.h" +#include "io/fs/path.h" +#include "io/fs/s3_file_bufferpool.h" + +namespace doris::io { + +class OSSFileWriter final : public FileWriter { +public: + OSSFileWriter(std::shared_ptr client, std::string bucket, std::string key, + const FileWriterOptions* opts); + ~OSSFileWriter() override; + + Status appendv(const Slice* data, size_t data_cnt) override; + + const Path& path() const override { return _path; } + size_t bytes_appended() const override { return _bytes_appended; } + State state() const override { return _state; } + + Status close(bool non_block = false) override; + +private: + Status _close_impl(); + Status _create_multipart_upload(); + + void _upload_one_part(int64_t part_num, UploadFileBuffer& buf); + + Status _complete_multipart_upload(); + Status _abort_multipart_upload(); + Status _put_object(UploadFileBuffer& buf); + + Status _build_upload_buffer(); + + bool _complete_part_task_callback(Status s); + + Status _set_upload_to_remote_less_than_buffer_size(); + + void _wait_until_finish(std::string_view task_name); + + Path _path; + std::string _bucket; + std::string _key; + std::shared_ptr _client; + + std::string _upload_id; + int64_t _cur_part_num = 1; + std::vector _completed_parts; + std::mutex _completed_lock; + + std::shared_ptr _pending_buf; + size_t _buffer_size; + size_t _bytes_appended = 0; + + bthread::CountdownEvent _countdown_event {0}; + + std::unique_ptr _async_close_pack; + + std::mutex _close_lock; + + // OSS committer mode: FE completes multipart upload + bool _used_by_oss_committer; + + State _state {State::OPENED}; + std::atomic_bool _failed {false}; + Status _st; +}; + +} // namespace doris::io diff --git a/be/src/io/fs/oss_v2_obj_storage_client.cpp b/be/src/io/fs/oss_v2_obj_storage_client.cpp new file mode 100644 index 00000000000000..9a2b51c15bded0 --- /dev/null +++ b/be/src/io/fs/oss_v2_obj_storage_client.cpp @@ -0,0 +1,444 @@ +// 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 "io/fs/oss_v2_obj_storage_client.h" + +#ifdef USE_OSS + +#include +#include +#include + +#include +#include +#include +#include +#include + +#include "common/config.h" +#include "common/status.h" +#include "io/fs/file_system.h" +#include "io/fs/obj_storage_client.h" +#include "util/bvar_helper.h" +#include "util/oss_util.h" +#include "util/s3_util.h" + +namespace doris::io { + +namespace { + +// Read-only iostream wrapper around an existing buffer (no AWS SDK dependency). +class OSSStringViewStream : public std::streambuf, public std::iostream { +public: + OSSStringViewStream(const char* data, size_t size) : std::streambuf(), std::iostream(this) { + // const_cast is safe: read-only streambuf (get-area only, no put-area). + char* p = const_cast(data); + setg(p, p, p + size); + } +}; + +// Rate limiting helpers — reuse shared S3RateLimiter infrastructure (parallel to s3_obj_storage_client.cpp). +inline AlibabaCloud::OSS::OssError oss_rate_limit_error() { + return {"ExceedsRateLimit", "request rate exceeds configured limit"}; +} + +template +auto oss_rate_limit(doris::S3RateLimitType op, Func callback) -> decltype(callback()) { + using T = decltype(callback()); + if (!doris::config::enable_s3_rate_limiter) { + return callback(); + } + auto sleep_duration = doris::S3ClientFactory::instance().rate_limiter(op)->add(1); + if (sleep_duration < 0) { + return T(oss_rate_limit_error()); + } + return callback(); +} + +template +auto oss_get_rate_limit(Func callback) -> decltype(callback()) { + return oss_rate_limit(doris::S3RateLimitType::GET, std::move(callback)); +} + +template +auto oss_put_rate_limit(Func callback) -> decltype(callback()) { + return oss_rate_limit(doris::S3RateLimitType::PUT, std::move(callback)); +} + +// Build a failed ObjectStorageResponse from an OSS error +inline ObjectStorageResponse oss_error_response(const std::string& op, const std::string& bucket, + const std::string& key, const std::string& code, + const std::string& msg) { + ObjectStorageResponse resp; + resp.status.code = -1; + resp.status.msg = fmt::format("OSS {} failed [{}/{}]: {} - {}", op, bucket, key, code, msg); + return resp; +} + +// OssError has no HTTP status code; Code() string is the only discriminator. +inline ObjectStorageHeadResponse oss_head_error_response(const std::string& bucket, + const std::string& key, + const std::string& code, + const std::string& msg) { + if (code == "NoSuchKey") { + return {.resp = {convert_to_obj_response(Status::Error(""))}}; + } + ObjectStorageHeadResponse r; + r.resp.status.code = -1; + r.resp.status.msg = + fmt::format("OSS HeadObject failed [{}/{}]: {} - {}", bucket, key, code, msg); + return r; +} + +} // namespace + +OSSv2ObjStorageClient::OSSv2ObjStorageClient(std::shared_ptr client, + std::string bucket) + : _client(std::move(client)), _bucket(std::move(bucket)) {} + +// ---------------------------------------------------------------------------- +// create_multipart_upload +// ---------------------------------------------------------------------------- +ObjectStorageUploadResponse OSSv2ObjStorageClient::create_multipart_upload( + const ObjectStoragePathOptions& opts) { + ObjectStorageUploadResponse resp; + const std::string& bucket = resolve_bucket(opts); + + AlibabaCloud::OSS::InitiateMultipartUploadRequest request(bucket, opts.key); + SCOPED_BVAR_LATENCY(oss_bvar::oss_multi_part_upload_latency); + auto outcome = oss_put_rate_limit([&]() { return _client->InitiateMultipartUpload(request); }); + + if (!outcome.isSuccess()) { + resp.resp = oss_error_response("InitiateMultipartUpload", bucket, opts.key, + outcome.error().Code(), outcome.error().Message()); + LOG(WARNING) << resp.resp.status.msg; + return resp; + } + + resp.upload_id = outcome.result().UploadId(); + VLOG(2) << "OSS InitiateMultipartUpload: " << bucket << "/" << opts.key + << " upload_id=" << *resp.upload_id; + return resp; +} + +// ---------------------------------------------------------------------------- +// put_object +// ---------------------------------------------------------------------------- +ObjectStorageResponse OSSv2ObjStorageClient::put_object(const ObjectStoragePathOptions& opts, + std::string_view stream) { + const std::string& bucket = resolve_bucket(opts); + auto content = std::make_shared(stream.data(), stream.size()); + + AlibabaCloud::OSS::PutObjectRequest request(bucket, opts.key, content); + request.MetaData().setContentLength(static_cast(stream.size())); + SCOPED_BVAR_LATENCY(oss_bvar::oss_put_latency); + auto outcome = oss_put_rate_limit([&]() { return _client->PutObject(request); }); + + if (!outcome.isSuccess()) { + return oss_error_response("PutObject", bucket, opts.key, outcome.error().Code(), + outcome.error().Message()); + } + + VLOG(2) << "OSS PutObject: " << bucket << "/" << opts.key << " size=" << stream.size(); + return ObjectStorageResponse::OK(); +} + +// ---------------------------------------------------------------------------- +// upload_part +// ---------------------------------------------------------------------------- +ObjectStorageUploadResponse OSSv2ObjStorageClient::upload_part(const ObjectStoragePathOptions& opts, + std::string_view stream, + int part_num) { + ObjectStorageUploadResponse resp; + const std::string& bucket = resolve_bucket(opts); + + if (!opts.upload_id.has_value()) { + resp.resp.status.code = -1; + resp.resp.status.msg = "upload_id is required for OSS UploadPart"; + return resp; + } + + auto content = std::make_shared(stream.data(), stream.size()); + + AlibabaCloud::OSS::UploadPartRequest request(bucket, opts.key, content); + request.setPartNumber(part_num); + request.setUploadId(*opts.upload_id); + request.setContentLength(static_cast(stream.size())); + + SCOPED_BVAR_LATENCY(oss_bvar::oss_multi_part_upload_latency); + auto outcome = oss_put_rate_limit([&]() { return _client->UploadPart(request); }); + + if (!outcome.isSuccess()) { + resp.resp = oss_error_response("UploadPart", bucket, opts.key, outcome.error().Code(), + outcome.error().Message()); + LOG(WARNING) << resp.resp.status.msg; + return resp; + } + + resp.etag = outcome.result().ETag(); + VLOG(2) << "OSS UploadPart " << part_num << ": " << bucket << "/" << opts.key + << " etag=" << *resp.etag; + return resp; +} + +// ---------------------------------------------------------------------------- +// complete_multipart_upload +// ---------------------------------------------------------------------------- +ObjectStorageResponse OSSv2ObjStorageClient::complete_multipart_upload( + const ObjectStoragePathOptions& opts, + const std::vector& completed_parts) { + const std::string& bucket = resolve_bucket(opts); + + if (!opts.upload_id.has_value()) { + return oss_error_response("CompleteMultipartUpload", bucket, opts.key, "InvalidArgument", + "upload_id is required"); + } + + AlibabaCloud::OSS::PartList part_list; + part_list.reserve(completed_parts.size()); + for (const auto& part : completed_parts) { + part_list.emplace_back(part.part_num, part.etag); + } + // Server requires ascending part-number order; SDK does not enforce this locally. + std::sort(part_list.begin(), part_list.end(), + [](const AlibabaCloud::OSS::Part& a, const AlibabaCloud::OSS::Part& b) { + return a.PartNumber() < b.PartNumber(); + }); + + AlibabaCloud::OSS::CompleteMultipartUploadRequest request(bucket, opts.key, part_list, + *opts.upload_id); + SCOPED_BVAR_LATENCY(oss_bvar::oss_multi_part_upload_latency); + auto outcome = oss_put_rate_limit([&]() { return _client->CompleteMultipartUpload(request); }); + + if (!outcome.isSuccess()) { + return oss_error_response("CompleteMultipartUpload", bucket, opts.key, + outcome.error().Code(), outcome.error().Message()); + } + + VLOG(1) << "OSS CompleteMultipartUpload: " << bucket << "/" << opts.key + << " parts=" << completed_parts.size(); + return ObjectStorageResponse::OK(); +} + +// ---------------------------------------------------------------------------- +// head_object +// ---------------------------------------------------------------------------- +ObjectStorageHeadResponse OSSv2ObjStorageClient::head_object(const ObjectStoragePathOptions& opts) { + ObjectStorageHeadResponse resp; + const std::string& bucket = resolve_bucket(opts); + + SCOPED_BVAR_LATENCY(oss_bvar::oss_head_latency); + auto outcome = oss_get_rate_limit([&]() { return _client->HeadObject(bucket, opts.key); }); + + if (!outcome.isSuccess()) { + return oss_head_error_response(bucket, opts.key, outcome.error().Code(), + outcome.error().Message()); + } + + resp.file_size = outcome.result().ContentLength(); + return resp; +} + +// ---------------------------------------------------------------------------- +// get_object +// ---------------------------------------------------------------------------- +ObjectStorageResponse OSSv2ObjStorageClient::get_object(const ObjectStoragePathOptions& opts, + void* buffer, size_t offset, + size_t bytes_read, size_t* size_return) { + const std::string& bucket = resolve_bucket(opts); + + if (bytes_read == 0) { + *size_return = 0; + return ObjectStorageResponse::OK(); + } + + AlibabaCloud::OSS::GetObjectRequest request(bucket, opts.key); + request.setRange(static_cast(offset), static_cast(offset + bytes_read - 1)); + + SCOPED_BVAR_LATENCY(oss_bvar::oss_get_latency); + auto outcome = oss_get_rate_limit([&]() { return _client->GetObject(request); }); + + if (!outcome.isSuccess()) { + return oss_error_response("GetObject", bucket, opts.key, outcome.error().Code(), + outcome.error().Message()); + } + + auto& content = outcome.result().Content(); + content->read(static_cast(buffer), static_cast(bytes_read)); + *size_return = static_cast(content->gcount()); + return ObjectStorageResponse::OK(); +} + +// ---------------------------------------------------------------------------- +// list_objects +// ---------------------------------------------------------------------------- +ObjectStorageResponse OSSv2ObjStorageClient::list_objects(const ObjectStoragePathOptions& opts, + std::vector* files) { + const std::string& bucket = resolve_bucket(opts); + bool is_truncated = true; + std::string marker; + + while (is_truncated) { + AlibabaCloud::OSS::ListObjectsRequest request(bucket); + request.setPrefix(opts.prefix); + request.setMaxKeys(1000); + if (!marker.empty()) { + request.setMarker(marker); + } + + SCOPED_BVAR_LATENCY(oss_bvar::oss_list_latency); + auto outcome = oss_get_rate_limit([&]() { return _client->ListObjects(request); }); + if (!outcome.isSuccess()) { + return oss_error_response("ListObjects", bucket, opts.prefix, outcome.error().Code(), + outcome.error().Message()); + } + + const auto& result = outcome.result(); + for (const auto& obj : result.ObjectSummarys()) { + FileInfo fi; + fi.file_name = obj.Key(); + fi.file_size = obj.Size(); + fi.is_file = true; + files->push_back(std::move(fi)); + } + + is_truncated = result.IsTruncated(); + if (is_truncated) { + // OSS v1 ListObjects may omit NextMarker without a delimiter; fall back to last key. + const auto& summaries = result.ObjectSummarys(); + marker = result.NextMarker(); + if (marker.empty() && !summaries.empty()) { + marker = summaries.back().Key(); + } + if (marker.empty()) { + return oss_error_response("ListObjects", bucket, opts.prefix, "InvalidResponse", + "IsTruncated=true but no continuation marker available"); + } + } + } + + return ObjectStorageResponse::OK(); +} + +// ---------------------------------------------------------------------------- +// delete_objects (batch, up to 1000 per OSS call) +// ---------------------------------------------------------------------------- +ObjectStorageResponse OSSv2ObjStorageClient::delete_objects(const ObjectStoragePathOptions& opts, + std::vector objs) { + if (objs.empty()) { + return ObjectStorageResponse::OK(); + } + + const std::string& bucket = resolve_bucket(opts); + constexpr size_t OSS_BATCH_DELETE_LIMIT = 1000; + + for (size_t i = 0; i < objs.size(); i += OSS_BATCH_DELETE_LIMIT) { + AlibabaCloud::OSS::DeletedKeyList keys; + size_t end = std::min(i + OSS_BATCH_DELETE_LIMIT, objs.size()); + for (size_t j = i; j < end; ++j) { + keys.push_back(objs[j]); + } + + AlibabaCloud::OSS::DeleteObjectsRequest request(bucket); + request.setKeyList(keys); + // Verbose mode (setQuiet=false) makes OSS return the list of successfully deleted keys, + // enabling partial-failure detection. Quiet mode (default) returns only errors. + request.setQuiet(false); + SCOPED_BVAR_LATENCY(oss_bvar::oss_delete_objects_latency); + auto outcome = oss_put_rate_limit([&]() { return _client->DeleteObjects(request); }); + + if (!outcome.isSuccess()) { + return oss_error_response("DeleteObjects", bucket, opts.prefix, outcome.error().Code(), + outcome.error().Message()); + } + // keyList() is populated in Verbose mode with the keys that were successfully deleted. + const auto& confirmed = outcome.result().keyList(); + if (confirmed.size() < keys.size()) { + LOG(WARNING) << "OSS BatchDelete partial failure: sent " << keys.size() + << " keys, only " << confirmed.size() << " confirmed deleted in bucket " + << bucket << "; some objects may still exist"; + } + } + + return ObjectStorageResponse::OK(); +} + +// ---------------------------------------------------------------------------- +// delete_object (single object, idempotent on NoSuchKey) +// ---------------------------------------------------------------------------- +ObjectStorageResponse OSSv2ObjStorageClient::delete_object(const ObjectStoragePathOptions& opts) { + const std::string& bucket = resolve_bucket(opts); + SCOPED_BVAR_LATENCY(oss_bvar::oss_delete_object_latency); + auto outcome = oss_put_rate_limit([&]() { return _client->DeleteObject(bucket, opts.key); }); + + if (!outcome.isSuccess()) { + const std::string& code = outcome.error().Code(); + if (code == "NoSuchKey") { + return ObjectStorageResponse::OK(); // idempotent + } + return oss_error_response("DeleteObject", bucket, opts.key, code, + outcome.error().Message()); + } + + return ObjectStorageResponse::OK(); +} + +// ---------------------------------------------------------------------------- +// delete_objects_recursively (list all under prefix then batch delete) +// ---------------------------------------------------------------------------- +ObjectStorageResponse OSSv2ObjStorageClient::delete_objects_recursively( + const ObjectStoragePathOptions& opts) { + std::vector files; + if (auto resp = list_objects(opts, &files); resp.status.code != 0) { + return resp; + } + std::vector keys; + keys.reserve(files.size()); + for (auto& fi : files) { + keys.push_back(std::move(fi.file_name)); + } + return delete_objects(opts, std::move(keys)); +} + +// ---------------------------------------------------------------------------- +// generate_presigned_url +// ---------------------------------------------------------------------------- +std::string OSSv2ObjStorageClient::generate_presigned_url(const ObjectStoragePathOptions& opts, + int64_t expiration_secs, + const S3ClientConf& /*conf*/) { + // `conf` unused: OSS SDK signs with credentials embedded at client construction time. + // Unlike S3/Azure, GeneratePresignedUrl does not accept a per-call credential override. + const std::string& bucket = resolve_bucket(opts); + time_t expiration_time = std::time(nullptr) + static_cast(expiration_secs); + + AlibabaCloud::OSS::GeneratePresignedUrlRequest request(bucket, opts.key, + AlibabaCloud::OSS::Http::Get); + request.setExpires(expiration_time); + + auto outcome = _client->GeneratePresignedUrl(request); + if (!outcome.isSuccess()) { + LOG(WARNING) << "OSS GeneratePresignedUrl failed [" << bucket << "/" << opts.key + << "]: " << outcome.error().Code() << " - " << outcome.error().Message(); + return ""; + } + + return outcome.result(); +} + +} // namespace doris::io + +#endif // USE_OSS diff --git a/be/src/io/fs/oss_v2_obj_storage_client.h b/be/src/io/fs/oss_v2_obj_storage_client.h new file mode 100644 index 00000000000000..838d017f59b265 --- /dev/null +++ b/be/src/io/fs/oss_v2_obj_storage_client.h @@ -0,0 +1,93 @@ +// 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. + +#pragma once + +#ifdef USE_OSS + +#include +#include +#include + +#include "io/fs/obj_storage_client.h" + +// Forward declare OSS SDK types to avoid heavy header in .h +namespace AlibabaCloud { +namespace OSS { +class OssClient; +} // namespace OSS +} // namespace AlibabaCloud + +namespace doris::io { + +// OSSv2ObjStorageClient implements ObjStorageClient using the native Alibaba Cloud OSS SDK. +// +// This is used for all non-cloud-vault OSS paths where previously the S3-compatible AWS SDK +// was used (cold/tiered storage, external tables, load/export, CREATE RESOURCE OSS). +// It is plugged into S3ClientFactory::create() when provider == ObjStorageType::OSS. +// +// Cloud vault OSS continues to use OSSFileSystem → OSSClientHolder directly (unchanged). +class OSSv2ObjStorageClient final : public ObjStorageClient { +public: + OSSv2ObjStorageClient(std::shared_ptr client, std::string bucket); + ~OSSv2ObjStorageClient() override = default; + + ObjectStorageUploadResponse create_multipart_upload( + const ObjectStoragePathOptions& opts) override; + + ObjectStorageResponse put_object(const ObjectStoragePathOptions& opts, + std::string_view stream) override; + + ObjectStorageUploadResponse upload_part(const ObjectStoragePathOptions& opts, + std::string_view stream, int part_num) override; + + ObjectStorageResponse complete_multipart_upload( + const ObjectStoragePathOptions& opts, + const std::vector& completed_parts) override; + + ObjectStorageHeadResponse head_object(const ObjectStoragePathOptions& opts) override; + + ObjectStorageResponse get_object(const ObjectStoragePathOptions& opts, void* buffer, + size_t offset, size_t bytes_read, + size_t* size_return) override; + + ObjectStorageResponse list_objects(const ObjectStoragePathOptions& opts, + std::vector* files) override; + + ObjectStorageResponse delete_objects(const ObjectStoragePathOptions& opts, + std::vector objs) override; + + ObjectStorageResponse delete_object(const ObjectStoragePathOptions& opts) override; + + ObjectStorageResponse delete_objects_recursively(const ObjectStoragePathOptions& opts) override; + + std::string generate_presigned_url(const ObjectStoragePathOptions& opts, + int64_t expiration_secs, const S3ClientConf& conf) override; + +private: + // Returns bucket: opts.bucket if set, else falls back to _bucket + const std::string& resolve_bucket(const ObjectStoragePathOptions& opts) const { + return opts.bucket.empty() ? _bucket : opts.bucket; + } + + std::shared_ptr _client; + std::string _bucket; +}; + +} // namespace doris::io + +#endif // USE_OSS diff --git a/be/src/util/CMakeLists.txt b/be/src/util/CMakeLists.txt index f23c3ad865c89b..6a11a4e3069fa2 100644 --- a/be/src/util/CMakeLists.txt +++ b/be/src/util/CMakeLists.txt @@ -31,6 +31,10 @@ endif() list(REMOVE_ITEM UTIL_FILES ${CMAKE_CURRENT_SOURCE_DIR}/libjvm_loader.cpp) +if(BUILD_OSS STREQUAL "OFF") + list(REMOVE_ITEM UTIL_FILES ${CMAKE_CURRENT_SOURCE_DIR}/oss_util.cpp) +endif() + if (ENABLE_TLS) list(REMOVE_ITEM UTIL_FILES ${CMAKE_CURRENT_SOURCE_DIR}/oss/client_connection_provider.cpp) list(APPEND UTIL_FILES ${CMAKE_CURRENT_SOURCE_DIR}/../${TLS_MODULE_DIR}/client_connection_provider.cpp) diff --git a/be/src/util/oss_util.cpp b/be/src/util/oss_util.cpp new file mode 100644 index 00000000000000..f071c634b39f86 --- /dev/null +++ b/be/src/util/oss_util.cpp @@ -0,0 +1,247 @@ +// 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 "util/oss_util.h" + +#include +#include +#include + +#include "common/config.h" +#include "common/logging.h" +#include "cpp/oss_credential_provider.h" +#include "util/s3_util.h" +#include "util/string_util.h" + +namespace doris { + +namespace oss_bvar { +bvar::LatencyRecorder oss_get_latency("oss_get"); +bvar::LatencyRecorder oss_put_latency("oss_put"); +bvar::LatencyRecorder oss_delete_object_latency("oss_delete_object"); +bvar::LatencyRecorder oss_delete_objects_latency("oss_delete_objects"); +bvar::LatencyRecorder oss_head_latency("oss_head"); +bvar::LatencyRecorder oss_list_latency("oss_list"); +bvar::LatencyRecorder oss_multi_part_upload_latency("oss_multi_part_upload"); +} // namespace oss_bvar + +OSSConf OSSConf::get_oss_conf(const cloud::ObjectStoreInfoPB& obj_info) { + OSSConf conf; + conf.bucket = obj_info.bucket(); + conf.prefix = obj_info.prefix(); + + conf.client_conf.endpoint = normalize_oss_endpoint(obj_info.endpoint()); + conf.client_conf.region = obj_info.region(); + conf.client_conf.bucket = obj_info.bucket(); + conf.client_conf.role_arn = obj_info.role_arn(); + conf.client_conf.external_id = obj_info.external_id(); + + if (obj_info.has_cred_provider_type()) { + switch (obj_info.cred_provider_type()) { + case cloud::CredProviderTypePB::DEFAULT: + conf.client_conf.cred_provider_type = OSSCredProviderType::DEFAULT; + VLOG(2) << "Using OSS DEFAULT credential provider"; + break; + case cloud::CredProviderTypePB::INSTANCE_PROFILE: + conf.client_conf.cred_provider_type = OSSCredProviderType::INSTANCE_PROFILE; + VLOG(2) << "Using OSS INSTANCE_PROFILE credential provider"; + break; + case cloud::CredProviderTypePB::SIMPLE: + conf.client_conf.cred_provider_type = OSSCredProviderType::SIMPLE; + conf.client_conf.ak = obj_info.ak(); + conf.client_conf.sk = obj_info.sk(); + VLOG(2) << "Using OSS SIMPLE credential provider"; + break; + default: + conf.client_conf.cred_provider_type = OSSCredProviderType::DEFAULT; + VLOG(2) << "Unknown credential provider type, defaulting to DEFAULT"; + break; + } + } else { + if (!obj_info.ak().empty() && !obj_info.sk().empty()) { + conf.client_conf.cred_provider_type = OSSCredProviderType::SIMPLE; + conf.client_conf.ak = obj_info.ak(); + conf.client_conf.sk = obj_info.sk(); + VLOG(2) << "Using OSS SIMPLE credential provider (from AK/SK)"; + } else { + conf.client_conf.cred_provider_type = OSSCredProviderType::DEFAULT; + VLOG(2) << "No AK/SK provided, using OSS DEFAULT credential provider"; + } + } + + if (!conf.client_conf.role_arn.empty()) { + VLOG(2) << "OSS AssumeRole enabled with role_arn: " << conf.client_conf.role_arn; + } + + return conf; +} + +OSSClientFactory::OSSClientFactory() { + static std::once_flag init_flag; + std::call_once(init_flag, []() { + AlibabaCloud::OSS::InitializeSdk(); + LOG(INFO) << "Alibaba Cloud OSS SDK initialized"; + }); + _ca_cert_file_path = get_valid_ca_cert_path(doris::split(config::ca_cert_file_paths, ";")); +} + +OSSClientFactory::~OSSClientFactory() { + AlibabaCloud::OSS::ShutdownSdk(); + LOG(INFO) << "Alibaba Cloud OSS SDK shut down"; +} + +OSSClientFactory& OSSClientFactory::instance() { + static OSSClientFactory instance; + return instance; +} + +std::shared_ptr OSSClientFactory::create( + const OSSClientConf& oss_conf) { + if (oss_conf.endpoint.empty() || oss_conf.region.empty() || oss_conf.bucket.empty()) { + LOG(ERROR) << "Invalid OSS conf: endpoint, region and bucket are required"; + return nullptr; + } + // SIMPLE + non-empty role_arn is valid: role_arn takes priority in the if-chain below + // (STS path), using AK/SK as the base credential. Only reject SIMPLE with no AK/SK and no role_arn. + if (oss_conf.cred_provider_type == OSSCredProviderType::SIMPLE && oss_conf.role_arn.empty() && + (oss_conf.ak.empty() || oss_conf.sk.empty())) { + LOG(ERROR) << "Invalid OSS conf: ak and sk required for SIMPLE credential provider"; + return nullptr; + } + + uint64_t hash = oss_conf.get_hash(); + + { + std::lock_guard lock(_lock); + auto it = _cache.find(hash); + if (it != _cache.end()) { + VLOG(2) << "Reusing cached OSS client for endpoint: " << oss_conf.endpoint; + return it->second; + } + } + // TOCTOU: concurrent calls with the same hash may both miss the cache and create duplicate + // clients. Accepted pattern matching S3ClientFactory; last writer wins under the lock below. + + AlibabaCloud::OSS::ClientConfiguration oss_client_config; + oss_client_config.maxConnections = oss_conf.max_connections; + oss_client_config.requestTimeoutMs = oss_conf.request_timeout_ms; + oss_client_config.connectTimeoutMs = oss_conf.connect_timeout_ms; + + if (_ca_cert_file_path.empty()) { + _ca_cert_file_path = get_valid_ca_cert_path(doris::split(config::ca_cert_file_paths, ";")); + } + if (!_ca_cert_file_path.empty()) { + oss_client_config.caFile = _ca_cert_file_path; + } + + std::shared_ptr client; + + try { + if (!oss_conf.role_arn.empty()) { + std::shared_ptr sts_provider; + { + std::lock_guard lock(_lock); + auto it = _sts_credential_providers.find(hash); + if (it != _sts_credential_providers.end()) { + sts_provider = it->second; + } else { + std::string region = oss_conf.region.empty() ? "cn-hangzhou" : oss_conf.region; + sts_provider = std::make_shared( + oss_conf.role_arn, region, oss_conf.external_id, _ca_cert_file_path, + oss_conf.sts_endpoint); + _sts_credential_providers[hash] = sts_provider; + } + } + client = std::make_shared( + oss_conf.endpoint, + std::static_pointer_cast(sts_provider), + oss_client_config); + LOG(INFO) << "OSS client created with AssumeRole, endpoint=" << oss_conf.endpoint + << ", role_arn=" << oss_conf.role_arn; + } else if (oss_conf.cred_provider_type == OSSCredProviderType::INSTANCE_PROFILE) { + std::shared_ptr provider; + { + std::lock_guard lock(_lock); + auto it = _ecs_credential_providers.find(hash); + if (it != _ecs_credential_providers.end()) { + provider = it->second; + } else { + provider = std::make_shared(); + _ecs_credential_providers[hash] = provider; + } + } + client = std::make_shared( + oss_conf.endpoint, + std::static_pointer_cast(provider), + oss_client_config); + LOG(INFO) << "OSS client created with INSTANCE_PROFILE, endpoint=" << oss_conf.endpoint; + } else if (oss_conf.cred_provider_type == OSSCredProviderType::DEFAULT) { + std::shared_ptr provider; + { + std::lock_guard lock(_lock); + auto it = _default_credential_providers.find(hash); + if (it != _default_credential_providers.end()) { + provider = it->second; + } else { + provider = std::make_shared(); + _default_credential_providers[hash] = provider; + } + } + client = std::make_shared( + oss_conf.endpoint, + std::static_pointer_cast(provider), + oss_client_config); + LOG(INFO) << "OSS client created with DEFAULT provider, endpoint=" << oss_conf.endpoint; + } else if (oss_conf.cred_provider_type == OSSCredProviderType::ENV) { + // Read OSS_ACCESS_KEY_ID and OSS_ACCESS_KEY_SECRET from environment + const char* env_ak = std::getenv("OSS_ACCESS_KEY_ID"); + const char* env_sk = std::getenv("OSS_ACCESS_KEY_SECRET"); + if (!env_ak || !env_sk) { + LOG(ERROR) << "ENV credential type requires OSS_ACCESS_KEY_ID and " + "OSS_ACCESS_KEY_SECRET environment variables"; + return nullptr; + } + AlibabaCloud::OSS::Credentials creds(env_ak, env_sk); + client = std::make_shared(oss_conf.endpoint, creds, + oss_client_config); + LOG(INFO) << "OSS client created with ENV credentials, endpoint=" << oss_conf.endpoint; + } else if (oss_conf.cred_provider_type == OSSCredProviderType::ANONYMOUS) { + AlibabaCloud::OSS::Credentials creds("", ""); + client = std::make_shared(oss_conf.endpoint, creds, + oss_client_config); + LOG(INFO) << "OSS client created with ANONYMOUS access, endpoint=" << oss_conf.endpoint; + } else { + AlibabaCloud::OSS::Credentials creds(oss_conf.ak, oss_conf.sk, oss_conf.token); + client = std::make_shared(oss_conf.endpoint, creds, + oss_client_config); + LOG(INFO) << "OSS client created with SIMPLE credentials, endpoint=" + << oss_conf.endpoint; + } + } catch (const std::exception& e) { + LOG(ERROR) << "Failed to create OSS client: " << e.what(); + return nullptr; + } + + { + std::lock_guard lock(_lock); + _cache[hash] = client; + } + + return client; +} + +} // namespace doris diff --git a/be/src/util/oss_util.h b/be/src/util/oss_util.h new file mode 100644 index 00000000000000..7c2171b9b03982 --- /dev/null +++ b/be/src/util/oss_util.h @@ -0,0 +1,133 @@ +// 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. + +#pragma once + +#include +#include +#include + +#include +#include +#include + +#include "common/status.h" +#include "cpp/aws_common.h" +#include "cpp/oss_common.h" +#include "util/s3_util.h" + +namespace AlibabaCloud { +namespace OSS { +class OssClient; +} // namespace OSS +} // namespace AlibabaCloud + +namespace doris { + +class ECSMetadataCredentialsProvider; +class OSSSTSCredentialProvider; +class OSSDefaultCredentialsProvider; + +namespace oss_bvar { +extern bvar::LatencyRecorder oss_get_latency; +extern bvar::LatencyRecorder oss_put_latency; +extern bvar::LatencyRecorder oss_delete_object_latency; +extern bvar::LatencyRecorder oss_delete_objects_latency; +extern bvar::LatencyRecorder oss_head_latency; +extern bvar::LatencyRecorder oss_list_latency; +extern bvar::LatencyRecorder oss_multi_part_upload_latency; +} // namespace oss_bvar + +struct OSSClientConf { + std::string endpoint; + std::string region; + std::string ak; + std::string sk; + std::string token; + std::string bucket; + std::string role_arn; + std::string external_id; + std::string sts_endpoint; + + int max_connections = 100; + int request_timeout_ms = 30000; + int connect_timeout_ms = 10000; + + OSSCredProviderType cred_provider_type = OSSCredProviderType::DEFAULT; + + uint64_t get_hash() const { + uint64_t hash_code = 0; + hash_code ^= crc32_hash(ak + sk); + hash_code ^= crc32_hash(token); + hash_code ^= crc32_hash(endpoint); + hash_code ^= crc32_hash(region); + hash_code ^= crc32_hash(bucket); + hash_code ^= crc32_hash(role_arn); + hash_code ^= crc32_hash(external_id); + hash_code ^= crc32_hash(sts_endpoint); + hash_code ^= max_connections; + hash_code ^= request_timeout_ms; + hash_code ^= connect_timeout_ms; + hash_code ^= static_cast(cred_provider_type); + return hash_code; + } + + std::string to_string() const { + return fmt::format( + "(ak={}, endpoint={}, region={}, bucket={}, role_arn={}, " + "max_connections={}, cred_provider_type={})", + hide_access_key(ak), endpoint, region, bucket, role_arn, max_connections, + static_cast(cred_provider_type)); + } +}; + +struct OSSConf { + std::string bucket; + std::string prefix; + OSSClientConf client_conf; + + static OSSConf get_oss_conf(const cloud::ObjectStoreInfoPB& obj_info); + + std::string to_string() const { + return fmt::format("(bucket={}, prefix={}, client_conf={})", bucket, prefix, + client_conf.to_string()); + } +}; + +class OSSClientFactory { +public: + ~OSSClientFactory(); + + static OSSClientFactory& instance(); + + std::shared_ptr create(const OSSClientConf& oss_conf); + +private: + OSSClientFactory(); + + std::mutex _lock; + std::unordered_map> _cache; + std::unordered_map> + _ecs_credential_providers; + std::unordered_map> + _sts_credential_providers; + std::unordered_map> + _default_credential_providers; + std::string _ca_cert_file_path; +}; + +} // namespace doris diff --git a/be/src/util/s3_util.cpp b/be/src/util/s3_util.cpp index c6e86a7f4b290f..151579d5e157ac 100644 --- a/be/src/util/s3_util.cpp +++ b/be/src/util/s3_util.cpp @@ -17,6 +17,11 @@ #include "util/s3_util.h" +#ifdef USE_OSS +#include "io/fs/oss_v2_obj_storage_client.h" +#include "util/oss_util.h" +#endif + #include #include #include @@ -309,9 +314,17 @@ std::shared_ptr S3ClientFactory::create(const S3ClientConf } } - auto obj_client = (s3_conf.provider == io::ObjStorageType::AZURE) - ? _create_azure_client(s3_conf) - : _create_s3_client(s3_conf); + std::shared_ptr obj_client; +#ifdef USE_OSS + if (s3_conf.provider == io::ObjStorageType::OSS && config::enable_oss_native_sdk) { + obj_client = _create_oss_client(s3_conf); + } else +#endif + if (s3_conf.provider == io::ObjStorageType::AZURE) { + obj_client = _create_azure_client(s3_conf); + } else { + obj_client = _create_s3_client(s3_conf); + } { uint64_t hash = s3_conf.get_hash(); @@ -578,6 +591,10 @@ Status S3ClientFactory::convert_properties_to_s3_conf( // S3 Provider properties should be case insensitive. if (0 == strcasecmp(it->second.c_str(), AZURE_PROVIDER_STRING)) { s3_conf->client_conf.provider = io::ObjStorageType::AZURE; +#ifdef USE_OSS + } else if (0 == strcasecmp(it->second.c_str(), "OSS")) { + s3_conf->client_conf.provider = io::ObjStorageType::OSS; +#endif } } @@ -793,4 +810,57 @@ std::string hide_access_key(const std::string& ak) { return key; } +#ifdef USE_OSS +// Maps S3/AWS CredProviderType to OSSCredProviderType +static OSSCredProviderType s3_cred_to_oss_cred(CredProviderType type) { + switch (type) { + case CredProviderType::InstanceProfile: + return OSSCredProviderType::INSTANCE_PROFILE; + case CredProviderType::Simple: + return OSSCredProviderType::SIMPLE; + case CredProviderType::Env: + return OSSCredProviderType::ENV; + case CredProviderType::Anonymous: + return OSSCredProviderType::ANONYMOUS; + // WebIdentity (AWS IRSA) and Container (ECS task role) have no direct OSS equivalent; + // fall through to DEFAULT which will use the Alibaba Cloud credential chain. + // SystemProperties is also AWS-specific. + default: + return OSSCredProviderType::DEFAULT; + } +} + +std::shared_ptr S3ClientFactory::_create_oss_client( + const S3ClientConf& s3_conf) { + OSSClientConf oss_conf; + oss_conf.endpoint = normalize_oss_endpoint(s3_conf.endpoint); + oss_conf.region = s3_conf.region; + oss_conf.ak = s3_conf.ak; + oss_conf.sk = s3_conf.sk; + oss_conf.token = s3_conf.token; + oss_conf.bucket = s3_conf.bucket; + oss_conf.role_arn = s3_conf.role_arn; + oss_conf.external_id = s3_conf.external_id; + oss_conf.max_connections = s3_conf.max_connections > 0 ? s3_conf.max_connections : 100; + oss_conf.request_timeout_ms = + s3_conf.request_timeout_ms > 0 ? s3_conf.request_timeout_ms : 30000; + oss_conf.connect_timeout_ms = + s3_conf.connect_timeout_ms > 0 ? s3_conf.connect_timeout_ms : 10000; + oss_conf.cred_provider_type = s3_cred_to_oss_cred(s3_conf.cred_provider_type); + // role_arn alone triggers STS AssumeRole; no explicit INSTANCE_PROFILE required. + if (!oss_conf.role_arn.empty() && oss_conf.cred_provider_type == OSSCredProviderType::DEFAULT) { + oss_conf.cred_provider_type = OSSCredProviderType::INSTANCE_PROFILE; + } + + auto native_client = OSSClientFactory::instance().create(oss_conf); + if (!native_client) { + LOG(WARNING) << "failed to create native OSS client with conf " << s3_conf.to_string(); + return nullptr; + } + + LOG(INFO) << "created native OSS client with " << s3_conf.to_string(); + return std::make_shared(std::move(native_client), s3_conf.bucket); +} +#endif + } // end namespace doris diff --git a/be/src/util/s3_util.h b/be/src/util/s3_util.h index 064f88accd8539..0e262de72e918d 100644 --- a/be/src/util/s3_util.h +++ b/be/src/util/s3_util.h @@ -173,6 +173,9 @@ class S3ClientFactory { private: std::shared_ptr _create_s3_client(const S3ClientConf& s3_conf); std::shared_ptr _create_azure_client(const S3ClientConf& s3_conf); +#ifdef USE_OSS + std::shared_ptr _create_oss_client(const S3ClientConf& s3_conf); +#endif std::shared_ptr _get_aws_credentials_provider_v1( const S3ClientConf& s3_conf); std::shared_ptr _get_aws_credentials_provider_v2( diff --git a/be/test/util/s3_util_test.cpp b/be/test/util/s3_util_test.cpp index a87d0e9a2d2980..acf06f4faaa21f 100644 --- a/be/test/util/s3_util_test.cpp +++ b/be/test/util/s3_util_test.cpp @@ -121,4 +121,56 @@ TEST_F(S3UTILTest, check_s3_rate_limiter_config_changed_rebuilds_limiter) { check_s3_rate_limiter_config_changed(); } +#ifdef USE_OSS +// Tests for OSS provider detection and credential type mapping. +// These run without any real OSS credentials — pure logic tests. + +TEST_F(S3UTILTest, convert_properties_sets_oss_provider) { + // When FE sends "provider"="OSS", BE must set ObjStorageType::OSS + // so _create_oss_client() is reached when enable_oss_native_sdk=true. + std::map props; + props["AWS_ENDPOINT"] = "oss-cn-hangzhou.aliyuncs.com"; + props["AWS_REGION"] = "cn-hangzhou"; + props["AWS_ACCESS_KEY"] = "ak"; + props["AWS_SECRET_KEY"] = "sk"; + props["provider"] = "OSS"; + + S3Conf s3_conf; + S3URI uri("oss://my-bucket/prefix"); + auto st = S3ClientFactory::convert_properties_to_s3_conf(props, uri, &s3_conf); + EXPECT_TRUE(st.ok()) << st.to_string(); + EXPECT_EQ(io::ObjStorageType::OSS, s3_conf.client_conf.provider); +} + +TEST_F(S3UTILTest, convert_properties_oss_provider_case_insensitive) { + std::map props; + props["AWS_ENDPOINT"] = "oss-cn-hangzhou.aliyuncs.com"; + props["AWS_REGION"] = "cn-hangzhou"; + props["AWS_ACCESS_KEY"] = "ak"; + props["AWS_SECRET_KEY"] = "sk"; + props["provider"] = "oss"; // lowercase + + S3Conf s3_conf; + S3URI uri("oss://my-bucket/prefix"); + auto st = S3ClientFactory::convert_properties_to_s3_conf(props, uri, &s3_conf); + EXPECT_TRUE(st.ok()) << st.to_string(); + EXPECT_EQ(io::ObjStorageType::OSS, s3_conf.client_conf.provider); +} + +TEST_F(S3UTILTest, convert_properties_missing_provider_defaults_to_aws) { + // Without provider key, defaults remain AWS (S3-compat path unchanged). + std::map props; + props["AWS_ENDPOINT"] = "oss-cn-hangzhou.aliyuncs.com"; + props["AWS_REGION"] = "cn-hangzhou"; + props["AWS_ACCESS_KEY"] = "ak"; + props["AWS_SECRET_KEY"] = "sk"; + + S3Conf s3_conf; + S3URI uri("oss://my-bucket/prefix"); + auto st = S3ClientFactory::convert_properties_to_s3_conf(props, uri, &s3_conf); + EXPECT_TRUE(st.ok()) << st.to_string(); + EXPECT_EQ(io::ObjStorageType::AWS, s3_conf.client_conf.provider); +} +#endif // USE_OSS + } // end namespace doris diff --git a/common/cpp/CMakeLists.txt b/common/cpp/CMakeLists.txt index 81a600e57bf84d..64574dae7580ed 100644 --- a/common/cpp/CMakeLists.txt +++ b/common/cpp/CMakeLists.txt @@ -19,4 +19,9 @@ set(LIBRARY_OUTPUT_PATH "${BUILD_DIR}/src/common_cpp") file(GLOB_RECURSE SRC_FILES CONFIGURE_DEPENDS *.cpp) + +if(BUILD_OSS STREQUAL "OFF") + list(REMOVE_ITEM SRC_FILES "${CMAKE_CURRENT_SOURCE_DIR}/oss_credential_provider.cpp") +endif() + add_library(CommonCPP STATIC ${SRC_FILES}) diff --git a/common/cpp/oss_common.h b/common/cpp/oss_common.h new file mode 100644 index 00000000000000..840df80d7800a6 --- /dev/null +++ b/common/cpp/oss_common.h @@ -0,0 +1,44 @@ +// 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. + +#pragma once + +#include + +namespace doris { + +enum class OSSCredProviderType { + SIMPLE = 0, // Static AK/SK credentials + INSTANCE_PROFILE = 1, // ECS instance profile (metadata service) + DEFAULT = 2, // Default credential provider chain + ENV = 3, // OSS_ACCESS_KEY_ID / OSS_ACCESS_KEY_SECRET env vars + ANONYMOUS = 4, // No credentials — public bucket access only +}; + +// Normalize OSS endpoint by ensuring it has a scheme (http:// or https://). +// Defaults to HTTPS if no scheme is specified. +inline std::string normalize_oss_endpoint(const std::string& endpoint) { + if (endpoint.empty()) { + return endpoint; + } + if (endpoint.starts_with("https://") || endpoint.starts_with("http://")) { + return endpoint; + } + return "https://" + endpoint; +} + +} // namespace doris diff --git a/common/cpp/oss_credential_provider.cpp b/common/cpp/oss_credential_provider.cpp new file mode 100644 index 00000000000000..defc513a38f79b --- /dev/null +++ b/common/cpp/oss_credential_provider.cpp @@ -0,0 +1,417 @@ +// 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 "cpp/oss_credential_provider.h" + +#ifdef USE_OSS + +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include + +#include "common/logging.h" + +namespace { +std::string mask_credential(const std::string& cred) { + if (cred.empty()) return ""; + size_t len = cred.length(); + if (len <= 8) { + if (len <= 4) return std::string(len, '*'); + return cred.substr(0, 2) + std::string(len - 4, '*') + cred.substr(len - 2); + } + return cred.substr(0, 4) + std::string(len - 8, '*') + cred.substr(len - 4); +} + +size_t oss_curl_write_cb(void* contents, size_t size, size_t nmemb, std::string* userp) { + userp->append(static_cast(contents), size * nmemb); + return size * nmemb; +} +} // namespace + +namespace doris { + +// ── ECSMetadataCredentialsProvider ─────────────────────────────────────────── + +ECSMetadataCredentialsProvider::ECSMetadataCredentialsProvider() + : _cached_credentials(nullptr), _expiration(std::chrono::system_clock::now()) { + LOG(INFO) << "ECSMetadataCredentialsProvider initialized"; +} + +bool ECSMetadataCredentialsProvider::_is_expired() const { + auto now = std::chrono::system_clock::now(); + return std::chrono::duration_cast(_expiration - now).count() <= + REFRESH_BEFORE_EXPIRY_SECONDS; +} + +AlibabaCloud::OSS::Credentials ECSMetadataCredentialsProvider::getCredentials() { + { + std::lock_guard lock(_mtx); + if (_cached_credentials && !_is_expired()) { + return *_cached_credentials; + } + if (_cached_credentials) { + auto t = std::chrono::system_clock::to_time_t(_expiration); + struct tm tm_buf; + LOG(INFO) << "ECS credentials expiring (" + << std::put_time(localtime_r(&t, &tm_buf), "%Y-%m-%d %H:%M:%S") + << "), refreshing"; + } else { + LOG(INFO) << "Fetching ECS metadata credentials (first time)"; + } + } + + std::unique_ptr new_creds; + std::chrono::system_clock::time_point new_expiry; + + if (_fetch_credentials_outside_lock(new_creds, new_expiry) != 0) { + std::lock_guard lock(_mtx); + if (_cached_credentials) { + LOG(WARNING) << "Using expired ECS credentials as fallback"; + return *_cached_credentials; + } + LOG(ERROR) << "Failed to fetch ECS credentials with no cached fallback"; + return AlibabaCloud::OSS::Credentials("", "", ""); + } + + std::lock_guard lock(_mtx); + if (_cached_credentials && !_is_expired()) { + return *_cached_credentials; + } + _cached_credentials = std::move(new_creds); + _expiration = new_expiry; + auto t = std::chrono::system_clock::to_time_t(_expiration); + struct tm tm_buf; + LOG(INFO) << "ECS credentials refreshed, expiry: " + << std::put_time(localtime_r(&t, &tm_buf), "%Y-%m-%d %H:%M:%S"); + return *_cached_credentials; +} + +int ECSMetadataCredentialsProvider::_http_get(const std::string& url, std::string& response) { + CURL* curl = curl_easy_init(); + if (!curl) { + LOG(ERROR) << "Failed to initialize CURL"; + return -1; + } + response.clear(); + curl_easy_setopt(curl, CURLOPT_URL, url.c_str()); + curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, oss_curl_write_cb); + curl_easy_setopt(curl, CURLOPT_WRITEDATA, &response); + curl_easy_setopt(curl, CURLOPT_TIMEOUT_MS, METADATA_SERVICE_TIMEOUT_MS); + curl_easy_setopt(curl, CURLOPT_NOSIGNAL, 1L); + CURLcode res = curl_easy_perform(curl); + if (res != CURLE_OK) { + LOG(ERROR) << "ECS metadata HTTP GET failed: " << curl_easy_strerror(res); + curl_easy_cleanup(curl); + return -1; + } + long http_code = 0; + curl_easy_getinfo(curl, CURLINFO_RESPONSE_CODE, &http_code); + curl_easy_cleanup(curl); + if (http_code != 200) { + LOG(ERROR) << "ECS metadata service returned HTTP " << http_code; + return -1; + } + return 0; +} + +int ECSMetadataCredentialsProvider::_get_role_name(std::string& role_name) { + std::string url = std::string("http://") + METADATA_SERVICE_HOST + METADATA_SERVICE_PATH; + std::string response; + if (_http_get(url, response) != 0) return -1; + role_name = response; + // trim whitespace + role_name.erase(role_name.begin(), + std::find_if(role_name.begin(), role_name.end(), + [](unsigned char ch) { return !std::isspace(ch); })); + role_name.erase(std::find_if(role_name.rbegin(), role_name.rend(), + [](unsigned char ch) { return !std::isspace(ch); }) + .base(), + role_name.end()); + if (role_name.empty()) { + LOG(ERROR) << "No RAM role attached to this ECS instance"; + return -1; + } + size_t nl = role_name.find('\n'); + if (nl != std::string::npos) { + LOG(WARNING) << "Multiple ECS RAM roles; using first: " << role_name.substr(0, nl); + role_name = role_name.substr(0, nl); + } + return 0; +} + +int ECSMetadataCredentialsProvider::_fetch_credentials_outside_lock( + std::unique_ptr& out_credentials, + std::chrono::system_clock::time_point& out_expiration) { + std::string role_name; + if (_get_role_name(role_name) != 0) return -1; + + std::string url = + std::string("http://") + METADATA_SERVICE_HOST + METADATA_SERVICE_PATH + role_name; + std::string response; + if (_http_get(url, response) != 0) return -1; + + rapidjson::Document doc; + doc.Parse(response.c_str()); + if (doc.HasParseError() || !doc.HasMember("Code") || + std::string(doc["Code"].GetString()) != "Success") { + LOG(ERROR) << "ECS metadata parse error or non-Success code"; + return -1; + } + if (!doc.HasMember("AccessKeyId") || !doc.HasMember("AccessKeySecret") || + !doc.HasMember("SecurityToken") || !doc.HasMember("Expiration")) { + LOG(ERROR) << "ECS metadata response missing required fields"; + return -1; + } + + std::string ak = doc["AccessKeyId"].GetString(); + std::string sk = doc["AccessKeySecret"].GetString(); + std::string token = doc["SecurityToken"].GetString(); + std::string expiry_str = doc["Expiration"].GetString(); + + if (ak.empty() || sk.empty() || token.empty()) { + LOG(ERROR) << "ECS metadata returned empty credentials"; + return -1; + } + + std::tm tm = {}; + std::istringstream ss(expiry_str); + ss >> std::get_time(&tm, "%Y-%m-%dT%H:%M:%SZ"); + if (ss.fail()) { + LOG(ERROR) << "Failed to parse ECS expiration: " << expiry_str; + return -1; + } + + out_expiration = std::chrono::system_clock::from_time_t(timegm(&tm)); + out_credentials = std::make_unique(ak, sk, token); + VLOG(1) << "ECS credentials: ak=" << mask_credential(ak) << ", expiry=" << expiry_str; + return 0; +} + +// ── OSSSTSCredentialProvider ────────────────────────────────────────────────── + +OSSSTSCredentialProvider::OSSSTSCredentialProvider(const std::string& role_arn, + const std::string& region, + const std::string& external_id, + const std::string& ca_cert_path, + const std::string& sts_endpoint) + : _cached_credentials(nullptr), + _expiration(std::chrono::system_clock::now()), + _role_arn(role_arn), + _region(region), + _external_id(external_id), + _ca_cert_path(ca_cert_path), + _sts_endpoint(sts_endpoint) { + if (_role_arn.empty()) { + throw std::invalid_argument("RAM role ARN cannot be empty for STS AssumeRole"); + } + LOG(INFO) << "OSSSTSCredentialProvider: role_arn=" << _role_arn << ", region=" << _region; +} + +bool OSSSTSCredentialProvider::_is_expired() const { + auto now = std::chrono::system_clock::now(); + return std::chrono::duration_cast(_expiration - now).count() <= + REFRESH_BEFORE_EXPIRY_SECONDS; +} + +AlibabaCloud::OSS::Credentials OSSSTSCredentialProvider::getCredentials() { + { + std::lock_guard lock(_mtx); + if (_cached_credentials && !_is_expired()) { + return *_cached_credentials; + } + if (!_cached_credentials) { + LOG(INFO) << "Fetching STS AssumeRole credentials (first time)"; + } + } + + std::unique_ptr new_creds; + std::chrono::system_clock::time_point new_expiry; + + if (_fetch_credentials_from_sts(new_creds, new_expiry) != 0) { + std::lock_guard lock(_mtx); + if (_cached_credentials) { + LOG(WARNING) << "Using expired STS credentials as fallback"; + return *_cached_credentials; + } + LOG(ERROR) << "Failed to fetch STS credentials with no cached fallback"; + return AlibabaCloud::OSS::Credentials("", "", ""); + } + + std::lock_guard lock(_mtx); + if (_cached_credentials && !_is_expired()) { + return *_cached_credentials; + } + _cached_credentials = std::move(new_creds); + _expiration = new_expiry; + auto t = std::chrono::system_clock::to_time_t(_expiration); + struct tm tm_buf; + LOG(INFO) << "STS credentials refreshed, expiry: " + << std::put_time(localtime_r(&t, &tm_buf), "%Y-%m-%d %H:%M:%S"); + return *_cached_credentials; +} + +int OSSSTSCredentialProvider::_fetch_credentials_from_sts( + std::unique_ptr& out_credentials, + std::chrono::system_clock::time_point& out_expiration) { + try { + AlibabaCloud::Credentials::Models::Config cred_config; + cred_config.setType("ecs_ram_role"); + AlibabaCloud::Credentials::Client cred_client(cred_config); + AlibabaCloud::Credentials::Models::CredentialModel base_cred = cred_client.getCredential(); + + AlibabaCloud::OpenApi::Utils::Models::Config config; + config.setAccessKeyId(base_cred.getAccessKeyId()); + config.setAccessKeySecret(base_cred.getAccessKeySecret()); + if (!base_cred.getSecurityToken().empty()) { + config.setSecurityToken(base_cred.getSecurityToken()); + } + config.setRegionId(_region); + config.setEndpoint(_sts_endpoint.empty() ? "sts." + _region + ".aliyuncs.com" + : _sts_endpoint); + + AlibabaCloud::Sts20150401::Client client(config); + + AlibabaCloud::Sts20150401::Models::AssumeRoleRequest request; + request.setRoleArn(_role_arn); + char hostname_buf[64] = {}; + if (gethostname(hostname_buf, sizeof(hostname_buf) - 1) != 0) { + strncpy(hostname_buf, "unknown", 7); + } + std::string session_name = "doris-"; + for (int i = 0; hostname_buf[i] != '\0' && session_name.size() < 32; ++i) { + char c = hostname_buf[i]; + if (std::isalnum(static_cast(c)) || c == '-' || c == '_' || c == '.') { + session_name += c; + } + } + if (session_name.size() < 2) session_name = "doris-oss-session"; + request.setRoleSessionName(session_name); + request.setDurationSeconds(SESSION_DURATION_SECONDS); + if (!_external_id.empty()) request.setExternalId(_external_id); + + Darabonba::RuntimeOptions runtime; + if (!_ca_cert_path.empty()) { + runtime.setCa(_ca_cert_path); + } else { + runtime.setIgnoreSSL(true); + } + + auto response = client.assumeRoleWithOptions(request, runtime); + auto creds = response.getBody().getCredentials(); + + std::string ak = creds.getAccessKeyId(); + std::string sk = creds.getAccessKeySecret(); + std::string token = creds.getSecurityToken(); + std::string expiry_str = creds.getExpiration(); + + if (ak.empty() || sk.empty() || token.empty()) { + LOG(ERROR) << "STS AssumeRole returned empty credentials"; + return -1; + } + + std::tm tm = {}; + std::istringstream ss(expiry_str); + ss >> std::get_time(&tm, "%Y-%m-%dT%H:%M:%SZ"); + if (ss.fail()) { + LOG(ERROR) << "Failed to parse STS expiration: " << expiry_str; + return -1; + } + + out_expiration = std::chrono::system_clock::from_time_t(timegm(&tm)); + out_credentials = std::make_unique(ak, sk, token); + VLOG(1) << "STS credentials: ak=" << mask_credential(ak) << ", expiry=" << expiry_str; + return 0; + } catch (const std::exception& e) { + LOG(ERROR) << "STS AssumeRole failed: " << e.what(); + return -1; + } +} + +// ── OSSDefaultCredentialsProvider ───────────────────────────────────────────── + +OSSDefaultCredentialsProvider::OSSDefaultCredentialsProvider() + : _cached_credentials(nullptr), _expiration(std::chrono::system_clock::now()) { + LOG(INFO) << "OSSDefaultCredentialsProvider initialized"; +} + +bool OSSDefaultCredentialsProvider::_is_expired() const { + auto now = std::chrono::system_clock::now(); + return std::chrono::duration_cast(_expiration - now).count() <= + REFRESH_BEFORE_EXPIRY_SECONDS; +} + +AlibabaCloud::OSS::Credentials OSSDefaultCredentialsProvider::getCredentials() { + { + std::lock_guard lock(_mtx); + if (_cached_credentials && !_is_expired()) { + return *_cached_credentials; + } + } + + try { + AlibabaCloud::Credentials::Client cred_client; + auto cred_model = cred_client.getCredential(); + std::string ak = cred_model.getAccessKeyId(); + std::string sk = cred_model.getAccessKeySecret(); + std::string token = cred_model.getSecurityToken(); + + if (ak.empty() || sk.empty()) { + std::lock_guard lock(_mtx); + if (_cached_credentials) { + LOG(WARNING) << "Default provider returned empty credentials, using fallback"; + return *_cached_credentials; + } + LOG(ERROR) << "Default provider returned empty credentials with no fallback"; + return AlibabaCloud::OSS::Credentials("", "", ""); + } + + auto new_creds = token.empty() + ? std::make_unique(ak, sk) + : std::make_unique(ak, sk, token); + auto new_expiry = std::chrono::system_clock::now() + std::chrono::hours(1); + + std::lock_guard lock(_mtx); + _cached_credentials = std::move(new_creds); + _expiration = new_expiry; + LOG(INFO) << "Default provider credentials fetched, provider: " + << cred_model.getProviderName(); + return *_cached_credentials; + } catch (const std::exception& e) { + std::lock_guard lock(_mtx); + if (_cached_credentials) { + LOG(WARNING) << "Default provider failed, using fallback: " << e.what(); + return *_cached_credentials; + } + LOG(ERROR) << "Default provider failed with no fallback: " << e.what(); + return AlibabaCloud::OSS::Credentials("", "", ""); + } +} + +} // namespace doris + +#endif // USE_OSS diff --git a/common/cpp/oss_credential_provider.h b/common/cpp/oss_credential_provider.h new file mode 100644 index 00000000000000..e07caab727030e --- /dev/null +++ b/common/cpp/oss_credential_provider.h @@ -0,0 +1,111 @@ +// 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. + +#pragma once + +#include +#include +#include +#include + +#ifdef USE_OSS + +#include +#include + +namespace doris { + +// Fetches temporary credentials from the ECS instance metadata service (100.100.100.200) +// and auto-refreshes before expiry. +class ECSMetadataCredentialsProvider : public AlibabaCloud::OSS::CredentialsProvider { +public: + ECSMetadataCredentialsProvider(); + ~ECSMetadataCredentialsProvider() override = default; + + AlibabaCloud::OSS::Credentials getCredentials() override; + +private: + int _fetch_credentials_outside_lock( + std::unique_ptr& out_credentials, + std::chrono::system_clock::time_point& out_expiration); + int _get_role_name(std::string& role_name); + bool _is_expired() const; + int _http_get(const std::string& url, std::string& response); + + mutable std::mutex _mtx; + std::unique_ptr _cached_credentials; + std::chrono::system_clock::time_point _expiration; + + static constexpr const char* METADATA_SERVICE_HOST = "100.100.100.200"; + static constexpr const char* METADATA_SERVICE_PATH = + "/latest/meta-data/ram/security-credentials/"; + static constexpr int METADATA_SERVICE_TIMEOUT_MS = 5000; + static constexpr int REFRESH_BEFORE_EXPIRY_SECONDS = 300; +}; + +// Obtains temporary credentials via Alibaba STS AssumeRole and auto-refreshes before expiry. +// Uses the alibabacloud-sdk-cpp stack (sts-20150401 + credentials-cpp). +class OSSSTSCredentialProvider : public AlibabaCloud::OSS::CredentialsProvider { +public: + explicit OSSSTSCredentialProvider(const std::string& role_arn, const std::string& region, + const std::string& external_id = "", + const std::string& ca_cert_path = "", + const std::string& sts_endpoint = ""); + ~OSSSTSCredentialProvider() override = default; + + AlibabaCloud::OSS::Credentials getCredentials() override; + +private: + int _fetch_credentials_from_sts( + std::unique_ptr& out_credentials, + std::chrono::system_clock::time_point& out_expiration); + bool _is_expired() const; + + mutable std::mutex _mtx; + std::unique_ptr _cached_credentials; + std::chrono::system_clock::time_point _expiration; + std::string _role_arn; + std::string _region; + std::string _external_id; + std::string _ca_cert_path; + std::string _sts_endpoint; + + static constexpr int REFRESH_BEFORE_EXPIRY_SECONDS = 300; + static constexpr int SESSION_DURATION_SECONDS = 3600; +}; + +// Default Alibaba Cloud credential provider chain (env vars, config file, ECS metadata). +class OSSDefaultCredentialsProvider : public AlibabaCloud::OSS::CredentialsProvider { +public: + OSSDefaultCredentialsProvider(); + ~OSSDefaultCredentialsProvider() override = default; + + AlibabaCloud::OSS::Credentials getCredentials() override; + +private: + bool _is_expired() const; + + mutable std::mutex _mtx; + std::unique_ptr _cached_credentials; + std::chrono::system_clock::time_point _expiration; + + static constexpr int REFRESH_BEFORE_EXPIRY_SECONDS = 300; +}; + +} // namespace doris + +#endif // USE_OSS diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/property/storage/OSSProperties.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/property/storage/OSSProperties.java index 9a5c11ce0f5771..41ef5fec0bab26 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/property/storage/OSSProperties.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/property/storage/OSSProperties.java @@ -133,6 +133,18 @@ public class OSSProperties extends AbstractS3CompatibleProperties { @Getter protected String forceParsingByStandardUrl = "false"; + @Getter + @ConnectorProperty(names = {"OSS_ROLE_ARN", "AWS_ROLE_ARN", "oss.role_arn"}, + required = false, + description = "RAM role ARN for Alibaba STS AssumeRole.") + protected String roleArn = ""; + + @Getter + @ConnectorProperty(names = {"AWS_EXTERNAL_ID", "oss.external_id"}, + required = false, + description = "External ID for cross-account AssumeRole trust policy.") + protected String externalId = ""; + private static final Pattern STANDARD_ENDPOINT_PATTERN = Pattern .compile("^(?:https?://)?(?:s3\\.)?oss-([a-z0-9-]+?)(?:-internal)?\\.aliyuncs\\.com$"); @@ -300,12 +312,24 @@ public AwsCredentialsProvider getAwsCredentialsProvider() { return credentialsProvider; } if (StringUtils.isBlank(accessKey) && StringUtils.isBlank(secretKey)) { - // For anonymous access (no credentials required) return AnonymousCredentialsProvider.create(); } return null; } + @Override + public Map getBackendConfigProperties() { + Map props = generateBackendS3Configuration(); + props.put("provider", "OSS"); + if (StringUtils.isNotBlank(roleArn)) { + props.put("AWS_ROLE_ARN", roleArn); + } + if (StringUtils.isNotBlank(externalId)) { + props.put("AWS_EXTERNAL_ID", externalId); + } + return props; + } + @Override protected Set schemas() { return ImmutableSet.of("oss"); diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/property/storage/OSSPropertiesTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/property/storage/OSSPropertiesTest.java index 7c9ec53b177d25..a50ae372844b40 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/property/storage/OSSPropertiesTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/property/storage/OSSPropertiesTest.java @@ -294,4 +294,38 @@ public void testOSSBucketEndpointPathProperties() throws UserException { Assertions.assertEquals("s3://my-bucket/path/to/dir/file.txt", OSSProperties.rewriteOssBucketIfNecessary("s3://my-bucket.oss-cn-hangzhou.aliyuncs.com/path/to/dir/file.txt")); Assertions.assertEquals("https://bucket-name.oss-cn-hangzhou.aliyuncs.com/path/to/dir/file.txt", OSSProperties.rewriteOssBucketIfNecessary("https://bucket-name.oss-cn-hangzhou.aliyuncs.com/path/to/dir/file.txt")); } + + @Test + public void testBackendConfigIncludesProviderOss() { + Map props = new HashMap<>(); + props.put("oss.endpoint", "oss-cn-hangzhou.aliyuncs.com"); + props.put("oss.access_key", "ak"); + props.put("oss.secret_key", "sk"); + OSSProperties ossProperties = (OSSProperties) StorageProperties.createPrimary(props); + + Map backend = ossProperties.getBackendConfigProperties(); + + Assertions.assertEquals("OSS", backend.get("provider")); + Assertions.assertNotNull(backend.get("AWS_ENDPOINT")); + Assertions.assertEquals("ak", backend.get("AWS_ACCESS_KEY")); + Assertions.assertFalse(backend.containsKey("AWS_ROLE_ARN")); + Assertions.assertFalse(backend.containsKey("AWS_EXTERNAL_ID")); + } + + @Test + public void testBackendConfigForwardsRoleArnAndExternalId() { + Map props = new HashMap<>(); + props.put("oss.endpoint", "oss-cn-hangzhou.aliyuncs.com"); + props.put("oss.access_key", "ak"); + props.put("oss.secret_key", "sk"); + props.put("OSS_ROLE_ARN", "acs:ram::123:role/MyRole"); + props.put("AWS_EXTERNAL_ID", "ext-id"); + OSSProperties ossProperties = (OSSProperties) StorageProperties.createPrimary(props); + + Map backend = ossProperties.getBackendConfigProperties(); + + Assertions.assertEquals("OSS", backend.get("provider")); + Assertions.assertEquals("acs:ram::123:role/MyRole", backend.get("AWS_ROLE_ARN")); + Assertions.assertEquals("ext-id", backend.get("AWS_EXTERNAL_ID")); + } } diff --git a/fe/fe-filesystem/fe-filesystem-oss/src/main/java/org/apache/doris/filesystem/oss/OssFileSystemProperties.java b/fe/fe-filesystem/fe-filesystem-oss/src/main/java/org/apache/doris/filesystem/oss/OssFileSystemProperties.java index 663b5596c242cd..07a00de04c5d44 100644 --- a/fe/fe-filesystem/fe-filesystem-oss/src/main/java/org/apache/doris/filesystem/oss/OssFileSystemProperties.java +++ b/fe/fe-filesystem/fe-filesystem-oss/src/main/java/org/apache/doris/filesystem/oss/OssFileSystemProperties.java @@ -237,6 +237,7 @@ public Map toMap() { private Map toBackendKv() { Map kv = new HashMap<>(); + kv.put("provider", "OSS"); putIfNotBlank(kv, "AWS_ENDPOINT", endpoint); putIfNotBlank(kv, "AWS_REGION", region); putIfNotBlank(kv, "AWS_ACCESS_KEY", accessKey);