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-08-18 14:24

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