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 |