作者簡介範振（花名辰繁），阿裡雲開源大資料-OLAP 方向負責人。
内容架構
- 背景
- 架構圖
- 核心流程圖梳理
- 内部代碼流程梳理
- Nuraft 關鍵配置排坑
- 結論
- 關于我們
- Reference
注：以下代碼分析版本為開源版本 ClickHouse v21.8.10.19-lts。類圖、順序圖未嚴格按照 UML 規範；為友善表意，函數名、函數參數等未嚴格按照原版代碼。
HouseKeeper Vs Zookeeper
- Zookeeper java 開發，有 JVM 痛點，執行效率不如 C++；Znode 數量太多容易出現性能問題，Full GC 比較多。
- Zookeeper 運維複雜，需要獨立部署元件，之前出問題比較多。HouseKeeper 部署形态比較多，可以 standalone 模式和內建模式。
- Zookeeper ZXID overflow 問題，HouseKeeper 沒有該問題。
- HouseKeeper 讀寫性能均有提升，支援讀寫線性一緻性，關于一緻性的級别參見 https://xzhu0027.gitbook.io/blog/misc/index/consistency-models-in-distributed-system 。
- HouseKeeper 代碼與 CK 統一，自主閉環可控。未來可擴充能力強，可以基于此做 MetaServer 的設計開發。主流的的 MetaServer 基本都是 Raft+rocksDB 的組合，可以借助該 codebase 進行開發。
Zookeeper Client
- Zookeeper Client 完全不需要修改，HouseKeeper 完全适配 Zookeeper 的協定。
- Zookeeper Client 由 CK 自己開發，放棄使用 libZookeeper（是一個bad smell代碼庫），CK 自己從 TCP 層進行封裝遵循 Zookeeper Protocol。
- 3種部署模式，推薦第一種 standalone 方式，可以選擇小機型 SSD 磁盤，最大程度發揮 Keeper 的性能。
類圖關系
- 入口 main 函數，主要做2件事：
-
- 初始化 Poco::Net::TCPServer，定義處理請求的 KeeperTCPHandler。
- 執行個體化 keeper_storage_dispatcher，并且調用 KeeperStorageDispatcher->initialize()。該函數主要作用是以下幾個：
-
-
- 執行個體化類圖中的幾個 Threads，以及相關的 ThreadSafeQueue，保證不同線程間同步資料。
- 執行個體化 KeeperServer 對象，該對象是核心資料結構，是整個 Raft 的最重要部分。KeeperServer 主要由 state_machine，state_manager，raft_instance，log_store（間接）組合成，他們分别繼承了 nuraft 庫中的父類。一般來說，所有 raft based 應用均應該實作這幾個類。
- 調用 KeeperServer::startup()，主要是初始化 state_machine，state_manager。啟動過程中會調用 state_machine->init(), state_manager->loadLogStore(...)，分别進行 snapshot 和 log 的加載。從最新的 raft snapshot 中恢複到最新送出的 latest_log_index，并形成記憶體資料結構（最關鍵是 Container 資料結構，即KeeperStorage::SnapshotableHashTable），然後再繼續加載 raft log 檔案中的每一條記錄至 logs (即資料結構 std::unordered_map)，這兩個粗體的唯二的資料結構，是整個 HouseKeeper 的核心，也是記憶體大戶，後邊會提及。
-
- KeeperTCPHandler 主循環是讀取 socket 請求，将請求 dispatcher->putRequest(req) 交給 requests_queue，然後通過 responses.tryPop(res) 從中讀到 response，最終寫 socket 将 response 傳回給用戶端。主要經曆以下幾個步驟：
-
- 确認整個叢集是否有 leader，如果有，sendHandshake。注意：HouseKeeper利用了 naraft 的 auto_forwarding 選項，是以如果接受請求的是非 leader，會承擔 proxy 的作用，将請求 forward 到 leader，讀寫請求都會經過 proxy。
- 獲得請求的 session_id。新來的 connection 擷取 session_id 的過程是服務端 keeper_dispatcher->internal_session_id_counter 自增的過程。
- keeper_dispatcher->registerSession(session_id，response_callback)，将對應的 session_id 和回調函數綁定。
- 将請求 keeper_dispatcher->putRequest(req) 交給 requests_queue。
- 通過循環 responses.tryPop(res) 從中讀到 response，最終寫 socket 将 response 傳回給用戶端。
處理請求的線程模型
- 從 TCPHandler 線程開始經曆順序圖中的不同線程調用，完成全鍊路的請求處理。
- 讀請求直接由 requests_thread 調用 state_machine->processReadRequest 處理，在該函數中，調用 storage->processRequest(...) 接口。
- 寫請求通過 raft_instance->append_entries(entries) 這個 nuraft 庫的 User API 進行 log 寫入。達成 consensus 之後，通過 nuraft 庫内部線程調用 commit 接口，執行 storage->processRequest(...) 接口。
- Nuraft 庫的 normal log replication 處理流程如下圖：
- Nuraft 庫内部維護兩個核心線程（或線程池），分别是：
-
- raft_server::append_entries_in_bg，leader 角色負責檢視 log_store 中是否有新的 entries，對 follower 進行 replication。
- raft_server::commit_in_bg，所有角色（role，follower）檢視自己的狀态機 sm_commit_index 是否落後于 leader 的 leader_commit_index，如果是，則 apply_entries 到狀态機中。
總體上nuraft實作了一個程式設計架構，需要對類圖中标紅的幾個class進行實作。
LogStore與Snapshot
- LogStore 負責持久化 logs，繼承自 nuraft::log_store，這一系列接口中比較重要的是：
-
- 寫：包括順序寫 KeeperLogStore::append(entry)，覆寫寫（截斷寫） KeeperLogStore::write_at(index, entry)，批量寫 KeeperLogStore::apply_pack(index, pack)等。
- 讀：last_entry()，entry_at(index) 等。
- 合并後清理：KeeperLogStore::compact(last_log_index)，主要會在 snapshot 之後進行調用。當 KeeperStateMachine::create_snapshot(last_log_idx) 調用時，當所有的 snapshot 将資料序列化到磁盤後，會調用 log_store_->compact(compact_upto)，其中 compact_upto = new_snp->get_last_log_idx() - params->reserved_log_items_。這是一個小坑， compact 的 compact_upto index 不是已經做過 snapshot 的最新 index，需要有一部分的保留，對應的配置是 reserved_log_items。
- ChangeLog 是 LogStore 的 pimpl，提供了所有的 LogStore/nuraft::log_store 的接口。ChangeLog 主要是由 current_wirter（log file writer）和 logs（記憶體std::unordered_map資料結構）組成。
-
- 每插入一條 log，會将 log 序列化到 file buffer 中，并且插入到記憶體 logs 中。是以可以确定，在未做 snapshot 之前，logs 占用記憶體會一直增加。
- 當做完 snaphost 之後，會把已經序列化磁盤中的 compact_upto 的 index 從記憶體 logs 中 erase 掉。是以，我們需要 trade off 兩個配置項 snapshot_distance 和 reserved_log_items。目前兩個配置項預設值都是10w條，容易大量占用記憶體，推薦值是：
-
-
- 10000
- 5000
-
- KeeperSnapshotManager 提供了一系列 ser/deser 的接口：
-
- KeeperStorageSnapshot 主要是提供了 KeeperStorage 和 file buffer 互相 ser/deser 的操作。
- 初始化時，直接通過 Snapshot 檔案進行 deser 操作，恢複到檔案訓示的 index（如 snapshot_200000.bin，訓示的 index 為200000）所對應的 KeeperStorage 資料結構。
- KeeperStateMachine::create_snapshot 時，根據提供的 snapshot 中繼資料（index，term等），執行 ser 操作，将 KeeperStorage 資料結構序列化到磁盤。
- Nuraft 庫中提供的 snapshot transmission：當新加入的 follower 節點或者 follower 節點的日志落後很多（已經落後于最新一次 log compaction upto_index），leader 會主動發起 InstallSnapshot 流程，如下圖：
- Nuraft 庫針對 InstallSnapshot 流程提供了幾個接口。KeeperStateMachine 對此進行了簡單的實作：
-
- read_logical_snp_obj(...)，leader 直接将記憶體中最新的快照 latest_snapshot_buf 發送。
- save_logical_snp_obj(...)，follower 接收并序列化落盤，更新自身的 latest_snapshot_buf。
- apply_snapshot(...)，将最新的快照 latest_snapshot_buf，生成最新版本的 storage。
KeeperStorage
這個類用來模拟與 Zookeeper 對等的功能。
- 最核心的資料結構是 Zookeeper 的 Znode 存儲：
-
- using Container = SnapshotableHashTable，由 std::unordered_map 和 std::list 組合來實作一種無鎖資料結構。key 為 Zookeeper path，value 為 Zookeeper Znode（包括存儲 Znode 的 stat 中繼資料），Node 定義為：
struct Node
{
String data;
uint64_t acl_id = 0; /// 0 -- no ACL by default
bool is_sequental = false;
Coordination::Stat stat{};
int32_t seq_num = 0;
ChildrenSet children{};
}; -
- SnapshotableHashTable 結構中的 map 總是儲存最新的資料結構，用來滿足讀需求。list 提供兩段資料結構，保障新插入的資料不影響正在做 snapshot 的資料。實作很簡單，具體見： https://github.com/ClickHouse/ClickHouse/blob/v21.8.12.29-lts/src/Coordination/SnapshotableHashTable.h
- 提供了 ephemerals，sessions_and_watchers，session_and_timeout，acl_map，watches 等資料結構，實作都很簡單，就不一一介紹了。
- 所有的 Request 都實作自 KeeperStorageRequest 父類，包括下圖的所有子類，每一個 Request 實作了純虛函數，用來對 KeeperStorage 的記憶體資料結構進行操作。
virtual std::pair<Coordination::ZooKeeperResponsePtr, Undo> process(KeeperStorage & storage, int64_t zxid, int64_t session_id) const = 0; - 阿裡雲 EMR ECS 機器對應的作業系統版本比較老（新版本已經解決），對于 ipv6 支援不好，server 啟動不了。workaround 方法是先将 nuraft 庫 hard coding 的 tcp port 改成 ipv4。
- 做5輪 zookeeper 壓測，發現記憶體一直上漲，現象接近記憶體洩露。結論是：不是記憶體洩露，需要調整參數，使 logs 記憶體資料結構不占用過多記憶體。
-
- 每一輪先建立500w個 Znode，每個 Znode 資料是256，再删除500w Znode。具體過程是：利用 ZookeeperClient 的 multi 模式，每一輪發起5000次請求，每個請求 transaction 建立1000個 Znode，達到500w個 Znode 後，再發起5000次請求，每個請求删除1000個 Znode，這樣保證每一輪所有的 Znode 全部删除。這樣即每一輪插入10000條 logEntry。
- 過程中發現每一輪記憶體都會上漲，經過5輪之後記憶體上漲到20G以上，懷疑是記憶體洩露。
- 加入代碼 profile 列印 showStatus 之後，發現每一輪 ChangeLog::logs 資料結構一直增長，而 KeeperStorage::Container 資料結構會随着 Znode 數量而周期變化，最終回歸0。結論是：由于 snapshot_distance 預設配置是10w條，是以，一直沒有發生 create_snapshot，也即沒有發生 compact logs，ChangeLog::logs 記憶體占用會越來越多。是以建議配置為：
- 通過配置 auto_forwarding，可以讓 leader 把請求轉發給 follower，對 ZookeeperClient 是透明實作。但是這個配置 nuraft 不推薦，後續版本應該會改善該做法。
- 去掉 Zookeeper 依賴會讓 ClickHouse 不再依賴外部元件，無論從穩定性和性能都向前邁進了一大步，為逐漸走向雲原生化提供了前提。
- 基于該 codebase，後續将會逐漸衍生出基于 Raft 的 MetaServer，為支援存算分離、支援分布式 Join 的 MPP 架構等方向提供了前提。
計算平台開源大資料團隊緻力于開源引擎的核心研發工作，OLAP 方向包括 ClickHouse，Starrocks，Trino(PrestoDB) 等。
期待熱愛核心深度開發的小夥伴們加入，郵箱請聯系[email protected]。
對開源大資料 OLAP 引擎感興趣的小夥伴，歡迎釘釘掃描文末二維碼加入交流群一起交流探讨！
[1]
https://github.com/eBay/NuRaft[2]
[3]
https://zhuanlan.zhihu.com/p/425072031我們會在釘釘群定期推送精彩文章，邀請技術大牛直播分享。
歡迎釘釘掃碼加入産品交流群一起參與讨論~