一篇文章带你学习etcd-wal模块分析

Part1常见数据库日志。传统的数据库日志，比如重做日志（redologs），记录的是被修改的数据。其实就是MySQL中经常提到的WAL技术。它的关键点是先写日志，再写磁盘。redolog是InnoDB引擎独有的；binlog由MySQLServer层实现，所有引擎都可以使用。Redolog是物理日志，记录的是“对某个数据页做了什么修改”；binlog是逻辑日志，记录了这条语句的原始逻辑，比如“ID=2行的c字段加1”。redolog是循环写入的，空间会一直用完；可以附加二进制日志。“追加写入”是指binlog文件写入到一定大小后，会切换到下一个，不会覆盖之前的日志。Redis使用AOF（AppendOnlyFile）。这样做的好处是命令执行成功后会保存下来，不需要事先验证命令是否正确。AOF会将服务器执行的所有写操作保存到日志文件中。服务重启后，会执行这些命令来恢复数据。AOF记录了Redis收到的每条命令，这些命令以文本形式保存。etcd会判断命令是否合法，然后Leader收到提议后，通过Raft模块的事件总线保存要发送给Follower节点的消息和要持久化的日志条目。日志条目是一个封装的条目。etcdserver从Raft模块获取到上述消息和日志条目后，作为leader会向集群中的各个节点广播putproposal消息，同时需要持久化集群leader的任期号、投票信息、提交索引、提案内容到一个WAL（WriteAheadLog）日志文件，用于保证集群的一致性和可恢复性。Part2wal源码分析etcd服务器启动时会根据wal目录是否存在来判断etcd之前是否创建过wal。如果还没有创建wal，etcd会尝试调用wal.Create方法创建wal。否则，wal.Open和wal.ReadAll方法是重新加载之前的wal。逻辑在etcd/etcdserver/server.go的NewServer方法中。当wal存在时，将调用restartNode。下面分别介绍创建wal和加载wal的两种情况。wal的关键对象介绍如下wal日志结构.pngdir：wal文件的保存路径dirFile：dir打开后的目录fd对象metadata：创建wal时传入的字节序列，etcd主要序列化节点id和clusterid相关的信息，每次创建wal文件时都会写入wal的头部。state：wal在append过程中保存的hardState信息。每次raft发送的hardState发生变化，都会及时更新刷新。当wal被切掉时，最新的hardState信息会保存在新wal的head中，etcd重启后会读取上次保存的hardState，在机器宕机或重启时恢复存储中的hardState状态信息。hardState的结构如下：typeHardStatestruct{Termuint64`protobuf:"varint,1,opt,name=term"json:"term"`Voteuint64`protobuf:"varint,2,opt,name=vote"json:"vote"`Commituint64`protobuf:"varint,3,opt,name=commit"json:"commit"`XXX_unrecognized[]byte`json:"-"`}start：记录上次保存快照的元数据信息，主要是索引andtermofthelastlogEntryinthesnapshot，walpb.Snapshot的结构如下：,2,opt,name=term"json:"term"`XXX_unrecognized[]byte`json:"-"`}解码器：负责在读取WAL日志文件时，将protobuf反序列化为一个Record实例。readClose：用于关闭解码器关联的reader，关闭wal读取模式。它是在readALL之后调用这个函数来实现的。enti：最后保存到wal的logEntry的indexencoder：负责将Record实例写入WAL日志文件序列化到protobuf中。size：创建临时文件时预分配空间的大小，默认为64MB（由wal.SegmentSizeBytes指定，也是每个日志文件的大小）。locks：当前WAL实例管理的所有WAL日志文件对应的句柄。fp：负责创建新的临时文件的filePipeline实例。WAL的创建我们先来看看wal.Create()方法。这个方法不仅创建了一个WAL实例，还做了很多初始化工作。大致步骤如下：(1)创建临时目录，创建名为“0-0”的临时目录，WAL日志文件名由两部分组成，一部分是seq（单调递增），另一部分是日志文件中第一条日志记录的索引值。(2)尝试为WAL日志文件预分配磁盘空间。(3)将crcType类型日志记录、metadataType类型日志记录和snapshotType类型日志记录写入WAL日志文件。(4)创建与WAL实例关联的filePipeline实例。(5)将临时目录重命名为WAL.dir字段指定的名称。之所以在重命名之前使用临时目录完成初始化操作，主要是为了让整个初始化过程看起来像一个原子操作。这样上层模块只需要检查wal目录是否存在即可。wal.Create()方法的具体实现如下：ationappearsatomic目录初始化然后重命名的方式主要是为了让整个初始化过程看起来像一个原子操作。tmpdirpath:=filepath.Clean(dirpath)+".tmp"iffileutil.Exist(tmpdirpath){iferr:=os.RemoveAll(tmpdirpath);err!=nil{returnnil,err}}iferr:=fileutil.CreateDirAll(tmpdirpath);err!=nil{returnnil,err}//dir/filename,filenamegetsseq-indexfromwalName.walp:=filepath.Join(tmpdirpath,walName(0,0))//在文件f上创建互斥锁，err:=fileutil.LockFile(p,os.O_WRONLY|os.O_CREATE,fileutil.PrivateFileMode)iferr!=nil{returnnil,err}//定位到文件末尾if_,err=f.Seek(0,io.SeekEnd);err!=nil{returnnil,err}//预分配文件，大小为SegmentSizeBytes(64MB)iferr=fileutil.Preallocate(f.File,SegmentSizeBytes,true);err!=nil{returnnil,err}//新的WAL结构w:=&WAL{dir:dirpath,metadata:metadata,//metadata可以为nil}//在这个wal文件上创建一个编码器w.encoder,err=newFileEncoder(f.File,0)iferr!=nil{returnnil,err}//将互斥文件添加到锁数组w.locks=append(w.locks,f)iferr=w.saveCrc(0);err!=nil{returnnil,err}//记录wal头中metadataType类型的记录iferr=w.encoder.encode(&walpb.Record{Type:metadataType,Data:metadata});err!=nil{returnnil,err}//存空snapshotiferr=w.SaveSnapshot(walpb.Snapshot{});err!=nil{returnnil,err}//之前以.tmp结尾的文件，初始化后重命名ifw,err=w.renameWal(tmpdirpath);err!=nil{returnnil,err}//目录被重命名；syncparentdirtopersistrenamepdir,perr:=fileutil.OpenDir(filepath.Dir(w.dir))ifperr!=nil{returnnil,perr}//将parentdir同步到磁盘ifperr=fileutil.fsync(pdir);perr!=nil{returnnil,perr}returnw,nil}WAL日志文件遵循一定的命名规则，由walName实现，格式为“序号--raftlogindex.wal”//根据seq和index生成wal文件名funcwalName(seq,indexuint64)string{returnfmt.Sprintf("%016x-%016x.wal",seq,index)}在创建过程中，Create函数还写入了WAL日志写入了两份数据，一份是记录元数据，一份是记录快照。WAL中的数据是以Record为单位存储的。结构体定义如下：//walstablestorage中存储了两种类型的消息。它是第一种常见的记录格式typeRecordstruct{Typeint64`protobuf:"varint,1,opt,name=type"json:"type"`Crcuint32`protobuf:"varint,2,opt,name=crc"json:"crc"`Data[]byte`protobuf:"bytes,3,opt,name=data"json:"data,omitempty"`XXX_unrecognized[]byte`json:"-"`}记录类型，其中Type字段表示类型Record，取值可以是如下：const(metadataTypeint64=iota+1entryTypestateTypecrcTypesnapshotType//warnSyncDurationistheamountoftimeallottedtoanfsyncbefore//loggingwarningwarnSyncDuration=time.Second)对应raft中的Snapshot（应用状态机的Snapshot），也会记录一些Snapshot信息在WAL中（但不会记录完整的应用状态机的Snapshot数据），WAL中的Snapshot格式定义如下：//wal中存储的第二种Record消息，snapshottypeSnapshotstruct{Indexuint64`protobuf:"varint,1,opt,name=index"json:"index"`Termuint64`protobuf:"varint,2,opt,name=term"json:"term"`XXX_unrecognized[]byte`json:"-"`}正在保存快照（注意这里的快照在WAL记录类型，不是raft中应用状态机的Snapshot)SaveSnapshot函数中：//Persistentwalpb.Snapshotfunc(w*WAL)SaveSnapshot(ewalpb.Snapshot)error{//pb序列化，此时e可以为空timeb:=pbutil.MustMarshal(&e)w.mu.Lock()deferw.mu.Unlock()//创建snapshotTyperecorderec:=&walpb.Record{Type:snapshotType,Data:b}//持久化到iferrinwal:=w.encoder.encode(rec);err!=nil{returnerr}//updateentionlywhensnapshotisaheodoflastindexifw.entilen(e.buf){//如果超过预分配的buf，使用动态分配数据，err=rec.Marshal()iferr!=nil{返回err}}else{//否则使用分配的bufn,err=rec.MarshalTo(e.buf)iferr!=nil{returnerr}data=e.buf[:n]}lenField,padBytes:=encodeFrameSize(len(data))//先写入recode后的长度iferr=writeUint64(e.bw,lenField,e.uint64buf);err!=nil{returnerr}ifpadBytes!=0{//如果有附加数据（对齐要求）data=append(data,make([]byte,padBytes)...)}//writerecodecontent_,err=e.bw.Write(data)returnerr}从代码中我们可以看到一个Record被序列化后（这里是protobuf格式），它将以Frame格式持久化。Frame首先是一个长度字段（encodeFrameSize完成，在encoder.go文件中），64bit，56bit用来存放数据。其中MSB表示Frame是否有paddingbytes，然后才是真正序列化后的数据。一般一个page是4096字节，对齐到8字节，一个double不会被拆分成两个page。在缓存中，不会拆分：funcencodeFrameSize(dataBytesint)(lenFielduint64,padBytesint){lenField=uint64(dataBytes)//force8bytealignmentsollengthnevergetsatornwritepadBytes=(8-(dataBytes%8))%8ifpadBytes!=0{//lenField高56recordpaddinglengthlenField|=uint64(0x80|padBytes)<<56//最高位1表示包含padding，方便解码时判断}returnlenField,padBytes}。最后，下图展示了包含两个WAL文件的示例图片。filePipeline类型filePipeline采用“饿汉式”，即提前创建一些文件进行备份，可以加快文件的创建速度。filePipeline，负责预先创建日志文件，并为日志文件预先分配空间。在filePipeline中，会启动一个独立的后台goroutine，创建一个以“.tmp”结尾的临时文件。日志文件切换时，临时文件重命名即可直接使用。结构体filePipeline中各个字段的含义如下。dir（字符串类型）：存放临时文件的目录。size（int64类型）：创建临时文件时预分配空间的大小，默认为64MB（由wal.SegmentSizeBytes指定，也是每个日志文件的大小）。count（int类型）：当前filePipeline实例创建的临时文件数量。filec(chan*fileutil.LockedFiletype):新创建的临时文件句柄会通过filec通道返回给WAL实例。errc(chanerrortype)：当创建临时文件时出现异常，将异常传递给errc通道。donec(chanstruct{}type)：当调用filePipeline.Close()时，将关闭donec通道，从而通知filePipeline实例删除最后创建的临时文件。在newFilePipeline()方法中，除了创建一个filePipeline实例外，还会启动一个后台goroutine来执行filePipeline.run()方法。将在后台goroutine中创建一个新的临时文件，并将其句柄传递给filec通道。filePipeline.run()方法的具体实现如下：//filePipelinepipelinesallocatingdiskspacetypefilePipelinestruct{//dirtoputfilesdirstring//sizeoffilestomake,inbytessizeint64//countnumberoffilesgeneratedcountintfilecchan*fileutil.LockedFileerrcchanerrordonecchanstruct{}}funcnewFilePipeline(dirstring,fileSizeint64)*filePipeline{fp:=&filePipeline{dir:dir,size:fileSize,filec:make(chan*fileutil.LockedFile),errc:make(chanerror,1),donec:make(chanstruct{}),}//总是执行预分配gofp.run()returnfp}//Openreturnsafreshfileforwriting.Renamethefilebeforecalling//Openagainortherewillbefilecollisions.func(fp*filePipeline)Open()(f*fileutil.LockedFile,errerror){select{casef=<-fp.filec://从filec通道获取文件描述符并返回caseerr=<-fp.errc://如果创建失败，从errc通道获取并返回}returnf,err}func(fp*filePipeline)Close()error{close(fp.donec)return<-fp.errc//发生错误，关闭donec通道并向errc通道发送错误}func(fp*filePipeline)alloc()(f*fileutil.LockedFile,errerror){//count%2sothisfileisn'tthesameastheonelastpublished//创建的临时文件个数为0或1。fpath:=filepath.Join(fp.dir,fmt.Sprintf("%d.tmp",fp.count%2))//创建临时文件，注意文件的模式和权限。iff,err=fileutil.LockFile(fpath,os.O_CREATE|os.O_WRONLY,fileutil.PrivateFileMode);err!=nil{returnnil,err}//尝试预分配，如果当前文件系统不支持预分配分配空间，那么no会报错。iferr=fileutil.Preallocate(f.File,fp.size,true);err!=nil{f.Close()//如果出现异常，done通道将被关闭returnnil,err}//已经存在的文件数allocatedfp.count++returnf,nil//返回创建的临时文件}//goroutinefunc(fp*filePipeline)run(){deferclose(fp.errc)for{//调用alloc()创建临时文件f,err:=fp.alloc()iferr!=nil{fp.errc<-errreturn}select{casefp.filec<-f://等待消费者取出预先创建的锁定文件case<-fp。donec://关闭时触发，删除上次创建的临时文件os.Remove(f.Name())f.Close()return}}}关注下方二维码。转载本文请联系运维发展故事公众号。