Coverage Report

Created: 2026-07-20 13:36

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/util/s3_util.cpp
Line
Count
Source
1
// Licensed to the Apache Software Foundation (ASF) under one
2
// or more contributor license agreements.  See the NOTICE file
3
// distributed with this work for additional information
4
// regarding copyright ownership.  The ASF licenses this file
5
// to you under the Apache License, Version 2.0 (the
6
// "License"); you may not use this file except in compliance
7
// with the License.  You may obtain a copy of the License at
8
//
9
//   http://www.apache.org/licenses/LICENSE-2.0
10
//
11
// Unless required by applicable law or agreed to in writing,
12
// software distributed under the License is distributed on an
13
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14
// KIND, either express or implied.  See the License for the
15
// specific language governing permissions and limitations
16
// under the License.
17
18
#include "util/s3_util.h"
19
20
#include <aws/core/auth/AWSAuthSigner.h>
21
#include <aws/core/auth/AWSCredentials.h>
22
#include <aws/core/auth/AWSCredentialsProviderChain.h>
23
#include <aws/core/auth/STSCredentialsProvider.h>
24
#include <aws/core/client/DefaultRetryStrategy.h>
25
#include <aws/core/platform/Environment.h>
26
#include <aws/core/utils/logging/LogLevel.h>
27
#include <aws/core/utils/logging/LogSystemInterface.h>
28
#include <aws/core/utils/memory/stl/AWSStringStream.h>
29
#include <aws/identity-management/auth/STSAssumeRoleCredentialsProvider.h>
30
#include <aws/s3/S3Client.h>
31
#include <aws/sts/STSClient.h>
32
#include <bvar/reducer.h>
33
#include <cpp/token_bucket_rate_limiter.h>
34
35
#include <atomic>
36
37
#include "util/string_util.h"
38
39
#ifdef USE_AZURE
40
#include <azure/core/diagnostics/logger.hpp>
41
#include <azure/core/http/curl_transport.hpp>
42
#include <azure/storage/blobs/blob_container_client.hpp>
43
#endif
44
#include <cstdlib>
45
#include <filesystem>
46
#include <fstream>
47
#include <functional>
48
#include <memory>
49
#include <ostream>
50
#include <utility>
51
52
#include "common/config.h"
53
#include "common/logging.h"
54
#include "common/status.h"
55
#include "cpp/aws_logger.h"
56
#include "cpp/custom_aws_credentials_provider_chain.h"
57
#include "cpp/obj_retry_strategy.h"
58
#include "cpp/sync_point.h"
59
#include "cpp/util.h"
60
#ifdef USE_AZURE
61
#include "io/fs/azure_obj_storage_client.h"
62
#endif
63
#include "cloud/config.h"
64
#include "exec/scan/scanner_scheduler.h"
65
#include "io/fs/obj_storage_client.h"
66
#include "io/fs/rate_limited_obj_storage_client.h"
67
#include "io/fs/s3_obj_storage_client.h"
68
#include "runtime/exec_env.h"
69
#include "util/s3_uri.h"
70
71
namespace doris {
72
namespace s3_bvar {
73
bvar::LatencyRecorder s3_get_latency("s3_get");
74
bvar::LatencyRecorder s3_put_latency("s3_put");
75
bvar::LatencyRecorder s3_delete_object_latency("s3_delete_object");
76
bvar::LatencyRecorder s3_delete_objects_latency("s3_delete_objects");
77
bvar::LatencyRecorder s3_head_latency("s3_head");
78
bvar::LatencyRecorder s3_multi_part_upload_latency("s3_multi_part_upload");
79
bvar::LatencyRecorder s3_list_latency("s3_list");
80
bvar::LatencyRecorder s3_list_object_versions_latency("s3_list_object_versions");
81
bvar::LatencyRecorder s3_get_bucket_version_latency("s3_get_bucket_version");
82
bvar::LatencyRecorder s3_copy_object_latency("s3_copy_object");
83
}; // namespace s3_bvar
84
85
namespace {
86
87
35
doris::Status is_s3_conf_valid(const S3ClientConf& conf) {
88
35
    if (conf.endpoint.empty()) {
89
0
        return Status::InvalidArgument<false>("Invalid s3 conf, empty endpoint");
90
0
    }
91
35
    if (conf.region.empty()) {
92
0
        return Status::InvalidArgument<false>("Invalid s3 conf, empty region");
93
0
    }
94
95
35
    if (conf.role_arn.empty()) {
96
        // Allow anonymous access when both ak and sk are empty
97
29
        bool hasAk = !conf.ak.empty();
98
29
        bool hasSk = !conf.sk.empty();
99
100
        // Either both credentials are provided or both are empty (anonymous access)
101
29
        if (hasAk && conf.sk.empty()) {
102
1
            return Status::InvalidArgument<false>("Invalid s3 conf, empty sk");
103
1
        }
104
28
        if (hasSk && conf.ak.empty()) {
105
1
            return Status::InvalidArgument<false>("Invalid s3 conf, empty ak");
106
1
        }
107
28
    }
108
33
    return Status::OK();
109
35
}
110
111
// Return true is convert `str` to int successfully
112
0
bool to_int(std::string_view str, int& res) {
113
0
    auto [_, ec] = std::from_chars(str.data(), str.data() + str.size(), res);
114
0
    return ec == std::errc {};
115
0
}
116
117
#ifdef USE_AZURE
118
0
std::string env_or_empty(const char* env_name) {
119
0
    if (const char* value = std::getenv(env_name); value != nullptr) {
120
0
        return value;
121
0
    }
122
0
    return "";
123
0
}
124
125
0
std::string build_azure_tls_debug_context(const std::string& selected_ca_file) {
126
0
    bool selected_ca_exists = false;
127
0
    bool selected_ca_readable = false;
128
0
    if (!selected_ca_file.empty()) {
129
0
        std::error_code ec;
130
0
        selected_ca_exists = std::filesystem::exists(selected_ca_file, ec) && !ec;
131
0
        std::ifstream input(selected_ca_file);
132
0
        selected_ca_readable = input.good();
133
0
    }
134
135
0
    return fmt::format(
136
0
            "tls_debug(ca_cert_file_paths='{}', selected_ca_file='{}', selected_ca_exists={}, "
137
0
            "selected_ca_readable={}, SSL_CERT_FILE='{}', CURL_CA_BUNDLE='{}', SSL_CERT_DIR='{}')",
138
0
            config::ca_cert_file_paths, selected_ca_file, selected_ca_exists, selected_ca_readable,
139
0
            env_or_empty("SSL_CERT_FILE"), env_or_empty("CURL_CA_BUNDLE"),
140
0
            env_or_empty("SSL_CERT_DIR"));
141
0
}
142
#endif
143
144
constexpr char USE_PATH_STYLE[] = "use_path_style";
145
146
constexpr char AZURE_PROVIDER_STRING[] = "AZURE";
147
constexpr char S3_PROVIDER[] = "provider";
148
constexpr char S3_AK[] = "AWS_ACCESS_KEY";
149
constexpr char S3_SK[] = "AWS_SECRET_KEY";
150
constexpr char S3_ENDPOINT[] = "AWS_ENDPOINT";
151
constexpr char S3_REGION[] = "AWS_REGION";
152
constexpr char S3_TOKEN[] = "AWS_TOKEN";
153
constexpr char S3_MAX_CONN_SIZE[] = "AWS_MAX_CONNECTIONS";
154
constexpr char S3_REQUEST_TIMEOUT_MS[] = "AWS_REQUEST_TIMEOUT_MS";
155
constexpr char S3_CONN_TIMEOUT_MS[] = "AWS_CONNECTION_TIMEOUT_MS";
156
constexpr char S3_NEED_OVERRIDE_ENDPOINT[] = "AWS_NEED_OVERRIDE_ENDPOINT";
157
158
constexpr char S3_ROLE_ARN[] = "AWS_ROLE_ARN";
159
constexpr char S3_EXTERNAL_ID[] = "AWS_EXTERNAL_ID";
160
constexpr char S3_CREDENTIALS_PROVIDER_TYPE[] = "AWS_CREDENTIALS_PROVIDER_TYPE";
161
} // namespace
162
163
1
S3ClientFactory::S3ClientFactory() {
164
1
    _aws_options = Aws::SDKOptions {};
165
1
    auto logLevel = static_cast<Aws::Utils::Logging::LogLevel>(config::aws_log_level);
166
1
    _aws_options.loggingOptions.logLevel = logLevel;
167
1
    _aws_options.loggingOptions.logger_create_fn = [logLevel] {
168
1
        return std::make_shared<DorisAWSLogger>(logLevel);
169
1
    };
170
1
    Aws::InitAPI(_aws_options);
171
1
    _ca_cert_file_path = get_valid_ca_cert_path(doris::split(config::ca_cert_file_paths, ";"));
172
173
1
#ifdef USE_AZURE
174
1
    auto azureLogLevel =
175
1
            static_cast<Azure::Core::Diagnostics::Logger::Level>(config::azure_log_level);
176
1
    Azure::Core::Diagnostics::Logger::SetLevel(azureLogLevel);
177
1
    Azure::Core::Diagnostics::Logger::SetListener(
178
1
            [&](Azure::Core::Diagnostics::Logger::Level level, const std::string& message) {
179
0
                switch (level) {
180
0
                case Azure::Core::Diagnostics::Logger::Level::Verbose:
181
0
                    LOG(INFO) << message;
182
0
                    break;
183
0
                case Azure::Core::Diagnostics::Logger::Level::Informational:
184
0
                    LOG(INFO) << message;
185
0
                    break;
186
0
                case Azure::Core::Diagnostics::Logger::Level::Warning:
187
0
                    LOG(WARNING) << message;
188
0
                    break;
189
0
                case Azure::Core::Diagnostics::Logger::Level::Error:
190
0
                    LOG(ERROR) << message;
191
0
                    break;
192
0
                default:
193
0
                    LOG(WARNING) << "Unknown level: " << static_cast<int>(level)
194
0
                                 << ", message: " << message;
195
0
                    break;
196
0
                }
197
0
            });
198
1
#endif
199
1
}
200
201
1
S3ClientFactory::~S3ClientFactory() {
202
1
    Aws::ShutdownAPI(_aws_options);
203
1
}
204
205
38
S3ClientFactory& S3ClientFactory::instance() {
206
38
    static S3ClientFactory ret;
207
38
    return ret;
208
38
}
209
210
20
std::shared_ptr<io::ObjStorageClient> S3ClientFactory::create(const S3ClientConf& s3_conf) {
211
20
    if (!is_s3_conf_valid(s3_conf).ok()) {
212
0
        return nullptr;
213
0
    }
214
215
20
#ifdef BE_TEST
216
20
    {
217
20
        std::lock_guard l(_lock);
218
20
        if (_test_client_creator) {
219
4
            return _test_client_creator(s3_conf);
220
4
        }
221
20
    }
222
16
#endif
223
224
16
    {
225
16
        std::lock_guard l(_lock);
226
16
        auto it = _cache.find(s3_conf);
227
16
        if (it != _cache.end()) {
228
9
            return it->second;
229
9
        }
230
16
    }
231
232
7
    auto obj_client = (s3_conf.provider == io::ObjStorageType::AZURE)
233
7
                              ? _create_azure_client(s3_conf)
234
7
                              : _create_s3_client(s3_conf);
235
236
    // Rate limiting lives in one decorator, decided here at construction time:
237
    // in cloud mode only internal storage-vault buckets are limited; external buckets
238
    // (S3 load, TVF, external catalogs) get the bare client. In non-cloud mode every
239
    // client is wrapped, preserving the legacy behavior.
240
7
    if (obj_client != nullptr && (!config::is_cloud_mode() || s3_conf.is_internal_bucket)) {
241
6
        obj_client = std::make_shared<io::RateLimitedObjStorageClient>(std::move(obj_client));
242
6
    }
243
244
7
    {
245
7
        std::lock_guard l(_lock);
246
7
        auto [it, _] = _cache.emplace(s3_conf, std::move(obj_client));
247
7
        return it->second;
248
16
    }
249
16
}
250
251
#ifdef BE_TEST
252
void S3ClientFactory::set_client_creator_for_test(
253
3
        std::function<std::shared_ptr<io::ObjStorageClient>(const S3ClientConf&)> creator) {
254
3
    std::lock_guard l(_lock);
255
3
    _test_client_creator = std::move(creator);
256
3
}
257
258
13
void S3ClientFactory::clear_client_creator_for_test() {
259
13
    std::lock_guard l(_lock);
260
13
    _test_client_creator = nullptr;
261
13
}
262
#endif
263
264
std::shared_ptr<io::ObjStorageClient> S3ClientFactory::_create_azure_client(
265
0
        const S3ClientConf& s3_conf) {
266
0
#ifdef USE_AZURE
267
0
    auto cred =
268
0
            std::make_shared<Azure::Storage::StorageSharedKeyCredential>(s3_conf.ak, s3_conf.sk);
269
270
0
    const std::string container_name = s3_conf.bucket;
271
0
    std::string uri = fmt::format("{}/{}", s3_conf.endpoint, container_name);
272
0
    if (s3_conf.endpoint.find("://") == std::string::npos) {
273
0
        uri = "https://" + uri;
274
0
    }
275
276
0
    Azure::Storage::Blobs::BlobClientOptions options;
277
0
    options.Retry.StatusCodes.insert(Azure::Core::Http::HttpStatusCode::TooManyRequests);
278
0
    options.Retry.MaxRetries = config::max_s3_client_retry;
279
0
    options.PerRetryPolicies.emplace_back(std::make_unique<AzureRetryRecordPolicy>());
280
0
    if (_ca_cert_file_path.empty()) {
281
0
        _ca_cert_file_path = get_valid_ca_cert_path(doris::split(config::ca_cert_file_paths, ";"));
282
0
    }
283
0
    if (!_ca_cert_file_path.empty()) {
284
0
        Azure::Core::Http::CurlTransportOptions curl_options;
285
0
        curl_options.CAInfo = _ca_cert_file_path;
286
0
        options.Transport.Transport =
287
0
                std::make_shared<Azure::Core::Http::CurlTransport>(std::move(curl_options));
288
0
    }
289
290
0
    std::string normalized_uri = normalize_http_uri(uri);
291
0
    VLOG_DEBUG << "uri:" << uri << ", normalized_uri:" << normalized_uri;
292
0
    std::string tls_debug_context = build_azure_tls_debug_context(_ca_cert_file_path);
293
294
0
    auto containerClient = std::make_shared<Azure::Storage::Blobs::BlobContainerClient>(
295
0
            uri, cred, std::move(options));
296
0
    LOG_INFO("create one azure client with {}", s3_conf.to_string());
297
0
    return std::make_shared<io::AzureObjStorageClient>(std::move(containerClient),
298
0
                                                       std::move(tls_debug_context));
299
#else
300
    LOG_FATAL("BE is not compiled with azure support, export BUILD_AZURE=ON before building");
301
    return nullptr;
302
#endif
303
0
}
304
305
std::shared_ptr<Aws::Auth::AWSCredentialsProvider>
306
6
S3ClientFactory::_get_aws_credentials_provider_v1(const S3ClientConf& s3_conf) {
307
6
    if (!s3_conf.ak.empty() && !s3_conf.sk.empty()) {
308
2
        Aws::Auth::AWSCredentials aws_cred(s3_conf.ak, s3_conf.sk);
309
2
        DCHECK(!aws_cred.IsExpiredOrEmpty());
310
2
        if (!s3_conf.token.empty()) {
311
0
            aws_cred.SetSessionToken(s3_conf.token);
312
0
        }
313
2
        return std::make_shared<Aws::Auth::SimpleAWSCredentialsProvider>(std::move(aws_cred));
314
2
    }
315
316
4
    if (s3_conf.cred_provider_type == CredProviderType::InstanceProfile) {
317
2
        if (s3_conf.role_arn.empty()) {
318
1
            return std::make_shared<Aws::Auth::InstanceProfileCredentialsProvider>();
319
1
        }
320
321
1
        Aws::Client::ClientConfiguration clientConfiguration =
322
1
                S3ClientFactory::getClientConfiguration();
323
324
1
        if (_ca_cert_file_path.empty()) {
325
0
            _ca_cert_file_path =
326
0
                    get_valid_ca_cert_path(doris::split(config::ca_cert_file_paths, ";"));
327
0
        }
328
1
        if (!_ca_cert_file_path.empty()) {
329
1
            clientConfiguration.caFile = _ca_cert_file_path;
330
1
        }
331
332
1
        auto stsClient = std::make_shared<Aws::STS::STSClient>(
333
1
                std::make_shared<Aws::Auth::InstanceProfileCredentialsProvider>(),
334
1
                clientConfiguration);
335
336
1
        return std::make_shared<Aws::Auth::STSAssumeRoleCredentialsProvider>(
337
1
                s3_conf.role_arn, Aws::String(), s3_conf.external_id,
338
1
                Aws::Auth::DEFAULT_CREDS_LOAD_FREQ_SECONDS, stsClient);
339
2
    }
340
341
    // Support anonymous access for public datasets when no credentials are provided
342
2
    if (s3_conf.ak.empty() && s3_conf.sk.empty()) {
343
2
        return std::make_shared<Aws::Auth::AnonymousAWSCredentialsProvider>();
344
2
    }
345
346
0
    return std::make_shared<Aws::Auth::DefaultAWSCredentialsProviderChain>();
347
2
}
348
349
std::shared_ptr<Aws::Auth::AWSCredentialsProvider> S3ClientFactory::_create_credentials_provider(
350
22
        CredProviderType type) {
351
22
    switch (type) {
352
2
    case CredProviderType::Env:
353
2
        return std::make_shared<Aws::Auth::EnvironmentAWSCredentialsProvider>();
354
2
    case CredProviderType::SystemProperties:
355
2
        return std::make_shared<Aws::Auth::ProfileConfigFileAWSCredentialsProvider>();
356
3
    case CredProviderType::WebIdentity:
357
3
        return std::make_shared<Aws::Auth::STSAssumeRoleWebIdentityCredentialsProvider>();
358
2
    case CredProviderType::Container:
359
2
        return std::make_shared<Aws::Auth::TaskRoleCredentialsProvider>(
360
2
                Aws::Environment::GetEnv("AWS_CONTAINER_CREDENTIALS_RELATIVE_URI").c_str());
361
4
    case CredProviderType::InstanceProfile:
362
4
        return std::make_shared<Aws::Auth::InstanceProfileCredentialsProvider>();
363
6
    case CredProviderType::Anonymous:
364
6
        return std::make_shared<Aws::Auth::AnonymousAWSCredentialsProvider>();
365
3
    case CredProviderType::Default:
366
3
    default:
367
3
        return std::make_shared<CustomAwsCredentialsProviderChain>();
368
22
    }
369
22
}
370
371
std::shared_ptr<Aws::Auth::AWSCredentialsProvider>
372
26
S3ClientFactory::_get_aws_credentials_provider_v2(const S3ClientConf& s3_conf) {
373
26
    if (!s3_conf.ak.empty() && !s3_conf.sk.empty()) {
374
4
        Aws::Auth::AWSCredentials aws_cred(s3_conf.ak, s3_conf.sk);
375
4
        DCHECK(!aws_cred.IsExpiredOrEmpty());
376
4
        if (!s3_conf.token.empty()) {
377
0
            aws_cred.SetSessionToken(s3_conf.token);
378
0
        }
379
4
        return std::make_shared<Aws::Auth::SimpleAWSCredentialsProvider>(std::move(aws_cred));
380
4
    }
381
382
    // Handle role_arn for assume role scenario
383
22
    if (!s3_conf.role_arn.empty()) {
384
8
        Aws::Client::ClientConfiguration clientConfiguration =
385
8
                S3ClientFactory::getClientConfiguration();
386
387
8
        if (_ca_cert_file_path.empty()) {
388
0
            _ca_cert_file_path =
389
0
                    get_valid_ca_cert_path(doris::split(config::ca_cert_file_paths, ";"));
390
0
        }
391
8
        if (!_ca_cert_file_path.empty()) {
392
8
            clientConfiguration.caFile = _ca_cert_file_path;
393
8
        }
394
395
8
        auto baseProvider = _create_credentials_provider(s3_conf.cred_provider_type);
396
8
        auto stsClient = std::make_shared<Aws::STS::STSClient>(baseProvider, clientConfiguration);
397
398
8
        return std::make_shared<Aws::Auth::STSAssumeRoleCredentialsProvider>(
399
8
                s3_conf.role_arn, Aws::String(), s3_conf.external_id,
400
8
                Aws::Auth::DEFAULT_CREDS_LOAD_FREQ_SECONDS, stsClient);
401
8
    }
402
403
    // Return provider based on cred_provider_type
404
14
    return _create_credentials_provider(s3_conf.cred_provider_type);
405
22
}
406
407
std::shared_ptr<Aws::Auth::AWSCredentialsProvider> S3ClientFactory::get_aws_credentials_provider(
408
32
        const S3ClientConf& s3_conf) {
409
32
    if (config::aws_credentials_provider_version == "v2") {
410
26
        return _get_aws_credentials_provider_v2(s3_conf);
411
26
    }
412
6
    return _get_aws_credentials_provider_v1(s3_conf);
413
32
}
414
415
std::shared_ptr<io::ObjStorageClient> S3ClientFactory::_create_s3_client(
416
7
        const S3ClientConf& s3_conf) {
417
7
    TEST_SYNC_POINT_RETURN_WITH_VALUE(
418
6
            "s3_client_factory::create",
419
6
            std::make_shared<io::S3ObjStorageClient>(std::make_shared<Aws::S3::S3Client>()));
420
6
    Aws::Client::ClientConfiguration aws_config = S3ClientFactory::getClientConfiguration();
421
6
    if (s3_conf.need_override_endpoint) {
422
6
        aws_config.endpointOverride = s3_conf.endpoint;
423
6
    }
424
6
    aws_config.region = s3_conf.region;
425
426
6
    if (_ca_cert_file_path.empty()) {
427
0
        _ca_cert_file_path = get_valid_ca_cert_path(doris::split(config::ca_cert_file_paths, ";"));
428
0
    }
429
430
6
    if (!_ca_cert_file_path.empty()) {
431
6
        aws_config.caFile = _ca_cert_file_path;
432
6
    }
433
434
6
    if (s3_conf.max_connections > 0) {
435
0
        aws_config.maxConnections = s3_conf.max_connections;
436
6
    } else {
437
6
        aws_config.maxConnections = 102400;
438
6
    }
439
440
6
    aws_config.requestTimeoutMs = 30000;
441
6
    if (s3_conf.request_timeout_ms > 0) {
442
0
        aws_config.requestTimeoutMs = s3_conf.request_timeout_ms;
443
0
    }
444
445
6
    if (s3_conf.connect_timeout_ms > 0) {
446
0
        aws_config.connectTimeoutMs = s3_conf.connect_timeout_ms;
447
0
    }
448
449
6
    if (config::s3_client_http_scheme == "http") {
450
6
        aws_config.scheme = Aws::Http::Scheme::HTTP;
451
6
    }
452
453
6
    aws_config.retryStrategy = std::make_shared<S3CustomRetryStrategy>(
454
6
            config::max_s3_client_retry /*scaleFactor = 25*/, /*retry_slow_down=*/true);
455
456
6
    std::shared_ptr<Aws::S3::S3Client> new_client = std::make_shared<Aws::S3::S3Client>(
457
6
            get_aws_credentials_provider(s3_conf), std::move(aws_config),
458
6
            Aws::Client::AWSAuthV4Signer::PayloadSigningPolicy::Never,
459
6
            s3_conf.use_virtual_addressing);
460
461
6
    auto obj_client = std::make_shared<io::S3ObjStorageClient>(std::move(new_client));
462
6
    LOG_INFO("create one s3 client with {}", s3_conf.to_string());
463
6
    return obj_client;
464
7
}
465
466
Status S3ClientFactory::convert_properties_to_s3_conf(
467
15
        const std::map<std::string, std::string>& prop, const S3URI& s3_uri, S3Conf* s3_conf) {
468
15
    StringCaseMap<std::string> properties(prop.begin(), prop.end());
469
15
    if (auto it = properties.find(S3_AK); it != properties.end()) {
470
2
        s3_conf->client_conf.ak = it->second;
471
2
    }
472
15
    if (auto it = properties.find(S3_SK); it != properties.end()) {
473
2
        s3_conf->client_conf.sk = it->second;
474
2
    }
475
15
    if (auto it = properties.find(S3_TOKEN); it != properties.end()) {
476
0
        s3_conf->client_conf.token = it->second;
477
0
    }
478
15
    if (auto it = properties.find(S3_ENDPOINT); it != properties.end()) {
479
15
        s3_conf->client_conf.endpoint = it->second;
480
15
    }
481
15
    if (auto it = properties.find(S3_NEED_OVERRIDE_ENDPOINT); it != properties.end()) {
482
0
        s3_conf->client_conf.need_override_endpoint = (it->second == "true");
483
0
    }
484
15
    if (auto it = properties.find(S3_REGION); it != properties.end()) {
485
15
        s3_conf->client_conf.region = it->second;
486
15
    }
487
15
    if (auto it = properties.find(S3_MAX_CONN_SIZE); it != properties.end()) {
488
0
        if (!to_int(it->second, s3_conf->client_conf.max_connections)) {
489
0
            return Status::InvalidArgument("invalid {} value \"{}\"", S3_MAX_CONN_SIZE, it->second);
490
0
        }
491
0
    }
492
15
    if (auto it = properties.find(S3_REQUEST_TIMEOUT_MS); it != properties.end()) {
493
0
        if (!to_int(it->second, s3_conf->client_conf.request_timeout_ms)) {
494
0
            return Status::InvalidArgument("invalid {} value \"{}\"", S3_REQUEST_TIMEOUT_MS,
495
0
                                           it->second);
496
0
        }
497
0
    }
498
15
    if (auto it = properties.find(S3_CONN_TIMEOUT_MS); it != properties.end()) {
499
0
        if (!to_int(it->second, s3_conf->client_conf.connect_timeout_ms)) {
500
0
            return Status::InvalidArgument("invalid {} value \"{}\"", S3_CONN_TIMEOUT_MS,
501
0
                                           it->second);
502
0
        }
503
0
    }
504
15
    if (auto it = properties.find(S3_PROVIDER); it != properties.end()) {
505
        // S3 Provider properties should be case insensitive.
506
0
        if (0 == strcasecmp(it->second.c_str(), AZURE_PROVIDER_STRING)) {
507
0
            s3_conf->client_conf.provider = io::ObjStorageType::AZURE;
508
0
        }
509
0
    }
510
511
15
    if (s3_uri.get_bucket().empty()) {
512
0
        return Status::InvalidArgument("Invalid S3 URI {}, bucket is not specified",
513
0
                                       s3_uri.to_string());
514
0
    }
515
15
    s3_conf->bucket = s3_uri.get_bucket();
516
    // For azure's compatibility
517
15
    s3_conf->client_conf.bucket = s3_uri.get_bucket();
518
15
    s3_conf->prefix = "";
519
520
    // See https://sdk.amazonaws.com/cpp/api/LATEST/class_aws_1_1_s3_1_1_s3_client.html
521
15
    s3_conf->client_conf.use_virtual_addressing = true;
522
15
    if (auto it = properties.find(USE_PATH_STYLE); it != properties.end()) {
523
0
        s3_conf->client_conf.use_virtual_addressing = it->second != "true";
524
0
    }
525
526
15
    if (auto it = properties.find(S3_ROLE_ARN); it != properties.end()) {
527
        // Keep provider type as Default unless explicitly configured by
528
        // AWS_CREDENTIALS_PROVIDER_TYPE, consistent with FE behavior.
529
5
        s3_conf->client_conf.role_arn = it->second;
530
5
    }
531
532
15
    if (auto it = properties.find(S3_EXTERNAL_ID); it != properties.end()) {
533
0
        s3_conf->client_conf.external_id = it->second;
534
0
    }
535
536
15
    if (auto it = properties.find(S3_CREDENTIALS_PROVIDER_TYPE); it != properties.end()) {
537
8
        s3_conf->client_conf.cred_provider_type = cred_provider_type_from_string(it->second);
538
8
    }
539
540
15
    if (auto st = is_s3_conf_valid(s3_conf->client_conf); !st.ok()) {
541
2
        return st;
542
2
    }
543
13
    return Status::OK();
544
15
}
545
546
0
static CredProviderType cred_provider_type_from_thrift(TCredProviderType::type cred_provider_type) {
547
0
    switch (cred_provider_type) {
548
0
    case TCredProviderType::DEFAULT:
549
0
        return CredProviderType::Default;
550
0
    case TCredProviderType::SIMPLE:
551
0
        return CredProviderType::Simple;
552
0
    case TCredProviderType::INSTANCE_PROFILE:
553
0
        return CredProviderType::InstanceProfile;
554
0
    case TCredProviderType::ENV:
555
0
        return CredProviderType::Env;
556
0
    case TCredProviderType::SYSTEM_PROPERTIES:
557
0
        return CredProviderType::SystemProperties;
558
0
    case TCredProviderType::WEB_IDENTITY:
559
0
        return CredProviderType::WebIdentity;
560
0
    case TCredProviderType::CONTAINER:
561
0
        return CredProviderType::Container;
562
0
    case TCredProviderType::ANONYMOUS:
563
0
        return CredProviderType::Anonymous;
564
0
    default:
565
0
        __builtin_unreachable();
566
0
        LOG(WARNING) << "Invalid TCredProviderType value: " << cred_provider_type
567
0
                     << ", use default instead.";
568
0
        return CredProviderType::Default;
569
0
    }
570
0
}
571
572
0
S3Conf S3Conf::get_s3_conf(const cloud::ObjectStoreInfoPB& info) {
573
0
    S3Conf ret {
574
0
            .bucket = info.bucket(),
575
0
            .prefix = info.prefix(),
576
0
            .client_conf {
577
0
                    .endpoint = info.endpoint(),
578
0
                    .region = info.region(),
579
0
                    .ak = info.ak(),
580
0
                    .sk = info.sk(),
581
0
                    .token {},
582
0
                    .bucket = info.bucket(),
583
0
                    .provider = io::ObjStorageType::AWS,
584
0
                    .use_virtual_addressing =
585
0
                            info.has_use_path_style() ? !info.use_path_style() : true,
586
587
0
                    .role_arn = info.role_arn(),
588
0
                    .external_id = info.external_id(),
589
                    // ObjectStoreInfoPB always describes a storage vault, i.e. a Doris
590
                    // internal bucket in cloud mode.
591
0
                    .is_internal_bucket = true,
592
0
            },
593
0
            .sse_enabled = info.sse_enabled(),
594
0
    };
595
596
0
    if (info.has_cred_provider_type()) {
597
0
        ret.client_conf.cred_provider_type = cred_provider_type_from_pb(info.cred_provider_type());
598
0
    }
599
600
0
    io::ObjStorageType type = io::ObjStorageType::AWS;
601
0
    switch (info.provider()) {
602
0
    case cloud::ObjectStoreInfoPB_Provider_OSS:
603
0
        type = io::ObjStorageType::OSS;
604
0
        break;
605
0
    case cloud::ObjectStoreInfoPB_Provider_S3:
606
0
        type = io::ObjStorageType::AWS;
607
0
        break;
608
0
    case cloud::ObjectStoreInfoPB_Provider_COS:
609
0
        type = io::ObjStorageType::COS;
610
0
        break;
611
0
    case cloud::ObjectStoreInfoPB_Provider_OBS:
612
0
        type = io::ObjStorageType::OBS;
613
0
        break;
614
0
    case cloud::ObjectStoreInfoPB_Provider_BOS:
615
0
        type = io::ObjStorageType::BOS;
616
0
        break;
617
0
    case cloud::ObjectStoreInfoPB_Provider_GCP:
618
0
        type = io::ObjStorageType::GCP;
619
0
        break;
620
0
    case cloud::ObjectStoreInfoPB_Provider_AZURE:
621
0
        type = io::ObjStorageType::AZURE;
622
0
        break;
623
0
    case cloud::ObjectStoreInfoPB_Provider_TOS:
624
0
        type = io::ObjStorageType::TOS;
625
0
        break;
626
0
    default:
627
0
        __builtin_unreachable();
628
0
        LOG_FATAL("unknown provider type {}, info {}", info.provider(), ret.to_string());
629
0
    }
630
0
    ret.client_conf.provider = type;
631
0
    return ret;
632
0
}
633
634
0
S3Conf S3Conf::get_s3_conf(const TS3StorageParam& param) {
635
0
    S3Conf ret {
636
0
            .bucket = param.bucket,
637
0
            .prefix = param.root_path,
638
0
            .client_conf = {
639
0
                    .endpoint = param.endpoint,
640
0
                    .region = param.region,
641
0
                    .ak = param.ak,
642
0
                    .sk = param.sk,
643
0
                    .token = param.token,
644
0
                    .bucket = param.bucket,
645
0
                    .provider = io::ObjStorageType::AWS,
646
0
                    .max_connections = param.max_conn,
647
0
                    .request_timeout_ms = param.request_timeout_ms,
648
0
                    .connect_timeout_ms = param.conn_timeout_ms,
649
                    // When using cold heat separation in minio, user might use ip address directly,
650
                    // which needs enable use_virtual_addressing to true
651
0
                    .use_virtual_addressing = !param.use_path_style,
652
0
                    .role_arn = param.role_arn,
653
0
                    .external_id = param.external_id,
654
0
            }};
655
656
0
    if (param.__isset.cred_provider_type) {
657
0
        ret.client_conf.cred_provider_type =
658
0
                cred_provider_type_from_thrift(param.cred_provider_type);
659
0
    }
660
661
0
    io::ObjStorageType type = io::ObjStorageType::AWS;
662
0
    switch (param.provider) {
663
0
    case TObjStorageType::UNKNOWN:
664
0
        LOG_INFO("Receive one legal storage resource, set provider type to aws, param detail {}",
665
0
                 ret.to_string());
666
0
        type = io::ObjStorageType::AWS;
667
0
        break;
668
0
    case TObjStorageType::AWS:
669
0
        type = io::ObjStorageType::AWS;
670
0
        break;
671
0
    case TObjStorageType::AZURE:
672
0
        type = io::ObjStorageType::AZURE;
673
0
        break;
674
0
    case TObjStorageType::BOS:
675
0
        type = io::ObjStorageType::BOS;
676
0
        break;
677
0
    case TObjStorageType::COS:
678
0
        type = io::ObjStorageType::COS;
679
0
        break;
680
0
    case TObjStorageType::OBS:
681
0
        type = io::ObjStorageType::OBS;
682
0
        break;
683
0
    case TObjStorageType::OSS:
684
0
        type = io::ObjStorageType::OSS;
685
0
        break;
686
0
    case TObjStorageType::GCP:
687
0
        type = io::ObjStorageType::GCP;
688
0
        break;
689
0
    case TObjStorageType::TOS:
690
0
        type = io::ObjStorageType::TOS;
691
0
        break;
692
0
    default:
693
0
        LOG_FATAL("unknown provider type {}, info {}", param.provider, ret.to_string());
694
0
        __builtin_unreachable();
695
0
    }
696
0
    ret.client_conf.provider = type;
697
0
    return ret;
698
0
}
699
700
18
std::string hide_access_key(const std::string& ak) {
701
18
    std::string key = ak;
702
18
    size_t key_len = key.length();
703
18
    size_t reserved_count;
704
18
    if (key_len > 7) {
705
3
        reserved_count = 6;
706
15
    } else if (key_len > 2) {
707
6
        reserved_count = key_len - 2;
708
9
    } else {
709
9
        reserved_count = 0;
710
9
    }
711
712
18
    size_t x_count = key_len - reserved_count;
713
18
    size_t left_x_count = (x_count + 1) / 2;
714
715
18
    if (left_x_count > 0) {
716
12
        key.replace(0, left_x_count, left_x_count, 'x');
717
12
    }
718
719
18
    if (x_count - left_x_count > 0) {
720
11
        key.replace(key_len - (x_count - left_x_count), x_count - left_x_count,
721
11
                    x_count - left_x_count, 'x');
722
11
    }
723
18
    return key;
724
18
}
725
726
} // end namespace doris