MongoDB集群均衡

01-29

作者:孔德雨騰訊雲資料庫工程師

mongodb 可以以單複製集的方式運行，client 直連mongod讀取數據。

單複製集的方式下，數據的水平擴展的責任推給了業務層解決（分實例，分庫分表），mongodb原生提供集群方案，該方案的簡要架構如下：

mongodb集群是一個典型的去中心化分散式集群。mongodb集群主要為用戶解決了如下問題：

元數據的一致性與高可用（Consistency + Partition Torrence）
業務數據的多備份容災(由複製集技術保證)
動態自動分片
動態自動數據均衡

下文通過介紹mongodb集群中各個組成部分，逐步深入剖析mongodb集群原理。

ConfigServer

mongodb元數據全部存放在configServer中，configServer 是由一組（至少三個）mongod實例組成的集群。

configServer 的唯一功能是提供元數據的增刪改查。和大多數元數據管理系統（etcd，zookeeper）類似，也是保證一致性與分區容錯性。本身不具備中心化的調度功能。

ConfigServer與複製集

ConfigServer的分區容錯性(P)和數據一致性(C)是複製集本身的性質。

MongoDb的讀寫一致性由WriteConcern和ReadConcern兩個參數保證。

writeConcern Write Concern - MongoDB Manual 3.2

readConcern Read Concern - MongoDB Manual 3.2

兩者組合可以得到不同的一致性等級。

指定 writeConcern:majority 可以保證寫入數據不丟失，不會因選舉新主節點而被回滾掉。

readConcern:majority + writeConcern:majority 可以保證強一致性的讀

readConcern:local + writeConcern:majority 可以保證最終一致性的讀

mongodb 對configServer全部指定writeConcern:majority 的寫入方式，因此元數據可以保證不丟失。

對configServer的讀指定了ReadPreference:PrimaryOnly的方式，在CAP中捨棄了A與P得到了元數據的強一致性讀。

Mongos

數據自動分片

對於一個讀寫操作，mongos需要知道應該將其路由到哪個複製集上，mongos通過將片鍵空間劃分為若干個區間，計算出一個操作的片鍵的所屬區間對應的複製集來實現路由。

Collection1 被劃分為4個chunk，其中

chunk1 包含（-INF，1) , chunk3 包含[20, 99) 的數據，放在shard1上。

chunk2 包含 [1,20), chunk4 包含[99, INF) 的數據，放在shard2上。

chunk 的信息存放在configServer 的mongod實例的 config.chunks 表中，格式如下：

{ n "_id" : "mydb.foo-a_"cat"", n "lastmod" : Timestamp(1000, 3), n "lastmodEpoch" : ObjectId("5078407bd58b175c5c225fdc"), n "ns" : "mydb.foo", n "min" : { "animal" : "cat" }, n "max" : { "animal" : "dog" }, n "shard" : "shard0004"n}n

值得注意的是：chunk是一個邏輯上的組織結構，並不涉及到底層的文件組織方式。

啟發式觸發chunk分裂

mongodb 默認配置下，每個chunk大小為16MB。超過該大小就需要執行chunk分裂。chunk分裂是由mongos發起的，而數據放在mongod處，因此mongos無法準確判斷每個增刪改操作後某個chunk的數據實際大小。因此mongos採用了一種啟發式的觸發分裂方式：

mongos在內存中記錄一份 chunk_id -> incr_delta 的哈希表。

對於insert和update操作，估算出incr_delta的上界(WriteOp::targetWrites), 當incr_delta超過閾值時，執行chunk分裂。

值得注意的是：

1) chunk_id->incr_delta 是維護在mongos內存里的一份數據，重啟後丟失
2) 不同mongos之間的這份數據相互獨立
3) 不帶shardkey的update 無法對 chunk_id->incr_delta 作用

因此這個啟發式的分裂方式很不精確，而除了手工以命令的方式分裂之外，這是mongos自帶的唯一的chunk分裂方式。

chunk分裂的執行過程

1) 向對應的mongod 發起splitVector 命令，獲得一個chunk的可分裂點
2) mongos 拿到這些分裂點後，向mongod發起splitChunk 命令

splitVector執行過程：

1) 計算出collection的文檔的 avgRecSize= coll.size/ coll.count
2) 計算出分裂後的chunk中，每個chunk應該有的count數， split_count = maxChunkSize / (2 * avgRecSize)
3) 線性遍歷collection 的shardkey 對應的index的 [chunk_min_index, chunk_max_index] 範圍，在遍歷過程中利用split_count 分割出若干spli

splitChunk執行過程：

1) 獲得待執行collection的分散式鎖（向configSvr 的mongod中寫入一條記錄實現）
2) 刷新（向configSvr讀取）本shard的版本號，檢查是否和命令發起者攜帶的版本號一致
3) 向configSvr中寫入分裂後的chunk信息，成功後修改本地的chunk信息與shard的版本號
4) 向configSvr中寫入變更日誌
5) 通知mongos操作完成，mongos修改自身元數據

chunk分裂的執行流程圖：

問題與思考

問題一：為何mongos在接收到splitVector的返回後，執行splitChunk 要放在mongod執行而不是mongos中呢，為何不是mongos自己執行完了splitChunk再通知mongod 修改元數據？

我們知道chunk元數據在三個地方持有，分別是configServer，mongos，mongod。如果chunk元信息由mongos更改，則其他mongos與mongod都無法第一時間獲得最新元數據。可能會發生這樣的問題，如下圖描述：

Mongos對元數據的修改還沒有被mongod與其他mongos感知，其他mongos與mongod的版本號保持一致，導致其他mongos寫入錯誤的chunk。

如果chunk元信息由mongod更改，mongod 先於所有的mongos感知到本shard的元數據被更改，由於mongos對mongod的寫入請求都會帶有版本號（以發起者mongos的POV 持有的版本號），mongod發現一個讀寫帶有的版本號低於自身版本號時就會返回 StaleShardingError，從而避免對錯誤的chunk進行讀寫。

Mongos對讀寫的路由

讀請求：

mongos將讀請求路由到對應的shard上，如果得到StaleShardingError，則刷新本地的元數據（從configServer讀取最新元數據）並重試。

寫請求：

mongos將寫請求路由到對應的shard上，如果得到StaleShardingError，並不會像讀請求一樣重試，這樣做並不合理，截至當前版本，mongos也只是列出了一個TODO(batch_write_exec.cpp:185)

185 // TODO: It may be necessary to refresh the cache if stale, or maybe justn186 // cancel and retarget the batchn

chunk遷移

chunk遷移由balancer模塊執行，balancer模塊並不是一個獨立的service，而是mongos的一個線程模塊。同一時間只有一個balancer模塊在執行，這一點是mongos在configServer中註冊分散式鎖來保證的。

balancer 對於每一個collection的chunk 分布，計算出這個collection需要進行遷移的chunk，以及每個chunk需要遷移到哪個shard上。計算的過程在BalancerPolicy 類中，比較瑣碎。

chunk遷移.Step1

MigrationManager::scheduleMigrations balancer對於每一個collection，嘗試獲得該collection的分散式鎖（向configSvr申請），如果獲得失敗，表明該collection已有正在執行的搬遷任務。這一點說明對於同一張表統一時刻只能有一個搬遷任務。如果這張表分布在不同的shard上，完全隔離的IO條件可以提高並發，不過mongos並沒有利用起來這一點。

如果獲得鎖成功，則向源shard發起moveChunk 命令

chunk遷移.Step2

mongod 執行moveChunk命令

cloneStage

1) 源mongod 根據需要遷移的chunk 的上下限構造好查詢計劃，基於分片索引的掃描查詢。並向目標mongod發起recvChunkStart 指令，讓目標chunk 開始進入數據拉取階段。

2) 源mongod對此階段的修改，將id欄位buffer在內存里(MigrationChunkClonerSourceLegacy類)，為了防止搬遷時速度過慢buffer無限制增長，buffer大小設置為500MB，在搬遷過程中key的更改量超過buffer大小會導致搬遷失敗。

3) 目標mongod 在接收到recvChunkStart命令後

a. 基於chunk的range，將本mongod上的可能臟數據清理掉
b. 向源發起_migrateClone指定，通過1）中構造好的基於分配索引的掃描查詢得到該chunk 數據的snapshot
c. 拷貝完snapshot後，向源發起_transferMods命令，將2）中維護在內存buffer中的修改
d. 源在收到_transferMods後，通過記錄的objid查詢對應的collection，將真實數據返回給目標。
e. 目標在收完_transferMods 階段的數據後，進入steady狀態，等待源接下來的命令。這裡有必要說明的是：用戶數據源源不斷的寫入，理論上_transferMods 階段會一直有新數據，但是必須要找到一個點截斷數據流，將源的數據（搬遷對應的chunk的數據）設置為不可寫，才能發起路由更改。因此這裡所說的「_transferMods階段的所有數據」只是針對於某個時間點，這個時間點過後依然會有新數據進來。
f. 源心跳檢查目標是否已經處於steady狀態，如果是，則封禁chunk的寫入，向目標發起_recvChunkCommit命令，之後源的chunk上就無修改了。
g. 目標收到_recvChunkCommit命令後，拉取源chunk上的修改並執行，執行成功後源解禁路由並清理源chunk的數據

流程圖如下：

總結

經過分析，我們發現Mongos在遷移方面有很大的待提升空間

1) 一張表同一時間只能有一個chunk在搬遷，沒有充分利用不同機器之間的IO隔離來做並發提速。

2) 搬遷時需要掃描源的數據集，一方面會與業務爭QPS，一方面會破壞(如果是Mmap引擎)熱點讀寫的working-set

3) Mongos啟發式分裂chunk的方式極不靠譜，mongos重啟後，啟發信息就丟失了，而且部分常見的寫入模式也不會記錄啟發信息

經過CMongo團隊的測試，mongos自帶的搬遷方案處理100GB的數據需要33小時。CMongo團隊分析了mongos自帶的搬遷方案的缺陷，自研了一套基於備份的搬遷方案，速度有30倍以上的提升,敬請期待！