彻底搞懂 etcd 系列文章（九）：etcd compact 和 watch API

etcd 是云原生架构中重要的基础组件，由 CNCF 孵化托管。etcd 在微服务和 Kubernates 集群中不仅可以作为服务注册与发现，还可以作为 key-value 存储的中间件。

《彻底搞懂 etcd 系列文章》将会从 etcd 的基本功能实践、API 接口、实现原理、源码分析，以及实现中的踩坑经验等几方面具体展开介绍 etcd。预计会有 20 篇左右的文章，笔者将会每周持续更新，欢迎关注。

1 Compact方法

Compact 方法压缩 etcd 键值对存储中的事件历史。键值对存储应该定期压缩，否则事件历史会无限制的持续增长。

rpc Compact(CompactionRequest) returns (CompactionResponse) {}

请求的消息体是 CompactionRequest， CompactionRequest 压缩键值对存储到给定修订版本。所有修订版本比压缩修订版本小的键都将被删除：

message CompactionRequest {
  // 键值存储的修订版本，用于比较操作
  int64 revision = 1;

  bool physical = 2;
}

physical 设置为 true 时 RPC 将会等待直到压缩物理性的应用到本地数据库，到这程度被压缩的项将完全从后端数据库中移除。应答的消息体 CompactionResponse 定义为：

message CompactionResponse {
  ResponseHeader header = 1;
}

CompactionResponse 只有一个通用的响应头。

2 Watch 服务

Watch API 提供了一个基于事件的接口，用于异步监视键的更改。etcd3 监视程序通过从给定的修订版本（当前版本或历史版本）持续监视 key 更改，并将 key 更新流回客户端。

事件

每个键的更改都用事件消息表示。 事件消息会同时提供更新数据和更新类型，mvccpb.Event 的消息体定义如下：

message Event {
  enum EventType {
    PUT = 0;
    DELETE = 1;
  }

  EventType type = 1;

  KeyValue kv = 2;

  // prev_kv 持有在事件发生前的键值对
  KeyValue prev_kv = 3;
}

type 是事件的类型。如果类型是 PUT，表明新的数据已经存储到 key；如果类型是 DELETE， 表明 key 已经被删除。

kv 为事件持有 KeyValue。PUT 事件包含当前的 kv 键值对。kv.Version=1 的 PUT 事件表明 key 的创建。DELETE/EXPIRE 事件包含被删除的 key，它的修改修订版本设置为删除的修订版本。

监视流

Watch API 提供了一个基于事件的接口，用于异步监视键的更改。etcd 监视程序通过从给定的修订版本（当前版本或历史版本）连续监视来等待密钥更改，并将密钥更新流回客户端。

监视持续运行，并使用 gRPC 来流式传输事件数据。监视流是双向的，客户端写入流以建立监视事件，并读取以接收监视事件。单个监视流可以通过使用每个观察器标识符标记事件来复用许多不同的观察。这种多路复用有助于减少 etcd 群集上的内存占用量和连接开销。

Watch 事件具有如下三个特性：

• 有序，事件按修订顺序排序；如果事件早于已发布的事件，它将永远不会出现在手表上。
• 可靠，事件序列永远不会丢弃任何事件子序列；如果按时间顺序为 a < b < c 三个事件，那么如果 Watch 接收到事件 a 和 c，则可以保证接收到 b。
• 原子，保证事件清单包含完整的修订版；同一修订版中通过多个键进行的更新不会拆分为多个事件列表。

Watch service 定义

在 rpc.proto 中 Watch service 定义如下：

service Watch {
  rpc Watch(stream WatchRequest) returns (stream WatchResponse) {}
}

Watch 观察将要发生或者已经发生的事件。输入和输出都是流;输入流用于创建和取消观察，而输出流发送事件。一个观察 RPC 可以在一次性在多个 key 范围上观察，并为多个观察流化事件。整个事件历史可以从最后压缩修订版本开始观察。

WatchService 只有一个 Watch 方法。

请求的消息体 WatchRequest 定义如下：

message WatchRequest {
  oneof request_union {
    WatchCreateRequest create_request = 1;
    WatchCancelRequest cancel_request = 2;
  }
}

request_union 要么是创建新的观察者的请求，要么是取消一个已经存在的观察者的请求。创建新的观察者的请求 WatchCreateRequest：

message WatchCreateRequest {
  // key 是注册要观察的 key
  bytes key = 1;

  bytes range_end = 2;

  // start_revision 是可选的开始(包括)观察的修订版本。不设置 start_revision 则表示 "现在".
  int64 start_revision = 3;

  bool progress_notify = 4;

  enum FilterType {
  // 过滤掉 put 事件
  NOPUT = 0;

  // 过滤掉 delete 事件
  NODELETE = 1;
  }

  // 过滤器，在服务器端发送事件给回观察者之前，过滤掉事件。
  repeated FilterType filters = 5;

  // 如果 prev_kv 被设置，被创建的观察者在事件发生前获取上一次的KV。
  // 如果上一次的KV已经被压缩，则不会返回任何东西
  bool prev_kv = 6;
}

range_end 是要观察的范围 [key, range_end) 的终点。如果 range_end 没有设置，则只有参数 key 被观察；如果 range_end 等同于 ‘\0’， 则大于等于参数 key 的所有 key 都将被观察；如果 range_end 比给定 key 大1， 则所有以给定 key 为前缀的 key 都将被观察。

progress_notify 设置，这样如果最近没有事件，etcd 服务器将定期的发送不带任何事件的 WatchResponse 给新的观察者。当客户端希望从最近已知的修订版本开始恢复断开的观察者时有用。etcd 服务器将基于当前负载决定它发送通知的频率。

取消已有观察者的 WatchCancelRequest：

message WatchCancelRequest {
  int64 watch_id = 1;
}

watch_id 是要取消的观察者的id，这样就不再有更多事件传播过来了。

应答的消息体 WatchResponse：

message WatchResponse {
  ResponseHeader header = 1;
  // watch_id 是和应答相关的观察者的ID
  int64 watch_id = 2;

  bool created = 3;

  bool canceled = 4;

  int64 compact_revision  = 5;

  // cancel_reason 指出取消观察者的理由.
  string cancel_reason = 6;

  repeated mvccpb.Event events = 11;
}

如果应答是用于创建观察者请求的，则 created 设置为 true。客户端应该记录 watch_id 并期待从同样的流中为创建的观察者接收事件。所有发送给被创建的观察者的事件将附带同样的 watch_id；如果应答是用于取消观察者请求的，则 canceled 设置为true。不会再有事件发送给被取消的观察者。

compact_revision 被设置为最小 index，如果观察者试图观察被压缩的 index。当在被压缩的修订版本上创建观察者或者观察者无法追上键值对存储的进展时发生。客户端应该视观察者为被取消，并不应该试图再次创建任何带有相同 start_revision 的观察者。