/root/doris/cloud/src/recycler/checker.h
Line | Count | Source (jump to first uncovered line) |
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 "resource-manager/resource_manager.h" |
21 | | #if defined(USE_LIBCPP) && _LIBCPP_ABI_VERSION <= 1 |
22 | | #define _LIBCPP_ABI_INCOMPLETE_TYPES_IN_DEQUE |
23 | | #endif |
24 | | #include <atomic> |
25 | | #include <condition_variable> |
26 | | #include <cstdint> |
27 | | #include <deque> |
28 | | #include <functional> |
29 | | #include <memory> |
30 | | #include <string_view> |
31 | | #include <thread> |
32 | | #include <unordered_map> |
33 | | #include <unordered_set> |
34 | | |
35 | | #include "meta-store/txn_kv_error.h" |
36 | | #include "recycler/storage_vault_accessor.h" |
37 | | #include "recycler/white_black_list.h" |
38 | | #include "snapshot/snapshot_manager.h" |
39 | | |
40 | | namespace doris { |
41 | | class RowsetMetaCloudPB; |
42 | | } // namespace doris |
43 | | |
44 | | namespace doris::cloud { |
45 | | class StorageVaultAccessor; |
46 | | class InstanceChecker; |
47 | | class TxnKv; |
48 | | class InstanceInfoPB; |
49 | | |
50 | | struct PendingTableStreamDrop { |
51 | | int64_t base_db_id; |
52 | | int64_t base_table_id; |
53 | | int64_t stream_db_id; |
54 | | |
55 | | bool matches(int64_t offset_base_db_id, int64_t offset_base_table_id, |
56 | 2 | int64_t offset_stream_db_id) const { |
57 | 2 | return base_db_id == offset_base_db_id && base_table_id == offset_base_table_id && |
58 | 2 | stream_db_id == offset_stream_db_id; |
59 | 2 | } |
60 | | }; |
61 | | |
62 | | TxnErrorCode collect_pending_table_stream_drops( |
63 | | const std::shared_ptr<TxnKv>& txn_kv, std::string_view instance_id, |
64 | | std::unordered_map<int64_t, PendingTableStreamDrop>* pending_drops); |
65 | | |
66 | | class Checker { |
67 | | public: |
68 | | explicit Checker(std::shared_ptr<TxnKv> txn_kv); |
69 | | ~Checker(); |
70 | | |
71 | | int start(); |
72 | | |
73 | | void stop(); |
74 | 323 | bool stopped() const { return stopped_.load(std::memory_order_acquire); } |
75 | | |
76 | | private: |
77 | | void lease_check_jobs(); |
78 | | void inspect_instance_check_interval(); |
79 | | void do_inspect(const InstanceInfoPB& instance); |
80 | | |
81 | | private: |
82 | | friend class RecyclerServiceImpl; |
83 | | |
84 | | std::shared_ptr<TxnKv> txn_kv_; |
85 | | std::atomic_bool stopped_ {false}; |
86 | | std::string ip_port_; |
87 | | std::vector<std::thread> workers_; |
88 | | |
89 | | std::mutex mtx_; |
90 | | // notify check workers |
91 | | std::condition_variable pending_instance_cond_; |
92 | | std::deque<InstanceInfoPB> pending_instance_queue_; |
93 | | // instance_id -> enqueue_timestamp |
94 | | std::unordered_map<std::string, long> pending_instance_map_; |
95 | | std::unordered_map<std::string, std::shared_ptr<InstanceChecker>> working_instance_map_; |
96 | | // notify instance scanner and lease thread |
97 | | std::condition_variable notifier_; |
98 | | |
99 | | WhiteBlackList instance_filter_; |
100 | | }; |
101 | | |
102 | | class InstanceChecker { |
103 | | public: |
104 | | explicit InstanceChecker(std::shared_ptr<TxnKv> txn_kv, const std::string& instance_id); |
105 | | // Return 0 if success, otherwise error |
106 | | int init(const InstanceInfoPB& instance); |
107 | | // Return 0 if success. |
108 | | // Return 1 if data leak is identified. |
109 | | // Return negative if a temporary error occurred during the check process. |
110 | | int do_inverted_check(); |
111 | | |
112 | | // Return 0 if success. |
113 | | // Return 1 if data loss is identified. |
114 | | // Return negative if a temporary error occurred during the check process. |
115 | | int do_check(); |
116 | | |
117 | | // Return 0 if success. |
118 | | // Return 1 if delete bitmap leak is identified. |
119 | | // Return negative if a temporary error occurred during the check process. |
120 | | int do_delete_bitmap_inverted_check(); |
121 | | |
122 | | // version = 1 : https://github.com/apache/doris/pull/40204 |
123 | | // checks if https://github.com/apache/doris/pull/40204 works as expected |
124 | | // the stale delete bitmap will be cleared in MS when BE delete expired stale rowsets |
125 | | // NOTE: stale rowsets will be lost after BE restarts, so there may be some stale delete bitmaps |
126 | | // which will not be cleared. |
127 | | // version = 2 : https://github.com/apache/doris/pull/49822 |
128 | | int do_delete_bitmap_storage_optimize_check(int version = 2); |
129 | | |
130 | | int do_mow_job_key_check(); |
131 | | |
132 | | int do_tablet_stats_key_check(); |
133 | | |
134 | | int do_restore_job_check(); |
135 | | |
136 | | int do_txn_key_check(); |
137 | | |
138 | | // check table and partition version key |
139 | | // table version should be greater than the versions of all its partitions |
140 | | // Return 0 if success, otherwise error |
141 | | int do_version_key_check(); |
142 | | |
143 | | // Return 0 if success. |
144 | | // Return 1 if meta rowset key leak or loss is identified. |
145 | | // Return negative if a temporary error occurred during the check process. |
146 | | int do_meta_rowset_key_check(); |
147 | | |
148 | | // Return 0 if success. |
149 | | // Return 1 if snapshot key and file leak or loss is identified. |
150 | | // Return negative if a temporary error occurred during the check process. |
151 | | int do_snapshots_check(); |
152 | | |
153 | | // Return 0 if success. |
154 | | // Return 1 if mvcc meta key and data leak or loss is identified. |
155 | | // Return negative if a temporary error occurred during the check process. |
156 | | int do_mvcc_meta_key_check(); |
157 | | |
158 | | // Return 0 if all Table Stream mappings and offsets are consistent. |
159 | | // Return 1 if an inconsistent mapping or offset is identified. |
160 | | // Return negative if a temporary error occurred during the check process. |
161 | | int do_table_stream_check(); |
162 | | |
163 | | // Return 0 if success. |
164 | | // Return 1 if packed file metadata leak or loss is identified. |
165 | | // Return negative if a temporary error occurred during the check process. |
166 | | int do_packed_file_check(); |
167 | | |
168 | | StorageVaultAccessor* get_accessor(const std::string& id); |
169 | | |
170 | 0 | ResourceManager* resource_mgr() const { return resource_mgr_.get(); } |
171 | | |
172 | | void get_all_accessor(std::vector<StorageVaultAccessor*>* accessors); |
173 | | |
174 | 0 | std::string_view instance_id() const { return instance_id_; } |
175 | | |
176 | 0 | void TEST_add_accessor(std::string_view id, std::shared_ptr<StorageVaultAccessor> accessor) { |
177 | 0 | accessor_map_.insert({std::string(id), std::move(accessor)}); |
178 | 0 | } |
179 | | |
180 | | // If there are multiple buckets, return the minimum lifecycle; if there are no buckets (i.e. |
181 | | // all accessors are HdfsAccessor), return INT64_MAX. |
182 | | // Return 0 if success, otherwise error |
183 | | int get_bucket_lifecycle(int64_t* lifecycle_days); |
184 | 0 | void stop() { stopped_.store(true, std::memory_order_release); } |
185 | 6.02k | bool stopped() const { return stopped_.load(std::memory_order_acquire); } |
186 | | |
187 | | private: |
188 | | struct RowsetIndexesFormatV1 { |
189 | | std::string rowset_id; |
190 | | std::unordered_set<int64_t> segment_ids; |
191 | | std::unordered_set<std::string> index_ids; |
192 | | }; |
193 | | |
194 | | struct RowsetIndexesFormatV2 { |
195 | | std::string rowset_id; |
196 | | std::unordered_set<int64_t> segment_ids; |
197 | | }; |
198 | | |
199 | | private: |
200 | | // returns 0 for success otherwise error |
201 | | int init_obj_store_accessors(const InstanceInfoPB& instance); |
202 | | |
203 | | // returns 0 for success otherwise error |
204 | | int init_storage_vault_accessors(const InstanceInfoPB& instance); |
205 | | |
206 | | int traverse_mow_tablet(const std::function<int(int64_t, bool)>& check_func); |
207 | | int traverse_rowset_delete_bitmaps( |
208 | | int64_t tablet_id, std::string rowset_id, |
209 | | const std::function<int(int64_t, std::string_view, int64_t, int64_t)>& callback); |
210 | | int collect_tablet_rowsets( |
211 | | int64_t tablet_id, |
212 | | const std::function<void(const doris::RowsetMetaCloudPB&)>& collect_cb); |
213 | | int collect_unexpired_job_tmp_rowsets( |
214 | | std::unordered_map<int64_t, std::unordered_set<std::string>>& tmp_rowsets); |
215 | | int get_pending_delete_bitmap_keys(int64_t tablet_id, |
216 | | std::unordered_set<std::string>& pending_delete_bitmaps); |
217 | | int check_delete_bitmap_storage_optimize_v2(int64_t tablet_id, bool has_sequence_col, |
218 | | int64_t& abnormal_rowsets_num); |
219 | | |
220 | | int check_inverted_index_file_storage_format_v1(int64_t tablet_id, const std::string& file_path, |
221 | | const std::string& rowset_info, |
222 | | RowsetIndexesFormatV1& rowset_index_cache_v1); |
223 | | |
224 | | int check_inverted_index_file_storage_format_v2(int64_t tablet_id, const std::string& file_path, |
225 | | const std::string& rowset_info, |
226 | | RowsetIndexesFormatV2& rowset_index_cache_v2); |
227 | | |
228 | | // Return 0 if success. |
229 | | // Return 1 if key loss is abnormal. |
230 | | // Return negative if a temporary error occurred during the check process. |
231 | | int check_stats_tablet_key(std::string_view key, std::string_view value); |
232 | | |
233 | | // Return 0 if success. |
234 | | // Return 1 if key loss is identified. |
235 | | // Return negative if a temporary error occurred during the check process. |
236 | | int check_stats_tablet_key_exists(std::string_view key, std::string_view value); |
237 | | |
238 | | // Return 0 if success. |
239 | | // Return 1 if key leak is identified. |
240 | | // Return negative if a temporary error occurred during the check process. |
241 | | int check_stats_tablet_key_leaked(std::string_view key, std::string_view value); |
242 | | int check_txn_info_key(std::string_view key, std::string_view value); |
243 | | |
244 | | int check_txn_label_key(std::string_view key, std::string_view value); |
245 | | |
246 | | int check_txn_index_key(std::string_view key, std::string_view value); |
247 | | |
248 | | int check_txn_running_key(std::string_view key, std::string_view value); |
249 | | |
250 | | // Only check whether the meta rowset key is leak |
251 | | // in do_inverted_check() function, check whether the key is lost by comparing data file with key |
252 | | // Return 0 if success. |
253 | | // Return 1 if meta rowset key leak is identified. |
254 | | // Return negative if a temporary error occurred during the check process. |
255 | | int check_meta_rowset_key(std::string_view key, std::string_view value); |
256 | | |
257 | | // if TxnInfoKey's finish time > current time, it should not find tmp rowset |
258 | | // Return 0 if success. |
259 | | // Return 1 if meta tmp rowset key is abnormal. |
260 | | // Return negative if a temporary error occurred during the check process. |
261 | | int check_meta_tmp_rowset_key(std::string_view key, std::string_view value); |
262 | | |
263 | | /** |
264 | | * It is used to scan the key in the range from start_key to end_key |
265 | | * and then perform handle operations on each group of kv |
266 | | * |
267 | | * @param start_key Range begining. Note that this function will modify the `start_key` |
268 | | * @param end_key Range ending |
269 | | * @param handle_kv Operations on kv |
270 | | * @return code int 0 for success to scan and hanle, 1 for success to scan but handle abnormally, -1 for failed to handle |
271 | | */ |
272 | | int scan_and_handle_kv(std::string& start_key, const std::string& end_key, |
273 | | std::function<int(std::string_view, std::string_view)> handle_kv); |
274 | | |
275 | | std::atomic_bool stopped_ {false}; |
276 | | std::shared_ptr<TxnKv> txn_kv_; |
277 | | std::string instance_id_; |
278 | | // id -> accessor |
279 | | std::unordered_map<std::string, std::shared_ptr<StorageVaultAccessor>> accessor_map_; |
280 | | std::shared_ptr<SnapshotManager> snapshot_manager_; |
281 | | std::shared_ptr<ResourceManager> resource_mgr_; |
282 | | bool table_stream_versioned_write_ {false}; |
283 | | }; |
284 | | |
285 | | } // namespace doris::cloud |