etcd源码阅读与分析(一):raftexample

开始读etcd的源代码,今天首先来看的是 raftexample,这是一个基于 raft 的简单内存KV,希望通过 raftexample 能对 etcd 有一个大概的认识。

首先看一下目录结构:

$ tree -d -L 1 .
.
├── Documentation  # 文档
├── auth  # 认证?还没细看
├── bin  # 编译出来的二进制文件
├── client  # 应该是v2版本的客户端代码
├── clientv3  # 应该是v3版本的客户端代码
├── contrib  # 今天我们要看的raftexample就在这里面
├── default.etcd  # 运行编译好的etcd产生的,忽略之
├── docs  # 文档
├── embed  # 封装了etcd的函数,以便别的程序封装
├── etcdctl  # etcdctl命令,也就是客户端
├── etcdmain  # main.go 调用了这里
├── etcdserver  # 服务端代码
├── functional  # 不知道是干啥的,看起来是用来验证功能的测试套件
├── hack  # 开发者用的,不知道干啥的
├── integration  # 不知道干啥的,忽略
├── lease  # 实现etcd的租约
├── logos
├── mvcc # MVCC存储的实现
├── pkg  # 通用库
├── proxy  # 代理
├── raft  # raft一致性协议的实现
├── scripts  # 各种脚本
├── tests  # 不晓得干啥的,忽略
├── tools  # 一些工具,不知道干啥的,忽略
├── vendor  # go的vendor,忽略
├── version  # 版本信息
└── wal  # Write-Ahead-Log的实现

27 directories

我的 如何阅读源代码 这篇文章里介绍过几种阅读 源代码的方式,今天我们就要用上。

首先,看 main.go 文件:

package main

import (
	"flag"
	"strings"

	"go.etcd.io/etcd/raft/raftpb"
)

func main() {
	cluster := flag.String("cluster", "http://127.0.0.1:9021", "comma separated cluster peers")
	id := flag.Int("id", 1, "node ID")
	kvport := flag.Int("port", 9121, "key-value server port")
	join := flag.Bool("join", false, "join an existing cluster")
	flag.Parse()

	proposeC := make(chan string)
	defer close(proposeC)
	confChangeC := make(chan raftpb.ConfChange)
	defer close(confChangeC)

	// raft provides a commit stream for the proposals from the http api
	var kvs *kvstore
	getSnapshot := func() ([]byte, error) { return kvs.getSnapshot() }
	commitC, errorC, snapshotterReady := newRaftNode(*id, strings.Split(*cluster, ","), *join, getSnapshot, proposeC, confChangeC)

	kvs = newKVStore(<-snapshotterReady, proposeC, commitC, errorC)

	// the key-value http handler will propose updates to raft
	serveHttpKVAPI(kvs, *kvport, confChangeC, errorC)
}

可以看出来,大概就是弄了两个channel,然后呢,新建了一个Raft的Node,新建了一个KV存储,然后就开始提供HTTP服务。

然后跟进去,读 newRaftNode:

func newRaftNode(id int, peers []string, join bool, getSnapshot func() ([]byte, error), proposeC <-chan string,
	confChangeC <-chan raftpb.ConfChange) (<-chan *string, <-chan error, <-chan *snap.Snapshotter) {

	commitC := make(chan *string)
	errorC := make(chan error)

	rc := &raftNode{
		proposeC:    proposeC,
		confChangeC: confChangeC,
		commitC:     commitC,
		errorC:      errorC,
		id:          id,
		peers:       peers,
		join:        join,
		waldir:      fmt.Sprintf("raftexample-%d", id),
		snapdir:     fmt.Sprintf("raftexample-%d-snap", id),
		getSnapshot: getSnapshot,
		snapCount:   defaultSnapshotCount,
		stopc:       make(chan struct{}),
		httpstopc:   make(chan struct{}),
		httpdonec:   make(chan struct{}),

		snapshotterReady: make(chan *snap.Snapshotter, 1),
		// rest of structure populated after WAL replay
	}
	go rc.startRaft() // 启动raft
	return commitC, errorC, rc.snapshotterReady
}

就是实力化了一个 raftNode,然后呢,调用了 raftNode.startRaft 这个方法,那就继续跟进去:

func (rc *raftNode) startRaft() {
	if !fileutil.Exist(rc.snapdir) {
		if err := os.Mkdir(rc.snapdir, 0750); err != nil {
			log.Fatalf("raftexample: cannot create dir for snapshot (%v)", err)
		}
	}
	rc.snapshotter = snap.New(zap.NewExample(), rc.snapdir)
	rc.snapshotterReady <- rc.snapshotter

	oldwal := wal.Exist(rc.waldir)
	rc.wal = rc.replayWAL()

	rpeers := make([]raft.Peer, len(rc.peers))
	for i := range rpeers {
		rpeers[i] = raft.Peer{ID: uint64(i + 1)}
	}
	c := &raft.Config{
		ID:                        uint64(rc.id),
		ElectionTick:              10,
		HeartbeatTick:             1,
		Storage:                   rc.raftStorage,
		MaxSizePerMsg:             1024 * 1024,
		MaxInflightMsgs:           256,
		MaxUncommittedEntriesSize: 1 << 30,
	}

	// 设置 rc.node
	if oldwal {
		rc.node = raft.RestartNode(c)
	} else {
		startPeers := rpeers
		if rc.join {
			startPeers = nil
		}
		rc.node = raft.StartNode(c, startPeers) // 配置节点
	}

	rc.transport = &rafthttp.Transport{
		Logger:      zap.NewExample(),
		ID:          types.ID(rc.id),
		ClusterID:   0x1000,
		Raft:        rc,
		ServerStats: stats.NewServerStats("", ""),
		LeaderStats: stats.NewLeaderStats(strconv.Itoa(rc.id)),
		ErrorC:      make(chan error),
	}

	rc.transport.Start()
	for i := range rc.peers {
		if i+1 != rc.id {
			rc.transport.AddPeer(types.ID(i+1), []string{rc.peers[i]})
		}
	}

	go rc.serveRaft()     // 启动HTTP服务
	go rc.serveChannels() // 开始监听各个channel然后消费
}

可以看出来,这个方法呢,就是先检查是不是有快照,是不是有WAL日志,如果有的话,就恢复到上一个状态,如果没有的话,就新建。 然后调用 raft.RestartNode,这里就是真正启用raft一致性协议的地方了,这里的raft就是最开始我们看的目录里的raft。这里 接下来做的事情就是启动一个transport,这嘎达呢,就跟指定的集群里其他节点通信。然后起一个循环去消费之前建立的的channel里的数据。 可以看到 rc.serveChannels 的代码:

func (rc *raftNode) serveChannels() {
	snap, err := rc.raftStorage.Snapshot()
	if err != nil {
		panic(err)
	}
	rc.confState = snap.Metadata.ConfState
	rc.snapshotIndex = snap.Metadata.Index
	rc.appliedIndex = snap.Metadata.Index

	defer rc.wal.Close()

	ticker := time.NewTicker(100 * time.Millisecond)
	defer ticker.Stop()

	// send proposals over raft
	go func() {
		confChangeCount := uint64(0)

		for rc.proposeC != nil && rc.confChangeC != nil {
			select {
			case prop, ok := <-rc.proposeC:
				log.Printf("received from rc.proposeC: prop: %+v, ok: %t", prop, ok)
				if !ok {
					rc.proposeC = nil
				} else {
					// blocks until accepted by raft state machine
					// 在此处,是kvstore.go里的kvstore
					rc.node.Propose(context.TODO(), []byte(prop))
				}

			case cc, ok := <-rc.confChangeC:
				if !ok {
					rc.confChangeC = nil
				} else {
					confChangeCount++
					cc.ID = confChangeCount
					rc.node.ProposeConfChange(context.TODO(), cc)
				}
			}
		}
		// client closed channel; shutdown raft if not already
		close(rc.stopc)
	}()

	// event loop on raft state machine updates
	for {
		select {
		case <-ticker.C:
			rc.node.Tick()

		// store raft entries to wal, then publish over commit channel
		case rd := <-rc.node.Ready():
			rc.wal.Save(rd.HardState, rd.Entries)
			if !raft.IsEmptySnap(rd.Snapshot) {
				rc.saveSnap(rd.Snapshot)
				rc.raftStorage.ApplySnapshot(rd.Snapshot)
				rc.publishSnapshot(rd.Snapshot)
			}
			rc.raftStorage.Append(rd.Entries)
			rc.transport.Send(rd.Messages)
			if ok := rc.publishEntries(rc.entriesToApply(rd.CommittedEntries)); !ok {
				rc.stop()
				return
			}
			rc.maybeTriggerSnapshot()
			rc.node.Advance()

		case err := <-rc.transport.ErrorC:
			rc.writeError(err)
			return

		case <-rc.stopc:
			rc.stop()
			return
		}
	}
}

这段代码就比较长了,先看第一个 go func() 里的循环,就是监听最开始建立的两个channel,然后分别调用对应的接口,要注意, rc.node 的类型是 raft.Node,这是一个接口。上面说了,实例化的时候,是调用 raft.StartNode 或者 raft.RestartNode, 其返回结果是一个 raft.Node,实际上代码返回的是 raft.node,而 raft.node 实现了 raft.Node 这个接口,会不会有点晕?

所以呢,接下来我们要看看 rc.proposeC 这个channel,rc.confChangeC 我们就不看了,虽然不知道是干啥的,但是呢,从名字我们先 猜测它是用来做配置变更的(实际上就是)。看 rc.proposeC,我们就要看这个channel在哪些地方用到了,也就是说,哪里有生产者, 哪里有消费者。搜索一下:

$ ack proposeC
raftexample_test.go
29:	proposeC    []chan string
44:		proposeC:    make([]chan string, len(peers)),
51:		clus.proposeC[i] = make(chan string, 1)
53:		clus.commitC[i], clus.errorC[i], _ = newRaftNode(i+1, clus.peers, false, nil, clus.proposeC[i], clus.confChangeC[i])
73:		close(clus.proposeC[i])
123:		}(clus.proposeC[i], clus.commitC[i], clus.errorC[i])
126:		go func(i int) { clus.proposeC[i] <- "foo" }(i)
151:		clus.proposeC[0] <- "foo"
152:		clus.proposeC[0] <- "bar"

kvstore.go
29:	proposeC    chan<- string // channel for proposing updates
40:func newKVStore(snapshotter *snap.Snapshotter, proposeC chan<- string, commitC <-chan *string, errorC <-chan error) *kvstore {
41:	s := &kvstore{proposeC: proposeC, kvStore: make(map[string]string), snapshotter: snapshotter}
61:	log.Printf("s.proposeC <- %s", buf.String())
62:	s.proposeC <- buf.String()

raft.go
42:	proposeC    <-chan string            // proposed messages (k,v)
80:// current), then new log entries. To shutdown, close proposeC and read errorC.
81:func newRaftNode(id int, peers []string, join bool, getSnapshot func() ([]byte, error), proposeC <-chan string,
88:		proposeC:    proposeC,
401:		for rc.proposeC != nil && rc.confChangeC != nil {
403:			case prop, ok := <-rc.proposeC:
404:				log.Printf("received from rc.proposeC: prop: %+v, ok: %t", prop, ok)
406:					rc.proposeC = nil

main.go
31:	proposeC := make(chan string)
32:	defer close(proposeC)
39:	commitC, errorC, snapshotterReady := newRaftNode(*id, strings.Split(*cluster, ","), *join, getSnapshot, proposeC, confChangeC)
41:	kvs = newKVStore(<-snapshotterReady, proposeC, commitC, errorC)

可以看出来,kvstore.go 的62行应该是生产者,而 raft.go 的403行应该是消费者。基于这种假设,我们要去验证一下,所以我加了几行日志, 那就运行一下:

$ curl -v -L http://127.0.0.1:9121/my-key -XPUT -d hello

下图是服务端的日志输出:

raft example

可以看出来,顺序是先调用了 s.proposeC <- 然后 received from rc.proposeC,然后raft对消息进行处理,把它还原成函数调用,就是:

  • func (s *kvstore) Propose(k string, v string)
  • func (rc *raftNode) serveChannels()
  • func (n *node) Propose(ctx context.Context, data []byte) error,这里的node是 etcd/raft/node.go 里的结构体 type node struct

为啥呢,我们要和raftexample里的代码对应上。继续看 raftexample/main.go

commitC, errorC, snapshotterReady := newRaftNode(*id, strings.Split(*cluster, ","), *join, getSnapshot, proposeC, confChangeC)

kvs = newKVStore(<-snapshotterReady, proposeC, commitC, errorC)

// the key-value http handler will propose updates to raft
serveHttpKVAPI(kvs, *kvport, confChangeC, errorC)

可以看到,main.go 最后其实是执行了 serveHttpKVAPI,然后跳进去一看,会发现,调用了 httpKVAPI,我们知道,go的HTTP服务 代码满足 ServeHTTP 这个接口就可以了,如果你不知道的话,说明没有读过 net/http 的代码,那么你现在知道了。所以直接跳过去看 httpKVAPIServeHTTP 这个方法,可以看到:

type httpKVAPI struct {
	store       *kvstore
	confChangeC chan<- raftpb.ConfChange
}

func (h *httpKVAPI) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	key := r.RequestURI
	switch {
	case r.Method == "PUT":
		v, err := ioutil.ReadAll(r.Body)
		if err != nil {
			log.Printf("Failed to read on PUT (%v)\n", err)
			http.Error(w, "Failed on PUT", http.StatusBadRequest)
			return
		}

		h.store.Propose(key, string(v))

		// Optimistic-- no waiting for ack from raft. Value is not yet
		// committed so a subsequent GET on the key may return old value
		w.WriteHeader(http.StatusNoContent)

h.store.Propose(key, string(v)) 这里就是入口了。按照我们之前说的顺序,h.store.Propose(key, string(v)) 会调用 s.proposeC <- buf.String(), 然后这个时候就会唤醒 serveChannels 里的 case prop, ok := <-rc.proposeC,然后调用 rc.node.Propose(context.TODO(), []byte(prop)), 接下来就会调用 etcd/raft/node.go 里的 func (n *node) Propose(ctx context.Context, data []byte) error 这个函数。

好了,到此为止我们就知道etcd大概是怎么一个工作法,这篇博客到此结束。

接下来我们会继续探索真正的etcd里的各个细节。


etcd源码阅读与分析系列文章


更多文章
  • Go的custom import path
  • 如何挖掘二级子域名?
  • Go Module 简明教程
  • 写了一个Telegram Bot:自动化分享高质量内容
  • Vim打开很慢,怎么找出最慢的插件?怎么解决?
  • 为什么我选择放弃运营微信公众号?
  • ArchLinux 怎么降级 package ?
  • Web后端工程师进阶指南(2018)
  • How to implement fork syscall in Golang?
  • macOS ansible 遇到 rsync: --chown=www-data: unknown option
  • 关于运营的思考-运营要怎么做?
  • Python中实现单例模式的n种方式和原理
  • Golang defer中修改返回值
  • Python dataclass 源码阅读与分析
  • gRPC-gateway 源码阅读与分析