Coverage Report

Created: 2025-12-31 18:02

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