etcd 存储：如何实现键值对的读写操作？

aoho求索

发布于 2022-06-23 14:47:29

1.5K0

发布于 2022-06-23 14:47:29

文章被收录于专栏：aoho求索

你好，我是 aoho，今天我和你分享的主题是 etcd 存储：如何实现键值对的读写操作？

我们在前面课时介绍了 etcd 的整体架构以及 etcd 常用的通信接口。在介绍 etcd 整体架构时，我们梳理了 etcd 的分层架构以及交互概览。本课时将会聚焦于 etcd 存储是如何实现键值对的读写操作。

本课时围绕 etcd 底层读写的实现，首先会简要介绍客户端访问 etcd 服务端读写的整个过程，然后是重点介绍读写的实现细节。

读操作

在 etcd 中读请求占了大部分，是高频的操作。我们使用 etcdctl 命令行工具进行读操作：

$ etcdctl --endpoints http://localhost:2379 get foo

foo
bar

将整个读操作划分成如下几个步骤：

etcdctl 会创建一个 clientv3 库对象，选取一个合适的 etcd 节点；
调用 KVServer 模块的 Range RPC 方法（上一课时有讲解），发送请求；
拦截器拦截，主要做一些校验和监控；
调用 KVServer 模块的 Range 接口获取数据；

接着就进入了读请求的核心步骤，会经过线性读 ReadIndex 模块、MVCC (包含 treeIndex 和 BlotDB）模块。

这里提一下线性读，线性读是相对串行读来说。集群模式下会有多个 etcd 节点，不同节点之间可能存在一致性问题，串行读直接返回状态数据，不需要与集群中其他节点交互。这种方式速度快，开销小，但是会存在数据不一致的情况。而线性读则需要集群成员之间达成共识，存在开销，响应速度相对慢。但是能够保证数据的一致性，etcd 默认读模式是线性读。我们将在后面的课时重点介绍分布式一致性的实现。

继续往下，看看如何读取 etcd 中的数据。etcd 中查询请求，查询单个键或者一组键，亦或是查询数量，到了底层实际都会调用 rangeKeys 方法，我们来分析下这个方式的实现。range 请求的结构图如下所示：

从上至下，查询键值对的流程包括：

在 treeIndex 中根据键利用 BTree 快速查询该键对应的索引项 keyIndex，索引项中包含 Revision；
根据查询到的版本号信息 Revision，在 Backend 的缓存 buffer 中利用二分法查找，如果命中则直接返回；
若缓存中不符合条件，在 BlotDB 中查找（基于 BlotDB 的索引），查询之后返回键值对信息。

图中的 ReadTx 和 BatchTx 是两个接口，用于负责读写请求。我们在创建 backend 结构时，都会创建 readTx 和 batchTx 结构体，这两个结构体分别实现了 ReadTx 和 BatchTx 接口，一个负责处理只读请求，另一个负责处理读写请求。。

rangeKeys 方法的实现如下所示：

// 位于 mvcc/kvstore_txn.go:117
func (tr *storeTxnRead) rangeKeys(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
 }
 if rev < tr.s.compactMainRev {
  return &RangeResult{KVs: nil, Count: -1, Rev: 0}, ErrCompacted
 }
  // 获取索引项 keyIndex，索引项中包含 Revision
 revpairs := tr.s.kvindex.Revisions(key, end, rev)
 tr.trace.Step("range keys from in-memory index tree")
  // 结果为空，直接返回
 if len(revpairs) == 0 {
  return &RangeResult{KVs: nil, Count: 0, Rev: curRev}, nil
 }
 if ro.Count {
  return &RangeResult{KVs: nil, Count: len(revpairs), Rev: curRev}, nil
 }

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

 kvs := make([]mvccpb.KeyValue, limit)
 revBytes := newRevBytes()
 for i, revpair := range revpairs[:len(kvs)] {
  revToBytes(revpair, revBytes)
    // UnsafeRange 实现了 ReadTx，查询对应的键值对
  _, vs := tr.tx.UnsafeRange(keyBucketName, 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: len(revpairs), Rev: curRev}, nil
}

在上述代码的实现中，我们需要通过 Revisions 方法从 Btree 中获取范围内所有的 keyIndex，以此才能获取一个范围内的所有键值对。Revisions 方法实现如下：

// 位于 mvcc/index.go:106
func (ti *treeIndex) Revisions(key, end []byte, atRev int64) (revs []revision) {
 if end == nil {
  rev, _, _, err := ti.Get(key, atRev)
  if err != nil {
   return nil
  }
  return []revision{rev}
 }
 ti.visit(key, end, func(ki *keyIndex) {
    // 使用 keyIndex.get 来遍历整颗树
  if rev, _, _, err := ki.get(ti.lg, atRev); err == nil {
   revs = append(revs, rev)
  }
 })
 return revs
}

如果只是获取一个键对应的版本，使用 treeIndex 方法即可，但是一般会从 Btree 索引中获取多个 Revision 值时，需要调用 keyIndex.get 方法来遍历整颗树并选取合适的版本。这是因为 BoltDB 保存一个 key 的多个历史版本。每一个 Key 的 keyIndex 中其实都存储着多个历史版本，我们需要根据传入的参数返回正确的版本。

对于上层的键值存储来说，它会利用这里返回的 Revision 从真正存储数据的 BoltDB 中查询当前 Key 对应 Revision 的结果。BoltDB 内部使用的也是类似 bucket 的方式存储，其实就是对应 MySQL 中的表结构，用户的 key 数据存放的 bucket 名字的是 key，etcd MVCC 元数据存放的 bucket 是 meta。

写操作

介绍完读请求，我们回忆一下写操作的实现。使用 etcdctl 命令行工具进行写操作：

$ etcdctl --endpoints http://localhost:2379 put foo bar

将整个写操作划分成如下几个步骤：

客户端通过负载均衡算法选择一个 etcd 节点，发起 gRPC 调用；
etcd server 收到客户端请求；
经过 gRPC 拦截、Quota 校验，Quota 模块用于校验 etcd db 文件大小是否超过了配额；
接着 KVServer 模块将请求发送给本模块中的 raft，这里负责与 etcd raft 模块进行通信。发起一个提案，命令为 put foo bar，即使用 put 方法将 foo 更新为 bar；
提案经过转发之后，半数节点成功持久化；
MVCC 模块更新状态机。

我们重点关注最后一步，学习如何更新和插入键值对。与上面一张图相对应，我们来看下 put 接口的执行过程：

调用 put 向 etcd 写入数据时，首先会使用传入的键构建 keyIndex 结构体，基于 currentRevision 自增生成新的 revision 如{1,0}，并从 treeIndex 中获取相关版本 revision 等信息；写事务提交之后，将本次写操作的缓存 buffer 合并到（merge）到读缓存上（图中 ReadTx 中的缓存）。代码实现如下所示：

//位于 mvcc/index.go:53
func (ti *treeIndex) Put(key []byte, rev revision) {
 keyi := &keyIndex{key: key}
  // 加锁，互斥
 ti.Lock()
 defer ti.Unlock()
  // 获取版本信息
 item := ti.tree.Get(keyi)
 if item == nil {
  keyi.put(ti.lg, rev.main, rev.sub)
  ti.tree.ReplaceOrInsert(keyi)
  return
 }
 okeyi := item.(*keyIndex)
 okeyi.put(ti.lg, rev.main, rev.sub)
}

treeIndex.Put 在获取 Btree 中的 keyIndex 结构之后，会通过 keyIndex.put 在其中加入新的 revision，方法实现如下：

// 位于 mvcc/key_index.go:77
func (ki *keyIndex) put(lg *zap.Logger, main int64, sub int64) {
 rev := revision{main: main, sub: sub}
  // 校验版本号
 if !rev.GreaterThan(ki.modified) {
  lg.Panic(
   "'put' with an unexpected smaller revision",
   zap.Int64("given-revision-main", rev.main),
   zap.Int64("given-revision-sub", rev.sub),
   zap.Int64("modified-revision-main", ki.modified.main),
   zap.Int64("modified-revision-sub", ki.modified.sub),
  )
 }
 if len(ki.generations) == 0 {
  ki.generations = append(ki.generations, generation{})
 }
 g := &ki.generations[len(ki.generations)-1]
 if len(g.revs) == 0 { // 创建一个新的键
  keysGauge.Inc()
  g.created = rev
 }
 g.revs = append(g.revs, rev)
 g.ver++
 ki.modified = rev
}

新的 revision 结构体写入 keyIndex 键值索引时，都会改变 generation 结构体中的创建版本、修改次数等参数，因此，基于 put 方法，我们就可以知道 generation 结构体中的各个成员如何定义和赋值。

revision{1,0} 是生成的全局版本号，作为 BoltDB 的 key，经过序列化包括 key 名称、key 创建时的版本号（create_revision）、value 值和租约等信息为二进制数据之后，将填充到 BoltDB 的 value 中，同时将该键和 revision 等信息存储到 Btree。