簡介：ClickHouse 社區在21.8版本中引入了 ClickHouse Keeper。ClickHouse Keeper 是完全兼容 Zookeeper 協議的分布式協調服務。本文對開源版本 ClickHouse v21.8.10.19-lts 源碼進行了解析。

作者簡介：范振（花名辰繁），阿里云開源大數據-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 讀寫性能均有提升，支持讀寫線性一致性，關于一致性的級別參見Consistency Models in Distributed System - Random Notes。
• 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 defaultbool 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;

Nuraft 關鍵配置排坑

• 阿里云 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 內存占用會越來越多。所以建議配置為：

• 10000
• 5000
• 通過配置 auto_forwarding，可以讓 leader 把請求轉發給 follower，對 ZookeeperClient 是透明實現。但是這個配置 nuraft 不推薦，后續版本應該會改善該做法。

結論

• 去掉 Zookeeper 依賴會讓 ClickHouse 不再依賴外部組件，無論從穩定性和性能都向前邁進了一大步，為逐漸走向云原生化提供了前提。
• 基于該 codebase，后續將會逐步衍生出基于 Raft 的 MetaServer，為支持存算分離、支持分布式 Join 的 MPP 架構等方向提供了前提。

關于我們

計算平臺開源大數據團隊致力于開源引擎的內核研發工作，OLAP 方向包括 ClickHouse，Starrocks，Trino(PrestoDB) 等。

原文鏈接
本文為阿里云原創內容，未經允許不得轉載。?

總結

以上是生活随笔為你收集整理的ClickHouse Keeper 源码解析的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。