add snapshot check and retry (#159)

Author: lgqss

include/raft/rocksdb_file_system_adaptor.h

@@ -130,6 +130,8 @@ friend class RocksdbFileSystemAdaptor;
     bool _is_meta_reader = false;
     bool _closed = true;
     size_t _num_lines = 0;
+    butil::IOPortal _last_package;
+    off_t _last_offset = 0;
 };
 
 class SstWriterAdaptor : public braft::FileAdaptor {
@@ -153,7 +155,7 @@ friend class RocksdbFileSystemAdaptor;
 
 private:
     bool finish_sst();
-    int parse_from_iobuf(const butil::IOBuf& data, std::vector<std::string>& keys, std::vector<std::string>& values);
+    int iobuf_to_sst(butil::IOBuf data);
     int64_t _region_id;
     SmartRegion _region_ptr;
     std::string _path;

include/store/region.h

@@ -735,6 +735,18 @@ friend class Backup;
     std::shared_ptr<Region> get_ptr() {
         return shared_from_this();
     }
+    uint64_t snapshot_data_size() const {
+        return _snapshot_data_size;
+    }
+    void set_snapshot_data_size(size_t size) {
+        _snapshot_data_size = size;
+    }
+    uint64_t snapshot_meta_size() const {
+        return _snapshot_meta_size;
+    }
+    void set_snapshot_meta_size(size_t size) {
+        _snapshot_meta_size = size;
+    }
     uint64_t get_approx_size() const {
         //分裂后一段时间每超过10分钟，或者超过10%的数据量diff则需要重新获取
         if (_approx_info.time_cost.get_time() > 10 * 60 * 1000 * 1000LL && 
@@ -899,6 +911,8 @@ friend class Backup;
     std::mutex          _region_lock;    
     //split后缓存分裂出去的region信息供baikaldb使用
     std::vector<pb::RegionInfo> _new_region_infos;
+    size_t _snapshot_data_size = 0;
+    size_t _snapshot_meta_size = 0;
     pb::RegionInfo      _new_region_info;
     int64_t             _region_id;
 

include/store/rpc_sender.h

@@ -25,6 +25,8 @@ class RpcSender {
                             int64_t request_version);
 
     static int64_t get_peer_applied_index(const std::string& peer, int64_t region_id);
+    static void get_peer_snapshot_size(const std::string& peer, int64_t region_id, 
+            uint64_t* data_size, uint64_t* meta_size);
     static int send_query_method(const pb::StoreReq& request, 
                 const std::string& instance, 
                 int64_t receive_region_id);

proto/store.interface.proto

@@ -121,6 +121,12 @@ message IndexRecords {
     repeated bytes  records         = 2;  
 };
 
+message RegionRaftStat {
+    optional int64 applied_index    = 1;
+    optional uint64 snapshot_data_size = 2;
+    optional uint64 snapshot_meta_size = 3;
+};
+
 message StoreRes {
     required ErrCode errcode        = 1;
     optional bytes errmsg          = 2;
@@ -145,6 +151,7 @@ message StoreRes {
     repeated int64 commit_ts     = 20; //与上面的binlog一一对应
     optional PeerStatus region_status = 21;
     optional int64 last_insert_id = 22;
+    optional RegionRaftStat region_raft_stat = 23;
 };
 message InitRegion {
     required RegionInfo region_info     = 1;
@@ -222,6 +229,7 @@ service StoreService {
     rpc add_peer(AddPeer) returns (StoreRes);
 
     //get region的applied_index, 用来分裂完成后优化选择transfer leader的对象
+    //复用，获取snapshot数据大小，用作校验
     rpc get_applied_index(GetAppliedIndex) returns (StoreRes);
 
     rpc compact_region(RegionIds) returns (StoreRes);

src/meta_server/region_manager.cpp

@@ -21,15 +21,23 @@
 #include "meta_util.h"
 #include "table_manager.h"
 #include "meta_rocksdb.h"
+#ifdef BAIDU_INTERNAL
+#include "baidu/rpc/reloadable_flags.h"
+#else
 #include "brpc/reloadable_flags.h"
+#endif
 
 namespace baikaldb {
 DECLARE_int32(concurrency_num);
 DECLARE_int64(store_heart_beat_interval_us);
 DECLARE_int32(store_dead_interval_times);
 DECLARE_int32(region_faulty_interval_times);
 DEFINE_int32(balance_add_peer_num, 10, "add peer num each time, default(10)");
+#ifdef BAIDU_INTERNAL
+BAIDU_RPC_VALIDATE_GFLAG(balance_add_peer_num, brpc::PositiveInteger);
+#else
 BRPC_VALIDATE_GFLAG(balance_add_peer_num, brpc::PositiveInteger);
+#endif
 
 //增加或者更新region信息
 //如果是增加，则需要更新表信息, 只有leader的上报会调用该接口