Coverage Report

Created: 2026-08-13 12:06

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/client/DefaultRetryStrategy.h>
24
#include <aws/core/utils/logging/LogLevel.h>
25
#include <aws/core/utils/logging/LogSystemInterface.h>
26
#include <aws/core/utils/memory/stl/AWSStringStream.h>
27
#include <aws/s3/S3Client.h>
28
29
#include "util/string_util.h"
30
31
#ifdef USE_AZURE
32
#include <azure/core/diagnostics/logger.hpp>
33
#include <azure/core/http/curl_transport.hpp>
34
#include <azure/storage/blobs/blob_container_client.hpp>
35
#endif
36
#include <cstdlib>
37
#include <filesystem>
38
#include <fstream>
39
#include <functional>
40
#include <memory>
41
#include <ostream>
42
#include <utility>
43
44
#include "common/config.h"
45
#include "common/logging.h"
46
#include "common/status.h"
47
#include "cpp/obj-client/auth/aws_credential_factory.h"
48
#ifdef USE_AZURE
49
#include "cpp/obj-client/auth/azure_auth_factory.h"
50
#include "cpp/obj-client/azure_obj_storage_client.h"
51
#endif
52
#include "cloud/config.h"
53
#include "cpp/aws_logger.h"
54
#include "cpp/obj-client/rate_limited_obj_storage_client.h"
55
#include "cpp/obj-client/s3_obj_storage_client.h"
56
#include "cpp/obj_retry_strategy.h"
57
#include "cpp/sync_point.h"
58
#include "cpp/util.h"
59
#include "exec/scan/scanner_scheduler.h"
60
#include "runtime/exec_env.h"
61
#include "util/s3_rate_limiter_manager.h"
62
#include "util/s3_uri.h"
63
64
namespace doris {
65
namespace {
66
67
97.0k
doris::Status is_s3_conf_valid(const S3ClientConf& conf) {
68
97.0k
    if (conf.endpoint.empty()) {
69
0
        return Status::InvalidArgument<false>("Invalid s3 conf, empty endpoint");
70
0
    }
71
97.0k
    if (conf.region.empty()) {
72
0
        return Status::InvalidArgument<false>("Invalid s3 conf, empty region");
73
0
    }
74
75
97.0k
    if (conf.role_arn.empty()) {
76
        // Allow anonymous access when both ak and sk are empty
77
97.0k
        bool hasAk = !conf.ak.empty();
78
97.0k
        bool hasSk = !conf.sk.empty();
79
80
        // Either both credentials are provided or both are empty (anonymous access)
81
97.2k
        if (hasAk && conf.sk.empty()) {
82
1
            return Status::InvalidArgument<false>("Invalid s3 conf, empty sk");
83
1
        }
84
97.1k
        if (hasSk && conf.ak.empty()) {
85
10
            return Status::InvalidArgument<false>("Invalid s3 conf, empty ak");
86
10
        }
87
97.0k
    }
88
97.0k
    return Status::OK();
89
97.0k
}
90
91
ObjStorageResponse make_be_rate_limit_response(S3RateLimitType type,
92
2
                                               S3RateLimitRejectReason reason) {
93
2
    const auto* limit_type = reason == S3RateLimitRejectReason::QPS ? "QPS" : "bytes";
94
    // A local admission rejection is not an S3 HTTP 429. Keep the merged #65420 behavior so S3
95
    // readers do not retry it as provider throttling.
96
2
    return ObjStorageResponse::rate_limit(
97
2
            ErrorCode::EXCEEDED_LIMIT, 0,
98
2
            fmt::format("s3 {} request exceeds {} limit, rejected by BE rate limiter",
99
2
                        to_string(type), limit_type));
100
2
}
101
102
class BeObjStorageRateLimitPolicy final : public ObjStorageRateLimitPolicy {
103
public:
104
480k
    ObjStorageAdmission acquire(S3RateLimitType type, size_t estimated_bytes) const override {
105
480k
        auto guard = std::make_shared<S3RateLimitGuard>(type, estimated_bytes);
106
480k
        if (!guard->ok()) {
107
2
            return ObjStorageAdmission {
108
2
                    .resp = make_be_rate_limit_response(type, guard->reject_reason()),
109
2
            };
110
2
        }
111
480k
        return ObjStorageAdmission {
112
480k
                .settle = [guard = std::move(guard)](
113
480k
                                  size_t actual_bytes) { guard->settle(actual_bytes); },
114
480k
        };
115
480k
    }
116
};
117
118
// Return true is convert `str` to int successfully
119
145k
bool to_int(std::string_view str, int& res) {
120
145k
    auto [_, ec] = std::from_chars(str.data(), str.data() + str.size(), res);
121
145k
    return ec == std::errc {};
122
145k
}
123
124
#ifdef USE_AZURE
125
0
std::string env_or_empty(const char* env_name) {
126
0
    if (const char* value = std::getenv(env_name); value != nullptr) {
127
0
        return value;
128
0
    }
129
0
    return "";
130
0
}
131
132
0
std::string build_azure_tls_debug_context(const std::string& selected_ca_file) {
133
0
    bool selected_ca_exists = false;
134
0
    bool selected_ca_readable = false;
135
0
    if (!selected_ca_file.empty()) {
136
0
        std::error_code ec;
137
0
        selected_ca_exists = std::filesystem::exists(selected_ca_file, ec) && !ec;
138
0
        std::ifstream input(selected_ca_file);
139
0
        selected_ca_readable = input.good();
140
0
    }
141
142
0
    return fmt::format(
143
0
            "tls_debug(ca_cert_file_paths='{}', selected_ca_file='{}', selected_ca_exists={}, "
144
0
            "selected_ca_readable={}, SSL_CERT_FILE='{}', CURL_CA_BUNDLE='{}', SSL_CERT_DIR='{}')",
145
0
            config::ca_cert_file_paths, selected_ca_file, selected_ca_exists, selected_ca_readable,
146
0
            env_or_empty("SSL_CERT_FILE"), env_or_empty("CURL_CA_BUNDLE"),
147
0
            env_or_empty("SSL_CERT_DIR"));
148
0
}
149
#endif
150
151
constexpr char USE_PATH_STYLE[] = "use_path_style";
152
153
constexpr char AZURE_PROVIDER_STRING[] = "AZURE";
154
constexpr char S3_PROVIDER[] = "provider";
155
constexpr char S3_AK[] = "AWS_ACCESS_KEY";
156
constexpr char S3_SK[] = "AWS_SECRET_KEY";
157
constexpr char S3_ENDPOINT[] = "AWS_ENDPOINT";
158
constexpr char S3_REGION[] = "AWS_REGION";
159
constexpr char S3_TOKEN[] = "AWS_TOKEN";
160
constexpr char S3_MAX_CONN_SIZE[] = "AWS_MAX_CONNECTIONS";
161
constexpr char S3_REQUEST_TIMEOUT_MS[] = "AWS_REQUEST_TIMEOUT_MS";
162
constexpr char S3_CONN_TIMEOUT_MS[] = "AWS_CONNECTION_TIMEOUT_MS";
163
constexpr char S3_NEED_OVERRIDE_ENDPOINT[] = "AWS_NEED_OVERRIDE_ENDPOINT";
164
165
constexpr char S3_ROLE_ARN[] = "AWS_ROLE_ARN";
166
constexpr char S3_EXTERNAL_ID[] = "AWS_EXTERNAL_ID";
167
constexpr char S3_CREDENTIALS_PROVIDER_TYPE[] = "AWS_CREDENTIALS_PROVIDER_TYPE";
168
} // namespace
169
170
8
S3ClientFactory::S3ClientFactory() {
171
8
    _aws_options = Aws::SDKOptions {};
172
8
    auto logLevel = static_cast<Aws::Utils::Logging::LogLevel>(config::aws_log_level);
173
8
    _aws_options.loggingOptions.logLevel = logLevel;
174
8
    _aws_options.loggingOptions.logger_create_fn = [logLevel] {
175
8
        return std::make_shared<DorisAWSLogger>(logLevel);
176
8
    };
177
8
    Aws::InitAPI(_aws_options);
178
8
    _ca_cert_file_path = get_valid_ca_cert_path(doris::split(config::ca_cert_file_paths, ";"));
179
180
8
#ifdef USE_AZURE
181
8
    auto azureLogLevel =
182
8
            static_cast<Azure::Core::Diagnostics::Logger::Level>(config::azure_log_level);
183
8
    Azure::Core::Diagnostics::Logger::SetLevel(azureLogLevel);
184
8
    Azure::Core::Diagnostics::Logger::SetListener(
185
8
            [&](Azure::Core::Diagnostics::Logger::Level level, const std::string& message) {
186
0
                switch (level) {
187
0
                case Azure::Core::Diagnostics::Logger::Level::Verbose:
188
0
                    LOG(INFO) << message;
189
0
                    break;
190
0
                case Azure::Core::Diagnostics::Logger::Level::Informational:
191
0
                    LOG(INFO) << message;
192
0
                    break;
193
0
                case Azure::Core::Diagnostics::Logger::Level::Warning:
194
0
                    LOG(WARNING) << message;
195
0
                    break;
196
0
                case Azure::Core::Diagnostics::Logger::Level::Error:
197
0
                    LOG(ERROR) << message;
198
0
                    break;
199
0
                default:
200
0
                    LOG(WARNING) << "Unknown level: " << static_cast<int>(level)
201
0
                                 << ", message: " << message;
202
0
                    break;
203
0
                }
204
0
            });
205
8
#endif
206
8
}
207
208
4
S3ClientFactory::~S3ClientFactory() {
209
4
    Aws::ShutdownAPI(_aws_options);
210
4
}
211
212
48.8k
S3ClientFactory& S3ClientFactory::instance() {
213
48.8k
    static S3ClientFactory ret;
214
48.8k
    return ret;
215
48.8k
}
216
217
48.7k
Result<std::shared_ptr<io::ObjStorageClient>> S3ClientFactory::create(const S3ClientConf& s3_conf) {
218
48.7k
    RETURN_IF_ERROR_RESULT(is_s3_conf_valid(s3_conf));
219
220
#ifdef BE_TEST
221
    {
222
        std::lock_guard l(_lock);
223
        if (_test_client_creator) {
224
            return _test_client_creator(s3_conf);
225
        }
226
    }
227
#endif
228
229
48.7k
    {
230
48.7k
        std::lock_guard l(_lock);
231
48.7k
        auto it = _cache.find(s3_conf);
232
49.2k
        if (it != _cache.end()) {
233
49.2k
            return it->second;
234
49.2k
        }
235
48.7k
    }
236
237
18.4E
    auto client_result = (s3_conf.provider == io::ObjStorageProvider::AZURE)
238
18.4E
                                 ? _create_azure_client(s3_conf)
239
18.4E
                                 : _create_s3_client(s3_conf);
240
18.4E
    if (!client_result.has_value()) {
241
0
        return ResultError(std::move(client_result).error());
242
0
    }
243
18.4E
    auto obj_client = std::move(client_result).value();
244
18.4E
    if (!config::is_cloud_mode() || s3_conf.is_internal_bucket) {
245
64
        obj_client = std::make_shared<io::RateLimitedObjStorageClient>(
246
64
                std::move(obj_client), std::make_shared<BeObjStorageRateLimitPolicy>());
247
64
    }
248
249
18.4E
    {
250
18.4E
        std::lock_guard l(_lock);
251
18.4E
        auto [it, _] = _cache.emplace(s3_conf, std::move(obj_client));
252
18.4E
        return it->second;
253
18.4E
    }
254
18.4E
}
255
256
#ifdef BE_TEST
257
void S3ClientFactory::set_client_creator_for_test(
258
        std::function<std::shared_ptr<io::ObjStorageClient>(const S3ClientConf&)> creator) {
259
    std::lock_guard l(_lock);
260
    _test_client_creator = std::move(creator);
261
}
262
263
void S3ClientFactory::clear_client_creator_for_test() {
264
    std::lock_guard l(_lock);
265
    _test_client_creator = nullptr;
266
}
267
#endif
268
269
Result<std::shared_ptr<io::ObjStorageClient>> S3ClientFactory::_create_azure_client(
270
0
        const S3ClientConf& s3_conf) {
271
0
#ifdef USE_AZURE
272
0
    const std::string container_name = s3_conf.bucket;
273
0
    std::string uri = fmt::format("{}/{}", s3_conf.endpoint, container_name);
274
0
    if (s3_conf.endpoint.find("://") == std::string::npos) {
275
0
        uri = "https://" + uri;
276
0
    }
277
278
0
    Azure::Storage::Blobs::BlobClientOptions options;
279
0
    options.Retry.StatusCodes.insert(Azure::Core::Http::HttpStatusCode::TooManyRequests);
280
0
    options.Retry.MaxRetries = config::max_s3_client_retry;
281
0
    options.PerRetryPolicies.emplace_back(std::make_unique<AzureRetryRecordPolicy>());
282
0
    if (_ca_cert_file_path.empty()) {
283
0
        _ca_cert_file_path = get_valid_ca_cert_path(doris::split(config::ca_cert_file_paths, ";"));
284
0
    }
285
0
    if (!_ca_cert_file_path.empty()) {
286
0
        Azure::Core::Http::CurlTransportOptions curl_options;
287
0
        curl_options.CAInfo = _ca_cert_file_path;
288
0
        options.Transport.Transport =
289
0
                std::make_shared<Azure::Core::Http::CurlTransport>(std::move(curl_options));
290
0
    }
291
292
0
    std::string normalized_uri = normalize_http_uri(uri);
293
0
    VLOG_DEBUG << "uri:" << uri << ", normalized_uri:" << normalized_uri;
294
0
    std::string tls_debug_context = build_azure_tls_debug_context(_ca_cert_file_path);
295
296
0
    auto built = AzureAuthFactory::create(uri,
297
0
                                          {
298
0
                                                  .type = AzureCredentialType::SHARED_KEY,
299
0
                                                  .account_name = s3_conf.ak,
300
0
                                                  .account_key = s3_conf.sk,
301
0
                                          },
302
0
                                          std::move(options));
303
0
    if (!built) {
304
0
        return ResultError(
305
0
                Status::InvalidArgument("failed to create Azure client: {}", built.error));
306
0
    }
307
0
    LOG_INFO("create one azure client with {}", s3_conf.to_string());
308
0
    return std::make_shared<io::AzureObjStorageClient>(
309
0
            std::move(built.container_client),
310
0
            ObjStorageEndpointInfo {
311
0
                    .endpoint = s3_conf.endpoint,
312
0
                    .ak = s3_conf.ak,
313
0
                    .sk = s3_conf.sk,
314
0
                    .tls_debug_context = std::move(tls_debug_context),
315
0
            },
316
0
            std::move(built.shared_key_credential));
317
#else
318
    return ResultError(Status::NotSupported(
319
            "BE is not compiled with azure support, export BUILD_AZURE=ON before building"));
320
#endif
321
0
}
322
323
109
AwsCredentialResult S3ClientFactory::create_aws_credentials_provider(const S3ClientConf& s3_conf) {
324
109
    auto sts_config = S3ClientFactory::getClientConfiguration();
325
109
    if (!_ca_cert_file_path.empty()) {
326
109
        sts_config.caFile = _ca_cert_file_path;
327
109
    }
328
109
    return AwsCredentialFactory::create({
329
109
            .version = config::aws_credentials_provider_version == "v2"
330
109
                               ? AwsCredentialProviderVersion::V2
331
109
                               : AwsCredentialProviderVersion::V1,
332
109
            .access_key = s3_conf.ak,
333
109
            .secret_key = s3_conf.sk,
334
109
            .session_token = s3_conf.token,
335
109
            .provider_type = s3_conf.cred_provider_type,
336
109
            .role_arn = s3_conf.role_arn,
337
109
            .external_id = s3_conf.external_id,
338
109
            .empty_credentials = EmptyCredentialsBehavior::ANONYMOUS,
339
109
            .sts_client_config = std::move(sts_config),
340
109
    });
341
109
}
342
343
Result<std::shared_ptr<io::ObjStorageClient>> S3ClientFactory::_create_s3_client(
344
75
        const S3ClientConf& s3_conf) {
345
75
    TEST_SYNC_POINT_RETURN_WITH_VALUE(
346
75
            "s3_client_factory::create",
347
75
            std::make_shared<io::S3ObjStorageClient>(std::make_shared<Aws::S3::S3Client>(),
348
75
                                                     ObjStorageEndpointInfo {}));
349
75
    Aws::Client::ClientConfiguration aws_config = S3ClientFactory::getClientConfiguration();
350
75
    if (s3_conf.need_override_endpoint) {
351
75
        aws_config.endpointOverride = s3_conf.endpoint;
352
75
    }
353
75
    aws_config.region = s3_conf.region;
354
355
75
    if (_ca_cert_file_path.empty()) {
356
0
        _ca_cert_file_path = get_valid_ca_cert_path(doris::split(config::ca_cert_file_paths, ";"));
357
0
    }
358
359
75
    if (!_ca_cert_file_path.empty()) {
360
75
        aws_config.caFile = _ca_cert_file_path;
361
75
    }
362
363
75
    if (s3_conf.max_connections > 0) {
364
73
        aws_config.maxConnections = s3_conf.max_connections;
365
73
    } else {
366
2
        aws_config.maxConnections = 102400;
367
2
    }
368
369
75
    aws_config.requestTimeoutMs = 30000;
370
75
    if (s3_conf.request_timeout_ms > 0) {
371
73
        aws_config.requestTimeoutMs = s3_conf.request_timeout_ms;
372
73
    }
373
374
75
    if (s3_conf.connect_timeout_ms > 0) {
375
73
        aws_config.connectTimeoutMs = s3_conf.connect_timeout_ms;
376
73
    }
377
378
75
    set_s3_client_default_http_scheme(aws_config, config::s3_client_http_scheme);
379
380
75
    aws_config.retryStrategy = std::make_shared<S3CustomRetryStrategy>(
381
75
            config::max_s3_client_retry /*scaleFactor = 25*/, /*retry_slow_down=*/true);
382
383
75
    auto credentials = create_aws_credentials_provider(s3_conf);
384
75
    if (!credentials) {
385
0
        return ResultError(Status::InvalidArgument("failed to create AWS credential provider: {}",
386
0
                                                   credentials.error));
387
0
    }
388
75
    std::shared_ptr<Aws::S3::S3Client> new_client = std::make_shared<Aws::S3::S3Client>(
389
75
            std::move(credentials.provider), std::move(aws_config),
390
75
            Aws::Client::AWSAuthV4Signer::PayloadSigningPolicy::Never,
391
75
            s3_conf.use_virtual_addressing);
392
393
75
    auto provider_client = std::make_shared<io::S3ObjStorageClient>(
394
75
            std::move(new_client), ObjStorageEndpointInfo {
395
75
                                           .endpoint = s3_conf.endpoint,
396
75
                                           .ak = s3_conf.ak,
397
75
                                           .sk = s3_conf.sk,
398
75
                                   });
399
75
    LOG_INFO("create one s3 client with {}", s3_conf.to_string());
400
75
    return provider_client;
401
75
}
402
403
Status S3ClientFactory::convert_properties_to_s3_conf(
404
48.4k
        const std::map<std::string, std::string>& prop, const S3URI& s3_uri, S3Conf* s3_conf) {
405
48.4k
    StringCaseMap<std::string> properties(prop.begin(), prop.end());
406
48.6k
    if (auto it = properties.find(S3_AK); it != properties.end()) {
407
48.6k
        s3_conf->client_conf.ak = it->second;
408
48.6k
    }
409
48.4k
    if (auto it = properties.find(S3_SK); it != properties.end()) {
410
48.4k
        s3_conf->client_conf.sk = it->second;
411
48.4k
    }
412
48.4k
    if (auto it = properties.find(S3_TOKEN); it != properties.end()) {
413
88
        s3_conf->client_conf.token = it->second;
414
88
    }
415
48.6k
    if (auto it = properties.find(S3_ENDPOINT); it != properties.end()) {
416
48.6k
        s3_conf->client_conf.endpoint = it->second;
417
48.6k
    }
418
48.4k
    if (auto it = properties.find(S3_NEED_OVERRIDE_ENDPOINT); it != properties.end()) {
419
0
        s3_conf->client_conf.need_override_endpoint = (it->second == "true");
420
0
    }
421
48.6k
    if (auto it = properties.find(S3_REGION); it != properties.end()) {
422
48.6k
        s3_conf->client_conf.region = it->second;
423
48.6k
    }
424
48.4k
    if (auto it = properties.find(S3_MAX_CONN_SIZE); it != properties.end()) {
425
48.4k
        if (!to_int(it->second, s3_conf->client_conf.max_connections)) {
426
0
            return Status::InvalidArgument("invalid {} value \"{}\"", S3_MAX_CONN_SIZE, it->second);
427
0
        }
428
48.4k
    }
429
48.5k
    if (auto it = properties.find(S3_REQUEST_TIMEOUT_MS); it != properties.end()) {
430
48.5k
        if (!to_int(it->second, s3_conf->client_conf.request_timeout_ms)) {
431
0
            return Status::InvalidArgument("invalid {} value \"{}\"", S3_REQUEST_TIMEOUT_MS,
432
0
                                           it->second);
433
0
        }
434
48.5k
    }
435
48.4k
    if (auto it = properties.find(S3_CONN_TIMEOUT_MS); it != properties.end()) {
436
48.3k
        if (!to_int(it->second, s3_conf->client_conf.connect_timeout_ms)) {
437
0
            return Status::InvalidArgument("invalid {} value \"{}\"", S3_CONN_TIMEOUT_MS,
438
0
                                           it->second);
439
0
        }
440
48.3k
    }
441
48.4k
    if (auto it = properties.find(S3_PROVIDER); it != properties.end()) {
442
        // S3 Provider properties should be case insensitive.
443
0
        if (0 == strcasecmp(it->second.c_str(), AZURE_PROVIDER_STRING)) {
444
0
            s3_conf->client_conf.provider = io::ObjStorageProvider::AZURE;
445
0
        }
446
0
    }
447
448
48.4k
    if (s3_uri.get_bucket().empty()) {
449
0
        return Status::InvalidArgument("Invalid S3 URI {}, bucket is not specified",
450
0
                                       s3_uri.to_string());
451
0
    }
452
48.4k
    s3_conf->bucket = s3_uri.get_bucket();
453
    // For azure's compatibility
454
48.4k
    s3_conf->client_conf.bucket = s3_uri.get_bucket();
455
48.4k
    s3_conf->prefix = "";
456
457
    // See https://sdk.amazonaws.com/cpp/api/LATEST/class_aws_1_1_s3_1_1_s3_client.html
458
48.4k
    s3_conf->client_conf.use_virtual_addressing = true;
459
48.6k
    if (auto it = properties.find(USE_PATH_STYLE); it != properties.end()) {
460
48.6k
        s3_conf->client_conf.use_virtual_addressing = it->second != "true";
461
48.6k
    }
462
463
48.4k
    if (auto it = properties.find(S3_ROLE_ARN); it != properties.end()) {
464
        // Keep provider type as Default unless explicitly configured by
465
        // AWS_CREDENTIALS_PROVIDER_TYPE, consistent with FE behavior.
466
5
        s3_conf->client_conf.role_arn = it->second;
467
5
    }
468
469
48.4k
    if (auto it = properties.find(S3_EXTERNAL_ID); it != properties.end()) {
470
0
        s3_conf->client_conf.external_id = it->second;
471
0
    }
472
473
48.4k
    if (auto it = properties.find(S3_CREDENTIALS_PROVIDER_TYPE); it != properties.end()) {
474
228
        s3_conf->client_conf.cred_provider_type = cred_provider_type_from_string(it->second);
475
228
    }
476
477
48.4k
    if (auto st = is_s3_conf_valid(s3_conf->client_conf); !st.ok()) {
478
2
        return st;
479
2
    }
480
48.4k
    return Status::OK();
481
48.4k
}
482
483
0
static CredProviderType cred_provider_type_from_thrift(TCredProviderType::type cred_provider_type) {
484
0
    switch (cred_provider_type) {
485
0
    case TCredProviderType::DEFAULT:
486
0
        return CredProviderType::Default;
487
0
    case TCredProviderType::SIMPLE:
488
0
        return CredProviderType::Simple;
489
0
    case TCredProviderType::INSTANCE_PROFILE:
490
0
        return CredProviderType::InstanceProfile;
491
0
    case TCredProviderType::ENV:
492
0
        return CredProviderType::Env;
493
0
    case TCredProviderType::SYSTEM_PROPERTIES:
494
0
        return CredProviderType::SystemProperties;
495
0
    case TCredProviderType::WEB_IDENTITY:
496
0
        return CredProviderType::WebIdentity;
497
0
    case TCredProviderType::CONTAINER:
498
0
        return CredProviderType::Container;
499
0
    case TCredProviderType::ANONYMOUS:
500
0
        return CredProviderType::Anonymous;
501
0
    default:
502
0
        __builtin_unreachable();
503
0
        LOG(WARNING) << "Invalid TCredProviderType value: " << cred_provider_type
504
0
                     << ", use default instead.";
505
0
        return CredProviderType::Default;
506
0
    }
507
0
}
508
509
98
S3Conf S3Conf::get_s3_conf(const cloud::ObjectStoreInfoPB& info) {
510
98
    S3Conf ret {
511
98
            .bucket = info.bucket(),
512
98
            .prefix = info.prefix(),
513
98
            .client_conf {
514
98
                    .endpoint = info.endpoint(),
515
98
                    .region = info.region(),
516
98
                    .ak = info.ak(),
517
98
                    .sk = info.sk(),
518
98
                    .token = {},
519
98
                    .bucket = info.bucket(),
520
98
                    .provider = io::ObjStorageProvider::AWS,
521
98
                    .use_virtual_addressing =
522
98
                            info.has_use_path_style() ? !info.use_path_style() : true,
523
524
98
                    .role_arn = info.role_arn(),
525
98
                    .external_id = info.external_id(),
526
                    // ObjectStoreInfoPB always describes a storage vault, i.e. a Doris
527
                    // internal bucket in cloud mode.
528
98
                    .is_internal_bucket = true,
529
98
            },
530
98
            .sse_enabled = info.sse_enabled(),
531
98
    };
532
533
98
    if (info.has_cred_provider_type()) {
534
0
        ret.client_conf.cred_provider_type = cred_provider_type_from_pb(info.cred_provider_type());
535
0
    }
536
537
98
    io::ObjStorageProvider type = io::ObjStorageProvider::AWS;
538
98
    switch (info.provider()) {
539
98
    case cloud::ObjectStoreInfoPB_Provider_OSS:
540
98
        type = io::ObjStorageProvider::OSS;
541
98
        break;
542
0
    case cloud::ObjectStoreInfoPB_Provider_S3:
543
0
        type = io::ObjStorageProvider::AWS;
544
0
        break;
545
0
    case cloud::ObjectStoreInfoPB_Provider_COS:
546
0
        type = io::ObjStorageProvider::COS;
547
0
        break;
548
0
    case cloud::ObjectStoreInfoPB_Provider_OBS:
549
0
        type = io::ObjStorageProvider::OBS;
550
0
        break;
551
0
    case cloud::ObjectStoreInfoPB_Provider_BOS:
552
0
        type = io::ObjStorageProvider::BOS;
553
0
        break;
554
0
    case cloud::ObjectStoreInfoPB_Provider_GCP:
555
0
        type = io::ObjStorageProvider::GCP;
556
0
        break;
557
0
    case cloud::ObjectStoreInfoPB_Provider_AZURE:
558
0
        type = io::ObjStorageProvider::AZURE;
559
0
        break;
560
0
    case cloud::ObjectStoreInfoPB_Provider_TOS:
561
0
        type = io::ObjStorageProvider::TOS;
562
0
        break;
563
0
    default:
564
0
        __builtin_unreachable();
565
0
        LOG_FATAL("unknown provider type {}, info {}", info.provider(), ret.to_string());
566
98
    }
567
98
    ret.client_conf.provider = type;
568
98
    return ret;
569
98
}
570
571
33
S3Conf S3Conf::get_s3_conf(const TS3StorageParam& param) {
572
33
    S3Conf ret {
573
33
            .bucket = param.bucket,
574
33
            .prefix = param.root_path,
575
33
            .client_conf = {
576
33
                    .endpoint = param.endpoint,
577
33
                    .region = param.region,
578
33
                    .ak = param.ak,
579
33
                    .sk = param.sk,
580
33
                    .token = param.token,
581
33
                    .bucket = param.bucket,
582
33
                    .provider = io::ObjStorageProvider::AWS,
583
33
                    .max_connections = param.max_conn,
584
33
                    .request_timeout_ms = param.request_timeout_ms,
585
33
                    .connect_timeout_ms = param.conn_timeout_ms,
586
                    // When using cold heat separation in minio, user might use ip address directly,
587
                    // which needs enable use_virtual_addressing to true
588
33
                    .use_virtual_addressing = !param.use_path_style,
589
33
                    .role_arn = param.role_arn,
590
33
                    .external_id = param.external_id,
591
33
            }};
592
593
33
    if (param.__isset.cred_provider_type) {
594
0
        ret.client_conf.cred_provider_type =
595
0
                cred_provider_type_from_thrift(param.cred_provider_type);
596
0
    }
597
598
33
    io::ObjStorageProvider type = io::ObjStorageProvider::AWS;
599
33
    switch (param.provider) {
600
33
    case TObjStorageType::UNKNOWN:
601
33
        LOG_INFO("Receive one legal storage resource, set provider type to aws, param detail {}",
602
33
                 ret.to_string());
603
33
        type = io::ObjStorageProvider::AWS;
604
33
        break;
605
0
    case TObjStorageType::AWS:
606
0
        type = io::ObjStorageProvider::AWS;
607
0
        break;
608
0
    case TObjStorageType::AZURE:
609
0
        type = io::ObjStorageProvider::AZURE;
610
0
        break;
611
0
    case TObjStorageType::BOS:
612
0
        type = io::ObjStorageProvider::BOS;
613
0
        break;
614
0
    case TObjStorageType::COS:
615
0
        type = io::ObjStorageProvider::COS;
616
0
        break;
617
0
    case TObjStorageType::OBS:
618
0
        type = io::ObjStorageProvider::OBS;
619
0
        break;
620
0
    case TObjStorageType::OSS:
621
0
        type = io::ObjStorageProvider::OSS;
622
0
        break;
623
0
    case TObjStorageType::GCP:
624
0
        type = io::ObjStorageProvider::GCP;
625
0
        break;
626
0
    case TObjStorageType::TOS:
627
0
        type = io::ObjStorageProvider::TOS;
628
0
        break;
629
0
    default:
630
0
        LOG_FATAL("unknown provider type {}, info {}", param.provider, ret.to_string());
631
0
        __builtin_unreachable();
632
33
    }
633
33
    ret.client_conf.provider = type;
634
33
    return ret;
635
33
}
636
637
223
std::string hide_access_key(const std::string& ak) {
638
223
    std::string key = ak;
639
223
    size_t key_len = key.length();
640
223
    size_t reserved_count;
641
223
    if (key_len > 7) {
642
74
        reserved_count = 6;
643
149
    } else if (key_len > 2) {
644
130
        reserved_count = key_len - 2;
645
130
    } else {
646
19
        reserved_count = 0;
647
19
    }
648
649
223
    size_t x_count = key_len - reserved_count;
650
223
    size_t left_x_count = (x_count + 1) / 2;
651
652
223
    if (left_x_count > 0) {
653
208
        key.replace(0, left_x_count, left_x_count, 'x');
654
208
    }
655
656
223
    if (x_count - left_x_count > 0) {
657
207
        key.replace(key_len - (x_count - left_x_count), x_count - left_x_count,
658
207
                    x_count - left_x_count, 'x');
659
207
    }
660
223
    return key;
661
223
}
662
663
} // end namespace doris