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 |