Databend Meta-Service 架构概述

Databend 的 meta-service 是一个基于 Raft 共识算法的分布式服务。其焦点架构可以概括为一个 Raft 应用，如图中虚线框所示。
Raft 协议的重要组成部分包括：

• 分布式日记（Log）
  
• 状态机（State Machine）
  

分布式日记

日记重要用于记录分布式体系的操作。当一条日记被写入本地并通过网络同步到其他 Raft 节点后，体系会等候日记在多数节点上完成恒久化。一旦达到多数派写入，该日记条目就被视为已提交（committed）。
状态机

提交的日记随后会被应用到状态机中。我们的状态机计划相对简洁，由两个重要部分组成：

• 内存表（MemTable）：一个基于内存的键值排序数组
  
• 磁盘快照（Snapshot）：恒久化在磁盘上的数据
  

工作流程

• 新的日记条目被提交后，会不断地应用到内存中的状态机。
  
• 状态机的数据不会立即写入磁盘，而是通过 Raft 的定机遇制，定期将整个状态机的数据保存为快照。
  
• 数据被保存到快照后，相应的日记条目可以从 Raft 日记中删除，以节流空间。
  

这种计划既包管了数据的一致性和恒久性，又进步了体系的效率和性能。
在体系架构中，恒久化组件包括日记（log）和快照（Snapshot）。其他组件重要存在于内存中。值得注意的是，快照体系还包含一个块缓存（block cache）机制。该机制将快照分割成多个块，并加载到内存中，从而进步访问速率。对于数据访问模式较为集中的场景，这种计划可以减少直接访问磁盘的频率，提升整体性能。
Raft 模块是在 datafuselabs/openraft 项目中维护的，而我们的 meta-service 可视为 Raft 的一个具体应用实现。在这个架构中，meta Node 是一个焦点数据结构，负责管理 Raft 节点和整个应用程序。它与 Raft 节点创建毗连，并向 Raft 发送指令。meta Node 本质上是一个封装层，重要功能是处理接口对接，而非实现复杂的业务逻辑。
体系对外提供两种 API 接口：

• client API（基于 gRPC ）：重要用于与 databend-query 进行交互，支持读写操作。
  
• Raft API：支持 Raft 协议内部通信，包括以下操作：
  
  
  
  • RequestVote
    
  • AppendEntries
    
  • Snapshot
    
  • Forward
    
  
  

这些 API 构成了体系的网络层，在架构图中以蓝色部分表现。
在本体系的读写操作计划中，所有的读写请求都会被路由到 Leader 节点处理。如图所示，虚线框内的部分代表Leader 节点。Leader 节点的 client API 直接接收来自 databend-query 的请求并进行处理。
对于其他非 Leader 节点（Follower），假如它们接收到客户端请求，会立即通过内部的 forward 接口将请求转发至 Leader 节点。在收到 Leader 的响应后，Follower 节点再将结果返回给客户端。这种计划使得客户端无需手动切换节点即可完成请求。
值得注意的是，当 Follower 节点处理转发请求时，会在返回给 databend-query 客户端的响应中包含一个特殊的 header。这个 header 包含以下信息：

• 指示该请求是被转发处理的
  
• 当前 Leader 节点的地点
  

通过提供这些信息，客户端在后续请求中可以直接访问 Leader 节点，而无需再颠末 Follower 节点的转发。这种机制确保了所有的读写请求终极都在 Leader 节点上执行，从而包管了数据的一致性。
接下来，我们将详细探讨体系支持的写操作接口。
Write 操作

Write 操作的焦点功能是 upsert（update or insert 的缩写），支持数据的更新和插入。以下是干系的数据结构界说：

1. pub struct LogEntry {
2.     pub time_ms: Option<u64>,
3.     pub cmd: Cmd,
4. }
6. pub enum Cmd {
7.     AddNode { node_id: NodeId, node: Node, },
8.     RemoveNode { node_id: NodeId },
9.     UpsertKV(UpsertKV),
10.     Transaction(TxnRequest),
11. }
13. pub struct UpsertKV {
14.     pub key: String,
15.     pub seq: MatchSeq,
16.     //  enum MatchSeq {
17.     //    Exact(u64),
18.     //    GE(u64),
19.     //  }
20.     pub value: Operation<Vec<u8>>,
21.     //    enum Operation<T> {
22.     //        Update(T),
23.     //        Delete,
24.     //        AsIs,
25.     //    }
26.     pub value_meta: Option<MetaSpec>,
27.     //              struct MetaSpec {
28.     //                  expire_at: Option<u64>,
29.     //                  ttl: Option<Interval>,
30.     //              }
31. }

复制代码

LogEntry 结构体包含一个可选的时间戳 time_ms 和一个 Cmd 枚举范例。
Cmd 枚举界说了大概的命令范例：

• AddNode: 添加节点
  
• RemoveNode: 移除节点
  
• UpsertKV: 执行 upsert 操作
  
• Transaction: 执行事件
  

UpsertKV 结构体界说了 upsert 操作的详细参数：

• key: 操作的键
  
• seq: 匹配序列，大概是 MatchSeq::Exact(u64) 或 MatchSeq::GE(u64)
  
• value: 值操作，大概是 Operation::Update(T)、Operation:elete 或 Operation::AsIs
  
• value_meta: 可选的元数据，包含 expire_at 和 ttl 信息
  

Raft LogEntry: 提供一致的时钟

1. pub struct LogEntry {
2.     pub time_ms: Option<u64>,
3.     pub cmd: Cmd,
4. }

复制代码

LogEntry 结构体包含两个关键字段：time_ms 和 cmd。time_ms 字段代表日记条目标天生时间，而 cmd 字段包含了实际必要执行的操作命令。
time_ms 字段在日记条目中的存在具有紧张意义。在分布式体系中，超时判断是一个关键操作，而这个判断通常发生在将日记应用到状态机的过程中。假如使用每个节点的本地时间来进行超时判断，大概会导致不一致性问题。这是因为 Leader 节点和 Follower 节点的本地时间大概存在差别，从而在判断某个条目是否超时时大概得出不同的结果。
例如，Leader 节点大概以为某个条目存在，而 Follower 节点以为该条目不存在。这种不一致性大概导致后续事件执行时出现偏差，终极使状态机陷入不一致状态。因此，时间在我们的体系中必须是可以在内部维护和复制的一致条件，而不能被视为随时大概变化的外部条件。
通过在每个日记条目中记录其天生时间，我们可以确保整个分布式状态机的一致性。当状态机应用一条日记时，我们将体系时间视为该日记条目中记录的时间，并基于这个时间进行超时判断。如许，只要应用的日记序列相同，我们就可以包管每次应用后的结果都是一致的。
这种计划确保了时间在体系内部是可控和一致的，从而维护了分布式状态机的整体一致性。
Cmd: 操作内容

1. pub enum Cmd {
2.     AddNode { node_id: NodeId, node: Node, },
3.     RemoveNode { node_id: NodeId },
4.     UpsertKV(UpsertKV),
5.     Transaction(TxnRequest),
6. }

复制代码

节点变更命令

Cmd 枚举范例界说了体系支持的具体操作。此中，AddNode 和 RemoveNode 这两个变体重要用于内部节点管理，不对外部业务逻辑开放。具体来说，databend-query 不会直接使用这两个命令，它们仅在 meta-service 内部使用。
固然 Raft 协议提供了成员配置变更（membership config change）算法来实现集群节点的动态变化，但该算法重要关注于确保配置变更前后均能维持连续的仲裁（quorum）。这种计划带来了一些限制：

• 每次变更都必要提交两个成员配置日记。
  
• 必须等候前一个配置日记提交后才气发起下一个，这导致了变更过程的延迟。
  

然而，在某些场景下，例如仅更改节点的对外服务地点和端口，并不涉及 Raft 成员的实质性变化。这种情况下，使用 Raft 的联合共识（joint consensus）算法进行成员配置变更显得过于复杂和低效。
考虑到这一点，我们选择将节点信息单独存储在状态机中，而不完全依赖 Raft 的成员配置变更流程。这种计划在包管一致性的同时，提供了更高的灵活性和效率，特殊是对于那些不影响 Raft 成员关系的节点信息更新操作。
数据更新命令

数据更新操作重要分为两类：单条数据更新（upsert）和多条数据更新（transaction）。transaction 可以视为多个 upsert 操作的聚集，同时包含一些必须预先满足的条件。
Upsert

1. pub struct UpsertKV {
2.     pub key: String,
3.     pub seq: MatchSeq,
4.     //  enum MatchSeq {
5.     //    Exact(u64),
6.     //    GE(u64),
7.     //  }
8.     pub value: Operation<Vec<u8>>,
9.     //    enum Operation<T> {
10.     //        Update(T),
11.     //        Delete,
12.     //        AsIs,
13.     //    }
14.     pub value_meta: Option<MetaSpec>,
15.     //              struct MetaSpec {
16.     //                  expire_at: Option<u64>,
17.     //                  ttl: Option<Interval>,
18.     //              }
19. }

复制代码

UpsertKV 结构体界说了单条数据更新操作的详细参数：

• key: 标识必要更新的具体数据项。
  
• seq: 全局序列号，范例为 MatchSeq，重要有两种匹配方式：
  
  
  
  • Exact(u64): 要求当前 key 的版本或序列号必须准确匹配指定值，用于支持比力并互换（Compare-And-Swap，CAS）操作。
    
  • GE(u64): "Greater than or Equal"的缩写，通常用于更新操作，要求记录的序列号大于或等于指定值，确保记录存在。
    
  
  
• value: 范例为 Operation<Vec<u8>>，形貌对值的操作：
  
  
  
  • Update(T): 直接用新值更换原值。
    
  • Delete: 删除该键值对。
    
  • AsIs: 保持值不变，通常用于修改 value_meta。
    
  
  
• value_meta: 可选的元数据，范例为 Option<MetaSpec>，大概包含逾期时间（expire_at）和生存时间（TTL）信息。
  

在 meta-service 中，对外提供的接口将 key 界说为 String 范例，value 界说为 Vec<u8> 范例。
AsIs 操作通常用于节点续期场景。每个节点大概有一个超时时间，通过延长租约（extend lease）来确保节点在超时之前继续存在。这种操作只修改 value_meta 而不改变 value 自己。
Expiration

超机遇制重要用于判断盘算节点是否存活。在这种场景下，通常采用 AsIs 操作，并更新 value_meta 中的超时信息。value_meta 的重要功能是设置超时时间，包含两个选项：

• expire_at: 绝对超时时间，指定一个具体的时间点，到达该时间点时立即超时。
  
• ttl (Time To Live): 相对超时时间，指定一个时间间隔。
  

超时时间的盘算与 LogEntry 中的 time_ms 字段密切干系。对于 ttl，实际的绝对超时时间是由 time_ms 加上 ttl得出。这个绝对超时时间由 meta-service 的 Leader 节点天生，确保不会出现时间回退问题。
使用 expire_at 大概会遇到一些潜在问题。由于客户端直接指定 expire_at，假如客户端的时间掉队于 meta-service 的时间，且指定的超时时间较短，大概会导致数据在写入时就立即超时。这种情况下，写入操作实际上没有产生预期结果，写入完成后客户端大概无法查看到该记录，从而引发一些问题。
因此，除非有特殊需求，通常建议使用 ttl 而非 expire_at。ttl 提供了更可靠和一致的超机遇制，能够有效避免因时间差别导致的不测超时问题。
Transaction

1. message TxnRequest {
2.   repeated TxnCondition condition = 1;
3.   repeated TxnOp if_then = 2;
4.   repeated TxnOp else_then = 3;
5. }
7. message TxnCondition {
8.   string key = 1;
9.   oneof target { bytes value = 2; uint64 seq = 3; }
10.   ConditionResult expected = 4;
12.   enum ConditionResult { EQ = 0; GT = 1; GE = 2; LT = 3; LE = 4; NE = 5; }
13. }
15. message TxnOp {
16.   oneof request {
17.     TxnGetRequest get = 1;
18.     TxnPutRequest put = 2;
19.     TxnDeleteRequest delete = 3;
20.     TxnDeleteByPrefixRequest delete_by_prefix = 4;
21.   }
22. }

复制代码

Transaction（事件）由一系列条件（conditions）和两组操作序列构成：

• condition: 界说事件执行的前提条件。
  
• if_then: 当所有条件满足时执行的操作序列。
  
• else_then: 当任一条件不满足时执行的操作序列。
  

TxnCondition 结构界说了单个条件的组成：

• key: 指定要比力的键。
  
• target: 比力目标，可以是键对应的值（value）或序列号（seq）。
  
• expected: 盼望的比力结果，包括等于（EQ）、大于（GT）、大于等于（GE）、小于（LT）、小于等于（LE）和不等于（NE）。
  

在实际应用中，比力序列号（seq）是最常见的用例。这种方式重要用于实现比力并互换（Compare-And-Swap，CAS）操作，确保在执行后续操作时，键的状态与之前读取时保持一致。
TxnOp 界说了事件中可执行的操作范例：

• get: 获取键值。
  
• put: 写入键值。
  
• delete: 删除指定键。
  
• delete_by_prefix: 删除指定前缀的所有键。
  

这些操作范例涵盖了常见的数据操作需求，为事件提供了灵活的操作空间。
通过这种结构，事件机制能够在包管数据一致性的前提下，实现复杂的条件判断和多步调操作。
数据写入流程

数据写入流程重要包含以下步调:

• 将日记条目落盘到本地磁盘
  
• 将日记复制到 quorum 数量的节点
  
• 确认日记提交状态 - 即后续任何 Leader 都能看到该日记条目
  
• 将已提交的日记应用(apply)到状态机
  

具体的 apply 操作代码如下:
apply()

1. pub async fn apply(&mut self, entry: &Entry) -> Result<AppliedState, io::Error> {
2.     let log_id = &entry.log_id;
3.     let log_time_ms = Self::get_log_time(entry);
5.     self.clean_expired_kvs(log_time_ms).await?;
7.     *self.sm.sys_data_mut().last_applied_mut() = Some(*log_id);
9.     let applied_state = match entry.payload {
10.         EntryPayload::Normal(ref data) => {
11.             info!("apply: normal: {} {}", log_id, data);
12.             assert!(data.txid.is_none(), "txid is disabled");
14.             self.apply_cmd(&data.cmd).await?
15.         }
16.         // ...
17.     };
19.     if let Some(subscriber) = &self.sm.subscriber {
20.         for event in self.changes.drain(..) {
21.             subscriber.kv_changed(event);
22.         }
23.     }
25.     Ok(applied_state)
26. }

复制代码

apply 操作的重要步调:

• 获取日记 ID,用于记录 apply 进度,避免重复 apply
  
• 从日记条目中获取时间戳,作为清理逾期键值对的阈值
  
• 清理所有逾期的键值对
  
• 记录当前 apply 到的日记位置
  
• 执行实际的 apply 命令
  
• 将此次 apply 产生的所有变更发送给 subscriber
  

末了一步是为了支持 watcher 接口,当监控的 key 范围发生变化时,可以接收到关照。
读操作

读操作与写操作不同，它不必要颠末分布式提交过程，而是直接在每个节点的本地状态机中执行。固然读操作可以在Leader 或 Follower 节点上进行，但为了减少数据不一致的风险，我们只允许在 Leader 节点上执行读操作。如许做可以最大程度地避免返回不一致的结果，因为使用逾期或不一致的数据进行后续操作大概会导致更多的冲突，引发问题或增加重试次数。
如前所述，所有的写操作都在 Leader 节点上执行。以下是读操作的简化抽象实现：

1. pub struct SMV003KVApi<'a> { sm: &'a SMV003, }
3. impl<'a> kvapi::KVApi for SMV003KVApi<'a> {
4.     type Error = io::Error;
6.     type KVStream<E> = BoxStream<'static, Result<StreamItem, E>>;
8.     async fn get_kv_stream(&self, keys: &[String]) -> Result<KVStream<Self::Error>, Self::Error> {
9.         let local_now_ms = SeqV::<()>::now_ms();
11.         let mut items = Vec::with_capacity(keys.len());
13.         for k in keys {
14.             let got = self.sm.get_maybe_expired_kv(k.as_str()).await?;
15.             let v = Self::non_expired(got, local_now_ms);
16.             items.push(Ok(StreamItem::from((k.clone(), v))));
17.         }
19.         Ok(futures::stream::iter(items).boxed())
20.     }
22.     async fn list_kv(&self, prefix: &str) -> Result<KVStream<Self::Error>, Self::Error> {
23.         let local_now_ms = SeqV::<()>::now_ms();
25.         let strm = self
26.             .sm
27.             .list_kv(prefix)
28.             .await?
29.             .try_filter(move |(_k, v)| future::ready(!v.is_expired(local_now_ms)))
30.             .map_ok(StreamItem::from);
32.         Ok(strm.boxed())
33.     }
34. }

复制代码

这段代码实现了状态机的 KVApi，包含两个重要接口：get_kv_stream和list_kv。前者用于按键获取值，后者用于按前缀列出键值对。
在执行读操作时，必须处理数据逾期的问题。纵然一段时间内日记没有更新，某些数据大概已颠末期。因此，在读取时必要清理逾期的键。然而，必要注意的是，这种逾期处理大概并不完全准确。
读操作的不一致性问题

• 写操作包管输出一致性结果：
  
  
  
  • 在应用新的日记条目之前，逾期的键会通过 LogEntry.time_ms 被清理。
    
  
  
• Leader 切换期间，读操作大概产生不一致结果：
  
  
  
  • 一条带有逾期时间的记录大概在Leader-A节点上不可见。
    
  • 但在切换到Leader-B节点后，同一条记录大概变得可见。
    
  
  

这种不一致性重要发生在 Leader 切换过程中。例如，当从 Leader-A 切换到 Leader-B 时，假如 Leader-B 的本地时间较小，那么在 Leader-A 上不可见的某条记录大概在 Leader-B 上变得可见。这种读操作的不一致性大概导致一些错误的重试操作。
然而，这个问题通常影响较小，因为节点间的数据不一致和 Leader 切换相对较少发生。写操作始终包管完全一致性，而读操作大概在某些情况下产生不一致的结果。
序列号（Seq number）

1. pub struct SeqV<T = Vec<u8>> {
2.     pub seq: u64,
3.     pub meta: Option<KVMeta>,
4.     pub data: T,
5. }

复制代码

序列号（Seq number）可以被明白为一个版本号。在我们的体系中，我们采用了全局序列号而非每个键（per-key）的版本号。每个键作用域内的版本号更新（例如，当前键的版本是 1，下次更新时版本肯定是 2）这种方式并未被采用。相反，我们使用全局序列号，因为它提供了更可靠的状态追踪本领。
以下是全局单调递增（Globally monotonic）与每键单调递增（Per-key monotonic）的对比示例：

1.                   per-key:        globally
2. insert key=foo -> version=1       seq=1
3.                                               Read   key=foo: version=1
4. update key=foo -> version=2       seq=2
5. delete key=foo -> version=ø       seq=2
6. insert key=foo -> version=1       seq=3
7.                                               Update key=foo(version=1)

复制代码

注意，在第三个操作（DELETE）中，假如使用每键计谋，版本号会变为空（ø）。而使用全局序列号计谋，序列号仍保持为 2。在下一次插入该键时，每键计谋的版本又变回 1，而全局序列号计谋中，序列号递增到 3。
这种差别在并发操作中尤为紧张。假设另一个线程想要执行比力并互换（CAS）操作，它首先读取键的值，获得版本 1。然后，当它尝试更新时，假如看到键的版本仍为 1，在每键计谋中，它无法确定该键是否已经发生变化，这会导致问题。
因此，每键计谋难以实现准确的 CAS 操作。而使用全局序列号，体系中任何键的变化都会导致序列号递增。假如序列号发生变化，就明确表现该键已经被修改。这为实现可靠的 CAS 操作提供了底子。
结语

Databend 的 meta-service 采用基于 Raft 的分布式架构,实现了高可用性和数据一致性。其焦点组件包括分布式日记和状态机,通过精心计划的写入和读取流程,确保了数据操作的可靠性。
该体系的一些关键特性包括:

• 使用 Raft 日记中的时间戳来提供一致的时钟,解决了分布式体系中的时间同步问题。
  
• 支持灵活的节点管理和数据更新操作,包括单条数据更新(upsert)和多条数据更新(transaction)。
  
• 采用全局序列号机制,为实现可靠的比力并互换(CAS)操作提供了底子。
  
• 通过只在 Leader 节点执行读操作,最大程度地减少了数据不一致的风险。
  
• 实现了高效的逾期数据清理机制,包管了数据的时效性。
  

尽管在某些特定情况下(如 Leader 切换期间)大概存在读操作的不一致性,但体系整体上包管了写操作的完全一致性,为分布式数据管理提供了强有力的支持。
随着 Databend 项目标不断发展,meta-service 将继续优化和改进,以应对更复杂的分布式体系挑战,为用户提供更可靠、高效的元数据管理服务。
关于 Databend

Databend 是一款开源、弹性、低成本，基于对象存储也可以做实时分析的新式数仓。期待您的关注，一起探索云原生数仓解决方案，打造新一代开源 Data Cloud。