簡(jiǎn)介：ClickHouse 社區(qū)在21.8版本中引入了 ClickHouse Keeper。ClickHouse Keeper 是完全兼容 Zookeeper 協(xié)議的分布式協(xié)調(diào)服務(wù)。本文對(duì)開(kāi)源版本 ClickHouse v21.8.10.19-lts 源碼進(jìn)行了解析。

作者簡(jiǎn)介：范振（花名辰繁），阿里云開(kāi)源大數(shù)據(jù)-OLAP 方向負(fù)責(zé)人。

內(nèi)容框架

• 背景
• 架構(gòu)圖
• 核心流程圖梳理
• 內(nèi)部代碼流程梳理
• Nuraft 關(guān)鍵配置排坑

• 結(jié)論
• 關(guān)于我們
• Reference

背景

注：以下代碼分析版本為開(kāi)源版本 ClickHouse v21.8.10.19-lts。類圖、順序圖未嚴(yán)格按照 UML 規(guī)范；為方便表意，函數(shù)名、函數(shù)參數(shù)等未嚴(yán)格按照原版代碼。

HouseKeeper Vs Zookeeper

• Zookeeper java 開(kāi)發(fā)，有 JVM 痛點(diǎn)，執(zhí)行效率不如 C++；Znode 數(shù)量太多容易出現(xiàn)性能問(wèn)題，Full GC 比較多。
• Zookeeper 運(yùn)維復(fù)雜，需要獨(dú)立部署組件，之前出問(wèn)題比較多。HouseKeeper 部署形態(tài)比較多，可以 standalone 模式和集成模式。
• Zookeeper ZXID overflow 問(wèn)題，HouseKeeper 沒(méi)有該問(wèn)題。
• HouseKeeper 讀寫性能均有提升，支持讀寫線性一致性，關(guān)于一致性的級(jí)別參見(jiàn)Consistency Models in Distributed System - Random Notes。
• HouseKeeper 代碼與 CK 統(tǒng)一，自主閉環(huán)可控。未來(lái)可擴(kuò)展能力強(qiáng)，可以基于此做 MetaServer 的設(shè)計(jì)開(kāi)發(fā)。主流的的 MetaServer 基本都是 Raft+rocksDB 的組合，可以借助該 codebase 進(jìn)行開(kāi)發(fā)。

Zookeeper Client

• Zookeeper Client 完全不需要修改，HouseKeeper 完全適配 Zookeeper 的協(xié)議。
• Zookeeper Client 由 CK 自己開(kāi)發(fā)，放棄使用 libZookeeper（是一個(gè)bad smell代碼庫(kù)），CK 自己從 TCP 層進(jìn)行封裝遵循 Zookeeper Protocol。

架構(gòu)圖

• 3種部署模式，推薦第一種 standalone 方式，可以選擇小機(jī)型 SSD 磁盤，最大程度發(fā)揮 Keeper 的性能。

核心流程圖梳理

類圖關(guān)系

• 入口 main 函數(shù)，主要做2件事：

• 初始化 Poco::Net::TCPServer，定義處理請(qǐng)求的 KeeperTCPHandler。
• 實(shí)例化 keeper_storage_dispatcher，并且調(diào)用 KeeperStorageDispatcher->initialize()。該函數(shù)主要作用是以下幾個(gè)：

• 實(shí)例化類圖中的幾個(gè) Threads，以及相關(guān)的 ThreadSafeQueue，保證不同線程間同步數(shù)據(jù)。
• 實(shí)例化 KeeperServer 對(duì)象，該對(duì)象是核心數(shù)據(jù)結(jié)構(gòu)，是整個(gè) Raft 的最重要部分。KeeperServer 主要由 state_machine，state_manager，raft_instance，log_store（間接）組合成，他們分別繼承了 nuraft 庫(kù)中的父類。一般來(lái)說(shuō)，所有 raft based 應(yīng)用均應(yīng)該實(shí)現(xiàn)這幾個(gè)類。
• 調(diào)用 KeeperServer::startup()，主要是初始化 state_machine，state_manager。啟動(dòng)過(guò)程中會(huì)調(diào)用 state_machine->init(), state_manager->loadLogStore(...)，分別進(jìn)行 snapshot 和 log 的加載。從最新的 raft snapshot 中恢復(fù)到最新提交的 latest_log_index，并形成內(nèi)存數(shù)據(jù)結(jié)構(gòu)（最關(guān)鍵是 Container 數(shù)據(jù)結(jié)構(gòu)，即KeeperStorage::SnapshotableHashTable），然后再繼續(xù)加載 raft log 文件中的每一條記錄至 logs (即數(shù)據(jù)結(jié)構(gòu) std::unordered_map)，這兩個(gè)粗體的唯二的數(shù)據(jù)結(jié)構(gòu)，是整個(gè) HouseKeeper 的核心，也是內(nèi)存大戶，后邊會(huì)提及。
• KeeperTCPHandler 主循環(huán)是讀取 socket 請(qǐng)求，將請(qǐng)求 dispatcher->putRequest(req) 交給 requests_queue，然后通過(guò) responses.tryPop(res) 從中讀到 response，最終寫 socket 將 response 返回給客戶端。主要經(jīng)歷以下幾個(gè)步驟：
• 確認(rèn)整個(gè)集群是否有 leader，如果有，sendHandshake。注意：HouseKeeper利用了 naraft 的 auto_forwarding 選項(xiàng)，所以如果接受請(qǐng)求的是非 leader，會(huì)承擔(dān) proxy 的作用，將請(qǐng)求 forward 到 leader，讀寫請(qǐng)求都會(huì)經(jīng)過(guò) proxy。
• 獲得請(qǐng)求的 session_id。新來(lái)的 connection 獲取 session_id 的過(guò)程是服務(wù)端 keeper_dispatcher->internal_session_id_counter 自增的過(guò)程。
• keeper_dispatcher->registerSession(session_id，response_callback)，將對(duì)應(yīng)的 session_id 和回調(diào)函數(shù)綁定。
• 將請(qǐng)求 keeper_dispatcher->putRequest(req) 交給 requests_queue。
• 通過(guò)循環(huán) responses.tryPop(res) 從中讀到 response，最終寫 socket 將 response 返回給客戶端。

處理請(qǐng)求的線程模型

• 從 TCPHandler 線程開(kāi)始經(jīng)歷順序圖中的不同線程調(diào)用，完成全鏈路的請(qǐng)求處理。
• 讀請(qǐng)求直接由 requests_thread 調(diào)用 state_machine->processReadRequest 處理，在該函數(shù)中，調(diào)用 storage->processRequest(...) 接口。
• 寫請(qǐng)求通過(guò) raft_instance->append_entries(entries) 這個(gè) nuraft 庫(kù)的 User API 進(jìn)行 log 寫入。達(dá)成 consensus 之后，通過(guò) nuraft 庫(kù)內(nèi)部線程調(diào)用 commit 接口，執(zhí)行 storage->processRequest(...) 接口。
• Nuraft 庫(kù)的 normal log replication 處理流程如下圖：

• Nuraft 庫(kù)內(nèi)部維護(hù)兩個(gè)核心線程（或線程池），分別是：

• raft_server::append_entries_in_bg，leader 角色負(fù)責(zé)查看 log_store 中是否有新的 entries，對(duì) follower 進(jìn)行 replication。
• raft_server::commit_in_bg，所有角色（role，follower）查看自己的狀態(tài)機(jī) sm_commit_index 是否落后于 leader 的 leader_commit_index，如果是，則 apply_entries 到狀態(tài)機(jī)中。

內(nèi)部代碼流程梳理

總體上nuraft實(shí)現(xiàn)了一個(gè)編程框架，需要對(duì)類圖中標(biāo)紅的幾個(gè)class進(jìn)行實(shí)現(xiàn)。

LogStore與Snapshot

• LogStore 負(fù)責(zé)持久化 logs，繼承自 nuraft::log_store，這一系列接口中比較重要的是：

• 寫：包括順序?qū)?KeeperLogStore::append(entry)，覆蓋寫（截?cái)鄬?#xff09; KeeperLogStore::write_at(index, entry)，批量寫 KeeperLogStore::apply_pack(index, pack)等。
• 讀：last_entry()，entry_at(index) 等。
• 合并后清理：KeeperLogStore::compact(last_log_index)，主要會(huì)在 snapshot 之后進(jìn)行調(diào)用。當(dāng) KeeperStateMachine::create_snapshot(last_log_idx) 調(diào)用時(shí)，當(dāng)所有的 snapshot 將數(shù)據(jù)序列化到磁盤后，會(huì)調(diào)用 log_store_->compact(compact_upto)，其中 compact_upto = new_snp->get_last_log_idx() - params->reserved_log_items_。這是一個(gè)小坑， compact 的 compact_upto index 不是已經(jīng)做過(guò) snapshot 的最新 index，需要有一部分的保留，對(duì)應(yīng)的配置是 reserved_log_items。
• ChangeLog 是 LogStore 的 pimpl，提供了所有的 LogStore/nuraft::log_store 的接口。ChangeLog 主要是由 current_wirter（log file writer）和 logs（內(nèi)存std::unordered_map數(shù)據(jù)結(jié)構(gòu)）組成。
• 每插入一條 log，會(huì)將 log 序列化到 file buffer 中，并且插入到內(nèi)存 logs 中。所以可以確定，在未做 snapshot 之前，logs 占用內(nèi)存會(huì)一直增加。
• 當(dāng)做完 snaphost 之后，會(huì)把已經(jīng)序列化磁盤中的 compact_upto 的 index 從內(nèi)存 logs 中 erase 掉。所以，我們需要 trade off 兩個(gè)配置項(xiàng) snapshot_distance 和 reserved_log_items。目前兩個(gè)配置項(xiàng)缺省值都是10w條，容易大量占用內(nèi)存，推薦值是：

• 10000
• 5000
• KeeperSnapshotManager 提供了一系列 ser/deser 的接口：
• KeeperStorageSnapshot 主要是提供了 KeeperStorage 和 file buffer 互相 ser/deser 的操作。
• 初始化時(shí)，直接通過(guò) Snapshot 文件進(jìn)行 deser 操作，恢復(fù)到文件指示的 index（如 snapshot_200000.bin，指示的 index 為200000）所對(duì)應(yīng)的 KeeperStorage 數(shù)據(jù)結(jié)構(gòu)。
• KeeperStateMachine::create_snapshot 時(shí)，根據(jù)提供的 snapshot 元數(shù)據(jù)（index，term等），執(zhí)行 ser 操作，將 KeeperStorage 數(shù)據(jù)結(jié)構(gòu)序列化到磁盤。
• Nuraft 庫(kù)中提供的 snapshot transmission：當(dāng)新加入的 follower 節(jié)點(diǎn)或者 follower 節(jié)點(diǎn)的日志落后很多（已經(jīng)落后于最新一次 log compaction upto_index），leader 會(huì)主動(dòng)發(fā)起 InstallSnapshot 流程，如下圖：

• Nuraft 庫(kù)針對(duì) InstallSnapshot 流程提供了幾個(gè)接口。KeeperStateMachine 對(duì)此進(jìn)行了簡(jiǎn)單的實(shí)現(xiàn)：

• read_logical_snp_obj(...)，leader 直接將內(nèi)存中最新的快照 latest_snapshot_buf 發(fā)送。
• save_logical_snp_obj(...)，follower 接收并序列化落盤，更新自身的 latest_snapshot_buf。
• apply_snapshot(...)，將最新的快照 latest_snapshot_buf，生成最新版本的 storage。

KeeperStorage

這個(gè)類用來(lái)模擬與 Zookeeper 對(duì)等的功能。

• 最核心的數(shù)據(jù)結(jié)構(gòu)是 Zookeeper 的 Znode 存儲(chǔ)：

• using Container = SnapshotableHashTable，由 std::unordered_map 和 std::list 組合來(lái)實(shí)現(xiàn)一種無(wú)鎖數(shù)據(jù)結(jié)構(gòu)。key 為 Zookeeper path，value 為 Zookeeper Znode（包括存儲(chǔ) Znode 的 stat 元數(shù)據(jù)），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 結(jié)構(gòu)中的 map 總是保存最新的數(shù)據(jù)結(jié)構(gòu)，用來(lái)滿足讀需求。list 提供兩段數(shù)據(jù)結(jié)構(gòu)，保障新插入的數(shù)據(jù)不影響正在做 snapshot 的數(shù)據(jù)。實(shí)現(xiàn)很簡(jiǎn)單，具體見(jiàn)：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 等數(shù)據(jù)結(jié)構(gòu)，實(shí)現(xiàn)都很簡(jiǎn)單，就不一一介紹了。
• 所有的 Request 都實(shí)現(xiàn)自 KeeperStorageRequest 父類，包括下圖的所有子類，每一個(gè) Request 實(shí)現(xiàn)了純虛函數(shù)，用來(lái)對(duì) KeeperStorage 的內(nèi)存數(shù)據(jù)結(jié)構(gòu)進(jìn)行操作。

virtual std::pair<Coordination::ZooKeeperResponsePtr, Undo> process(KeeperStorage & storage, int64_t zxid, int64_t session_id) const = 0;

Nuraft 關(guān)鍵配置排坑

• 阿里云 EMR ECS 機(jī)器對(duì)應(yīng)的操作系統(tǒng)版本比較老（新版本已經(jīng)解決），對(duì)于 ipv6 支持不好，server 啟動(dòng)不了。workaround 方法是先將 nuraft 庫(kù) hard coding 的 tcp port 改成 ipv4。
• 做5輪 zookeeper 壓測(cè)，發(fā)現(xiàn)內(nèi)存一直上漲，現(xiàn)象接近內(nèi)存泄露。結(jié)論是：不是內(nèi)存泄露，需要調(diào)整參數(shù)，使 logs 內(nèi)存數(shù)據(jù)結(jié)構(gòu)不占用過(guò)多內(nèi)存。
• 每一輪先創(chuàng)建500w個(gè) Znode，每個(gè) Znode 數(shù)據(jù)是256，再刪除500w Znode。具體過(guò)程是：利用 ZookeeperClient 的 multi 模式，每一輪發(fā)起5000次請(qǐng)求，每個(gè)請(qǐng)求 transaction 創(chuàng)建1000個(gè) Znode，達(dá)到500w個(gè) Znode 后，再發(fā)起5000次請(qǐng)求，每個(gè)請(qǐng)求刪除1000個(gè) Znode，這樣保證每一輪所有的 Znode 全部刪除。這樣即每一輪插入10000條 logEntry。
• 過(guò)程中發(fā)現(xiàn)每一輪內(nèi)存都會(huì)上漲，經(jīng)過(guò)5輪之后內(nèi)存上漲到20G以上，懷疑是內(nèi)存泄露。
• 加入代碼 profile 打印 showStatus 之后，發(fā)現(xiàn)每一輪 ChangeLog::logs 數(shù)據(jù)結(jié)構(gòu)一直增長(zhǎng)，而 KeeperStorage::Container 數(shù)據(jù)結(jié)構(gòu)會(huì)隨著 Znode 數(shù)量而周期變化，最終回歸0。結(jié)論是：由于 snapshot_distance 默認(rèn)配置是10w條，所以，一直沒(méi)有發(fā)生 create_snapshot，也即沒(méi)有發(fā)生 compact logs，ChangeLog::logs 內(nèi)存占用會(huì)越來(lái)越多。所以建議配置為：

• 10000
• 5000
• 通過(guò)配置 auto_forwarding，可以讓 leader 把請(qǐng)求轉(zhuǎn)發(fā)給 follower，對(duì) ZookeeperClient 是透明實(shí)現(xiàn)。但是這個(gè)配置 nuraft 不推薦，后續(xù)版本應(yīng)該會(huì)改善該做法。

結(jié)論

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

關(guān)于我們

計(jì)算平臺(tái)開(kāi)源大數(shù)據(jù)團(tuán)隊(duì)致力于開(kāi)源引擎的內(nèi)核研發(fā)工作，OLAP 方向包括 ClickHouse，Starrocks，Trino(PrestoDB) 等。

原文鏈接
本文為阿里云原創(chuàng)內(nèi)容，未經(jīng)允許不得轉(zhuǎn)載。?

總結(jié)

以上是生活随笔為你收集整理的ClickHouse Keeper 源码解析的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。