Lease(租約): 其實就是一個定時器。首先申請一個TTL=N的lease(定時器),然後創建key的時候傳入該lease,那麼就實現了一個定時的key。 在程式中可以定時為該lease續約,也就是不斷重覆的重置TTL=N。當lease過期的時候,其所關聯的所有key都會自動刪除。 Raft協議: ...
Lease(租約):
其實就是一個定時器。首先申請一個TTL=N的lease(定時器),然後創建key的時候傳入該lease,那麼就實現了一個定時的key。
在程式中可以定時為該lease續約,也就是不斷重覆的重置TTL=N。當lease過期的時候,其所關聯的所有key都會自動刪除。
Raft協議:
etcd基於Raft協議實現數據同步(K-V數據),集群由多個節點組成。
Raft協議理解起來相比Paxos並沒有簡單到哪裡,因為都很難理解,所以我簡單描述一下:
- 每次寫入都是在一個事務(tx)中完成的。
- 一個事務(tx)可以包含若幹put(寫入K-V鍵值對)操作。
- etcd集群有一個leader,寫入請求都會提交給它。
- leader先將數據保存成日誌形式,並定時的將日誌發往其他節點保存。
- 當超過1/2節點成功保存了日誌,則leader會將tx最終提交(也是一條日誌)。
- 一旦leader提交tx,則會在下一次心跳時將提交記錄發送給其他節點,其他節點也會提交。
- leader宕機後,剩餘節點協商找到擁有最大已提交tx ID(必須是被超過半數的節點已提交的)的節點作為新leader。
這裡最重要的是知道:
- Raft中,後提交的事務ID>先提交的事務ID,每個事務ID都是唯一的。
- 無論客戶端是在哪個etcd節點提交,整個集群對外表現出數據視圖最終都是一樣的。
K-V存儲
etcd根本上來說是一個K-V存儲,它在記憶體中維護了一個btree(B樹),就和MySQL的索引一樣,它是有序的。
在這個btree中,key就是用戶傳入的原始key,而value並不是用戶傳入的value,具體是什麼後面再說,整個k-v存儲大概就是這樣:
type treeIndex struct { sync.RWMutex tree *btree.BTree }
當存儲大量的K-V時,因為用戶的value一般比較大,全部放在記憶體btree里記憶體耗費過大,所以etcd將用戶value保存在磁碟中。
簡單的說,etcd是純記憶體索引,數據在磁碟持久化,這個模型整體來說並不複雜。在磁碟上,etcd使用了一個叫做bbolt的純K-V存儲引擎(可以理解為leveldb),那麼bbolt的key和value分別是什麼呢?
MVCC多版本
如果僅僅維護一個K-V模型,那麼連續的更新只能保存最後一個value,歷史版本無從追溯,而多版本可以解決這個問題,怎麼維護多個版本呢?下麵是幾條預備知識:
- 每個tx事務有唯一事務ID,在etcd中叫做main ID,全局遞增不重覆。
- 一個tx可以包含多個修改操作(put和delete),每一個操作叫做一個revision(修訂),共用同一個main ID。
- 一個tx內連續的多個修改操作會被從0遞增編號,這個編號叫做sub ID。
- 每個revision由(main ID,sub ID)唯一標識。
下麵是revision的定義:
// A revision indicates modification of the key-value space. // The set of changes that share same main revision changes the key-value space atomically. type revision struct { // main is the main revision of a set of changes that happen atomically. main int64 // sub is the the sub revision of a change in a set of changes that happen // atomically. Each change has different increasing sub revision in that // set. sub int64 }
在記憶體索引中,每個用戶原始key會關聯一個key_index結構,裡面維護了多版本信息:
type keyIndex struct { key []byte modified revision // the main rev of the last modification generations []generation }
key欄位就是用戶的原始key,modified欄位記錄這個key的最後一次修改對應的revision信息。
多版本(歷史修改)保存在Generations數組中,它的定義:
// generation contains multiple revisions of a key. type generation struct { ver int64 created revision // when the generation is created (put in first revision). revs []revision }
我稱generations[i]為第i代,當一個key從無到有的時候,generations[0]會被創建,其created欄位記錄了引起本次key創建的revision信息。
當用戶繼續更新這個key的時候,generations[0].revs數組會不斷追加記錄本次的revision信息(main,sub)。
在多版本中,每一次操作行為都被單獨記錄下來,那麼用戶value是怎麼存儲的呢?就是保存到bbolt中。
在bbolt中,每個revision將作為key,即序列化(revision.main+revision.sub)作為key。因此,我們先通過記憶體btree在keyIndex.generations[0].revs中找到最後一條revision,即可去bbolt中讀取對應的數據。
相應的,etcd支持按key首碼查詢,其實也就是遍歷btree的同時根據revision去bbolt中獲取用戶的value。
如果我們持續更新同一個key,那麼generations[0].revs就會一直變大,這怎麼辦呢?在多版本中的,一般採用compact來壓縮歷史版本,即當歷史版本到達一定數量時,會刪除一些歷史版本,只保存最近的一些版本。
下麵的是一個keyIndex在compact時,Generations數組的變化:
// For example: put(1.0);put(2.0);tombstone(3.0);put(4.0);tombstone(5.0) on key "foo" // generate a keyIndex: // key: "foo" // rev: 5 // generations: // {empty} // {4.0, 5.0(t)} // {1.0, 2.0, 3.0(t)} // // Compact a keyIndex removes the versions with smaller or equal to // rev except the largest one. If the generation becomes empty // during compaction, it will be removed. if all the generations get // removed, the keyIndex should be removed. // For example: // compact(2) on the previous example // generations: // {empty} // {4.0, 5.0(t)} // {2.0, 3.0(t)} // // compact(4) // generations: // {empty} // {4.0, 5.0(t)} // // compact(5): // generations: // {empty} -> key SHOULD be removed. // // compact(6): // generations: // {empty} -> key SHOULD be removed.
Tombstone就是指delete刪除key,一旦發生刪除就會結束當前的Generation,生成新的Generation,小括弧里的(t)標識Tombstone。
compact(n)表示壓縮掉revision.main <= n的所有歷史版本,會發生一系列的刪減操作,可以仔細觀察上述流程。
多版本總結來說:記憶體btree維護的是用戶key => keyIndex的映射,keyIndex內維護多版本的revision信息,而revision可以映射到磁碟bbolt中的用戶value。
最後,在bbolt中存儲的value是這樣一個json序列化後的結構,包括key創建時的revision(對應某一代generation的created),本次更新版本,sub ID(Version ver),Lease ID(租約ID):
kv := mvccpb.KeyValue{
Key: key,
Value: value,
CreateRevision: c,
ModRevision: rev,
Version: ver,
Lease: int64(leaseID),
}
watch機制
etcd的事件通知機制是基於MVCC多版本實現的。
客戶端可以提供一個要監聽的revision.main作為watch的起始ID,只要etcd當前的全局自增事務ID > watch起始ID,etcd就會將MVCC在bbolt中存儲的所有歷史revision數據,逐一順序的推送給客戶端。
這顯然和ZooKeeper是不同的,ZooKeeper總是獲取最新數據並建立一個一次性的監聽後續變化。而etcd支持客戶端從任意歷史版本開始訂閱事件,並且會推送當時的數據快照給客戶端。
那麼,etcd大概是如何實現基於MVCC的watch機制的呢?
etcd會保存每個客戶端發來的watch請求,watch請求可以關註一個key(單key),或者一個key首碼(區間)。
etcd會有一個協程持續不斷的遍歷所有的watch請求,每個watch對象都維護了其watch的key事件推送到了哪個revision。
etcd會拿著這個revision.main ID去bbolt中繼續向後遍歷,實際上bbolt類似於leveldb,是一個按key有序的K-V引擎,而bbolt中的key是revision.main+revision.sub組成的,所以遍歷就會依次經過歷史上發生過的所有事務(tx)記錄。
對於遍歷經過的每個k-v,etcd會反序列化其中的value,也就是mvccpb.KeyValue,判斷其中的Key是否為watch請求關註的key,如果是就發送給客戶端。
// syncWatchersLoop syncs the watcher in the unsynced map every 100ms. func (s *watchableStore) syncWatchersLoop() { defer s.wg.Done() for { s.mu.RLock() st := time.Now() lastUnsyncedWatchers := s.unsynced.size() s.mu.RUnlock() unsyncedWatchers := 0 if lastUnsyncedWatchers > 0 { unsyncedWatchers = s.syncWatchers() } syncDuration := time.Since(st) waitDuration := 100 * time.Millisecond // more work pending? if unsyncedWatchers != 0 && lastUnsyncedWatchers > unsyncedWatchers { // be fair to other store operations by yielding time taken waitDuration = syncDuration } select { case <-time.After(waitDuration): case <-s.stopc: return } } }
上述代碼是一個迴圈,不停的調用syncWatchers:
// syncWatchers syncs unsynced watchers by: // 1. choose a set of watchers from the unsynced watcher group // 2. iterate over the set to get the minimum revision and remove compacted watchers // 3. use minimum revision to get all key-value pairs and send those events to watchers // 4. remove synced watchers in set from unsynced group and move to synced group func (s *watchableStore) syncWatchers() int { s.mu.Lock() defer s.mu.Unlock() if s.unsynced.size() == 0 { return 0 } s.store.revMu.RLock() defer s.store.revMu.RUnlock() // in order to find key-value pairs from unsynced watchers, we need to // find min revision index, and these revisions can be used to // query the backend store of key-value pairs curRev := s.store.currentRev compactionRev := s.store.compactMainRev wg, minRev := s.unsynced.choose(maxWatchersPerSync, curRev, compactionRev) minBytes, maxBytes := newRevBytes(), newRevBytes() revToBytes(revision{main: minRev}, minBytes) revToBytes(revision{main: curRev + 1}, maxBytes) // UnsafeRange returns keys and values. And in boltdb, keys are revisions. // values are actual key-value pairs in backend. tx := s.store.b.ReadTx() tx.Lock() revs, vs := tx.UnsafeRange(keyBucketName, minBytes, maxBytes, 0) evs := kvsToEvents(wg, revs, vs) tx.Unlock()
代碼比較長不全貼,它會每次從所有的watcher選出一批watcher進行批處理(組成為一個group,叫做watchGroup),這批watcher中觀察的最小revision.main ID作為bbolt的遍歷起始位置,這是一種優化。
你可以想一下,如果為每個watcher單獨遍歷bbolt並從中摘出屬於自己關註的key,那性能就太差了。通過一次性遍歷,處理多個watcher,顯然可以有效減少遍歷的次數。
也許你覺得這樣在watcher數量多的情況下性能仍舊很差,但是你需要知道一般的用戶行為都是從最新的Revision開始watch,很少有需求關註到很古老的revision,這就是關鍵。
遍歷bbolt時,json反序列化每個mvccpb.KeyValue結構,判斷其中的key是否屬於watchGroup關註的key,這是由kvsToEvents函數完成的:
// kvsToEvents gets all events for the watchers from all key-value pairs func kvsToEvents(wg *watcherGroup, revs, vals [][]byte) (evs []mvccpb.Event) { for i, v := range vals { var kv mvccpb.KeyValue if err := kv.Unmarshal(v); err != nil { plog.Panicf("cannot unmarshal event: %v", err) } if !wg.contains(string(kv.Key)) { continue } ty := mvccpb.PUT if isTombstone(revs[i]) { ty = mvccpb.DELETE // patch in mod revision so watchers won't skip kv.ModRevision = bytesToRev(revs[i]).main } evs = append(evs, mvccpb.Event{Kv: &kv, Type: ty}) } return evs }
可見,刪除key對應的revision也會保存到bbolt中,只是bbolt的key比較特別:
put操作的key由main+sub構成:
ibytes := newRevBytes() idxRev := revision{main: rev, sub: int64(len(tw.changes))} revToBytes(idxRev, ibytes)
delete操作的key由main+sub+”t”構成:
idxRev := revision{main: tw.beginRev + 1, sub: int64(len(tw.changes))} revToBytes(idxRev, ibytes) ibytes = appendMarkTombstone(ibytes) // appendMarkTombstone appends tombstone mark to normal revision bytes. func appendMarkTombstone(b []byte) []byte { if len(b) != revBytesLen { plog.Panicf("cannot append mark to non normal revision bytes") } return append(b, markTombstone) } // isTombstone checks whether the revision bytes is a tombstone. func isTombstone(b []byte) bool { return len(b) == markedRevBytesLen && b[markBytePosition] == markTombstone }