Skip to content

Latest commit

 

History

History
310 lines (250 loc) · 9.1 KB

Etcd源码分析-存储.md

File metadata and controls

310 lines (250 loc) · 9.1 KB

存储数据结构

Etcd存储在集群搭建和使用篇有简介，总结起来有如下特点：

  • 采用kv型数据存储，一般情况下比关系型数据库快。
  • 支持动态存储(内存)以及静态存储(磁盘)。
  • 分布式存储，可集成为多节点集群。
  • 存储方式，采用类似目录结构。 1、只有叶子节点才能真正存储数据，相当于文件。 2、叶子节点的父节点一定是目录，目录不能存储数据。

叶子节点数据结构位于 /store/store.go

type store struct {
	Root           *node
	WatcherHub     *watcherHub
	CurrentIndex   uint64
	Stats          *Stats
	CurrentVersion int
	ttlKeyHeap     *ttlKeyHeap  // need to recovery manually
	worldLock      sync.RWMutex // stop the world lock
	clock          clockwork.Clock
	readonlySet    types.Set
}

其父节点数据结构位于/store/node.go

type node struct {
	Path string

	CreatedIndex  uint64
	ModifiedIndex uint64

	Parent *node `json:"-"` // should not encode this field! avoid circular dependency.

	ExpireTime time.Time
	Value      string           // for key-value pair
	Children   map[string]*node // for directory

	// A reference to the store this node is attached to.
	store *store
}

其中Path即为key

WAL

内存中的数据格式

 +-------------------------------+
 | F |  Pad  |                   |
 +-----------+  Length           |
 |                               |
 +-------------------------------+
 |             type              |
 +-------------------------------+
 |              CRC              |
 +-------------------------------+
 |             Data              |
 +-------------------------------+
字段名称 含义 占用大小
文本1 文本2 文本3
F 是否存在补齐数据 0:表示没有补齐字段 1:表示存在补齐字段 1bit
Pad 表示补齐长度。在F为1时有效 7bit
Length 表示数据有效负载长度,不包括F、Pad自身长度、补齐字段。 56bit
Type 类型 int64，8字节，有符号
CRC 校验 uint32，4字节，无符号
Data 私有数据

WAL文件数据格式

当我们持久化到文件系统中，数据格式并不是上面介绍，而是grpc格式。

WAL文件以小端序方式存储

启动一个全新etcd，默认会在目录：/var/lib/etcd/default.etcd/member/wal/中生成一个.wal文件。

WAL的定义和创建

定义在wal.go中, WAL日志文件遵循一定的命名规则，由walName()实现，格式为"序号--raft日志索引.wal"。

// 根据seq和index产生wal文件名
func walName(seq, index uint64) string {
	return fmt.Sprintf("%016x-%016x.wal", seq, index)
}

WAL对外暴露的创建接口就是Create()函数

// Create creates a WAL ready for appending records. The given metadata is
// recorded at the head of each WAL file, and can be retrieved(检索) with ReadAll.
func Create(dirpath string, metadata []byte) (*WAL, error) {
	if Exist(dirpath) {
		return nil, os.ErrExist
	}

	// 先在.tmp临时文件上做修改，修改完之后可以直接执行rename,这样起到了原子修改文件的效果
	tmpdirpath := filepath.Clean(dirpath) + ".tmp"
	if fileutil.Exist(tmpdirpath) {
		if err := os.RemoveAll(tmpdirpath); err != nil {
			return nil, err
		}
	}
	if err := fileutil.CreateDirAll(tmpdirpath); err != nil {
		return nil, err
	}

	// dir/filename  ,filename从walName获取   seq-index.wal
	p := filepath.Join(tmpdirpath, walName(0, 0))

	// 对文件上互斥锁
	f, err := fileutil.LockFile(p, os.O_WRONLY|os.O_CREATE, fileutil.PrivateFileMode)
	if err != nil {
		return nil, err
	}

	// 定位到文件末尾
	if _, err = f.Seek(0, io.SeekEnd); err != nil {
		return nil, err
	}

	// 预分配文件，大小为SegmentSizeBytes（64MB）
	if err = fileutil.Preallocate(f.File, SegmentSizeBytes, true); err != nil {
		return nil, err
	}

	// 新建WAL结构
	w := &WAL{
		dir:      dirpath,
		metadata: metadata,// metadata 可为nil
	}

	// 在这个wal文件上创建一个encoder
	w.encoder, err = newFileEncoder(f.File, 0)
	if err != nil {
		return nil, err
	}

	// 把这个上了互斥锁的文件加入到locks数组中
	w.locks = append(w.locks, f)
	if err = w.saveCrc(0); err != nil {
		return nil, err
	}

	// 将metadataType类型的record记录在wal的header处
	if err = w.encoder.encode(&walpb.Record{Type: metadataType, Data: metadata}); err != nil {
		return nil, err
	}

	// 保存空的snapshot
	if err = w.SaveSnapshot(walpb.Snapshot{}); err != nil {
		return nil, err
	}

	// 重命名，之前以.tmp结尾的文件，初始化完成之后重命名，类似原子操作
	if w, err = w.renameWal(tmpdirpath); err != nil {
		return nil, err
	}

	// directory was renamed; sync parent dir to persist rename
	pdir, perr := fileutil.OpenDir(filepath.Dir(w.dir))
	if perr != nil {
		return nil, perr
	}

    // 将上述的所有文件操作进行同步
	if perr = fileutil.Fsync(pdir); perr != nil {
		return nil, perr
	}

	// 关闭目录
	if perr = pdir.Close(); err != nil {
		return nil, perr
	}

	return w, nil
}

其中, SaveSnapshot()是做walpb.Snapshot持久化的, 里面的内容略过, 不过里面有一行代码if err := w.encoder.encode(rec)表示一条Record需要先把序列化后才能持久化，这个是通过encode()函数完成的（encoder.go）

一个Record被序列化之后（这里为JOSN格式），会以一个Frame的格式持久化。Frame首先是一个长度字段（encodeFrameSize()完成,在encoder.go文件），64bit，其中MSB表示这个Frame是否有padding字节，接下来才是真正的序列化后的数据

WAL存储

当raft模块收到一个proposal时就会调用Save方法完成（定义在wal.go）持久化

func (w *WAL) Save(st raftpb.HardState, ents []raftpb.Entry) error {
	w.mu.Lock()  // 上锁
	defer w.mu.Unlock()

	// short cut（捷径）, do not call sync
	// IsEmptyHardState returns true if the given HardState is empty.
	if raft.IsEmptyHardState(st) && len(ents) == 0 {
		return nil
	}

	// 是否需要同步刷新磁盘
	mustSync := raft.MustSync(st, w.state, len(ents))

	// TODO(xiangli): no more reference operator
	// 保存所有日志项
	for i := range ents {
		if err := w.saveEntry(&ents[i]); err != nil {
			return err
		}
	}

	// 持久化HardState, HardState表示服务器当前状态，定义在raft.pb.go，主要包含Term、Vote、Commit
	if err := w.saveState(&st); err != nil {
		return err
	}

	// 获取最后一个LockedFile的大小（已经使用的）
	curOff, err := w.tail().Seek(0, io.SeekCurrent)
	if err != nil {
		return err
	}
	// 如果小于64MB
	if curOff < SegmentSizeBytes {
		if mustSync {
			// 如果需要sync，就执行sync
			return w.sync()
		}
		return nil
	}

	// 否则执行切割（也就是说明，WAL文件是可以超过64MB的）
	return w.cut()
}

snapshot

snapshot比wal大小要小5倍左右，只有CRC和Data两个字段

etcd中对raft snapshot的定义如下（在文件raft.pb.go）：

type Snapshot struct {
	Data             []byte           `protobuf:"bytes,1,opt,name=data" json:"data,omitempty"`
	Metadata         SnapshotMetadata `protobuf:"bytes,2,opt,name=metadata" json:"metadata"`
	XXX_unrecognized []byte           `json:"-"`
}

Metadata则是snaoshot的元信息

// snapshot的元数据
type SnapshotMetadata struct {
      // 最后一次的配置状态
	ConfState        ConfState `protobuf:"bytes,1,opt,name=conf_state,json=confState" json:"conf_state"` 
       // 被快照取代的最后的条目在日志中的索引值(appliedIndex)
	Index            uint64    `protobuf:"varint,2,opt,name=index" json:"index"`
       // 该条目的任期号                     
	Term             uint64    `protobuf:"varint,3,opt,name=term" json:"term"`                           
	XXX_unrecognized []byte    `json:"-"`
}

snapshot持久化使用func (s *Snapshotter) SaveSnap(snapshot raftpb.Snapshot)

过程比较简单, 略去, 里面可以看到snapshot文件的命名规则

// 将raft snapshot序列化后持久化到磁盘
func (s *Snapshotter) save(snapshot *raftpb.Snapshot) error {
	// 产生snapshot的时间
	start := time.Now()
	// snapshot的文件名Term-Index.snap
	fname := fmt.Sprintf("%016x-%016x%s", snapshot.Metadata.Term, snapshot.Metadata.Index, snapSuffix)

动态存储

+--------------------+          +--------------------+
| etcdserver/raft.go |          |   raft/storge.go   |
|                    +<-------->+                    |
|   startNode()      |          | NewMemoryStorge()  |
+---------^----------+          +--------------------+
          |
          |
          |
+---------v----------+          +--------------------+
|     rafe/node.go   |          |     rafe/node.go   |
|                    +<-------->+                    |
|   StartNode()      |          |     newRaft()      |
+---------^----------+          +--------------------+
          |
          |
          |
+---------v----------+
|     raft/node.go   |
|                    |
|        run()       |
+--------------------+

首先调用NewMemoryStorage进行初始化，然后在newRaft()中生成raftLog对象并且调用InitialState()进行状态初始化，最后在node中run方法接收数据。