Etcd 日志复制流程概述

// etcd/raft/log.go
func (r *raft) appendEntry(es ...pb.Entry) (accepted bool) {
	li := r.raftLog.lastIndex()
	for i := range es {
		es[i].Term = r.Term
		es[i].Index = li + 1 + uint64(i)
	}
	// ...
	return true
}

// 根据 unstable 里的内容确定 lastIndex
func (l *raftLog) lastIndex() uint64 {
	if i, ok := l.unstable.maybeLastIndex(); ok {
		return i
	}
	i, err := l.storage.LastIndex()
	if err != nil {
		panic(err) // TODO(bdarnell)
	}
	return i
}

// etcd/raft/raft.go
func (r *raft) maybeSendAppend(to uint64, sendIfEmpty bool) bool {
	m := pb.Message{}
	// To 表示消息接收者
	m.To = to
	// 已经发送的最大消息的 term
	term, errt := r.raftLog.term(pr.Next - 1)
	// 取所有还没有发送的消息，这个 maxMsgSize 默认是 1024*1024，足够大
	ents, erre := r.raftLog.entries(pr.Next, r.maxMsgSize)
		//...

		// 消息的类型为 MsgApp
		m.Type = pb.MsgApp
		// 消息的 Index 为，已发送的最大的 index
		m.Index = pr.Next - 1
		m.LogTerm = term
		// 还没有发送的 entry
		m.Entries = ents
		// 当前leader，已经 commit 的 索引
		m.Commit = r.raftLog.committed
	}
	r.send(m)
	return true
}

// etcd/raft/node.go
func newReady(r *raft, prevSoftSt *SoftState, prevHardSt pb.HardState) Ready {
	rd := Ready{
		// 保存在临时存储中的 Entry
		Entries:          r.raftLog.unstableEntries(),
		CommittedEntries: r.raftLog.nextEnts(),
		// 待发送的消息
		Messages:         r.msgs,
	}
	// ...
	return rd
}

// etcd/etcdserver/raft.go
func (r *raftNode) start(rh *raftReadyHandler) {
	// ...
	// 发送消息
	if islead {
		r.transport.Send(r.processMessages(rd.Messages))
	}
	// 将 unstable 存储中的 Entry 写 wal 
	if err := r.storage.Save(rd.HardState, rd.Entries); err != nil {
		// ...
	}
}

func (r *raft) handleAppendEntries(m pb.Message) {
	// follower 记录的 committed index 已经大于 msg 发送过来的最小 index，旧消息
	if m.Index < r.raftLog.committed {
		r.send(pb.Message{To: m.From, Type: pb.MsgAppResp, Index: r.raftLog.committed})
		return
	}
	// 这里将消息拷贝到 unstable 存储，返回的 mlastIndex 是完成拷贝的最大的 Entry
	if mlastIndex, ok := r.raftLog.maybeAppend(m.Index, m.LogTerm, m.Commit, m.Entries...); ok {
		r.send(pb.Message{To: m.From, Type: pb.MsgAppResp, Index: mlastIndex})
	} else {
		// 处理失败...
	}
}

func (l *raftLog) maybeAppend(index, logTerm, committed uint64, ents ...pb.Entry) (lastnewi uint64, ok bool) {
	if l.matchTerm(index, logTerm) {
		lastnewi = index + uint64(len(ents))
		ci := l.findConflict(ents)
		switch {
		case ci == 0:
		case ci <= l.committed:
			l.logger.Panicf("entry %d conflict with committed entry [committed(%d)]", ci, l.committed)
		default:
			offset := index + 1
			// 这里是添加到本地 unstable 存储
			l.append(ents[ci-offset:]...)
		}
		// 对于 follower 来说，committed 标志是 min{leader的commit序号，已复制的 entry 最大序号} 
		l.commitTo(min(committed, lastnewi))
		return lastnewi, true
	}
	return 0, false
}

type Storage interface {
	InitialState() (pb.HardState, pb.ConfState, error)
	// 返回位于区间 [lo, hi) 的 Entry，最多返回 maxSize 哥
	Entries(lo, hi, maxSize uint64) ([]pb.Entry, error)
	// 返回 term i 的任期
	Term(i uint64) (uint64, error)
	LastIndex() (uint64, error)
	FirstIndex() (uint64, error)
	// 返回最近的快照
	Snapshot() (pb.Snapshot, error)
}

// etcd/etcdserver/raft.go
r.raftStorage.Append(rd.Entries)

func (ms *MemoryStorage) Append(entries []pb.Entry) error {
	// MemoryStorage 中的第一个 Index
	first := ms.firstIndex()
	// 新增 Entry 的 last
	last := entries[0].Index + uint64(len(entries)) - 1
	if last < first {
		return nil
	}
	// truncate compacted entries
	// 这个成立的条件是保证 Entry 是连续的
	if first > entries[0].Index {
		entries = entries[first-entries[0].Index:]
	}
	offset := entries[0].Index - ms.ents[0].Index
	switch {
	case uint64(len(ms.ents)) > offset:
		ms.ents = append([]pb.Entry{}, ms.ents[:offset]...)
		ms.ents = append(ms.ents, entries...)
	case uint64(len(ms.ents)) == offset:
		ms.ents = append(ms.ents, entries...)
		// ... 
	}
	return nil
}

// etcd/raft/raft.go
func stepLeader(r *raft, m pb.Message) error {
	// ...
	pr := r.prs.Progress[m.From]
	if pr == nil {
		r.logger.Debugf("%x no progress available for %x", r.id, m.From)
		return nil
	}
	switch m.Type {
	case pb.MsgAppResp:
		pr.RecentActive = true
		if m.Reject {
			// handle rejected
		} else {
			oldPaused := pr.IsPaused()
			// 这里的 MaybeUpdate 返回 true，如果 leader 记录的 match 小于 follower 已经复制的 index
			// 并且同时更新 follower 的 match
			if pr.MaybeUpdate(m.Index) { 
				// 计算最大的 committed index，如果有变化，则 bcastAppend
				if r.maybeCommit() {
					r.bcastAppend()
				} else if oldPaused {
					r.sendAppend(m.From)
				}
			}
		}
		// ... 
	}
	return nil
}

// etcd/etcdserver/raft.go
func (r *raftNode) start(rh *raftReadyHandler) {
	go func() {
		for {
			select {
			case <-r.ticker.C:
			case rd := <-r.Ready():
				notifyc := make(chan struct{}, 1)
				ap := apply{
					// 获取 committedEntries
					entries:  rd.CommittedEntries,
					snapshot: rd.Snapshot,
					notifyc:  notifyc,
				}
				select {
				case r.applyc <- ap: // 将 apply 结构体写入 channel，消费者取 apply 进行应用
				case <-r.stopped:
					return
				}
			}
		}
	}
}
// etcd/raft/node.go
func newReady(r *raft, prevSoftSt *SoftState, prevHardSt pb.HardState) Ready {
	rd := Ready{
		// 保存在临时存储中的 Entry，待写 wal
		Entries:          r.raftLog.unstableEntries(),
		// 已经 committed，待 apply 的 Entries
		CommittedEntries: r.raftLog.nextEnts(),
		// 待发送的消息
		Messages:         r.msgs,
	}
	// ...
	return rd
}

func (l *raftLog) nextEnts() (ents []pb.Entry) {
	off := max(l.applied+1, l.firstIndex())
	if l.committed+1 > off {
		ents, err := l.slice(off, l.committed+1, l.maxNextEntsSize)
		if err != nil {
			l.logger.Panicf("unexpected error when getting unapplied entries (%v)", err)
		}
		return ents
	}
	return nil
}

func (l *raftLog) slice(lo, hi, maxSize uint64) ([]pb.Entry, error) {
	err := l.mustCheckOutOfBounds(lo, hi)

	var ents []pb.Entry
	if lo < l.unstable.offset {
		// 从 MemoryStorage 中取
		storedEnts, err := l.storage.Entries(lo, min(hi, l.unstable.offset), maxSize)
		ents = storedEnts
	}
	if hi > l.unstable.offset {
		// 从 unstable 中取
		unstable := l.unstable.slice(max(lo, l.unstable.offset), hi)
	}
	return limitSize(ents, maxSize), nil
}

// etcd/etcdserver/server.go
func (s *EtcdServer) run() {
	sched := schedule.NewFIFOScheduler()
	// 启动 raftnode
	s.r.start(rh)
	// ...
	for {
		select {
		case ap := <-s.r.apply():
			// raftNode.applyc channel 中读取待 apply 的消息，进行 apply
			f := func(context.Context) { s.applyAll(&ep, &ap) }
			sched.Schedule(f)
		}
	}
}

// etcd/pkg/wait/wait.go
type Wait interface {
	// Register waits returns a chan that waits on the given ID.
	// The chan will be triggered when Trigger is called with
	// the same ID.
	Register(id uint64) <-chan interface{}
	// Trigger triggers the waiting chans with the given ID.
	Trigger(id uint64, x interface{})
	IsRegistered(id uint64) bool
}