common/cpp/obj-client/s3_obj_storage_client.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 "s3_obj_storage_client.h" |
19 | | |
20 | | #include <cpp/obj-client/obj_storage_client.h> |
21 | | #include <gen_cpp/Status_types.h> |
22 | | |
23 | | #include <algorithm> |
24 | | #include <chrono> |
25 | | |
26 | | #include "client_bvar.h" |
27 | | #include "cpp/obj_retry_strategy.h" |
28 | | |
29 | | namespace Aws::S3::Model { |
30 | | class DeleteObjectRequest; |
31 | | } // namespace Aws::S3::Model |
32 | | |
33 | | using Aws::S3::Model::CompletedPart; |
34 | | using Aws::S3::Model::CompletedMultipartUpload; |
35 | | using Aws::S3::Model::CompleteMultipartUploadRequest; |
36 | | using Aws::S3::Model::CreateMultipartUploadRequest; |
37 | | using Aws::S3::Model::UploadPartRequest; |
38 | | using Aws::S3::Model::UploadPartOutcome; |
39 | | |
40 | | namespace doris { |
41 | | using namespace Aws::S3::Model; |
42 | | namespace { |
43 | | |
44 | | constexpr int64_t S3_REQUEST_THRESHOLD_MS = 5000; |
45 | | |
46 | 22 | int64_t elapsed_time_milliseconds(std::chrono::steady_clock::time_point start) { |
47 | 22 | return std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now() - |
48 | 22 | start) |
49 | 22 | .count(); |
50 | 22 | } |
51 | | |
52 | 6 | void record_s3_request_failed(const Aws::S3::S3Error& error) { |
53 | 6 | record_object_request_failed(static_cast<int>(error.GetResponseCode())); |
54 | 6 | } |
55 | | |
56 | 3 | std::string object_identity(const ObjStoragePath& opts) { |
57 | 3 | return opts.path.empty() ? opts.key : opts.path.native(); |
58 | 3 | } |
59 | | |
60 | 10 | std::string s3_error_message(const Aws::S3::S3Error& error, std::string_view message) { |
61 | | // A failure raised by the client itself carries no request id, and a dangling |
62 | | // `request_id=` has been read as a request id of the object storage. |
63 | 10 | std::string request_id = |
64 | 10 | error.GetRequestId().empty() ? "<empty>" : error.GetRequestId().c_str(); |
65 | 10 | return fmt::format("{}: {} {} code={}, type={}, request_id={}", message, |
66 | 10 | error.GetExceptionName(), error.GetMessage(), |
67 | 10 | static_cast<int>(error.GetResponseCode()), |
68 | 10 | static_cast<int>(error.GetErrorType()), request_id); |
69 | 10 | } |
70 | | |
71 | | } // namespace |
72 | | |
73 | 10 | ObjStorageStatus s3fs_error(const Aws::S3::S3Error& err, std::string_view msg) { |
74 | 10 | return obj_storage_status_from_http_code(static_cast<int>(err.GetResponseCode()), |
75 | 10 | s3_error_message(err, msg)); |
76 | 10 | } |
77 | | |
78 | 8 | ObjStorageUploadResult S3ObjStorageClient::create_multipart_upload(const ObjStoragePath& opts) { |
79 | 8 | CreateMultipartUploadRequest request; |
80 | 8 | request.WithBucket(opts.bucket).WithKey(opts.key); |
81 | 8 | request.SetContentType("application/octet-stream"); |
82 | | |
83 | 8 | const auto start = std::chrono::steady_clock::now(); |
84 | 8 | auto outcome = SYNC_POINT_HOOK_RETURN_VALUE( |
85 | 8 | [&]() { |
86 | 8 | client_bvar::ScopedLatency scoped_latency( |
87 | 8 | client_bvar::s3_multi_part_upload_latency); |
88 | 8 | return _client->CreateMultipartUpload(request); |
89 | 8 | }(), |
90 | 8 | "s3_file_writer::create_multi_part_upload", std::cref(request).get()); |
91 | 8 | SYNC_POINT_CALLBACK("s3_file_writer::_open", &outcome); |
92 | 8 | const auto elapsed_ms = elapsed_time_milliseconds(start); |
93 | | |
94 | 8 | const auto& request_id = outcome.IsSuccess() ? outcome.GetResult().GetRequestId() |
95 | 8 | : outcome.GetError().GetRequestId(); |
96 | | |
97 | 8 | LOG_IF(INFO, elapsed_ms > S3_REQUEST_THRESHOLD_MS) |
98 | 0 | << "CreateMultipartUpload cost=" << elapsed_ms << "ms" |
99 | 0 | << ", request_id=" << request_id << ", bucket=" << opts.bucket << ", key=" << opts.key; |
100 | | |
101 | 8 | if (!outcome.IsSuccess()) { |
102 | 1 | record_s3_request_failed(outcome.GetError()); |
103 | 1 | auto st = s3fs_error(outcome.GetError(), fmt::format("failed to CreateMultipartUpload: {} ", |
104 | 1 | opts.path.native())); |
105 | 1 | LOG(WARNING) << st.code << " request_id=" << request_id; |
106 | 1 | return ObjStorageUploadResult { |
107 | 1 | .resp = {.status = st, |
108 | 1 | .http_code = static_cast<int>(outcome.GetError().GetResponseCode()), |
109 | 1 | .request_id = outcome.GetError().GetRequestId()}, |
110 | 1 | }; |
111 | 1 | } |
112 | | |
113 | 7 | return ObjStorageUploadResult {.resp = ObjStorageResponse::OK(), |
114 | 7 | .upload_id {outcome.GetResult().GetUploadId()}}; |
115 | 8 | } |
116 | | |
117 | | ObjStorageResponse S3ObjStorageClient::put_object(const ObjStoragePath& opts, |
118 | 1 | std::string_view stream) { |
119 | 1 | Aws::S3::Model::PutObjectRequest request; |
120 | 1 | request.WithBucket(opts.bucket).WithKey(opts.key); |
121 | 1 | auto string_view_stream = std::make_shared<StringViewStream>(stream.data(), stream.size()); |
122 | 1 | Aws::Utils::ByteBuffer part_md5(Aws::Utils::HashingUtils::CalculateMD5(*string_view_stream)); |
123 | 1 | request.SetContentMD5(Aws::Utils::HashingUtils::Base64Encode(part_md5)); |
124 | 1 | request.SetBody(string_view_stream); |
125 | 1 | request.SetContentLength(stream.size()); |
126 | 1 | request.SetContentType("application/octet-stream"); |
127 | | |
128 | 1 | const auto start = std::chrono::steady_clock::now(); |
129 | 1 | auto outcome = SYNC_POINT_HOOK_RETURN_VALUE( |
130 | 1 | [&]() { |
131 | 1 | client_bvar::ScopedLatency scoped_latency(client_bvar::s3_put_latency); |
132 | 1 | return _client->PutObject(request); |
133 | 1 | }(), |
134 | 1 | "s3_file_writer::put_object", std::cref(request).get(), &stream); |
135 | 1 | const auto elapsed_ms = elapsed_time_milliseconds(start); |
136 | | |
137 | 1 | const auto& request_id = outcome.IsSuccess() ? outcome.GetResult().GetRequestId() |
138 | 1 | : outcome.GetError().GetRequestId(); |
139 | | |
140 | 1 | if (!outcome.IsSuccess()) { |
141 | 0 | record_s3_request_failed(outcome.GetError()); |
142 | 0 | auto st = s3fs_error(outcome.GetError(), |
143 | 0 | fmt::format("failed to put object: {}", opts.path.native())); |
144 | 0 | LOG(WARNING) << st.code << ", request_id=" << request_id; |
145 | 0 | return ObjStorageResponse { |
146 | 0 | .status = st, |
147 | 0 | .http_code = static_cast<int>(outcome.GetError().GetResponseCode()), |
148 | 0 | .request_id = outcome.GetError().GetRequestId()}; |
149 | 0 | } |
150 | | |
151 | 1 | LOG_IF(INFO, elapsed_ms > S3_REQUEST_THRESHOLD_MS) |
152 | 0 | << "PutObject cost=" << elapsed_ms << "ms" |
153 | 0 | << ", request_id=" << request_id << ", bucket=" << opts.bucket << ", key=" << opts.key; |
154 | 1 | return ObjStorageResponse::OK(); |
155 | 1 | } |
156 | | |
157 | | ObjStorageUploadResult S3ObjStorageClient::upload_part(const ObjStoragePath& opts, |
158 | | const std::string& upload_id, |
159 | 11 | std::string_view stream, int part_num) { |
160 | 11 | UploadPartRequest request; |
161 | 11 | request.WithBucket(opts.bucket) |
162 | 11 | .WithKey(opts.key) |
163 | 11 | .WithPartNumber(part_num) |
164 | 11 | .WithUploadId(upload_id); |
165 | 11 | auto string_view_stream = std::make_shared<StringViewStream>(stream.data(), stream.size()); |
166 | | |
167 | 11 | request.SetBody(string_view_stream); |
168 | | |
169 | 11 | Aws::Utils::ByteBuffer part_md5(Aws::Utils::HashingUtils::CalculateMD5(*string_view_stream)); |
170 | 11 | request.SetContentMD5(Aws::Utils::HashingUtils::Base64Encode(part_md5)); |
171 | | |
172 | 11 | request.SetContentLength(stream.size()); |
173 | 11 | request.SetContentType("application/octet-stream"); |
174 | | |
175 | 11 | const auto start = std::chrono::steady_clock::now(); |
176 | 11 | auto outcome = SYNC_POINT_HOOK_RETURN_VALUE( |
177 | 11 | [&]() { |
178 | 11 | client_bvar::ScopedLatency scoped_latency( |
179 | 11 | client_bvar::s3_multi_part_upload_latency); |
180 | | |
181 | 11 | return _client->UploadPart(request); |
182 | 11 | }(), |
183 | 11 | "s3_file_writer::upload_part", std::cref(request).get(), &stream); |
184 | 11 | const auto elapsed_ms = elapsed_time_milliseconds(start); |
185 | | |
186 | 11 | const auto& request_id = outcome.IsSuccess() ? outcome.GetResult().GetRequestId() |
187 | 11 | : outcome.GetError().GetRequestId(); |
188 | | |
189 | 11 | TEST_SYNC_POINT_CALLBACK("S3FileWriter::_upload_one_part", &outcome); |
190 | 11 | if (!outcome.IsSuccess()) { |
191 | 1 | record_s3_request_failed(outcome.GetError()); |
192 | 1 | auto st = s3fs_error(outcome.GetError(), |
193 | 1 | fmt::format("failed to UploadPart: {}, part_num {}, upload_id={}", |
194 | 1 | opts.path.native(), part_num, upload_id)); |
195 | | |
196 | 1 | LOG(WARNING) << st.code << ", request_id=" << request_id; |
197 | 1 | return ObjStorageUploadResult { |
198 | 1 | .resp = {.status = st, |
199 | 1 | .http_code = static_cast<int>(outcome.GetError().GetResponseCode()), |
200 | 1 | .request_id = outcome.GetError().GetRequestId()}}; |
201 | 1 | } |
202 | 11 | LOG_IF(INFO, elapsed_ms > S3_REQUEST_THRESHOLD_MS) |
203 | 0 | << "UploadPart cost=" << elapsed_ms << "ms" |
204 | 0 | << ", request_id=" << request_id << ", bucket=" << opts.bucket << ", key=" << opts.key |
205 | 0 | << ", part_num=" << part_num << ", upload_id=" << upload_id; |
206 | 10 | return ObjStorageUploadResult {.resp = ObjStorageResponse::OK(), |
207 | 10 | .etag = outcome.GetResult().GetETag()}; |
208 | 11 | } |
209 | | |
210 | | ObjStorageResponse S3ObjStorageClient::complete_multipart_upload( |
211 | | const ObjStoragePath& opts, const std::string& upload_id, |
212 | 3 | const std::vector<ObjStorageCompletedPart>& completed_parts) { |
213 | 3 | CompleteMultipartUploadRequest request; |
214 | 3 | request.WithBucket(opts.bucket).WithKey(opts.key).WithUploadId(upload_id); |
215 | | |
216 | 3 | CompletedMultipartUpload completed_upload; |
217 | 3 | std::vector<CompletedPart> complete_parts; |
218 | 3 | std::ranges::transform(completed_parts, std::back_inserter(complete_parts), |
219 | 6 | [](const ObjStorageCompletedPart& part_ptr) { |
220 | 6 | CompletedPart part; |
221 | 6 | part.SetPartNumber(part_ptr.part_num); |
222 | 6 | part.SetETag(part_ptr.etag); |
223 | 6 | return part; |
224 | 6 | }); |
225 | 3 | completed_upload.SetParts(std::move(complete_parts)); |
226 | 3 | request.WithMultipartUpload(completed_upload); |
227 | | |
228 | 3 | TEST_SYNC_POINT_RETURN_WITH_VALUE("S3FileWriter::_complete:3", ObjStorageResponse(), this); |
229 | | |
230 | 2 | const auto start = std::chrono::steady_clock::now(); |
231 | 2 | auto outcome = SYNC_POINT_HOOK_RETURN_VALUE( |
232 | 2 | [&]() { |
233 | 2 | client_bvar::ScopedLatency scoped_latency( |
234 | 2 | client_bvar::s3_multi_part_upload_latency); |
235 | 2 | return _client->CompleteMultipartUpload(request); |
236 | 2 | }(), |
237 | 2 | "s3_file_writer::complete_multi_part", std::cref(request).get()); |
238 | 2 | const auto elapsed_ms = elapsed_time_milliseconds(start); |
239 | | |
240 | 2 | const auto& request_id = outcome.IsSuccess() ? outcome.GetResult().GetRequestId() |
241 | 2 | : outcome.GetError().GetRequestId(); |
242 | | |
243 | 2 | if (!outcome.IsSuccess()) { |
244 | 1 | record_s3_request_failed(outcome.GetError()); |
245 | 1 | auto st = s3fs_error(outcome.GetError(), |
246 | 1 | fmt::format("failed to CompleteMultipartUpload: {}, upload_id={}", |
247 | 1 | opts.path.native(), upload_id)); |
248 | 1 | LOG(WARNING) << st.code << ", request_id=" << request_id; |
249 | 1 | return {.status = st, |
250 | 1 | .http_code = static_cast<int>(outcome.GetError().GetResponseCode()), |
251 | 1 | .request_id = outcome.GetError().GetRequestId()}; |
252 | 1 | } |
253 | | |
254 | 2 | LOG_IF(INFO, elapsed_ms > S3_REQUEST_THRESHOLD_MS) |
255 | 0 | << "CompleteMultipartUpload cost=" << elapsed_ms << "ms" |
256 | 0 | << ", request_id=" << request_id << ", bucket=" << opts.bucket << ", key=" << opts.key |
257 | 0 | << ", upload_id=" << upload_id; |
258 | 1 | return ObjStorageResponse::OK(); |
259 | 2 | } |
260 | | |
261 | 15 | ObjStorageHeadResult S3ObjStorageClient::head_object(const ObjStoragePath& opts) { |
262 | 15 | Aws::S3::Model::HeadObjectRequest request; |
263 | 15 | request.WithBucket(opts.bucket).WithKey(opts.key); |
264 | | |
265 | 15 | auto outcome = SYNC_POINT_HOOK_RETURN_VALUE( |
266 | 15 | [&]() { |
267 | 15 | client_bvar::ScopedLatency scoped_latency(client_bvar::s3_head_latency); |
268 | 15 | return _client->HeadObject(request); |
269 | 15 | }(), |
270 | 15 | "s3_file_system::head_object", std::ref(request).get()); |
271 | | |
272 | 15 | if (outcome.IsSuccess()) { |
273 | 8 | return {.resp = ObjStorageResponse::OK(), |
274 | 8 | .file_size = outcome.GetResult().GetContentLength()}; |
275 | 8 | } else if (outcome.GetError().GetResponseCode() == Aws::Http::HttpResponseCode::NOT_FOUND) { |
276 | 4 | return {.resp = {.status = ObjStorageStatus::NOT_FOUND}, .file_size = 0}; |
277 | 4 | } else { |
278 | 3 | record_s3_request_failed(outcome.GetError()); |
279 | 3 | LOG(WARNING) << "failed to head object" |
280 | 3 | << "bucket " << opts.bucket << " key " << opts.key << " responseCode " |
281 | 3 | << outcome.GetError() << " error " << outcome.GetError().GetMessage() |
282 | 3 | << " request_id " << outcome.GetError().GetRequestId(); |
283 | 3 | return {.resp = {.status = s3fs_error( |
284 | 3 | outcome.GetError(), |
285 | 3 | fmt::format("failed to head object: {}", object_identity(opts))), |
286 | 3 | .http_code = static_cast<int>(outcome.GetError().GetResponseCode()), |
287 | 3 | .request_id = outcome.GetError().GetRequestId()}, |
288 | 3 | .file_size = -1}; |
289 | 3 | } |
290 | 15 | } |
291 | | |
292 | | ObjStorageResponse S3ObjStorageClient::get_object(const ObjStoragePath& opts, void* buffer, |
293 | | size_t offset, size_t bytes_read, |
294 | 0 | size_t* size_return) { |
295 | 0 | Aws::S3::Model::GetObjectRequest request; |
296 | 0 | request.WithBucket(opts.bucket).WithKey(opts.key); |
297 | 0 | request.SetRange(fmt::format("bytes={}-{}", offset, offset + bytes_read - 1)); |
298 | 0 | request.SetResponseStreamFactory(AwsWriteableStreamFactory(buffer, bytes_read)); |
299 | |
|
300 | 0 | auto outcome = [&]() { |
301 | 0 | client_bvar::ScopedLatency scoped_latency(client_bvar::s3_get_latency); |
302 | 0 | return _client->GetObject(request); |
303 | 0 | }(); |
304 | 0 | if (!outcome.IsSuccess()) { |
305 | 0 | record_s3_request_failed(outcome.GetError()); |
306 | 0 | return ObjStorageResponse { |
307 | 0 | .status = s3fs_error( |
308 | 0 | outcome.GetError(), |
309 | 0 | fmt::format("failed to get object: bucket={} object={} offset={} size={}", |
310 | 0 | opts.bucket, object_identity(opts), offset, bytes_read)), |
311 | 0 | .http_code = static_cast<int>(outcome.GetError().GetResponseCode()), |
312 | 0 | .request_id = outcome.GetError().GetRequestId(), |
313 | 0 | }; |
314 | 0 | } |
315 | 0 | *size_return = outcome.GetResult().GetContentLength(); |
316 | | // Short read, or a server or a proxy answering a ranged read with the whole object. |
317 | 0 | SYNC_POINT_CALLBACK("s3_obj_storage_client::get_object", size_return); |
318 | 0 | if (*size_return != bytes_read) { |
319 | 0 | const auto& request_id = outcome.GetResult().GetRequestId(); |
320 | 0 | return ObjStorageResponse { |
321 | 0 | .status = {ObjStorageStatus::INTERNAL_ERROR, |
322 | 0 | fmt::format("incomplete read from bucket={} object={} offset={}, expect " |
323 | 0 | "{}, got {}, request_id={}", |
324 | 0 | opts.bucket, object_identity(opts), offset, bytes_read, |
325 | 0 | *size_return, request_id)}, |
326 | 0 | .request_id = request_id}; |
327 | 0 | } |
328 | 0 | return ObjStorageResponse::OK(); |
329 | 0 | } |
330 | | |
331 | | ObjStorageListPageResult S3ObjStorageClient::list_objects_page( |
332 | 4 | const ObjStoragePath& opts, std::string_view continuation_token) { |
333 | 4 | const auto& prefix = opts.prefix.empty() ? opts.key : opts.prefix; |
334 | 4 | Aws::S3::Model::ListObjectsV2Request request; |
335 | 4 | request.WithBucket(opts.bucket) |
336 | 4 | .WithPrefix(prefix) |
337 | 4 | .WithMaxKeys(static_cast<int>(capabilities().max_list_page)); |
338 | 4 | if (!continuation_token.empty()) { |
339 | 2 | request.SetContinuationToken(std::string(continuation_token)); |
340 | 2 | } |
341 | 4 | TEST_SYNC_POINT_CALLBACK("S3ObjStorageClient::list_objects", &request); |
342 | | |
343 | 4 | auto outcome = [&]() { |
344 | 4 | client_bvar::ScopedLatency scoped_latency(client_bvar::s3_list_latency); |
345 | 4 | return _client->ListObjectsV2(request); |
346 | 4 | }(); |
347 | | |
348 | 4 | const auto& request_id = outcome.IsSuccess() ? outcome.GetResult().GetRequestId() |
349 | 4 | : outcome.GetError().GetRequestId(); |
350 | 4 | if (!outcome.IsSuccess()) { |
351 | | // Some S3-compatible providers (for example TOS) return NoSuchKey instead of an empty page |
352 | | // when a prefix does not exist. |
353 | 0 | if (outcome.GetError().GetErrorType() == Aws::S3::S3Errors::NO_SUCH_KEY) { |
354 | 0 | LOG(INFO) << fmt::format( |
355 | 0 | "NoSuchKey when listing objects, treat as empty response, endpoint: {}, " |
356 | 0 | "bucket: {}, prefix: {}, request_id: {}", |
357 | 0 | _config.endpoint, request.GetBucket(), request.GetPrefix(), request_id); |
358 | 0 | return {.resp = ObjStorageResponse::OK()}; |
359 | 0 | } |
360 | 0 | record_object_request_failed(static_cast<int>(outcome.GetError().GetResponseCode())); |
361 | 0 | const auto status = s3fs_error(outcome.GetError(), |
362 | 0 | fmt::format("failed to list objects: {}, prefix: {}", |
363 | 0 | request.GetBucket(), request.GetPrefix())); |
364 | 0 | LOG(WARNING) << fmt::format( |
365 | 0 | "failed to list objects, endpoint: {}, bucket: {}, prefix: {}, responseCode: {}, " |
366 | 0 | "error: {}, request_id: {}", |
367 | 0 | _config.endpoint, request.GetBucket(), request.GetPrefix(), |
368 | 0 | static_cast<int>(outcome.GetError().GetResponseCode()), |
369 | 0 | outcome.GetError().GetMessage(), request_id); |
370 | 0 | return { |
371 | 0 | .resp = {.status = status, |
372 | 0 | .http_code = static_cast<int>(outcome.GetError().GetResponseCode()), |
373 | 0 | .request_id = request_id}, |
374 | 0 | }; |
375 | 0 | } |
376 | | |
377 | 4 | const auto& result = outcome.GetResult(); |
378 | 4 | if (result.GetIsTruncated() && result.GetNextContinuationToken().empty()) { |
379 | 1 | LOG(WARNING) << fmt::format( |
380 | 1 | "failed to list objects, isTruncated but no continuation token, endpoint: {}, " |
381 | 1 | "bucket: {}, prefix: {}, request_id: {}", |
382 | 1 | _config.endpoint, request.GetBucket(), request.GetPrefix(), request_id); |
383 | 1 | return { |
384 | 1 | .resp = {.status = {ObjStorageStatus::INTERNAL_ERROR, |
385 | 1 | fmt::format("failed to list objects: {}, prefix: {}", |
386 | 1 | request.GetBucket(), request.GetPrefix())}, |
387 | 1 | .http_code = 0, |
388 | 1 | .request_id = request_id}, |
389 | 1 | }; |
390 | 1 | } |
391 | | |
392 | 3 | ObjStorageListPageResult page { |
393 | 3 | .resp = ObjStorageResponse::OK(), |
394 | 3 | .continuation_token = result.GetNextContinuationToken(), |
395 | 3 | .has_more = result.GetIsTruncated(), |
396 | 3 | }; |
397 | 3 | const auto& content = result.GetContents(); |
398 | 3 | page.objects.reserve(content.size()); |
399 | 5 | for (const auto& obj : content) { |
400 | 5 | DCHECK(obj.GetKey().starts_with(request.GetPrefix())) |
401 | 0 | << obj.GetKey() << ' ' << request.GetPrefix(); |
402 | 5 | page.objects.emplace_back(ObjectMeta {.key = obj.GetKey(), |
403 | 5 | .size = obj.GetSize(), |
404 | 5 | .mtime_s = obj.GetLastModified().Seconds()}); |
405 | 5 | } |
406 | 3 | return page; |
407 | 4 | } |
408 | | |
409 | | ObjStorageResponse S3ObjStorageClient::delete_objects(const ObjStoragePath& opts, |
410 | 1 | std::vector<std::string> objs) { |
411 | 1 | size_t max_delete_batch = 1000; |
412 | 1 | TEST_SYNC_POINT_CALLBACK("S3ObjClient::delete_objects", &max_delete_batch); |
413 | 1 | TEST_SYNC_POINT_CALLBACK("S3ObjStorageClient::delete_objects", &max_delete_batch); |
414 | 1 | max_delete_batch = std::max<size_t>(1, max_delete_batch); |
415 | 2 | for (size_t begin = 0; begin < objs.size(); begin += max_delete_batch) { |
416 | 1 | const size_t end = std::min(begin + max_delete_batch, objs.size()); |
417 | 1 | Aws::S3::Model::DeleteObjectsRequest delete_request; |
418 | 1 | delete_request.SetBucket(opts.bucket); |
419 | 1 | Aws::S3::Model::Delete del; |
420 | 1 | Aws::Vector<Aws::S3::Model::ObjectIdentifier> objects; |
421 | 1 | objects.reserve(end - begin); |
422 | 2 | for (size_t i = begin; i < end; ++i) { |
423 | 1 | Aws::S3::Model::ObjectIdentifier object; |
424 | 1 | object.SetKey(std::move(objs[i])); |
425 | 1 | objects.emplace_back(std::move(object)); |
426 | 1 | } |
427 | 1 | del.WithObjects(std::move(objects)).SetQuiet(true); |
428 | 1 | delete_request.SetDelete(std::move(del)); |
429 | | |
430 | 1 | auto delete_outcome = [&]() { |
431 | 1 | client_bvar::ScopedLatency scoped_latency(client_bvar::s3_delete_objects_latency); |
432 | 1 | return _client->DeleteObjects(delete_request); |
433 | 1 | }(); |
434 | 1 | SYNC_POINT_CALLBACK("s3_obj_storage_client::delete_objects", &delete_outcome); |
435 | 1 | SYNC_POINT_CALLBACK("s3_obj_storage_client::delete_objects_recursively", &delete_outcome); |
436 | 1 | if (!delete_outcome.IsSuccess()) { |
437 | 0 | record_s3_request_failed(delete_outcome.GetError()); |
438 | 0 | LOG(WARNING) << fmt::format( |
439 | 0 | "failed to delete objects, endpoint: {}, bucket: {}, key: {}, responseCode: " |
440 | 0 | "{}, error: {}, request_id: {}", |
441 | 0 | _config.endpoint, opts.bucket, |
442 | 0 | delete_request.GetDelete().GetObjects().front().GetKey(), |
443 | 0 | static_cast<int>(delete_outcome.GetError().GetResponseCode()), |
444 | 0 | delete_outcome.GetError().GetMessage(), |
445 | 0 | delete_outcome.GetError().GetRequestId()); |
446 | 0 | return ObjStorageResponse { |
447 | 0 | .status = s3fs_error(delete_outcome.GetError(), |
448 | 0 | fmt::format("failed to delete dir {}", opts.key)), |
449 | 0 | .http_code = static_cast<int>(delete_outcome.GetError().GetResponseCode()), |
450 | 0 | .request_id = delete_outcome.GetError().GetRequestId()}; |
451 | 0 | } |
452 | 1 | if (!delete_outcome.GetResult().GetErrors().empty()) { |
453 | 0 | const auto& error = delete_outcome.GetResult().GetErrors().front(); |
454 | 0 | LOG(WARNING) << fmt::format( |
455 | 0 | "failed to delete object in batch, endpoint: {}, bucket: {}, key: {}, error " |
456 | 0 | "code: {}, error: {}, request_id: {}", |
457 | 0 | _config.endpoint, opts.bucket, error.GetKey(), error.GetCode(), |
458 | 0 | error.GetMessage(), delete_outcome.GetResult().GetRequestId()); |
459 | 0 | return ObjStorageResponse { |
460 | 0 | .status = {ObjStorageStatus::INTERNAL_ERROR, |
461 | 0 | fmt::format("failed to delete object {}: {}, request_id={}", |
462 | 0 | error.GetKey(), error.GetMessage(), |
463 | 0 | delete_outcome.GetResult().GetRequestId())}, |
464 | 0 | .request_id = delete_outcome.GetResult().GetRequestId()}; |
465 | 0 | } |
466 | 1 | } |
467 | 1 | return ObjStorageResponse::OK(); |
468 | 1 | } |
469 | | |
470 | 1 | ObjStorageResponse S3ObjStorageClient::delete_object(const ObjStoragePath& opts) { |
471 | 1 | Aws::S3::Model::DeleteObjectRequest request; |
472 | 1 | request.WithBucket(opts.bucket).WithKey(opts.key); |
473 | | |
474 | 1 | auto outcome = [&]() { |
475 | 1 | client_bvar::ScopedLatency scoped_latency(client_bvar::s3_delete_object_latency); |
476 | | |
477 | 1 | return _client->DeleteObject(request); |
478 | 1 | }(); |
479 | 1 | TEST_SYNC_POINT_CALLBACK("S3ObjClient::delete_object", &outcome); |
480 | 1 | TEST_SYNC_POINT_CALLBACK("S3ObjStorageClient::delete_object", &outcome); |
481 | 1 | if (outcome.IsSuccess()) { |
482 | 0 | return ObjStorageResponse::OK(); |
483 | 0 | } |
484 | 1 | ObjStorageResponse response { |
485 | 1 | .status = s3fs_error(outcome.GetError(), |
486 | 1 | fmt::format("failed to delete object {}", opts.key)), |
487 | 1 | .http_code = static_cast<int>(outcome.GetError().GetResponseCode()), |
488 | 1 | .request_id = outcome.GetError().GetRequestId()}; |
489 | 1 | if (response.status.code == ObjStorageStatus::NOT_FOUND) { |
490 | 1 | return response; |
491 | 1 | } |
492 | 0 | record_s3_request_failed(outcome.GetError()); |
493 | 0 | LOG(WARNING) << fmt::format( |
494 | 0 | "failed to delete object, endpoint: {}, bucket: {}, key: {}, responseCode: {}, " |
495 | 0 | "error: {}, request_id: {}", |
496 | 0 | _config.endpoint, opts.bucket, opts.key, |
497 | 0 | static_cast<int>(outcome.GetError().GetResponseCode()), outcome.GetError().GetMessage(), |
498 | 0 | outcome.GetError().GetRequestId()); |
499 | 0 | return response; |
500 | 1 | } |
501 | | |
502 | | std::string S3ObjStorageClient::generate_presigned_url(const ObjStoragePath& opts, |
503 | 0 | int64_t expiration_secs) { |
504 | 0 | return _client->GeneratePresignedUrl(opts.bucket, opts.key, Aws::Http::HttpMethod::HTTP_GET, |
505 | 0 | expiration_secs); |
506 | 0 | } |
507 | | |
508 | 0 | ObjStorageResponse S3ObjStorageClient::check_versioning(const std::string& bucket) { |
509 | 0 | Aws::S3::Model::GetBucketVersioningRequest request; |
510 | 0 | request.SetBucket(bucket); |
511 | |
|
512 | 0 | auto outcome = _client->GetBucketVersioning(request); |
513 | |
|
514 | 0 | if (outcome.IsSuccess()) { |
515 | 0 | const auto& versioning_configuration = outcome.GetResult().GetStatus(); |
516 | 0 | if (versioning_configuration != Aws::S3::Model::BucketVersioningStatus::Enabled) { |
517 | 0 | LOG(WARNING) << "Err for check interval: bucket doesn't enable bucket versioning" |
518 | 0 | << " endpoint=" << _config.endpoint << " bucket=" << bucket; |
519 | 0 | return ObjStorageResponse { |
520 | 0 | .status = {ObjStorageStatus::INTERNAL_ERROR, |
521 | 0 | fmt::format("bucket versioning is not enabled: {}", bucket)}}; |
522 | 0 | } |
523 | 0 | } else { |
524 | 0 | record_s3_request_failed(outcome.GetError()); |
525 | 0 | LOG(WARNING) << "Err for check interval: failed to get status of bucket versioning" |
526 | 0 | << " endpoint=" << _config.endpoint << " bucket=" << bucket |
527 | 0 | << " responseCode=" << static_cast<int>(outcome.GetError().GetResponseCode()) |
528 | 0 | << " error=" << outcome.GetError().GetMessage() |
529 | 0 | << " request_id=" << outcome.GetError().GetRequestId(); |
530 | 0 | return ObjStorageResponse { |
531 | 0 | .status = s3fs_error(outcome.GetError(), |
532 | 0 | fmt::format("failed to get bucket versioning: {}", bucket)), |
533 | 0 | .http_code = static_cast<int>(outcome.GetError().GetResponseCode()), |
534 | 0 | .request_id = outcome.GetError().GetRequestId()}; |
535 | 0 | } |
536 | 0 | return ObjStorageResponse::OK(); |
537 | 0 | } |
538 | | |
539 | | ObjStorageResponse S3ObjStorageClient::abort_multipart_upload(const ObjStoragePath& opts, |
540 | 0 | const std::string& upload_id) { |
541 | 0 | Aws::S3::Model::AbortMultipartUploadRequest request; |
542 | 0 | request.WithBucket(opts.bucket).WithKey(opts.key).WithUploadId(upload_id); |
543 | |
|
544 | 0 | auto outcome = _client->AbortMultipartUpload(request); |
545 | 0 | if (!outcome.IsSuccess()) { |
546 | 0 | LOG(WARNING) << "failed to abort multipart upload" |
547 | 0 | << " endpoint=" << _config.endpoint << " bucket=" << opts.bucket |
548 | 0 | << " key=" << opts.key << " upload_id=" << upload_id |
549 | 0 | << " responseCode=" << static_cast<int>(outcome.GetError().GetResponseCode()) |
550 | 0 | << " error=" << outcome.GetError().GetMessage() |
551 | 0 | << " request_id=" << outcome.GetError().GetRequestId(); |
552 | 0 | if (outcome.GetError().GetResponseCode() == Aws::Http::HttpResponseCode::NOT_FOUND) { |
553 | 0 | return ObjStorageResponse::OK(); |
554 | 0 | } |
555 | 0 | record_s3_request_failed(outcome.GetError()); |
556 | 0 | return ObjStorageResponse { |
557 | 0 | .status = |
558 | 0 | s3fs_error(outcome.GetError(), |
559 | 0 | fmt::format("failed to abort multipart upload: {}, upload_id={}", |
560 | 0 | opts.path.native(), upload_id)), |
561 | 0 | .http_code = static_cast<int>(outcome.GetError().GetResponseCode()), |
562 | 0 | .request_id = outcome.GetError().GetRequestId(), |
563 | 0 | }; |
564 | 0 | } |
565 | 0 | return ObjStorageResponse::OK(); |
566 | 0 | } |
567 | | |
568 | | ObjStorageResponse S3ObjStorageClient::get_lifecycle(const std::string& bucket, |
569 | 0 | int64_t* expiration_days) { |
570 | 0 | Aws::S3::Model::GetBucketLifecycleConfigurationRequest request; |
571 | 0 | request.SetBucket(bucket); |
572 | |
|
573 | 0 | auto outcome = _client->GetBucketLifecycleConfiguration(request); |
574 | 0 | bool has_lifecycle = false; |
575 | 0 | if (outcome.IsSuccess()) { |
576 | 0 | const auto& rules = outcome.GetResult().GetRules(); |
577 | 0 | for (const auto& rule : rules) { |
578 | 0 | if (rule.NoncurrentVersionExpirationHasBeenSet()) { |
579 | 0 | has_lifecycle = true; |
580 | 0 | *expiration_days = rule.GetNoncurrentVersionExpiration().GetNoncurrentDays(); |
581 | 0 | } |
582 | 0 | } |
583 | 0 | } else { |
584 | 0 | record_s3_request_failed(outcome.GetError()); |
585 | 0 | LOG(WARNING) << "Err for check interval: failed to get bucket lifecycle" |
586 | 0 | << " endpoint=" << _config.endpoint << " bucket=" << bucket |
587 | 0 | << " responseCode=" << static_cast<int>(outcome.GetError().GetResponseCode()) |
588 | 0 | << " error=" << outcome.GetError().GetMessage() |
589 | 0 | << " request_id=" << outcome.GetError().GetRequestId(); |
590 | 0 | return ObjStorageResponse { |
591 | 0 | .status = s3fs_error(outcome.GetError(), |
592 | 0 | fmt::format("failed to get lift cycle: {}", bucket)), |
593 | 0 | .http_code = static_cast<int>(outcome.GetError().GetResponseCode()), |
594 | 0 | .request_id = outcome.GetError().GetRequestId()}; |
595 | 0 | } |
596 | | |
597 | 0 | if (!has_lifecycle) { |
598 | 0 | LOG(WARNING) << "Err for check interval: bucket doesn't have lifecycle configuration" |
599 | 0 | << " endpoint=" << _config.endpoint << " bucket=" << bucket; |
600 | 0 | return ObjStorageResponse { |
601 | 0 | .status = {ObjStorageStatus::NOT_FOUND, |
602 | 0 | fmt::format("bucket has no lifecycle configuration: {}", bucket)}}; |
603 | 0 | } |
604 | 0 | return ObjStorageResponse::OK(); |
605 | 0 | } |
606 | | |
607 | | } // namespace doris |