Etcd Raft库的日志存储

2021-06-28
14分钟阅读时长
分布式

概述

之前看etcd raft实现的时候，由于wal以及日志的落盘存储部分，没有放在raft模块中，对这部分没有扣的特别细致。而且，以前我的观点认为etcd raft把WAL这部分留给了上层的应用去实现，自身通过Ready结构体来通知应用层落盘的数据，这个观点也有失偏颇，etcd只是没有把这部分代码放在raft模块中，属于代码组织的范畴问题，并不是需要应用层自己来实现。

于是，决定专门写一篇文章把这部分内容给讲解一下，主要涉及以下内容：

  • 日志（包括快照）文件的格式。
  • 日志（包括快照）内容的落盘、恢复。

以前的系列文章可以在下面的链接中找到，本文不打算过多重复原理性的内容：

WAL及快照文件格式

首先来讲解这两种文件的格式，了解了格式才能继续展开下面的讲述。

WAL文件格式

wal文件的文件名格式为：seq-index.wal（见函数walName）。其中：

  • seq：序列号，从0开始递增。
  • index：该wal文件存储的第一条日志数据的索引。

因此，如果将一个目录下的所有wal文件按照名称排序之后，给定一个日志索引，很快就能知道该索引的日志落在哪个wal文件之中的。

WAL文件中每条记录的格式如下：

message Record {
	optional int64 type  = 1 [(gogoproto.nullable) = false];
	optional uint32 crc  = 2 [(gogoproto.nullable) = false];
	optional bytes data  = 3;
}
  • type：记录的类型，下面解释。
  • crc：后面data部分数据的crc32校验值。
  • data：数据部分，根据类型的不同有不同格式的数据。

记录数据的类型如下：

const (
	// 以下是WAL存放的数据类型
	// 元数据
	metadataType int64 = iota + 1
	// 日志数据
	entryType
	// 状态数据
	stateType
	// 校验初始值
	crcType
	// 快照数据
	snapshotType
)

下面展开解释。

元数据

元数据就是应用层自定义的数据，需要注意的是，一个服务中如果有多个wal文件，且这些文件中有多份元数据，那么这些元数据都必须一致，否则报错。

对于etcd这个服务而言，存储的元数据就是节点ID以及集群ID：

	metadata := pbutil.MustMarshal(
		&pb.Metadata{
			NodeID:    uint64(member.ID),
			ClusterID: uint64(cl.ID()),
		},
	)
	if w, err = wal.Create(cfg.WALDir(), metadata); err != nil {
		plog.Fatalf("create wal error: %v", err)
	}

日志数据

日志数据的格式，就是raft.proto中Entry的格式：

message Entry {
	optional uint64     Term  = 2 [(gogoproto.nullable) = false]; // must be 64-bit aligned for atomic operations
	optional uint64     Index = 3 [(gogoproto.nullable) = false]; // must be 64-bit aligned for atomic operations
	optional EntryType  Type  = 1 [(gogoproto.nullable) = false];
	optional bytes      Data  = 4;
}

状态数据

保存当前“硬状态（HardState）”的记录，HardState包括：当前任期号、当前给哪个节点ID投票、当前提交的最大日志索引。

message HardState {
	optional uint64 term   = 1 [(gogoproto.nullable) = false];
	optional uint64 vote   = 2 [(gogoproto.nullable) = false];
	optional uint64 commit = 3 [(gogoproto.nullable) = false];
}

校验初始值

校验数据这一块，挺有意思的，可以展开好好说一下。

使用CRC算法来计算数据的校验值，除了需要原始数据之外，还需要一个校验初始值（即校验种子seed），在每个wal文件中，类型为校验初始值的记录就用于存储这个值。其值和使用方式有以下几点需要注意：

  • 每个wal文件必须有校验初始值类型的数据，后续所有写入该wal文件的记录，都使用该初始值来计算CRC校验值。
  • 第一个wal文件，即序列号为0的wal文件，其校验初始值为0（见wal.go的Create函数）。
  • 当生成下一个wal文件时，以上一个wal文件的最后一条日志数据的CRC校验码来做为该文件的校验初始值，这样就要求类型为校验初始值的记录，必须存储在同一个wal文件中第一条日志数据的前面，否则计算出来该日志数据的crc校验码就不准。

wal文件的校验初始值

可以看到，通过这个机制，将多个连续的wal文件“串联”了起来：使用上一个wal文件的最后一个日志数据的crc校验值，来做为下一个wal文件的校验初始值，可以有效的校验同一个项目中wal文件的正确性。

快照数据

在wal文件中存储的快照数据类型的记录，其中仅存储了当前快照的索引和任期号，而快照的详细数据都放到快照数据文件中存储，下面讲到数据恢复时再展开讨论这部分内容：

message Snapshot {
	optional uint64 index = 1 [(gogoproto.nullable) = false];
	optional uint64 term  = 2 [(gogoproto.nullable) = false];
}

快照文件格式

快照文件的文件名格式为：任期号-索引号.snap（见函数Snapshotter::save）。每次来一个快照数据，都新建一个快照文件，文件中存储快照数据的格式为：

message snapshot {
	optional uint32 crc  = 1 [(gogoproto.nullable) = false];
	optional bytes data  = 2;
}

即：只存储快照数据及其校验值，数据的具体格式由存储快照数据的使用方来解释。在etcd这个服务里，这份快照数据的格式就是：

message Snapshot {
	optional bytes            data     = 1;
	optional SnapshotMetadata metadata = 2 [(gogoproto.nullable) = false];
}

数据恢复流程

日志、快照数据的落盘，都是为了重启时恢复数据，了解了上面wal以及快照文件的格式，可以来看看数据的恢复流程。

其大体流程如下:

  • 到快照目录中取出最新的一份无错的快照文件，首先取出这个文件中存储的快照数据。（见函数Snapshotter::Load）
  • 此时，从快照数据中可以反序列化出：快照数据、对应的任期号、索引号。
  • 根据第二步拿到的快照数据，到wal目录中拿到日志索引号在快照数据索引号之后的日志，遍历满足条件的记录进行数据恢复。（见函数WAL::ReadAll）。

下面具体来看每种wal记录格式数据在进行数据恢复时的流程：

  • 日志数据：由于还可能存在一小部分小于快照索引的日志，所以恢复时会忽略掉这部分数据。
  • 状态数据：每一条状态数据都会反序列化出来，以最后一条状态数据为准。
  • 元数据：前面提到过，同一个服务的元数据必须一致，所以这里会校验元数据前后是否一致，不一致将报错退出数据恢复流程。
  • 校验初始值数据：可以参见前面关于该类型数据的讲解。
  • 快照数据：下面详细解释。

举一个例子来描述前面根据快照文件和WAL文件恢复数据的流程：

WAL与快照文件关系

如上图中：

  • 快照文件集合为[1-50.snap,1-150.snap]，取最新的快照文件，即1-150.snap，而1-50.snap文件的数据为过期数据。
  • 由于快照文件中存储的日志索引到150，即在此之前的日志已经全部被压缩到了快照文件中，因此wal文件集合中：
    • 0-100.wal中的数据已经全部被压缩。
    • 1-200.wal中的数据部分被压缩，恢复数据时要忽略日志索引小于150的日志数据。
    • 3-300.wal中的数据都没有被压缩，恢复数据时要如实全部重放该文件的数据。

前面分析快照数据类型的时候，提到过这个类型的数据在wal文件中的记录，只会存储：

  • 当前快照时对应的任期号。
  • 当前快照时对应的索引号。
  • 而具体的快照数据内容存储在快照文件中。

也就是说，当生成一份新的快照数据时，将会把这份快照数据相关的以上三部分内容存储到wal和快照文件中。

所以当恢复数据的时候，此时已经反序列化出快照数据了，这时拿着快照数据读wal文件时，如果读到了快照类型的数据，就会去对比起任期号和索引号是否一致，不一致报错停止恢复流程：

		case snapshotType: // 快照数据
			var snap walpb.Snapshot
			pbutil.MustUnmarshal(&snap, rec.Data)
			if snap.Index == w.start.Index { // 两者的索引相同
				if snap.Term != w.start.Term { // 但是任期号不同
					state.Reset()
					// 返回ErrSnapshotMismatch错误
					return nil, state, nil, ErrSnapshotMismatch
				}
				// 保存快照数据匹配的标志位
				match = true
			}

以上，解释清楚了wal、快照文件的格式，以及数据恢复的流程。

因为wal文件和快照文件的读写，都与磁盘读写相关，所以在etcd服务中，将这两个结构体，统一到etcdserver/storage.go的storage结构体中：

type storage struct {
	*wal.WAL
	*snap.Snapshotter
}

由storage结构体统一对外提供wal、快照文件的读写接口：

type Storage interface {
	// Save function saves ents and state to the underlying stable storage.
	// Save MUST block until st and ents are on stable storage.
	Save(st raftpb.HardState, ents []raftpb.Entry) error
	// SaveSnap function saves snapshot to the underlying stable storage.
	SaveSnap(snap raftpb.Snapshot) error
	// DBFilePath returns the file path of database snapshot saved with given
	// id.
	DBFilePath(id uint64) (string, error)
	// Close closes the Storage and performs finalization.
	Close() error
}

下面，解释一下写wal文件中需要注意的一些细节。

写优化问题

数据对齐

每条写入wal的记录，都会将其大小向上8字节对齐，多出来的部分填零：

func encodeFrameSize(dataBytes int) (lenField uint64, padBytes int) {
	lenField = uint64(dataBytes)
	// force 8 byte alignment so length never gets a torn write
	padBytes = (8 - (dataBytes % 8)) % 8
	if padBytes != 0 {
		lenField |= uint64(0x80|padBytes) << 56
	}
	return
}

WAL记录数据需8字节对齐

写缓冲区

另外，为了缓解写文件的IO负担，etcd做了一个写优化：落盘的数据首先写到一个内存缓冲区中，只有每次填满了一个page的数据才会进行落盘操作。

etcd中定义了几个常量：

  • const minSectorSize = 512
  • const walPageBytes = 8 * minSectorSize

其中：minSectorSize表示一个sector的大小，而walPageBytes必须为minSectorSize的整数倍。

etcd中定义了一个PageWriter结构体，用于实现写入日志的操作，内部定义了一个循环缓冲区，只有填满一个walPageBytes大小的数据才会进行落盘。

下图是写入数据落盘后循环缓冲区的变化的示意图：

写入数据落盘后循环缓冲区的变化

  • 黄色方块表示一个page的空闲空间，绿色方块表示待写入数据，红色方块表示当前已经写入数据的缓冲区。
  • 刚开始，第一个page已经有部分数据写入，还剩余一部分空闲空间。因此，当写入数据时，只会把写入数据凑齐一个页面大小来落盘。
  • 落盘完毕之后，第一个page重新变成黄色，即空闲页面，而第二个页面存储了写入数据中没有落盘的部分。

代码流程见函数PageWriter::Write。

从上面的写入落盘流程可以看到，一次写入的数据可能会有一部分落盘，一部分还在内存中，这样当系统发生宕机这部分数据就是被损坏（corruption）的数据。

因此，etcd中还需要有办法来识别和恢复数据。

识别部分写入（partial write）数据

函数decoder::isTornEntry用于判断一条记录是否为部分写的损坏数据。

其原理是：

  • 每次新创建用于写入记录的wal文件，都会将剩余文件清零。
  • 读入记录的数据之后，将数据根据不大于每个chunk为minSectorSize大小的方式，存入chunk数组中。
  • 遍历这些chunk，如果有一个chunk的数据全部是零，则认为这块数据是部分写入的损坏数据。

这个地方要跟前面落盘流程来对照看：因为每次落盘都是以一个page为单位落盘，而page大小又是minSectorSize的整数倍，因此以minSectorSize为一个chunk的大小来判断是否损坏。

修复wal文件流程

当进行数据恢复时，可能会出现前面的部分写导致数据损坏问题，etcd会进行如下的修复操作：

  • 部分写导致数据损坏都只会出现在最后一个wal文件，因此打开最后一个wal文件进行处理（见函数openLast）。
  • 出现部分写导致损坏的记录，解析过程中都会返回ErrUnexpectedEOF错误，对于这样的文件：
    • 将损坏的文件重命名为原文件名.broken。
    • 记录下来最后一个无损记录的偏移量，将损坏之后的数据都截断（Truncate）。

只读和只写文件的区别

在etcd中，wal文件有两种并不能同时共存的模式：对于同一个wal文件而言，要么处于只读模式，要么处于append写模式，这两种模式不能同时存在。见WAL结构体的注释：

// WAL is a logical representation of the stable storage.
// WAL is either in read mode or append mode but not both.
// A newly created WAL is in append mode, and ready for appending records.
// A just opened WAL is in read mode, and ready for reading records.
// The WAL will be ready for appending after reading out all the previous records.

根据上面可能使用缓冲区优化写操作可知，两种模式下在读记录时能容忍的错误级别也不一样：

  • 读模式：读模式下可能读到部分写的数据，所以可以容忍这种错误。
  • 写模式：写模式下，不能容忍读到部分写的数据。
	switch w.tail() {
	case nil:
		// We do not have to read out all entries in read mode.
		// The last record maybe a partial written one, so
		// ErrunexpectedEOF might be returned.
		// 在只读模式下，可能没有读完全部的记录。最后一条记录可能是只写了一部分，此时就会返回ErrunexpectedEOF错误
		if err != io.EOF && err != io.ErrUnexpectedEOF { // 如果不是EOF以及ErrunexpectedEOF错误的情况就返回错误
			state.Reset()
			return nil, state, nil, err
		}
	default:
		// 写模式下必须读完全部的记录
		// We must read all of the entries if WAL is opened in write mode.
		if err != io.EOF { // 如果不是EOF错误，说明没有读完数据就报错了，这种情况也是返回错误
			state.Reset()
			return nil, state, nil, err
		}

数据落盘的全流程

以上了解了wal、快照文件的格式，以及写入流程，这里把之前写的不够好的数据落盘流程重新梳理一下。

在 etcd Raft库解析 - codedump的网络日志中，曾经指出etcd raft库是通过Ready结构体，来通知应用层的当前的数据的，不清楚的话可以回看一下之前的内容。在这里，只解释该结构体中与数据落盘相关的几个成员的数据走向流程，即日志数据（成员Entries）、快照数据（Snapshot）、已提交日志（CommittedEntries）。

日志数据

日志数据从客户端提交到落盘的走向是这样的：

  • 由客户端提交给服务器（注：只有leader节点才能接收客户端提交的日志数据，其他节点需转发给leader）。
  • 服务器收到之后，首先调用raftLog.append函数保存到unstable_log中，此时日志还是在内存中的，并未落地。
  • 通过newReady函数构建Ready结构体时，将上一步保存下来的日志数据保存到Ready结构体的Entries。
  • 应用层收到Ready结构体之后，调用wal的WAL.Save接口保存日志数据。这一步做完之后，可以认为日志数据已经落盘了。
  • 由于数据已经落盘到WAL日志中，所以在应用层通过Node.Advance接口回调通知raft库时，暂存在unstable_log中的日志就可以通过函数raftLog.stableTo删除了。

日志数据从提交到落盘的走向

已提交日志

raft日志中，需要保存两个日志索引：

  • appliedIndex：通知到应用层目前为止最大的日志索引；
  • commitIndex：当前已提交日志的最大索引。

在这里，总有appliedIndex <= commitIndex条件成立，即日志总是先被提交成功（即达成一致），才会通知给应用层。

通知应用层已提交日志的流程如下：

  • 调用raftLog.nextEnts()函数获得当前满足appliedIndex <= commitIndex条件的日志，存入到Ready.CommittedEntries通知应用层。
  • 应用层处理这部分已提交日志。
  • 调用raftLog.appliedTo()函数，这里会修改appliedIndex = commitIndex，即所有日志都已通知应用层。

通知应用层已提交日志流程

快照数据

快照数据由应用层生成，然后将生成的快照数据、当前appliedIndex、配置状态一起交给存储层，保存之后就可以把在该快照之前的数据给删除了：

func (rc *raftNode) maybeTriggerSnapshot() {
	// 生成快照数据
	data, err := rc.getSnapshot()

	// 通知存储层快照数据
	rc.raftStorage.CreateSnapshot(rc.appliedIndex, &rc.confState, data)

	// 保存快照数据
	rc.saveSnap(snap)

	// 将快照之前的数据压缩
	compactIndex := uint64(1)
	rc.raftStorage.Compact(compactIndex)

	// 更新快照数据索引，以便下一次生成新的快照数据
	rc.snapshotIndex = rc.appliedIndex
}

数据的修复

从上面的分析中可以看到，日志数据是在客户端提交之后，就马上落盘到WAL文件中的，不会等到日志在集群中达成一致。

这样会带来一个问题，比如：

  • 节点A认为自己还是集群的leader节点，此时收到客户端日志之后，将数据落盘到WAL文件中。
  • 落盘之后，节点A将日志同步给集群的其它节点，但是发现自己已经不再是集群的leader节点了。

在这种情况下，显然第一步已经落盘的日志是无效的，需要进行修复，这时候是怎么操作的呢？

etcd raft的做法是不回退日志，继续走正常的流程，用新的、正确的日志添加在错误的日志后面，这样回放数据的时候恢复数据。

继续以上面的例子为例：

  • 节点A在认为自己是leader的情况下落盘日志到本地WAL中，落盘完毕之后同步给集群内其他节点。
  • 同步到集群其他节点的过程中，才发现节点A已经不是集群的leader，此时节点A降级为follower节点，并开始从正确的集群节点那里同步日志。
  • 同步日志的流程中，节点A将收到来自leader节点的正确日志，这些日志也将落盘到节点A的WAL中。

第二步中同步日志的流程可以参见 Raft算法原理 - codedump的网络日志，这里不再阐述。

上面的流程之后，节点A的WAL中将存在：

  • 认为自己是leader时已落盘的日志；
  • 集群leader纠正节点A同步过来的日志。

这样，当重启恢复时，会一并将这些日志重放，应用层只要按顺序回放日志即可。

WAL日志的纠错机制

如上图中：

  • 节点认为自己是leader节点时，落盘到WAL文件中的日志是[(1,10),(1,11)]，列表中的二元组数据中，第一个元素是任期号，第二个元素是日志索引号。
  • 在落盘日志之后，节点将数据广播到集群，才发现自己已经不是集群的leader节点，此时集群的leader节点发现从日志10开始，该节点的数据就是不对的，开始同步正确的日志给节点，于是把正确的日志[(2,10),(2,11)]同步给了节点，这部分日志会添加到前面错误的日志之后。
  • 假设节点重启恢复，那么会依次重放前面这四条日志，其中前两条日志是错误的日志，但是由于有后面的两条正确日志，最终节点的状态还是会恢复正确状态。
  • 随着后面日志数据压缩成快照文件，冗余的错误日志的磁盘占用将被解决。

读者不妨在这里就着这个流程多思考一个问题：做为follower的节点，是什么时候将日志落盘到WAL文件中，是在收到leader节点同步过来的日志时，还是在leader节点通知某个日志已经在集群达成一致？为什么以及流程是怎样的？

总结

  • etcd的wal模块，虽然并没有和raft模块放在一起，但并不是说这一部分就需要应用者来自己实现，这两部分其实是一起打包做为整个etcd raft算法库提供给使用者的。可以认为raft模块提供算法，wal和快照模块提供日志存储读写的接口。
  • 日志落盘部分，包括wal文件以及快照文件读写这两部分内容，etcd将这两部分统一到Storage接口统一对外服务。
  • raft算法是在收到客户端日志之后就理解落盘日志到wal文件中保存的，如果后面发现出错，就走正常的同步正确日志的流程，将正确的日志添加到后面，这样恢复时重放整个日志，最终节点达成一致的正确状态。
#etcd #raft #分布式
下一页 Etcd Raft库的工程化实现

相关

Etcd Raft库的工程化实现
etcd Raft库解析
Etcd存储的实现
Raft算法原理
服务调用的演进历史
《数据密集型应用系统设计》第九章《一致性与共识》笔记