common/cpp/obj-client/obj_storage_client.cpp

Source
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

#include "obj_storage_client.h"

#include <cpp/sync_point.h>
#include <glog/logging.h>

#include <algorithm>
#include <chrono>

namespace doris {
ObjStorageStatus obj_storage_status_from_http_code(int http_code, std::string message) {
    switch (http_code) {
    case 401:
    case 403:
        return {ObjStorageStatus::PERMISSION_DENIED, std::move(message)};
    case 404:
        return {ObjStorageStatus::NOT_FOUND, std::move(message)};
    case 429:
        return {ObjStorageStatus::LIMIT_REACH, std::move(message)};
    default:
        return {http_code <= 0 ? ObjStorageStatus::NETWORK_ERROR : ObjStorageStatus::INTERNAL_ERROR,
                std::move(message)};
    }
}

std::unique_ptr<ObjStorageListIterator> ObjStorageClient::list_objects(const ObjStoragePath& opts) {
    return std::make_unique<ObjStorageListIterator>(shared_from_this(), opts);
}

ObjStorageResponse ObjStorageClient::list_objects(const ObjStoragePath& opts,
                                                  std::vector<ObjectMeta>* objects) {
    objects->clear();
    auto iter = list_objects(opts);
    for (;;) {
        auto result = iter->next();
        if (!result.object.has_value()) {
            if (!result.resp.ok()) {
                objects->clear();
            }
            return result.resp;
        }
        objects->emplace_back(std::move(*result.object));
    }
}

ObjStorageResponse ObjStorageListIterator::has_next() {
    if (!is_valid_) {
        return {
                .status = {ObjStorageStatus::INTERNAL_ERROR, "Iterator is invalid"},
                .http_code = 0,
        };
    }
    while (next_index_ == objects_.size()) {
        if (!has_more_) {
            return {
                    .status = {ObjStorageStatus::END_OF_FILE, "No more results"},
                    .http_code = 200,
            };
        }
        auto page = client_->list_objects_page(opts_, continuation_token_);
        if (!page.resp.ok()) {
            is_valid_ = false;
            return page.resp;
        }
        objects_ = std::move(page.objects);
        next_index_ = 0;
        continuation_token_ = std::move(page.continuation_token);
        has_more_ = page.has_more;
    }
    return ObjStorageResponse::OK();
}

ObjStorageListResult ObjStorageListIterator::next() {
    auto response = has_next();
    if (response.status.code == ObjStorageStatus::END_OF_FILE) {
        return {.resp = ObjStorageResponse::OK(), .object = {}};
    }
    if (!response.ok()) {
        return {.resp = std::move(response), .object = {}};
    }
    return {
            .resp = ObjStorageResponse::OK(),
            .object = std::move(objects_[next_index_++]),
    };
}

ObjStorageResponse delete_objects_recursively(
        std::shared_ptr<ObjStorageClient> client, const ObjStoragePath& path,
        const ObjStorageRecursiveDeleteOptions& delete_options) {
    const auto start_time = std::chrono::steady_clock::now();
    auto list_path = path;
    if (list_path.prefix.empty()) {
        list_path.prefix = list_path.key;
    }
    auto delete_batch_size = std::max<size_t>(1, client->capabilities().max_delete_batch);
    TEST_SYNC_POINT_CALLBACK("ObjStorageClient::delete_objects_recursively_", &delete_batch_size);
    delete_batch_size = std::max<size_t>(1, delete_batch_size);
    const auto max_tasks_per_batch = std::max<size_t>(1, delete_options.max_tasks_per_batch);
    std::vector<std::string> keys;
    keys.reserve(delete_batch_size);
    size_t pending_tasks = 0;
    size_t total_batches = 0;
    size_t num_deleted = 0;
    size_t error_count = 0;
    auto first_error = ObjStorageResponse::OK();

    auto elapsed_milliseconds = [&]() {
        return std::chrono::duration_cast<std::chrono::milliseconds>(
                       std::chrono::steady_clock::now() - start_time)
                .count();
    };
    auto finish = [&](ObjStorageResponse response) {
        LOG(INFO) << "delete objects under " << list_path.bucket << "/" << list_path.prefix
                  << " finished, ret=" << response.status.code
                  << ", total_batches=" << total_batches << ", num_deleted=" << num_deleted
                  << ", error_count=" << error_count << ", cost=" << elapsed_milliseconds()
                  << " ms";
        return response;
    };
    auto record_error = [&](ObjStorageResponse response) {
        if (response.ok()) {
            return;
        }
        ++error_count;
        if (first_error.ok()) {
            first_error = std::move(response);
        }
    };

    auto wait_for_tasks = [&]() {
        if (pending_tasks == 0) {
            return ObjStorageResponse::OK();
        }
        const auto tasks_in_batch = pending_tasks;
        pending_tasks = 0;
        auto response = delete_options.executor ? delete_options.executor->wait()
                                                : ObjStorageResponse::OK();
        ++total_batches;
        LOG(INFO) << "delete objects under " << list_path.bucket << "/" << list_path.prefix
                  << " batch " << total_batches << " completed"
                  << ", tasks_in_batch=" << tasks_in_batch << ", total_deleted=" << num_deleted
                  << ", elapsed=" << elapsed_milliseconds() << " ms";
        return response;
    };
    auto submit_delete_task = [&]() -> bool {
        ObjStorageDeleteTask task = [client, bucket = path.bucket,
                                     batch = std::move(keys)]() mutable {
            return client->delete_objects(ObjStoragePath {.bucket = std::move(bucket)},
                                          std::move(batch));
        };
        keys.clear();
        keys.reserve(delete_batch_size);

        ObjStorageResponse response;
        if (delete_options.executor) {
            response = delete_options.executor->submit(std::move(task));
        } else {
            response = task();
        }
        ++pending_tasks;
        if (!response.ok()) {
            record_error(std::move(response));
            record_error(wait_for_tasks());
            return false;
        }
        if (pending_tasks == max_tasks_per_batch) {
            record_error(wait_for_tasks());
        }
        // Match the pre-refactor Recycler behavior: do not scan the next task batch after the
        // current batch reports a submit, delete, or wait failure.
        return first_error.ok();
    };

    auto iter = client->list_objects(list_path);
    for (;;) {
        auto result = iter->next();
        if (!result.object.has_value()) {
            if (result.resp.ok()) {
                break;
            }
            if (!keys.empty()) {
                submit_delete_task();
            }
            record_error(wait_for_tasks());
            record_error(std::move(result.resp));
            return finish(std::move(first_error));
        }
        auto& object = *result.object;
        if (delete_options.expiration_time > 0 && object.mtime_s > delete_options.expiration_time) {
            continue;
        }
        ++num_deleted;
        keys.emplace_back(std::move(object.key));
        if (keys.size() == delete_batch_size && !submit_delete_task()) {
            return finish(std::move(first_error));
        }
    }
    if (!keys.empty() && !submit_delete_task()) {
        return finish(std::move(first_error));
    }
    record_error(wait_for_tasks());
    return finish(std::move(first_error));
}

} // namespace doris

Coverage Report

Created: 2026-11-02 20:37

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 "obj_storage_client.h"
19
20		#include <cpp/sync_point.h>
21		#include <glog/logging.h>
22
23		#include <algorithm>
24		#include <chrono>
25
26		namespace doris {
27	43	ObjStorageStatus obj_storage_status_from_http_code(int http_code, std::string message) {
28	43	switch (http_code) {
29	26	case 401:
30	30	case 403:
31	30	return {ObjStorageStatus::PERMISSION_DENIED, std::move(message)};
32	2	case 404:
33	2	return {ObjStorageStatus::NOT_FOUND, std::move(message)};
34	3	case 429:
35	3	return {ObjStorageStatus::LIMIT_REACH, std::move(message)};
36	8	default:
37	8	return {http_code <= 0 ? ObjStorageStatus::NETWORK_ERROR : ObjStorageStatus::INTERNAL_ERROR,
38	8	std::move(message)};
39	43	}
40	43	}
41
42	187	std::unique_ptr<ObjStorageListIterator> ObjStorageClient::list_objects(const ObjStoragePath& opts) {
43	187	return std::make_unique<ObjStorageListIterator>(shared_from_this(), opts);
44	187	}
45
46		ObjStorageResponse ObjStorageClient::list_objects(const ObjStoragePath& opts,
47	69	std::vector<ObjectMeta>* objects) {
48	69	objects->clear();
49	69	auto iter = list_objects(opts);
50	151	for (;;) {
51	151	auto result = iter->next();
52	151	if (!result.object.has_value()) {
53	69	if (!result.resp.ok()) {
54	2	objects->clear();
55	2	}
56	69	return result.resp;
57	69	}
58	82	objects->emplace_back(std::move(*result.object));
59	82	}
60	69	}
61
62	332	ObjStorageResponse ObjStorageListIterator::has_next() {
63	332	if (!is_valid_) {
64	0	return {
65	0	.status = {ObjStorageStatus::INTERNAL_ERROR, "Iterator is invalid"},
66	0	.http_code = 0,
67	0	};
68	0	}
69	520	while (next_index_ == objects_.size()) {
70	322	if (!has_more_) {
71	131	return {
72	131	.status = {ObjStorageStatus::END_OF_FILE, "No more results"},
73	131	.http_code = 200,
74	131	};
75	131	}
76	191	auto page = client_->list_objects_page(opts_, continuation_token_);
77	191	if (!page.resp.ok()) {
78	3	is_valid_ = false;
79	3	return page.resp;
80	3	}
81	188	objects_ = std::move(page.objects);
82	188	next_index_ = 0;
83	188	continuation_token_ = std::move(page.continuation_token);
84	188	has_more_ = page.has_more;
85	188	}
86	198	return ObjStorageResponse::OK();
87	332	}
88
89	332	ObjStorageListResult ObjStorageListIterator::next() {
90	332	auto response = has_next();
91	332	if (response.status.code == ObjStorageStatus::END_OF_FILE) {
92	131	return {.resp = ObjStorageResponse::OK(), .object = {}};
93	131	}
94	201	if (!response.ok()) {
95	3	return {.resp = std::move(response), .object = {}};
96	3	}
97	198	return {
98	198	.resp = ObjStorageResponse::OK(),
99	198	.object = std::move(objects_[next_index_++]),
100	198	};
101	201	}
102
103		ObjStorageResponse delete_objects_recursively(
104		std::shared_ptr<ObjStorageClient> client, const ObjStoragePath& path,
105	117	const ObjStorageRecursiveDeleteOptions& delete_options) {
106	117	const auto start_time = std::chrono::steady_clock::now();
107	117	auto list_path = path;
108	117	if (list_path.prefix.empty()) {
109	2	list_path.prefix = list_path.key;
110	2	}
111	117	auto delete_batch_size = std::max<size_t>(1, client->capabilities().max_delete_batch);
112	117	TEST_SYNC_POINT_CALLBACK("ObjStorageClient::delete_objects_recursively_", &delete_batch_size);
113	117	delete_batch_size = std::max<size_t>(1, delete_batch_size);
114	117	const auto max_tasks_per_batch = std::max<size_t>(1, delete_options.max_tasks_per_batch);
115	117	std::vector<std::string> keys;
116	117	keys.reserve(delete_batch_size);
117	117	size_t pending_tasks = 0;
118	117	size_t total_batches = 0;
119	117	size_t num_deleted = 0;
120	117	size_t error_count = 0;
121	117	auto first_error = ObjStorageResponse::OK();
122
123	230	auto elapsed_milliseconds = [&]() {
124	230	return std::chrono::duration_cast<std::chrono::milliseconds>(
125	230	std::chrono::steady_clock::now() - start_time)
126	230	.count();
127	230	};
128	117	auto finish = [&](ObjStorageResponse response) {
129	117	LOG(INFO) << "delete objects under " << list_path.bucket << "/" << list_path.prefix
130	117	<< " finished, ret=" << response.status.code
131	117	<< ", total_batches=" << total_batches << ", num_deleted=" << num_deleted
132	117	<< ", error_count=" << error_count << ", cost=" << elapsed_milliseconds()
133	117	<< " ms";
134	117	return response;
135	117	};
136	231	auto record_error = [&](ObjStorageResponse response) {
137	231	if (response.ok()) {
138	178	return;
139	178	}
140	53	++error_count;
141	53	if (first_error.ok()) {
142	53	first_error = std::move(response);
143	53	}
144	53	};
145
146	178	auto wait_for_tasks = [&]() {
147	178	if (pending_tasks == 0) {
148	65	return ObjStorageResponse::OK();
149	65	}
150	113	const auto tasks_in_batch = pending_tasks;
151	113	pending_tasks = 0;
152	113	auto response = delete_options.executor ? delete_options.executor->wait()
153	113	: ObjStorageResponse::OK();
154	113	++total_batches;
155	113	LOG(INFO) << "delete objects under " << list_path.bucket << "/" << list_path.prefix
156	113	<< " batch " << total_batches << " completed"
157	113	<< ", tasks_in_batch=" << tasks_in_batch << ", total_deleted=" << num_deleted
158	113	<< ", elapsed=" << elapsed_milliseconds() << " ms";
159	113	return response;
160	178	};
161	117	auto submit_delete_task = [&]() -> bool {
162	113	ObjStorageDeleteTask task = [client, bucket = path.bucket,
163	113	batch = std::move(keys)]() mutable {
164	113	return client->delete_objects(ObjStoragePath {.bucket = std::move(bucket)},
165	113	std::move(batch));
166	113	};
167	113	keys.clear();
168	113	keys.reserve(delete_batch_size);
169
170	113	ObjStorageResponse response;
171	113	if (delete_options.executor) {
172	0	response = delete_options.executor->submit(std::move(task));
173	113	} else {
174	113	response = task();
175	113	}
176	113	++pending_tasks;
177	113	if (!response.ok()) {
178	52	record_error(std::move(response));
179	52	record_error(wait_for_tasks());
180	52	return false;
181	52	}
182	61	if (pending_tasks == max_tasks_per_batch) {
183	61	record_error(wait_for_tasks());
184	61	}
185		// Match the pre-refactor Recycler behavior: do not scan the next task batch after the
186		// current batch reports a submit, delete, or wait failure.
187	61	return first_error.ok();
188	113	};
189
190	117	auto iter = client->list_objects(list_path);
191	178	for (;;) {
192	178	auto result = iter->next();
193	178	if (!result.object.has_value()) {
194	65	if (result.resp.ok()) {
195	64	break;
196	64	}
197	1	if (!keys.empty()) {
198	0	submit_delete_task();
199	0	}
200	1	record_error(wait_for_tasks());
201	1	record_error(std::move(result.resp));
202	1	return finish(std::move(first_error));
203	65	}
204	113	auto& object = *result.object;
205	113	if (delete_options.expiration_time > 0 && object.mtime_s > delete_options.expiration_time) {
206	0	continue;
207	0	}
208	113	++num_deleted;
209	113	keys.emplace_back(std::move(object.key));
210	113	if (keys.size() == delete_batch_size && !submit_delete_task()) {
211	52	return finish(std::move(first_error));
212	52	}
213	113	}
214	64	if (!keys.empty() && !submit_delete_task()) {
215	0	return finish(std::move(first_error));
216	0	}
217	64	record_error(wait_for_tasks());
218	64	return finish(std::move(first_error));
219	64	}
220
221		} // namespace doris