Coverage Report

Created: 2026-08-31 19:39

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
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