etcd教程(十二)---etcd mvcc 源码分析

意琦行 included in etcd

2022-01-14 4054 words 9 minutes

Contents

本文主要通过源码分析了 etcd v3 版本 MVCC 的具体实现。

对 etcd mvcc 不太了解的朋友，可以先阅读这篇文章：etcd教程(六)—etcd多版本并发控制

以下分析基于 etcd v3.5.1版本。

1. 概述

为什么选择MVCC

etcd v3 版本为了解决 v2 版本的并发性能问题和 watch 机制可靠性问题，因此选择了 MVCC 机制。

大致实现

etcd 借助 blotdb，以 revision 为 key，在 blotdb 中存储了 key 的多版本数据。

借助 treeIndex 模块，在内存中以 BTree 构建了 keyIndex 结构来关联 key 及其对应的 revisions。

用户操作时，先根据 key 查询 keyIndex 找到对应的 revisions，然后再操作 blotdb。

整体架构

2. treeIndex 模块

3. MVCC 模块

1. put

一个 put 命令流程如下图所示：

共分为以下几个步骤：

1）查询 keyIndex
- keyIndex 中存储了 key 的创建版本号、修改的次数等信息，这些信息在事务中发挥着重要作用，因此会存储在 boltdb 的 value 中。
2）写入 boltdb
3）更新 treeIndex
4）持久化
- 为了提升性能，具体实现为异步批量操作
- 为了提升 etcd 的写吞吐量、性能，一般情况下（默认堆积的写事务数大于 1 万才在写事务结束时同步持久化），数据持久化由 Backend 的异步 goroutine 完成，它通过事务批量提交，定时将 boltdb 页缓存中的脏数据提交到持久化存储磁盘中。

源码如下：

// server/storage/mvcc/kvstore_txn.go 108 行
func (tw *storeTxnWrite) Put(key, value []byte, lease lease.LeaseID) int64 {
	tw.put(key, value, lease)
	return tw.beginRev + 1
}

// server/storage/mvcc/kvstore_txn.go 182 行
func (tw *storeTxnWrite) put(key, value []byte, leaseID lease.LeaseID) {
	rev := tw.beginRev + 1
	c := rev
	oldLease := lease.NoLease

    // 1.查询keyIndex
	_, created, ver, err := tw.s.kvindex.Get(key, rev)
	if err == nil {
		c = created.main
		oldLease = tw.s.le.GetLease(lease.LeaseItem{Key: string(key)})
	}
	ibytes := newRevBytes()
	idxRev := revision{main: rev, sub: int64(len(tw.changes))}
	revToBytes(idxRev, ibytes)

	ver = ver + 1
	kv := mvccpb.KeyValue{
		Key:            key,
		Value:          value,
		CreateRevision: c,
		ModRevision:    rev,
		Version:        ver,
		Lease:          int64(leaseID),
	}

	d, err := kv.Marshal()
	if err != nil {
		tw.storeTxnRead.s.lg.Fatal(
			"failed to marshal mvccpb.KeyValue",
			zap.Error(err),
		)
	}

    // 2.写blotdb
	tw.tx.UnsafeSeqPut(schema.Key, ibytes, d)
    // 3.更新keyIndex
	tw.s.kvindex.Put(key, idxRev)
	tw.changes = append(tw.changes, kv)
    // lease 相关更新
    // 若存在旧lease则移除
	if oldLease != lease.NoLease {
		if tw.s.le == nil {
			panic("no lessor to detach lease")
		}
		err = tw.s.le.Detach(oldLease, []lease.LeaseItem{{Key: string(key)}})
		if err != nil {
			tw.storeTxnRead.s.lg.Error(
				"failed to detach old lease from a key",
				zap.Error(err),
			)
		}
	}
    // 若本次指定了 lease则关联上
	if leaseID != lease.NoLease {
		if tw.s.le == nil {
			panic("no lessor to attach lease")
		}
		err = tw.s.le.Attach(leaseID, []lease.LeaseItem{{Key: string(key)}})
		if err != nil {
			panic("unexpected error from lease Attach")
		}
	}
}

具体逻辑和前面分析的一致，不过这里需要注意的是 Lease 相关的处理。PUT 时会移除旧的 Lease 和 key 的关联。这就意味着如果想要一直让 key 关联 lease 的话需要每次 PUT 都指定Lease才行。

这和 Redis 的 TTL 还是有很大的不同

然后发现一个问题，如果更新的时候提交一个相同的 leaseID，岂不是会先 Detach 然后又 Attach 上去？可以说是白给了。

于是提了个 PR，现在已经合并进主干了。

2. get

具体流程如下：

1）查询版本号
2）查询 blotdb

具体如下：

// server/storage/mvcc/kvstore_txn.go 61行
func (tr *storeTxnRead) Range(ctx context.Context, key, end []byte, ro RangeOptions) (r *RangeResult, err error) {
	return tr.rangeKeys(ctx, key, end, tr.Rev(), ro)
}

// server/storage/mvcc/kvstore_txn.go 127行
func (tr *storeTxnRead) rangeKeys(ctx context.Context, key, end []byte, curRev int64, ro RangeOptions) (*RangeResult, error) {
	rev := ro.Rev
	if rev > curRev {
		return &RangeResult{KVs: nil, Count: -1, Rev: curRev}, ErrFutureRev
	}
    // 若没指定或指定了错误的版本号就会默认查最新的一个版本
	if rev <= 0 {
		rev = curRev
	}
    // 1.查找 revisions 
     // 这里如果当前查询的版本号比compactMainRev小说明这个版本已经被回收了 直接返回错误
	if rev < tr.s.compactMainRev {
		return &RangeResult{KVs: nil, Count: -1, Rev: 0}, ErrCompacted
	}
	if ro.Count {
		total := tr.s.kvindex.CountRevisions(key, end, rev)
		tr.trace.Step("count revisions from in-memory index tree")
		return &RangeResult{KVs: nil, Count: total, Rev: curRev}, nil
	}
    // 否则就查询比当前版本号大的所有版本号
	revpairs, total := tr.s.kvindex.Revisions(key, end, rev, int(ro.Limit))
	tr.trace.Step("range keys from in-memory index tree")
	if len(revpairs) == 0 {
		return &RangeResult{KVs: nil, Count: total, Rev: curRev}, nil
	}

	limit := int(ro.Limit)
	if limit <= 0 || limit > len(revpairs) {
		limit = len(revpairs)
	}

	kvs := make([]mvccpb.KeyValue, limit)
	revBytes := newRevBytes()
    // 2.查询 blotdb
    // 然后根据上面查到的版本号循环去blotdb中查找对应value
	for i, revpair := range revpairs[:len(kvs)] {
		select {
		case <-ctx.Done():
			return nil, ctx.Err()
		default:
		}
		revToBytes(revpair, revBytes)
		_, vs := tr.tx.UnsafeRange(schema.Key, revBytes, nil, 0)
		if len(vs) != 1 {
			tr.s.lg.Fatal(
				"range failed to find revision pair",
				zap.Int64("revision-main", revpair.main),
				zap.Int64("revision-sub", revpair.sub),
			)
		}
		if err := kvs[i].Unmarshal(vs[0]); err != nil {
			tr.s.lg.Fatal(
				"failed to unmarshal mvccpb.KeyValue",
				zap.Error(err),
			)
		}
	}
	tr.trace.Step("range keys from bolt db")
	return &RangeResult{KVs: kvs, Count: total, Rev: curRev}, nil
}

根据源码可以知道，当我们没有指定 Revision 时，etcd 会默认查询最新版本的数据。

3. del

当执行 del 命令时 etcd 实现的是延期删除模式，原理与 key 更新类似。

与更新 key 不一样之处在于：

一方面，生成的 boltdb key 版本号{4,0,t}追加了删除标识（tombstone, 简写 t），boltdb value 变成只含用户 key 的 KeyValue 结构体。
另一方面 treeIndex 模块也会给此 key hello 对应的 keyIndex 对象，追加一个空的 generation 对象，表示此索引对应的 key 被删除了。

当你再次查询 hello 的时候，treeIndex 模块根据 key hello 查找到 keyindex 对象后，若发现其存在空的 generation 对象，并且查询的版本号大于等于被删除时的版本号，则会返回空。

那么 key 打上删除标记后有哪些用途呢？什么时候会真正删除它呢？

一方面删除 key 时会生成 events，Watch 模块根据 key 的删除标识，会生成对应的 Delete 事件。
另一方面，当你重启 etcd，遍历 boltdb 中的 key 构建 treeIndex 内存树时，你需要知道哪些 key 是已经被删除的，并为对应的 key 索引生成 tombstone 标识。

而真正删除 treeIndex 中的索引对象、boltdb 中的 key 是通过压缩 (compactor) 组件异步完成。

正因为 etcd 的删除 key 操作是基于以上延期删除原理实现的，因此只要压缩组件未回收历史版本，我们就能从 etcd 中找回误删的数据。

具体如下：

// server/storage/mvcc/kvstore_txn.go 101行
func (tw *storeTxnWrite) DeleteRange(key, end []byte) (int64, int64) {
	if n := tw.deleteRange(key, end); n != 0 || len(tw.changes) > 0 {
		return n, tw.beginRev + 1
	}
	return 0, tw.beginRev
}

// server/storage/mvcc/kvstore_txn.go 247行
func (tw *storeTxnWrite) deleteRange(key, end []byte) int64 {
	rrev := tw.beginRev
	if len(tw.changes) > 0 {
		rrev++
	}
    // 1.先在 keyIndex 中找到 blotdb 中对应的key
	keys, _ := tw.s.kvindex.Range(key, end, rrev)
	if len(keys) == 0 {
		return 0
	}
    // 2. 循环删除
	for _, key := range keys {
		tw.delete(key)
	}
	return int64(len(keys))
}

具体 blotdb 删除逻辑如下：

// server/storage/mvcc/kvstore_txn.go 262行
func (tw *storeTxnWrite) delete(key []byte) {
	ibytes := newRevBytes()
	idxRev := revision{main: tw.beginRev + 1, sub: int64(len(tw.changes))}
	revToBytes(idxRev, ibytes)
    // 1.标记删除 blotdb
    // 在 blotdb 的 key上追加tombstone标识(标记删除)
	ibytes = appendMarkTombstone(tw.storeTxnRead.s.lg, ibytes)

	kv := mvccpb.KeyValue{Key: key}

	d, err := kv.Marshal()
	if err != nil {
		tw.storeTxnRead.s.lg.Fatal(
			"failed to marshal mvccpb.KeyValue",
			zap.Error(err),
		)
	}
    // 因为是标记删除,所以这里调用的是 put而不是delete
	tw.tx.UnsafeSeqPut(schema.Key, ibytes, d)
    // 2.处理keyIndex
	err = tw.s.kvindex.Tombstone(key, idxRev)
	if err != nil {
		tw.storeTxnRead.s.lg.Fatal(
			"failed to tombstone an existing key",
			zap.String("key", string(key)),
			zap.Error(err),
		)
	}
	tw.changes = append(tw.changes, kv)
    // 3.如果还有关联的 lease,则移除关联
	item := lease.LeaseItem{Key: string(key)}
	leaseID := tw.s.le.GetLease(item)
	
	if leaseID != lease.NoLease {
		err = tw.s.le.Detach(leaseID, []lease.LeaseItem{item})
		if err != nil {
			tw.storeTxnRead.s.lg.Error(
				"failed to detach old lease from a key",
				zap.Error(err),
			)
		}
	}
}

对 keyIndex 的处理如下：

// server/storage/mvcc/index.go 165 行
func (ti *treeIndex) Tombstone(key []byte, rev revision) error {
	keyi := &keyIndex{key: key}

	ti.Lock()
	defer ti.Unlock()
    // 如果 key 不存在，返回一个错误
	item := ti.tree.Get(keyi)
	if item == nil {
		return ErrRevisionNotFound
	}

	ki := item.(*keyIndex)
	return ki.tombstone(ti.lg, rev.main, rev.sub)
}

具体逻辑如下：

// server/storage/mvcc/key_index.go 119行
func (ki *keyIndex) tombstone(lg *zap.Logger, main int64, sub int64) error {
	if ki.isEmpty() {
		lg.Panic(
			"'tombstone' got an unexpected empty keyIndex",
			zap.String("key", string(ki.key)),
		)
	}
	if ki.generations[len(ki.generations)-1].isEmpty() {
		return ErrRevisionNotFound
	}
    // 首先是把当前删除也作为一个版本号写进入
	ki.put(lg, main, sub)
    // 然后新增了一个 generation，后续的操作就会记录到这个新的 generation 里
	ki.generations = append(ki.generations, generation{})
    // 这个是用于 prometheus 测量数据用的，标记着 etcd 中的 key的数量
    // 虽然是标记删除但还是把这个计数-1了，等后续这个key被再次创建的时候又会+1
	keysGauge.Dec()
	return nil
}

4. 小结

1）blotdb 中以 revision 作为 key，以存储多版本数据。
2）treeIndex 模块中构建 BTree 结构的 keyIndex 以关联 key 和 revisions 的关系，加快查询速度。
3）当你未带版本号查询 key 时，etcd 返回的是 key 最新版本数据。
4）删除一个数据时，etcd 并未真正删除它，而是基于 lazy delete 实现的异步删除，真正删除 key 是通过 etcd 的压缩组件去异步实现的。
- 具体为 del 时会在 keyIndex 中追加一个空的 generation
- 若查询时发送有空的 generation 且查询版本号大于 keyIndex 中的版本号则说明该 key 已经被删除了，当前查询会返回空数据

5. 参考

https://github.com/etcd-io/etcd