// 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/s3_util.h"

#include <aws/core/auth/AWSAuthSigner.h>

#include <aws/core/auth/AWSCredentials.h>

#include <aws/core/auth/AWSCredentialsProviderChain.h>

#include <aws/core/auth/STSCredentialsProvider.h>

#include <aws/core/client/DefaultRetryStrategy.h>

#include <aws/core/platform/Environment.h>

#include <aws/core/utils/logging/LogLevel.h>

#include <aws/core/utils/logging/LogSystemInterface.h>

#include <aws/core/utils/memory/stl/AWSStringStream.h>

#include <aws/identity-management/auth/STSAssumeRoleCredentialsProvider.h>

#include <aws/s3/S3Client.h>

#include <aws/sts/STSClient.h>

#include <bvar/reducer.h>

#include <cpp/token_bucket_rate_limiter.h>

#include <atomic>

#include <charconv>

#include <limits>

#include <mutex>

#include "util/cpu_info.h"

#include "util/string_util.h"

#ifdef USE_AZURE

#include <azure/core/diagnostics/logger.hpp>

#include <azure/core/http/curl_transport.hpp>

#include <azure/storage/blobs/blob_container_client.hpp>

#endif

#include <cstdlib>

#include <filesystem>

#include <fstream>

#include <functional>

#include <memory>

#include <ostream>

#include <utility>

#include "cloud/config.h"

#include "common/config.h"

#include "common/logging.h"

#include "common/status.h"

#include "cpp/aws_logger.h"

#include "cpp/custom_aws_credentials_provider_chain.h"

#include "cpp/obj_retry_strategy.h"

#include "cpp/sync_point.h"

#include "cpp/util.h"

#ifdef USE_AZURE

#include "io/fs/azure_obj_storage_client.h"

#endif

#include "exec/scan/scanner_scheduler.h"

#include "io/fs/obj_storage_client.h"

#include "io/fs/s3_obj_storage_client.h"

#include "runtime/exec_env.h"

#include "util/s3_uri.h"

namespace doris {

namespace s3_bvar {

bvar::LatencyRecorder s3_get_latency("s3_get");

bvar::LatencyRecorder s3_put_latency("s3_put");

bvar::LatencyRecorder s3_delete_object_latency("s3_delete_object");

bvar::LatencyRecorder s3_delete_objects_latency("s3_delete_objects");

bvar::LatencyRecorder s3_head_latency("s3_head");

bvar::LatencyRecorder s3_multi_part_upload_latency("s3_multi_part_upload");

bvar::LatencyRecorder s3_list_latency("s3_list");

bvar::LatencyRecorder s3_list_object_versions_latency("s3_list_object_versions");

bvar::LatencyRecorder s3_get_bucket_version_latency("s3_get_bucket_version");

bvar::LatencyRecorder s3_copy_object_latency("s3_copy_object");

}; // namespace s3_bvar

namespace {

doris::Status is_s3_conf_valid(const S3ClientConf& conf) {

    if (conf.endpoint.empty()) {

        return Status::InvalidArgument<false>("Invalid s3 conf, empty endpoint");

    }

    if (conf.region.empty()) {

        return Status::InvalidArgument<false>("Invalid s3 conf, empty region");

    }

    if (conf.role_arn.empty()) {

        // Allow anonymous access when both ak and sk are empty

        bool hasAk = !conf.ak.empty();

        bool hasSk = !conf.sk.empty();

        // Either both credentials are provided or both are empty (anonymous access)

        if (hasAk && conf.sk.empty()) {

            return Status::InvalidArgument<false>("Invalid s3 conf, empty sk");

        }

        if (hasSk && conf.ak.empty()) {

            return Status::InvalidArgument<false>("Invalid s3 conf, empty ak");

        }

    }

    return Status::OK();

}

// Return true is convert `str` to int successfully

bool to_int(std::string_view str, int& res) {

    auto [_, ec] = std::from_chars(str.data(), str.data() + str.size(), res);

    return ec == std::errc {};

}

#ifdef USE_AZURE

std::string env_or_empty(const char* env_name) {

    if (const char* value = std::getenv(env_name); value != nullptr) {

        return value;

    }

    return "";

}

std::string build_azure_tls_debug_context(const std::string& selected_ca_file) {

    bool selected_ca_exists = false;

    bool selected_ca_readable = false;

    if (!selected_ca_file.empty()) {

        std::error_code ec;

        selected_ca_exists = std::filesystem::exists(selected_ca_file, ec) && !ec;

        std::ifstream input(selected_ca_file);

        selected_ca_readable = input.good();

    }

    return fmt::format(

            "tls_debug(ca_cert_file_paths='{}', selected_ca_file='{}', selected_ca_exists={}, "

            "selected_ca_readable={}, SSL_CERT_FILE='{}', CURL_CA_BUNDLE='{}', SSL_CERT_DIR='{}')",

            config::ca_cert_file_paths, selected_ca_file, selected_ca_exists, selected_ca_readable,

            env_or_empty("SSL_CERT_FILE"), env_or_empty("CURL_CA_BUNDLE"),

            env_or_empty("SSL_CERT_DIR"));

}

#endif

constexpr char USE_PATH_STYLE[] = "use_path_style";

constexpr char AZURE_PROVIDER_STRING[] = "AZURE";

constexpr char S3_PROVIDER[] = "provider";

constexpr char S3_AK[] = "AWS_ACCESS_KEY";

constexpr char S3_SK[] = "AWS_SECRET_KEY";

constexpr char S3_ENDPOINT[] = "AWS_ENDPOINT";

constexpr char S3_REGION[] = "AWS_REGION";

constexpr char S3_TOKEN[] = "AWS_TOKEN";

constexpr char S3_MAX_CONN_SIZE[] = "AWS_MAX_CONNECTIONS";

constexpr char S3_REQUEST_TIMEOUT_MS[] = "AWS_REQUEST_TIMEOUT_MS";

constexpr char S3_CONN_TIMEOUT_MS[] = "AWS_CONNECTION_TIMEOUT_MS";

constexpr char S3_NEED_OVERRIDE_ENDPOINT[] = "AWS_NEED_OVERRIDE_ENDPOINT";

constexpr char S3_ROLE_ARN[] = "AWS_ROLE_ARN";

constexpr char S3_EXTERNAL_ID[] = "AWS_EXTERNAL_ID";

constexpr char S3_CREDENTIALS_PROVIDER_TYPE[] = "AWS_CREDENTIALS_PROVIDER_TYPE";

#ifdef BE_TEST

std::atomic<int64_t> s3_rate_limiter_cpu_cores_for_test {0};

#endif

int64_t get_s3_rate_limiter_cpu_cores() {

#ifdef BE_TEST

    const int64_t cpu_cores_for_test =

            s3_rate_limiter_cpu_cores_for_test.load(std::memory_order_relaxed);

    if (cpu_cores_for_test > 0) {

        return cpu_cores_for_test;

    }

#endif

    return CpuInfo::num_cores();

}

int64_t get_s3_qps_per_core(S3RateLimitType type) {

    switch (type) {

    case S3RateLimitType::GET:

        return config::s3_get_qps_per_core;

    case S3RateLimitType::PUT:

        return config::s3_put_qps_per_core;

    default:

        CHECK(false) << "unknown S3 rate limiter type: " << to_string(type);

        return -1;

    }

}

int64_t get_s3_bytes_per_second_per_core(S3RateLimitType type) {

    switch (type) {

    case S3RateLimitType::GET:

        return config::s3_get_bytes_per_second_per_core;

    case S3RateLimitType::PUT:

        return config::s3_put_bytes_per_second_per_core;

    default:

        CHECK(false) << "unknown S3 rate limiter type: " << to_string(type);

        return -1;

    }

}

int64_t compute_s3_rate_limit_from_core(int64_t per_core, int64_t cores, int64_t max_rate) {

    max_rate = max_rate > 0 ? max_rate : std::numeric_limits<int64_t>::max();

    if (per_core > max_rate / cores) {

        return max_rate;

    }

    return per_core * cores;

}

S3RateLimiterConfig get_effective_s3_rate_limiter_config_with_cpu_cores(S3RateLimitType type,

                                                                        int64_t cpu_cores) {

    S3RateLimiterConfig result;

    int64_t per_core;

    int64_t max_qps;

    switch (type) {

    case S3RateLimitType::GET:

        per_core = config::s3_get_qps_per_core;

        max_qps = config::s3_get_qps_max;

        if (per_core < 0) {

            result.token_limit = config::s3_get_token_limit;

            result.token_per_second = config::s3_get_token_per_second;

            result.bucket_tokens = config::s3_get_bucket_tokens;

        }

        break;

    case S3RateLimitType::PUT:

        per_core = config::s3_put_qps_per_core;

        max_qps = config::s3_put_qps_max;

        if (per_core < 0) {

            result.token_limit = config::s3_put_token_limit;

            result.token_per_second = config::s3_put_token_per_second;

            result.bucket_tokens = config::s3_put_bucket_tokens;

        }

        break;

    default:

        CHECK(false) << "unknown S3 rate limiter type: " << to_string(type);

        return result;

    }

    if (per_core >= 0) {

        result.token_per_second = compute_s3_rate_limit_from_core(per_core, cpu_cores, max_qps);

        result.bucket_tokens = result.token_per_second;

    }

    return result;

}

S3RateLimiterConfig get_effective_s3_bytes_rate_limiter_config_with_cpu_cores(S3RateLimitType type,

                                                                              int64_t cpu_cores) {

    int64_t per_core = 0;

    int64_t max_bytes_per_second = 0;

    switch (type) {

    case S3RateLimitType::GET:

        per_core = config::s3_get_bytes_per_second_per_core;

        max_bytes_per_second = config::s3_get_bytes_per_second_max;

        break;

    case S3RateLimitType::PUT:

        per_core = config::s3_put_bytes_per_second_per_core;

        max_bytes_per_second = config::s3_put_bytes_per_second_max;

        break;

    default:

        CHECK(false) << "unknown S3 rate limiter type: " << to_string(type);

    }

    S3RateLimiterConfig result;

    if (per_core > 0) {

        result.token_per_second =

                compute_s3_rate_limit_from_core(per_core, cpu_cores, max_bytes_per_second);

        result.bucket_tokens = result.token_per_second;

    }

    return result;

}

} // namespace

bvar::Adder<int64_t> get_bytes_rate_limit_ns("get_bytes_rate_limit_ns");

bvar::Adder<int64_t> get_bytes_rate_limit_exceed_req_num("get_bytes_rate_limit_exceed_req_num");

bvar::Adder<int64_t> put_bytes_rate_limit_ns("put_bytes_rate_limit_ns");

bvar::Adder<int64_t> put_bytes_rate_limit_exceed_req_num("put_bytes_rate_limit_exceed_req_num");

namespace {

enum class S3RateLimiterKind {

    QPS,

    BYTES,

};

struct S3RateLimiterUpdateState {

    S3RateLimiterConfig applied_config;

    std::mutex update_lock;

};

std::array<std::array<S3RateLimiterUpdateState, 2>, 2> update_states;

S3RateLimiterUpdateState& get_update_state(S3RateLimitType type, S3RateLimiterKind kind) {

    DCHECK(type == S3RateLimitType::GET || type == S3RateLimitType::PUT);

    return update_states[static_cast<size_t>(kind)][static_cast<size_t>(type)];

}

void update_s3_rate_limiter(S3ClientFactory& factory, S3RateLimitType type, S3RateLimiterKind kind,

                            int64_t cpu_cores) {

    auto& state = get_update_state(type, kind);

    std::lock_guard update_guard(state.update_lock);

    const int64_t per_core = kind == S3RateLimiterKind::BYTES

                                     ? get_s3_bytes_per_second_per_core(type)

                                     : get_s3_qps_per_core(type);

    const auto current =

            kind == S3RateLimiterKind::BYTES

                    ? get_effective_s3_bytes_rate_limiter_config_with_cpu_cores(type, cpu_cores)

                    : get_effective_s3_rate_limiter_config_with_cpu_cores(type, cpu_cores);

    const bool config_changed = state.applied_config.token_per_second != current.token_per_second ||

                                state.applied_config.bucket_tokens != current.bucket_tokens ||

                                state.applied_config.token_limit != current.token_limit;

    if (!config_changed) {

        return;

    }

    auto* limiter = kind == S3RateLimiterKind::BYTES ? factory.bytes_rate_limiter(type)

                                                     : factory.rate_limiter(type);

    const int ret =

            limiter->reset(current.token_per_second, current.bucket_tokens, current.token_limit);

    if (ret != 0) {

        LOG(WARNING) << "Failed to reset S3 " << to_string(type)

                     << (kind == S3RateLimiterKind::BYTES ? " bytes" : " QPS")

                     << " rate limiter, error code: " << ret;

        return;

    }

    const int64_t logged_cpu_cores = per_core > 0 ? cpu_cores : 1;

    if (kind == S3RateLimiterKind::BYTES) {

        LOG(INFO) << "Reset S3 " << to_string(type)

                  << " bytes rate limiter, bytes_per_second=" << current.token_per_second

                  << ", burst_bytes=" << current.bucket_tokens

                  << ", cpu_cores=" << logged_cpu_cores;

    } else if (per_core >= 0) {

        LOG(INFO) << "Reset S3 " << to_string(type)

                  << " QPS rate limiter, qps=" << current.token_per_second

                  << ", burst=" << current.bucket_tokens << ", limit=" << current.token_limit

                  << ", cpu_cores=" << logged_cpu_cores;

    }

    state.applied_config = current;

}

} // namespace

S3RateLimiterHolder* S3ClientFactory::rate_limiter(S3RateLimitType type) {

    CHECK(type == S3RateLimitType::GET || type == S3RateLimitType::PUT) << to_string(type);

    return _rate_limiters[static_cast<size_t>(type)].get();

}

S3RateLimiterHolder* S3ClientFactory::bytes_rate_limiter(S3RateLimitType type) {

    CHECK(type == S3RateLimitType::GET || type == S3RateLimitType::PUT) << to_string(type);

    return _bytes_rate_limiters[static_cast<size_t>(type)].get();

}

bool acquire_s3_qps_rate_limit(S3RateLimitType type) {

    if (!config::enable_s3_rate_limiter) {

        return true;

    }

    auto& factory = S3ClientFactory::instance();

    auto* qps_limiter = factory.rate_limiter(type);

    if (qps_limiter->is_enabled() &&

        doris::apply_s3_rate_limit(type, qps_limiter, config::s3_rate_limiter_log_interval) < 0) {

        return false;

    }

    return true;

}

bool acquire_s3_qps_rate_limit(bool is_internal_bucket, S3RateLimitType type) {

    // In cloud mode, the shared limiter only protects traffic to Doris internal object storage

    // buckets.

    if (config::is_cloud_mode() && !is_internal_bucket) {

        return true;

    }

    return acquire_s3_qps_rate_limit(type);

}

bool acquire_s3_bytes_rate_limit(S3RateLimitType type, size_t bytes) {

    if (!config::enable_s3_rate_limiter || bytes == 0) {

        return true;

    }

    auto* bytes_limiter = S3ClientFactory::instance().bytes_rate_limiter(type);

    if (!bytes_limiter->is_enabled()) {

        return true;

    }

    return bytes_limiter->add(bytes) >= 0;

}

bool acquire_s3_bytes_rate_limit(bool is_internal_bucket, S3RateLimitType type, size_t bytes) {

    // In cloud mode, the shared limiter only protects traffic to Doris internal object storage

    // buckets.

    if (config::is_cloud_mode() && !is_internal_bucket) {

        return true;

    }

    return acquire_s3_bytes_rate_limit(type, bytes);

}

bool acquire_s3_rate_limit(S3RateLimitType type, size_t bytes) {

    return acquire_s3_qps_rate_limit(type) && acquire_s3_bytes_rate_limit(type, bytes);

}

bool acquire_s3_rate_limit(bool is_internal_bucket, S3RateLimitType type, size_t bytes) {

    return acquire_s3_qps_rate_limit(is_internal_bucket, type) &&

           acquire_s3_bytes_rate_limit(is_internal_bucket, type, bytes);

}

void refresh_s3_rate_limiters() {

    auto& factory = S3ClientFactory::instance();

    const int64_t cpu_cores = get_s3_rate_limiter_cpu_cores();

    for (auto type : {S3RateLimitType::GET, S3RateLimitType::PUT}) {

        update_s3_rate_limiter(factory, type, S3RateLimiterKind::QPS, cpu_cores);

        update_s3_rate_limiter(factory, type, S3RateLimiterKind::BYTES, cpu_cores);

    }

}

#ifdef BE_TEST

S3RateLimiterConfig get_effective_s3_rate_limiter_config(S3RateLimitType type) {

    return get_effective_s3_rate_limiter_config_with_cpu_cores(type,

                                                               get_s3_rate_limiter_cpu_cores());

}

S3RateLimiterConfig get_effective_s3_bytes_rate_limiter_config(S3RateLimitType type) {

    return get_effective_s3_bytes_rate_limiter_config_with_cpu_cores(

            type, get_s3_rate_limiter_cpu_cores());

}

void set_s3_rate_limiter_cpu_cores_for_test(int64_t cpu_cores) {

    s3_rate_limiter_cpu_cores_for_test.store(cpu_cores, std::memory_order_relaxed);

}

void reset_s3_rate_limiters_from_config_for_test() {

    auto& factory = S3ClientFactory::instance();

    const int64_t cpu_cores = get_s3_rate_limiter_cpu_cores();

    auto reset = [&](S3RateLimitType type, S3RateLimiterKind kind) {

        auto& state = get_update_state(type, kind);

        std::lock_guard update_guard(state.update_lock);

        const auto config =

                kind == S3RateLimiterKind::BYTES

                        ? get_effective_s3_bytes_rate_limiter_config_with_cpu_cores(type, cpu_cores)

                        : get_effective_s3_rate_limiter_config_with_cpu_cores(type, cpu_cores);

        auto* limiter = kind == S3RateLimiterKind::BYTES ? factory.bytes_rate_limiter(type)

                                                         : factory.rate_limiter(type);

        CHECK_EQ(limiter->reset(config.token_per_second, config.bucket_tokens, config.token_limit),

                 0);

        state.applied_config = config;

    };

    for (auto type : {S3RateLimitType::GET, S3RateLimitType::PUT}) {

        reset(type, S3RateLimiterKind::QPS);

        reset(type, S3RateLimiterKind::BYTES);

    }

}

#endif

int reset_s3_rate_limiter(S3RateLimitType type, size_t max_speed, size_t max_burst, size_t limit) {

    if (type == S3RateLimitType::UNKNOWN) {

        return -1;

    }

    return S3ClientFactory::instance().rate_limiter(type)->reset(max_speed, max_burst, limit);

}

// The analyzer does not model destruction of the process-lifetime singleton's owned resources.

S3ClientFactory::S3ClientFactory() { // NOLINT(clang-analyzer-cplusplus.NewDeleteLeaks)

    _aws_options = Aws::SDKOptions {};

    auto logLevel = static_cast<Aws::Utils::Logging::LogLevel>(config::aws_log_level);

    _aws_options.loggingOptions.logLevel = logLevel;

    _aws_options.loggingOptions.logger_create_fn = [logLevel] {

        return std::make_shared<DorisAWSLogger>(logLevel);

    };

    Aws::InitAPI(_aws_options);

    _ca_cert_file_path = get_valid_ca_cert_path(doris::split(config::ca_cert_file_paths, ";"));

    const int64_t cpu_cores = get_s3_rate_limiter_cpu_cores();

    auto get_init_config =

            get_effective_s3_rate_limiter_config_with_cpu_cores(S3RateLimitType::GET, cpu_cores);

    auto put_init_config =

            get_effective_s3_rate_limiter_config_with_cpu_cores(S3RateLimitType::PUT, cpu_cores);

    _rate_limiters = {

            std::make_unique<S3RateLimiterHolder>(

                    get_init_config.token_per_second, get_init_config.bucket_tokens,

                    get_init_config.token_limit, s3_rate_limiter_metric_func(S3RateLimitType::GET)),

            std::make_unique<S3RateLimiterHolder>(

                    put_init_config.token_per_second, put_init_config.bucket_tokens,

                    put_init_config.token_limit,

                    s3_rate_limiter_metric_func(S3RateLimitType::PUT))};

    auto get_bytes_init_config = get_effective_s3_bytes_rate_limiter_config_with_cpu_cores(

            S3RateLimitType::GET, cpu_cores);

    auto put_bytes_init_config = get_effective_s3_bytes_rate_limiter_config_with_cpu_cores(

            S3RateLimitType::PUT, cpu_cores);

    _bytes_rate_limiters = {

            std::make_unique<S3RateLimiterHolder>(

                    get_bytes_init_config.token_per_second, get_bytes_init_config.bucket_tokens,

                    get_bytes_init_config.token_limit,

                    metric_func_factory(get_bytes_rate_limit_ns,

                                        get_bytes_rate_limit_exceed_req_num)),

            std::make_unique<S3RateLimiterHolder>(

                    put_bytes_init_config.token_per_second, put_bytes_init_config.bucket_tokens,

                    put_bytes_init_config.token_limit,

                    metric_func_factory(put_bytes_rate_limit_ns,

                                        put_bytes_rate_limit_exceed_req_num))};

    get_update_state(S3RateLimitType::GET, S3RateLimiterKind::QPS).applied_config = get_init_config;

    get_update_state(S3RateLimitType::PUT, S3RateLimiterKind::QPS).applied_config = put_init_config;

    get_update_state(S3RateLimitType::GET, S3RateLimiterKind::BYTES).applied_config =

            get_bytes_init_config;

    get_update_state(S3RateLimitType::PUT, S3RateLimiterKind::BYTES).applied_config =

            put_bytes_init_config;

#ifdef USE_AZURE

    auto azureLogLevel =

            static_cast<Azure::Core::Diagnostics::Logger::Level>(config::azure_log_level);

    Azure::Core::Diagnostics::Logger::SetLevel(azureLogLevel);

    Azure::Core::Diagnostics::Logger::SetListener(

            [&](Azure::Core::Diagnostics::Logger::Level level, const std::string& message) {

                switch (level) {

                case Azure::Core::Diagnostics::Logger::Level::Verbose:

                    LOG(INFO) << message;

                    break;

                case Azure::Core::Diagnostics::Logger::Level::Informational:

                    LOG(INFO) << message;

                    break;

                case Azure::Core::Diagnostics::Logger::Level::Warning:

                    LOG(WARNING) << message;

                    break;

                case Azure::Core::Diagnostics::Logger::Level::Error:

                    LOG(ERROR) << message;

                    break;

                default:

                    LOG(WARNING) << "Unknown level: " << static_cast<int>(level)

                                 << ", message: " << message;

                    break;

                }

            });

#endif

}

S3ClientFactory::~S3ClientFactory() {

    Aws::ShutdownAPI(_aws_options);

}

S3ClientFactory& S3ClientFactory::instance() {

    static S3ClientFactory ret;

    return ret;

}

std::shared_ptr<io::ObjStorageClient> S3ClientFactory::create(const S3ClientConf& s3_conf) {

    if (!is_s3_conf_valid(s3_conf).ok()) {

        return nullptr;

    }

#ifdef BE_TEST

    {

        std::lock_guard l(_lock);

        if (_test_client_creator) {

            return _test_client_creator(s3_conf);

        }

    }

#endif

    {

        std::lock_guard l(_lock);

        auto it = _cache.find(s3_conf);

        if (it != _cache.end()) {

            return it->second;

        }

    }

    auto obj_client = (s3_conf.provider == io::ObjStorageType::AZURE)

                              ? _create_azure_client(s3_conf)

                              : _create_s3_client(s3_conf);

    {

        std::lock_guard l(_lock);

        auto [it, _] = _cache.emplace(s3_conf, obj_client);

        return it->second;

    }

}

#ifdef BE_TEST

void S3ClientFactory::set_client_creator_for_test(

        std::function<std::shared_ptr<io::ObjStorageClient>(const S3ClientConf&)> creator) {

    std::lock_guard l(_lock);

    _test_client_creator = std::move(creator);

}

void S3ClientFactory::clear_client_creator_for_test() {

    std::lock_guard l(_lock);

    _test_client_creator = nullptr;

}

#endif

std::shared_ptr<io::ObjStorageClient> S3ClientFactory::_create_azure_client(

        const S3ClientConf& s3_conf) {

#ifdef USE_AZURE

    auto cred =

            std::make_shared<Azure::Storage::StorageSharedKeyCredential>(s3_conf.ak, s3_conf.sk);

    const std::string container_name = s3_conf.bucket;

    std::string uri = fmt::format("{}/{}", s3_conf.endpoint, container_name);

    if (s3_conf.endpoint.find("://") == std::string::npos) {

        uri = "https://" + uri;

    }

    Azure::Storage::Blobs::BlobClientOptions options;

    options.Retry.StatusCodes.insert(Azure::Core::Http::HttpStatusCode::TooManyRequests);

    options.Retry.MaxRetries = config::max_s3_client_retry;

    options.PerRetryPolicies.emplace_back(

            io::create_azure_bytes_rate_limit_policy(s3_conf.is_internal_bucket));

    options.PerRetryPolicies.emplace_back(std::make_unique<AzureRetryRecordPolicy>());

    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()) {

        Azure::Core::Http::CurlTransportOptions curl_options;

        curl_options.CAInfo = _ca_cert_file_path;

        options.Transport.Transport =

                std::make_shared<Azure::Core::Http::CurlTransport>(std::move(curl_options));

    }

    std::string normalized_uri = normalize_http_uri(uri);

    VLOG_DEBUG << "uri:" << uri << ", normalized_uri:" << normalized_uri;

    std::string tls_debug_context = build_azure_tls_debug_context(_ca_cert_file_path);

    auto containerClient = std::make_shared<Azure::Storage::Blobs::BlobContainerClient>(

            uri, cred, std::move(options));

    LOG_INFO("create one azure client with {}", s3_conf.to_string());

    return std::make_shared<io::AzureObjStorageClient>(

            std::move(containerClient), s3_conf.is_internal_bucket, std::move(tls_debug_context));

#else

    LOG_FATAL("BE is not compiled with azure support, export BUILD_AZURE=ON before building");

    return nullptr;

#endif

}

std::shared_ptr<Aws::Auth::AWSCredentialsProvider>

S3ClientFactory::_get_aws_credentials_provider_v1(const S3ClientConf& s3_conf) {

    if (!s3_conf.ak.empty() && !s3_conf.sk.empty()) {

        Aws::Auth::AWSCredentials aws_cred(s3_conf.ak, s3_conf.sk);

        DCHECK(!aws_cred.IsExpiredOrEmpty());

        if (!s3_conf.token.empty()) {

            aws_cred.SetSessionToken(s3_conf.token);

        }

        return std::make_shared<Aws::Auth::SimpleAWSCredentialsProvider>(std::move(aws_cred));

    }

    if (s3_conf.cred_provider_type == CredProviderType::InstanceProfile) {

        if (s3_conf.role_arn.empty()) {

            return std::make_shared<Aws::Auth::InstanceProfileCredentialsProvider>();

        }

        Aws::Client::ClientConfiguration clientConfiguration =

                S3ClientFactory::getClientConfiguration();

        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()) {

            clientConfiguration.caFile = _ca_cert_file_path;

        }

        auto stsClient = std::make_shared<Aws::STS::STSClient>(

                std::make_shared<Aws::Auth::InstanceProfileCredentialsProvider>(),

                clientConfiguration);

        return std::make_shared<Aws::Auth::STSAssumeRoleCredentialsProvider>(

                s3_conf.role_arn, Aws::String(), s3_conf.external_id,

                Aws::Auth::DEFAULT_CREDS_LOAD_FREQ_SECONDS, stsClient);

    }

    // Support anonymous access for public datasets when no credentials are provided

    if (s3_conf.ak.empty() && s3_conf.sk.empty()) {

        return std::make_shared<Aws::Auth::AnonymousAWSCredentialsProvider>();

    }

    return std::make_shared<Aws::Auth::DefaultAWSCredentialsProviderChain>();

}

std::shared_ptr<Aws::Auth::AWSCredentialsProvider> S3ClientFactory::_create_credentials_provider(

        CredProviderType type) {

    switch (type) {

    case CredProviderType::Env:

        return std::make_shared<Aws::Auth::EnvironmentAWSCredentialsProvider>();

    case CredProviderType::SystemProperties:

        return std::make_shared<Aws::Auth::ProfileConfigFileAWSCredentialsProvider>();

    case CredProviderType::WebIdentity:

        return std::make_shared<Aws::Auth::STSAssumeRoleWebIdentityCredentialsProvider>();

    case CredProviderType::Container:

        return std::make_shared<Aws::Auth::TaskRoleCredentialsProvider>(

                Aws::Environment::GetEnv("AWS_CONTAINER_CREDENTIALS_RELATIVE_URI").c_str());

    case CredProviderType::InstanceProfile:

        return std::make_shared<Aws::Auth::InstanceProfileCredentialsProvider>();

    case CredProviderType::Anonymous:

        return std::make_shared<Aws::Auth::AnonymousAWSCredentialsProvider>();

    case CredProviderType::Default:

    default:

        return std::make_shared<CustomAwsCredentialsProviderChain>();

    }

}

std::shared_ptr<Aws::Auth::AWSCredentialsProvider>

S3ClientFactory::_get_aws_credentials_provider_v2(const S3ClientConf& s3_conf) {

    if (!s3_conf.ak.empty() && !s3_conf.sk.empty()) {

        Aws::Auth::AWSCredentials aws_cred(s3_conf.ak, s3_conf.sk);

        DCHECK(!aws_cred.IsExpiredOrEmpty());

        if (!s3_conf.token.empty()) {

            aws_cred.SetSessionToken(s3_conf.token);

        }

        return std::make_shared<Aws::Auth::SimpleAWSCredentialsProvider>(std::move(aws_cred));

    }

    // Handle role_arn for assume role scenario

    if (!s3_conf.role_arn.empty()) {

        Aws::Client::ClientConfiguration clientConfiguration =

                S3ClientFactory::getClientConfiguration();

        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()) {

            clientConfiguration.caFile = _ca_cert_file_path;

        }

        auto baseProvider = _create_credentials_provider(s3_conf.cred_provider_type);

        auto stsClient = std::make_shared<Aws::STS::STSClient>(baseProvider, clientConfiguration);

        return std::make_shared<Aws::Auth::STSAssumeRoleCredentialsProvider>(

                s3_conf.role_arn, Aws::String(), s3_conf.external_id,

                Aws::Auth::DEFAULT_CREDS_LOAD_FREQ_SECONDS, stsClient);

    }

    // Return provider based on cred_provider_type

    return _create_credentials_provider(s3_conf.cred_provider_type);

}

std::shared_ptr<Aws::Auth::AWSCredentialsProvider> S3ClientFactory::get_aws_credentials_provider(

        const S3ClientConf& s3_conf) {

    if (config::aws_credentials_provider_version == "v2") {

        return _get_aws_credentials_provider_v2(s3_conf);

    }

    return _get_aws_credentials_provider_v1(s3_conf);

}

std::shared_ptr<io::ObjStorageClient> S3ClientFactory::_create_s3_client(

        const S3ClientConf& s3_conf) {

    TEST_SYNC_POINT_RETURN_WITH_VALUE(

            "s3_client_factory::create",

            std::make_shared<io::S3ObjStorageClient>(std::make_shared<Aws::S3::S3Client>(),

                                                     s3_conf.is_internal_bucket));

    Aws::Client::ClientConfiguration aws_config = S3ClientFactory::getClientConfiguration();

    if (s3_conf.need_override_endpoint) {

        aws_config.endpointOverride = s3_conf.endpoint;

    }

    aws_config.region = s3_conf.region;

    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()) {

        aws_config.caFile = _ca_cert_file_path;

    }

    if (s3_conf.max_connections > 0) {

        aws_config.maxConnections = s3_conf.max_connections;

    } else {

        aws_config.maxConnections = 102400;

    }

    aws_config.requestTimeoutMs = 30000;

    if (s3_conf.request_timeout_ms > 0) {

        aws_config.requestTimeoutMs = s3_conf.request_timeout_ms;

    }

    if (s3_conf.connect_timeout_ms > 0) {

        aws_config.connectTimeoutMs = s3_conf.connect_timeout_ms;

    }

    if (config::s3_client_http_scheme == "http") {

        aws_config.scheme = Aws::Http::Scheme::HTTP;

    }

    aws_config.retryStrategy = std::make_shared<S3CustomRetryStrategy>(

            config::max_s3_client_retry /*scaleFactor = 25*/, /*retry_slow_down=*/true);

    std::shared_ptr<Aws::S3::S3Client> new_client = std::make_shared<Aws::S3::S3Client>(

            get_aws_credentials_provider(s3_conf), std::move(aws_config),

            Aws::Client::AWSAuthV4Signer::PayloadSigningPolicy::Never,

            s3_conf.use_virtual_addressing);

    auto obj_client = std::make_shared<io::S3ObjStorageClient>(std::move(new_client),

                                                               s3_conf.is_internal_bucket);

    LOG_INFO("create one s3 client with {}", s3_conf.to_string());

    return obj_client;

}

Status S3ClientFactory::convert_properties_to_s3_conf(

        const std::map<std::string, std::string>& prop, const S3URI& s3_uri, S3Conf* s3_conf) {

    StringCaseMap<std::string> properties(prop.begin(), prop.end());

    if (auto it = properties.find(S3_AK); it != properties.end()) {

        s3_conf->client_conf.ak = it->second;

    }

    if (auto it = properties.find(S3_SK); it != properties.end()) {

        s3_conf->client_conf.sk = it->second;

    }

    if (auto it = properties.find(S3_TOKEN); it != properties.end()) {

        s3_conf->client_conf.token = it->second;

    }

    if (auto it = properties.find(S3_ENDPOINT); it != properties.end()) {

        s3_conf->client_conf.endpoint = it->second;

    }

    if (auto it = properties.find(S3_NEED_OVERRIDE_ENDPOINT); it != properties.end()) {

        s3_conf->client_conf.need_override_endpoint = (it->second == "true");

    }

    if (auto it = properties.find(S3_REGION); it != properties.end()) {

        s3_conf->client_conf.region = it->second;

    }

    if (auto it = properties.find(S3_MAX_CONN_SIZE); it != properties.end()) {

        if (!to_int(it->second, s3_conf->client_conf.max_connections)) {

            return Status::InvalidArgument("invalid {} value \"{}\"", S3_MAX_CONN_SIZE, it->second);

        }

    }

    if (auto it = properties.find(S3_REQUEST_TIMEOUT_MS); it != properties.end()) {

        if (!to_int(it->second, s3_conf->client_conf.request_timeout_ms)) {

            return Status::InvalidArgument("invalid {} value \"{}\"", S3_REQUEST_TIMEOUT_MS,

                                           it->second);

        }

    }

    if (auto it = properties.find(S3_CONN_TIMEOUT_MS); it != properties.end()) {

        if (!to_int(it->second, s3_conf->client_conf.connect_timeout_ms)) {

            return Status::InvalidArgument("invalid {} value \"{}\"", S3_CONN_TIMEOUT_MS,

                                           it->second);

        }

    }

    if (auto it = properties.find(S3_PROVIDER); it != properties.end()) {

        // 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;

        }

    }

    if (s3_uri.get_bucket().empty()) {

        return Status::InvalidArgument("Invalid S3 URI {}, bucket is not specified",

                                       s3_uri.to_string());

    }

    s3_conf->bucket = s3_uri.get_bucket();

    // For azure's compatibility

    s3_conf->client_conf.bucket = s3_uri.get_bucket();

    s3_conf->prefix = "";

    // See https://sdk.amazonaws.com/cpp/api/LATEST/class_aws_1_1_s3_1_1_s3_client.html

    s3_conf->client_conf.use_virtual_addressing = true;

    if (auto it = properties.find(USE_PATH_STYLE); it != properties.end()) {

        s3_conf->client_conf.use_virtual_addressing = it->second != "true";

    }

    if (auto it = properties.find(S3_ROLE_ARN); it != properties.end()) {

        // Keep provider type as Default unless explicitly configured by

        // AWS_CREDENTIALS_PROVIDER_TYPE, consistent with FE behavior.

        s3_conf->client_conf.role_arn = it->second;

    }

    if (auto it = properties.find(S3_EXTERNAL_ID); it != properties.end()) {

        s3_conf->client_conf.external_id = it->second;

    }

    if (auto it = properties.find(S3_CREDENTIALS_PROVIDER_TYPE); it != properties.end()) {

        s3_conf->client_conf.cred_provider_type = cred_provider_type_from_string(it->second);

    }

    if (auto st = is_s3_conf_valid(s3_conf->client_conf); !st.ok()) {

        return st;

    }

    return Status::OK();

}

static CredProviderType cred_provider_type_from_thrift(TCredProviderType::type cred_provider_type) {

    switch (cred_provider_type) {

    case TCredProviderType::DEFAULT:

        return CredProviderType::Default;

    case TCredProviderType::SIMPLE:

        return CredProviderType::Simple;

    case TCredProviderType::INSTANCE_PROFILE:

        return CredProviderType::InstanceProfile;

    case TCredProviderType::ENV:

        return CredProviderType::Env;

    case TCredProviderType::SYSTEM_PROPERTIES:

        return CredProviderType::SystemProperties;

    case TCredProviderType::WEB_IDENTITY:

        return CredProviderType::WebIdentity;

    case TCredProviderType::CONTAINER:

        return CredProviderType::Container;

    case TCredProviderType::ANONYMOUS:

        return CredProviderType::Anonymous;

    default:

        __builtin_unreachable();

        LOG(WARNING) << "Invalid TCredProviderType value: " << cred_provider_type

                     << ", use default instead.";

        return CredProviderType::Default;

    }

}

S3Conf S3Conf::get_s3_conf(const cloud::ObjectStoreInfoPB& info) {

    S3Conf ret {

            .bucket = info.bucket(),

            .prefix = info.prefix(),

            .client_conf {

                    .endpoint = info.endpoint(),

                    .region = info.region(),

                    .ak = info.ak(),

                    .sk = info.sk(),

                    .token {},

                    .bucket = info.bucket(),

                    .provider = io::ObjStorageType::AWS,

                    .use_virtual_addressing =

                            info.has_use_path_style() ? !info.use_path_style() : true,

                    .role_arn = info.role_arn(),

                    .external_id = info.external_id(),

                    .is_internal_bucket = true,

            },

            .sse_enabled = info.sse_enabled(),

    };

    if (info.has_cred_provider_type()) {

        ret.client_conf.cred_provider_type = cred_provider_type_from_pb(info.cred_provider_type());

    }

    io::ObjStorageType type = io::ObjStorageType::AWS;

    switch (info.provider()) {

    case cloud::ObjectStoreInfoPB_Provider_OSS:

        type = io::ObjStorageType::OSS;

        break;

    case cloud::ObjectStoreInfoPB_Provider_S3:

        type = io::ObjStorageType::AWS;

        break;

    case cloud::ObjectStoreInfoPB_Provider_COS:

        type = io::ObjStorageType::COS;

        break;

    case cloud::ObjectStoreInfoPB_Provider_OBS:

        type = io::ObjStorageType::OBS;

        break;

    case cloud::ObjectStoreInfoPB_Provider_BOS:

        type = io::ObjStorageType::BOS;

        break;

    case cloud::ObjectStoreInfoPB_Provider_GCP:

        type = io::ObjStorageType::GCP;

        break;

    case cloud::ObjectStoreInfoPB_Provider_AZURE:

        type = io::ObjStorageType::AZURE;

        break;

    case cloud::ObjectStoreInfoPB_Provider_TOS:

        type = io::ObjStorageType::TOS;

        break;

    default:

        __builtin_unreachable();

        LOG_FATAL("unknown provider type {}, info {}", info.provider(), ret.to_string());

    }

    ret.client_conf.provider = type;

    return ret;

}

S3Conf S3Conf::get_s3_conf(const TS3StorageParam& param) {

    S3Conf ret {

            .bucket = param.bucket,

            .prefix = param.root_path,

            .client_conf = {

                    .endpoint = param.endpoint,

                    .region = param.region,

                    .ak = param.ak,

                    .sk = param.sk,

                    .token = param.token,

                    .bucket = param.bucket,

                    .provider = io::ObjStorageType::AWS,

                    .max_connections = param.max_conn,

                    .request_timeout_ms = param.request_timeout_ms,