common/cpp/obj-client/obj_storage_client.h
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 | | #pragma once |
19 | | |
20 | | #include <gen_cpp/Status_types.h> |
21 | | |
22 | | #include <cstdint> |
23 | | #include <filesystem> |
24 | | #include <functional> |
25 | | #include <memory> |
26 | | #include <optional> |
27 | | #include <string> |
28 | | #include <string_view> |
29 | | #include <utility> |
30 | | #include <vector> |
31 | | |
32 | | namespace doris { |
33 | | // Names are in lexico order. |
34 | | enum class ObjStorageProvider : uint8_t { |
35 | | UNKNOWN = 0, |
36 | | AWS = 1, |
37 | | AZURE = 2, |
38 | | BOS = 3, |
39 | | COS = 4, |
40 | | OSS = 5, |
41 | | OBS = 6, |
42 | | GCP = 7, |
43 | | TOS = 8, |
44 | | }; |
45 | | |
46 | | /// eg: |
47 | | /// s3://bucket1/path/to/file.txt |
48 | | /// path: s3://bucket1/path/to/file.txt |
49 | | /// bucket: bucket1 |
50 | | /// key: path/to/file.txt |
51 | | struct ObjStoragePath { |
52 | | std::filesystem::path path = ""; |
53 | | std::string bucket {}; // blob container in azure |
54 | | std::string key {}; // blob name in azure |
55 | | std::string prefix {}; // for list and recursive delete |
56 | | }; |
57 | | |
58 | | struct ObjStorageEndpointInfo { |
59 | | std::string endpoint {}; |
60 | | std::string ak {}; |
61 | | std::string sk {}; |
62 | | std::string tls_debug_context {}; |
63 | | }; |
64 | | |
65 | | struct ObjectMeta { |
66 | | std::string key {}; |
67 | | int64_t size {0}; |
68 | | int64_t mtime_s {0}; |
69 | | }; |
70 | | |
71 | | struct ObjStorageCompletedPart { |
72 | | int part_num = 0; |
73 | | std::string etag {}; |
74 | | }; |
75 | | |
76 | | struct ObjStorageStatus { |
77 | | enum Code : int { |
78 | | OK = TStatusCode::OK, |
79 | | INTERNAL_ERROR = TStatusCode::INTERNAL_ERROR, |
80 | | LIMIT_REACH = TStatusCode::LIMIT_REACH, |
81 | | NOT_FOUND = TStatusCode::NOT_FOUND, |
82 | | END_OF_FILE = TStatusCode::END_OF_FILE, |
83 | | IO_ERROR = TStatusCode::IO_ERROR, |
84 | | NETWORK_ERROR = TStatusCode::NETWORK_ERROR, |
85 | | // Keep the legacy BE error code for access-denied object storage responses. |
86 | | PERMISSION_DENIED = -256, |
87 | | }; |
88 | | |
89 | 1.21k | ObjStorageStatus(int r = OK, std::string msg = "") : code(r), msg(std::move(msg)) {} |
90 | | // clang-format off |
91 | | int code {OK}; // To unify the error handle logic with BE, we'd better use the same error code as BE |
92 | | // clang-format on |
93 | | std::string msg; |
94 | | }; |
95 | | |
96 | | // We only store error code along with err_msg instead of Status to unify BE and recycler's error handle logic |
97 | | struct ObjStorageResponse { |
98 | | ObjStorageStatus status {ObjStorageStatus::OK, ""}; |
99 | | int http_code {200}; |
100 | | std::string request_id {}; |
101 | 0 | static ObjStorageResponse OK() { |
102 | | // clang-format off |
103 | 0 | return { |
104 | 0 | .status = ObjStorageStatus{ObjStorageStatus::OK, ""}, |
105 | 0 | .http_code = 200, |
106 | 0 | .request_id = "" |
107 | 0 | }; |
108 | | // clang-format on |
109 | 0 | } |
110 | | |
111 | 0 | static ObjStorageResponse rate_limit(int status_code, int http_code, std::string message) { |
112 | 0 | return { |
113 | 0 | .status = ObjStorageStatus {status_code, std::move(message)}, |
114 | 0 | .http_code = http_code, |
115 | 0 | }; |
116 | 0 | } |
117 | | |
118 | 365k | bool ok() const { return status.code == ObjStorageStatus::OK; } |
119 | | }; |
120 | | |
121 | | // Convert a provider HTTP response code into the object-storage status domain. A non-positive |
122 | | // response code means that no HTTP response was received. |
123 | | ObjStorageStatus obj_storage_status_from_http_code(int http_code, std::string message); |
124 | | |
125 | | struct ObjStorageUploadResult { |
126 | | ObjStorageResponse resp = ObjStorageResponse::OK(); |
127 | | std::optional<std::string> upload_id = std::nullopt; |
128 | | std::optional<std::string> etag = std::nullopt; |
129 | | }; |
130 | | |
131 | | struct ObjStorageHeadResult { |
132 | | ObjStorageResponse resp = ObjStorageResponse::OK(); |
133 | | long long file_size {0}; |
134 | | }; |
135 | | |
136 | | struct ObjStorageListResult { |
137 | | ObjStorageResponse resp = ObjStorageResponse::OK(); |
138 | | std::optional<ObjectMeta> object = std::nullopt; |
139 | | }; |
140 | | |
141 | | struct ObjStorageListPageResult { |
142 | | ObjStorageResponse resp = ObjStorageResponse::OK(); |
143 | | std::vector<ObjectMeta> objects {}; |
144 | | std::string continuation_token {}; |
145 | | bool has_more = false; |
146 | | }; |
147 | | |
148 | | struct ObjStorageCapabilities { |
149 | | size_t max_delete_batch = 1; |
150 | | size_t max_list_page = 1000; |
151 | | }; |
152 | | |
153 | | using ObjStorageDeleteTask = std::function<ObjStorageResponse()>; |
154 | | |
155 | | // Implementations must enqueue tasks immediately and remain reusable after wait() completes. |
156 | | // Recycler implements this in cloud/src/recycler/s3_accessor.cpp; BE uses the synchronous fallback. |
157 | | class ObjStorageDeleteExecutor { |
158 | | public: |
159 | | virtual ~ObjStorageDeleteExecutor() = default; |
160 | | virtual ObjStorageResponse submit(ObjStorageDeleteTask task) = 0; |
161 | | virtual ObjStorageResponse wait() = 0; |
162 | | }; |
163 | | |
164 | | struct ObjStorageRecursiveDeleteOptions { |
165 | | int64_t expiration_time = 0; |
166 | | size_t max_tasks_per_batch = 1; |
167 | | std::shared_ptr<ObjStorageDeleteExecutor> executor {}; |
168 | | }; |
169 | | |
170 | | class ObjStorageListIterator; |
171 | | class RateLimitedObjStorageClient; |
172 | | |
173 | | // Provider implementations are in common/cpp/obj-client/s3_obj_storage_client.cpp and |
174 | | // common/cpp/obj-client/azure_obj_storage_client.cpp. |
175 | | // Clients are shared-owned so a lazy list iterator can keep the complete decorator chain alive. |
176 | | class ObjStorageClient : public std::enable_shared_from_this<ObjStorageClient> { |
177 | | public: |
178 | | ObjStorageClient() = default; |
179 | 0 | virtual ~ObjStorageClient() = default; |
180 | | |
181 | | ObjStorageClient(const ObjStorageClient&) = delete; |
182 | | ObjStorageClient& operator=(const ObjStorageClient&) = delete; |
183 | | // Create a multipart upload request. The returned token may be provider-issued or local and |
184 | | // identifies this writer's parts. |
185 | | // The input parameters should include the bucket and key for the object storage. |
186 | | virtual ObjStorageUploadResult create_multipart_upload(const ObjStoragePath& opts) = 0; |
187 | | // To directly upload a piece of data to object storage and generate a user-visible file. |
188 | | // You need to clearly specify the bucket and key |
189 | | virtual ObjStorageResponse put_object(const ObjStoragePath& opts, std::string_view stream) = 0; |
190 | | // Upload one part of a large object without making the object visible to users. upload_id is |
191 | | // the provider-issued or local writer token returned by create_multipart_upload. |
192 | | // Reusing the same bucket, key, upload_id, and part_num replaces that writer's staged part. |
193 | | virtual ObjStorageUploadResult upload_part(const ObjStoragePath& opts, |
194 | | const std::string& upload_id, |
195 | | std::string_view stream, int part_num) = 0; |
196 | | // Combine the parts belonging to upload_id into the key passed in opts. After this succeeds, |
197 | | // the complete object is visible in object storage. |
198 | | virtual ObjStorageResponse complete_multipart_upload( |
199 | | const ObjStoragePath& opts, const std::string& upload_id, |
200 | | const std::vector<ObjStorageCompletedPart>& completed_parts) = 0; |
201 | | // According to the passed bucket and key, it will access whether the corresponding file exists in the object storage. |
202 | | // If it exists, it will return the corresponding file size |
203 | | virtual ObjStorageHeadResult head_object(const ObjStoragePath& opts) = 0; |
204 | | // According to the bucket and key, it finds the corresponding file in the object storage |
205 | | // and starting from the offset, it reads bytes_read bytes into the buffer, with size_return recording the actual number of bytes read |
206 | | virtual ObjStorageResponse get_object(const ObjStoragePath& opts, void* buffer, size_t offset, |
207 | | size_t bytes_read, size_t* size_return) = 0; |
208 | | // Return a lazy iterator that fetches one provider page at a time. |
209 | | virtual std::unique_ptr<ObjStorageListIterator> list_objects(const ObjStoragePath& opts); |
210 | | // Collect all objects by consuming the lazy iterator. This preserves the eager BE API. |
211 | | // **Notice**: The files returned by this function contain the full key in object storage. |
212 | | virtual ObjStorageResponse list_objects(const ObjStoragePath& opts, |
213 | | std::vector<ObjectMeta>* objects); |
214 | | |
215 | | // According to the bucket and prefix specified by the user, it performs batch deletion based on the object names in the object array. |
216 | | virtual ObjStorageResponse delete_objects(const ObjStoragePath& opts, |
217 | | std::vector<std::string> objs) = 0; |
218 | | // Delete the file named key in the object storage bucket. |
219 | | virtual ObjStorageResponse delete_object(const ObjStoragePath& opts) = 0; |
220 | | virtual ObjStorageCapabilities capabilities() const = 0; |
221 | | // Return a presigned URL for users to access the object |
222 | | virtual std::string generate_presigned_url(const ObjStoragePath& opts, |
223 | | int64_t expiration_secs) = 0; |
224 | | |
225 | | // Get the objects' expiration time on the bucket |
226 | | virtual ObjStorageResponse get_lifecycle(const std::string& bucket, |
227 | | int64_t* expiration_days) = 0; |
228 | | |
229 | | // Check if the objects' versioning is on or off |
230 | | // returns 0 when versioning is on, otherwise versioning is off or check failed |
231 | | virtual ObjStorageResponse check_versioning(const std::string& bucket) = 0; |
232 | | |
233 | | virtual ObjStorageResponse abort_multipart_upload(const ObjStoragePath& path, |
234 | | const std::string& upload_id) = 0; |
235 | | |
236 | | protected: |
237 | | // Fetch at most one page. One call corresponds to exactly one provider request. |
238 | | virtual ObjStorageListPageResult list_objects_page(const ObjStoragePath& opts, |
239 | | std::string_view continuation_token) = 0; |
240 | | |
241 | | private: |
242 | | friend class ObjStorageListIterator; |
243 | | friend class RateLimitedObjStorageClient; |
244 | | }; |
245 | | |
246 | | // A client-side iterator returned by ObjStorageClient. It requests one fixed-size page at a time. |
247 | | class ObjStorageListIterator { |
248 | | public: |
249 | | ObjStorageListIterator(std::shared_ptr<ObjStorageClient> client, ObjStoragePath opts) |
250 | 16 | : client_(std::move(client)), opts_(std::move(opts)) {} |
251 | | |
252 | 0 | bool is_valid() const { return is_valid_; } |
253 | | ObjStorageResponse has_next(); |
254 | | ObjStorageListResult next(); |
255 | | |
256 | | private: |
257 | | std::shared_ptr<ObjStorageClient> client_; |
258 | | ObjStoragePath opts_; |
259 | | std::vector<ObjectMeta> objects_; |
260 | | size_t next_index_ = 0; |
261 | | std::string continuation_token_; |
262 | | bool has_more_ = true; |
263 | | bool is_valid_ = true; |
264 | | }; |
265 | | |
266 | | // Provider-independent recursive deletion shared by concrete clients. Passing the complete client |
267 | | // decorator chain keeps it alive for asynchronous delete tasks and applies policy per list page and |
268 | | // delete batch. |
269 | | ObjStorageResponse delete_objects_recursively(std::shared_ptr<ObjStorageClient> client, |
270 | | const ObjStoragePath& path, |
271 | | const ObjStorageRecursiveDeleteOptions& options = {}); |
272 | | } // namespace doris |
273 | | |
274 | | // Keep the BE namespace spelling source-compatible while the implementation is |
275 | | // shared with Recycler in `doris`. |
276 | | namespace doris::io { |
277 | | using ::doris::ObjStorageCapabilities; |
278 | | using ::doris::ObjStorageClient; |
279 | | using ::doris::ObjStorageCompletedPart; |
280 | | using ::doris::ObjStorageDeleteExecutor; |
281 | | using ::doris::ObjStorageDeleteTask; |
282 | | using ::doris::ObjStorageEndpointInfo; |
283 | | using ::doris::ObjStorageHeadResult; |
284 | | using ::doris::ObjStorageListIterator; |
285 | | using ::doris::ObjStorageListPageResult; |
286 | | using ::doris::ObjStorageListResult; |
287 | | using ::doris::ObjStoragePath; |
288 | | using ::doris::ObjStorageProvider; |
289 | | using ::doris::ObjStorageRecursiveDeleteOptions; |
290 | | using ::doris::ObjStorageResponse; |
291 | | using ::doris::ObjStorageStatus; |
292 | | using ::doris::ObjStorageUploadResult; |
293 | | using ::doris::ObjectMeta; |
294 | | using ::doris::delete_objects_recursively; |
295 | | } // namespace doris::io |