Coverage Report

Created: 2025-09-30 22:51

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