PD源码阅读系列：PD节点启动

李雷，神州数码武汉云基地，目前在研究TiDB的PD模块。

在TiDB生态中，PD作为调度模块，负责整个集群的调度以及保存整个集群的云信息。这篇文章将从PD的启动作为入手点，简单剖析PD节点启动的步骤，了解PD启动的流程，学习PD读取配置、启动日志和监控、设置并启动PD节点服务并通过协程的方式监听退出命令等知识点。

PD简介

Placement Driver (后续以 PD 简称) 是 TiDB 里面全局中心总控节点，它负责整个集群的调度，负责全局 ID 的生成，以及全局时间戳 TSO 的生成等。PD 还保存着整个集群 TiKV 的元信息，负责给 client 提供路由功能。

在架构上面，PD 所有的数据都是通过 TiKV 主动上报获知的。同时，PD 对整个 TiKV 集群的调度等操作，也只会在 TiKV 发送 heartbeat 命令的结果里面返回相关的命令，让 TiKV 自行去处理，而不是主动去给 TiKV 发命令。这样设计上面就非常简单，我们完全可以认为 PD 是一个无状态的服务（当然，PD 仍然会将一些信息持久化到 etcd），所有的操作都是被动触发，即使 PD 挂掉，新选出的 PD leader 也能立刻对外服务，无需考虑任何之前的中间状态。

Why PD?

根据上文，我们了解到PD节点的主要作用在于元数据的存储以及TiKV节点的调度。那么我们不禁要问，为什么需要PD？

当我们只有一个TiKV时，那就根本不需要调度，因为数据只可能存在于这一台机器上，各种客户端也只可能与这一个TiKV节点进行交互。在分布式存储领域，这种情况不可能一直持续下去，因为数据的增量一定会超过这台机器的存储极限。到时我们必须将部分数据迁移到其他机器上去。

了解过TiKV的同学们都知道TiKV使用range的方式将数据进行切分。我们使用Region来表示一个数据range。每个Region都有多个副本Peer。通常为了数据可靠性，我们至少使用三个副本。

最开始系统初始化的时候，我们只有一个Region。当数据量持续增大而超过Region设置的最大Size（64MB）阈值时，Region就会分裂，生成两个新的Region。Region是调度TiKV的基本单位。当我们新增一个TiKV的时候，PD就会将原来TiKV中的一些Region调度到这个新增的TiKV中去。这样就能保证整个数据均衡的分布在TiKV集群上面。因为一个Region通常是64MB，将一个Region从一个TiKV移动到另一个TiKV的过程中，数据量变更其实不大。所以可以直接使用Region的数量来大概的做数据的平衡。

上面我们对TiKV数据的调度做了简单的介绍，但是实际的情况要比这个复杂很多。我们不仅要考虑数据的均衡，也要考虑计算的均衡。这样才能保证整个TiKV集群更快更好的对外提供服务。因为TiKV使用的是Raft一致性算法。Raft有一个强约束就是为了保证线性一致性。所有的读写都必须通过Leader发起。假设现在有三个TiKV，如果几乎所有的Leader都集中在某一个TiKV上，那么会造成这个TiKV成为性能瓶颈，最好的做法是Leader也能够均衡地分布在不同的TiKV上，这样整个系统都能对外提供服务。

总的来说，在分布式存储TiKV中，调度任务及其重要。这关乎系统向外提供服务的质量。我们必须同时考虑存储Storage和计算Leader等资源。所以我们得出一个观点，分布式存储系统是必须要有一个调度模块的。那么，调度模块的实现形式是什么样的？今天我们都知道了在TIDB生态中，有PD作为TiKV集群的调度模块。那么为什么需要单独拿出来作为一个项目？我认为这样做的最大好处就是降低耦合。TiDB生态中，TiDB server负责查询，TiKV负责存储，PD则负责TiKV调度。如果将调度模块写在TiDB或者TiKV里，当TiDB或TiKV扩展节点时，PD也会跟着1:1地扩展。这将会造成一定的性能浪费，因为我们实际上并不一定需要与TIDB或TiKV节点数一样多的PD模块。另外也可以说这是遵守了软件设计原则中的单一职责原则。

PD相关技术

Go：PD完全由Go开发。Go语言简单易用，天生支持高并发。PD源码体积很小，不到5M，但是性能相当不错。
Etcd：分布式系统中最关键的分布式可靠键值存储。PD将Region meta信息持久化在etcd，以保证切换 PD Leader 节点后能快速继续提供 Region 路由服务。
Raft：Etcd实现数据可靠性靠的是分布式一致性算法Raft。
Prometheus：PD集成Prometheus来达到指标监控的目的。每个PD启动时都会配置Prometheus，将系统运行的指标传给Prometheus。
Zap Logger：Go系统库自带的日志包存在一定的性能与功能缺乏。PD集成了由 Ubder 开源的高性能日志框架Zap Logger来提高PD的性能。
TOML：PD配置文件书写语法，由前GitHub CEO， Tom Preston-Werner，于2013年创建。其目标是成为一个小规模的易于使用的语义化配置文件格式。

PD本地编译运行

PD代码开源，可以从github获取：

https://github.com/tikv/pd

源码阅读需要在本地编译运行PD源码。首先需要准备PD所需环境。我本地运行的是Win10 系统，安装了如下依赖：go 1.14.7 + cmake3 + mingw64，使用intellij idea本地编译运行。

这里需要注意的是，我一开始安装的 go 版本为1.15。结果每次本地编译都会报类似于内存泄漏等问题。解决方法是降低 go 的版本。我降到1.14版本后即可正常编译运行PD server。

还有另外一点是PD源码有个ui模块中文件 embedded_assets_rewriter 可能会报错，报错原因是未识别的变量。我在相关论坛提问也没得到回应，于是只能选择注释掉未声明的变量并将相关方法返回nil。处理完这些问题就能跑起PD server来了。

PD源码阅读

今天将解读pd源码的开始部分：启动一个pd server。

阅读从根目录下的cmd/pd-server/main.go开始，由此展开。

一、读取配置

PD的配置信息有三个来源。分别是Config对象默认配置，外部配置文件和命令行参数。它们的优先级分别是命令行参数 > 外部配置文件 > 默认。下面第一块代码就是读取配置的两行代码。config.NewConfig()获取到系统的默认配置。系统默认配置文件在/conf/config.toml里。在Config 的结构体中，可以利用第三方包 http://github.com/BurntSushi/toml 直接读取 toml 格式的配置文件中的值。下面的第二段代码就是config结构体中使用 toml 工具包读取 toml 格式的配置文件中的值来设置属性的默认值的部分代码。通过 toml:"配置文件中属性名"的形式获取到配置的值。从而设置为该属性的默认值。Parse 方法读取命令行参数并将参数设置到config对象中去。

读取配置

cfg := config.NewConfig()err := cfg.Parse(os.Args[1:])

Config结构体部分代码

type Config struct {flagSet *flag.FlagSetVersion bool `json:"-"`ConfigCheck bool `json:"-"`ClientUrls          string`toml:"client-urls" json:"client-urls"`PeerUrls            string`toml:"peer-urls" json:"peer-urls"`AdvertiseClientUrlsstring `toml:"advertise-client-urls"json:"advertise-client-urls"`AdvertisePeerUrls   string`toml:"advertise-peer-urls" json:"advertise-peer-urls"`}

创建默认配置对象cfg时，NewConfig 方法内部还将利用 flagSet 对象对cfg各个属性做属性说明。对于bool类型的属性将调用flagSet的BoolVar方法对其进行说明。具体过程会声明该变量的简称，值以及用处。同理 StringVar 就是对 string 类型的变量做说明的。

下面的示例代码就展示了 BoolVar 和 StringVar 的内部逻辑以及使用这些方法对config对象的属性做说明的过程。我们可以看到使用 StringVar 对属性 configFile 做了说明。其简称为 config 。它的值默认为 "" 。它的用处就是作为配置文件。同理，BoolVar也对bool类型的属性 ConfigCheck 做了说明。说明它是检查配置文件的合规性的。

New Config()

cfg := &Config{}cfg.flagSet =flag.NewFlagSet("pd", flag.ContinueOnError)fs := cfg.flagSetfs.StringVar(&cfg.configFile,"config", "", "config file")fs.BoolVar(&cfg.ConfigCheck,"config-check", false, "check config file validity and exit")func (f *FlagSet) BoolVar(p *bool, namestring, value bool, usage string) {f.Var(newBoolValue(value, p), name, usage)}func (f *FlagSet) StringVar(p *string,name string, value string, usage string) {f.Var(newStringValue(value, p), name, usage)}

以上是默认配置的一些处理操作。接下来讲讲获取外部配置文件和命令行中的配置信息。

PD 在启动时可以携带外部的配置文件对 PD 的属性做配置。具体操作是用命令行启动 PD 时，使用命令行参数 --config 指明外部配置文件的位置。例如 --config "/usr/local/config.toml" 将指定 PD 启动时读取本机文件目录 /usr/local/config.toml 的配置文件。

接着我们在代码层面看一下这个过程：

首先在 main 方法中获取命令行参数信息。这一步骤是通过 go 的os包支持的。通过 os.Args获取命令行参数数组。然后传入到 config 对象的 Parse 方法中。

接着在 Parse 方法中，调用 flagSet 的 Parse 方法将命令行参数都设置到config对象对应的属性上。在随后的代码中将判断 config 对象中 configFile 属性是否非空。因为这个属性默认是空字符串，只有设置了值，才能进行下一步读取指定路径的配置文件。当它的值非空时将调用 configFromFile 方法读取指定目录的配置文件，读取的结果放到 toml.MetaData 对象中。然后将这个对象传入到 config 对象的Adjust 方法中用于调整 config 的各个属性值。

PD 的配置文件描述全面的资料可以参考：

PD 配置文件描述

命令行参数描述可以参考：

PD 配置参数

读取完配置后，Parse 方法将返回err对象以帮助判断Parse过程是否成功。err 如果是 nil，则说明Parse是没有问题的。如果是ErrHelp，则说明输入命令行的是-h或者是-help。输入这个命令说明我只是想查看pd启动时可以携带哪些配置参数而不是直接启动pd。所以在这个case下将调用 exit 方法退出启动程序。除此之外，其他情况就是parse过程错误，输出错误提示信息。

Parse结果检查

switch errors.Cause(err) {
case nil:
case flag.ErrHelp:exit(0)
default:log.Fatal("parse cmd flags error", errs.ZapError(errs.ErrParseFlags))
}

二、启动logger服务并打印PD Server的信息和警告信息

PD使用zap Logger替代go原生的log组件以此来提高整体运行的性能。我们都知道go原生的logger使用起来十分简单。我们通过设置任何io.writer作为日志记录输出并向其发送要写入的日志就行。但是简单归简单，原生logger也有很多不足的地方。例如：仅限基本日志级别、只有一个Print选项、Fatal日志通过调用os.Exit(1)来结束程序、Panic日志在写入日志消息之后抛出一个panic、不提供日志切割的能力、缺乏日志格式化能力等。综合这些原因，pd使用uber开源的日志框架zap logger来替换原生的logger。zap logger有两个优点。其一是提供了结构化日志记录和printf风格的日志记录。其二是它非常的快。关于zap logger高性能的设计思路可以参考它家github地址：

https://github.com/uber-go/zap#performance

下方代码就是PD创建zap logger来替换原生logger的过程：

首先调用cfg对象的 SetupLogger 方法设置cfg的logger和logProps属性。在SetupLogger 方法内部，使用PingCAP自家的log包里的初始化方法 InitLogger获得zap.logger 和ZapProperties对象并将二者分别赋给cfg的 logger 和 logProps属性。接着使用 ReplaceGlobals替换全局的logger。然后刷新缓存，最后使用 InitLogger 初始化zap logger。

logger组件设置启动好之后，打印PD信息和警告。

启动logger：

err = cfg.SetupLogger()
if err == nil {log.ReplaceGlobals(cfg.GetZapLogger(), cfg.GetZapLogProperties())
} else {log.Fatal("initialize logger error", errs.ZapError(err))
}
// Flushing any buffered log entries
defer log.Sync()// The old logger
err = logutil.InitLogger(&cfg.Log)
if err != nil {log.Fatal("initialize logger error", errs.ZapError(err))
}server.LogPDInfo()for _, msg := range cfg.WarningMsgs {log.Warn(msg)
}

三、Prometheus监控

在 main 方法中调用 EnableHandlingTimeHistogram 。在 PD 启动时，会初始化一个默认的 ServerMetrics 对象来记录 PD server服务运行的指标。默认不开启 Histogram metrics 这个指标监控。因为这个指标监控耗费性能较高。在源码的注释中也说明，开启 Histogram metrics 监控可能会耗费较大性能。如果机器性能有限，那么可以选择不开启。

接着就会调用 Push 方法将指标发送到 Prometheus 的推送网关上。具体推送方法是 prometheusPushClinet。在该方法内首先构造推送者对象pusher。pusher的构造使用了建造者模式。首先使用推送的地址和任务初始化pusher，添加了为其添加了收集器以及分组标签。

Prometheus监控：

grpcprometheus.EnableHandlingTimeHistogram()metricutil.Push(&cfg.Metric)
Gatherer(prometheus.DefaultGatherer).Grouping("instance", instanceName())for {err := pusher.Push()if err != nil {log.Error("could not push metrics to Prometheus Pushgateway", errs.ZapError(errs.ErrPrometheusPushMetrics, err))}time.Sleep(interval)}
}

四、动态添加节点

PD使用 PrepareJoinCluster 方法将当前节点 Join指定的集群当中去并且在Join成功后持久化Join配置，当PD节点宕机后重启时，读取本地配置就能快速重新加入集群。

下面简单聊聊从PD节点首次加入到一个集群以及PD停机再次加入集群的情况。

当PD节点首次Join某集群时，我们进入PrepareJoinCluster 方法，携带的参数时cfg，也就是PD的配置对象。当我们想Join某个集群时，首先保证目标集群能够正常工作。在启动PD节点时。命令行携带参数--join="target-urls"，target-urls就是目标集群里任意PD的advertise-clinet-url。PD启动时通过os.Args读取这些额外参数并设置到cfg对象中去。首先要做基本的差错检测，排除Join信息错误的情况。然后尝试读取本地保存的Join信息。我们是第一次Join到一个陌生的集群，这些信息以及目录还没有创建。接下来将创建一个etcd的client，创建时传入Join信息、TLS凭证配置、超时限制等信息。下一步，ListEtcdMember 方法列出目标集群所有的etcd成员。随后判断当前PD节点是否与集群中的节点重名。重名则无法加入集群，直接退出。如果满足条件名字不冲突。随后使用 AddEtcdMenber方法尝试加入集群。结果将返回到类型为*clientv3.MenberAddResponse的对象中。随后再次调用 ListEtcdMenber 获取最新的etcd集群成员信息并对集群情况进行验证，并将最新的集群信息更新到cfg对象中。最后将节点配置信息保存到本地。

当PD停机再次重启时，直接读取本地文件获取集群信息并加入到集群中去。

Join节点：

err = join.PrepareJoinCluster(cfg)
if err != nil {log.Fatal("join meet error", errs.ZapError(err))
}

五、创建并运行PD Server

这一步骤主要做两件事情。第一个就是创建PD Server并运行。第二就是监听退出信号。

首先使用 CreateServer 方法创建Server对象并且传入所需要的参数：上下文对象ctx、配置cfg、服务数组servcieBuilders。接着调用server的Run方法启动Server。在Run方法内，首先会通过协程开启监控。随后开启etcd和Server服务。最后通过Server的startServerLoop方法使得服务处于不断运行的状态而不退出。

另外一个部分就是监听退出信号。通过监听四种信号来判断是否要中止服务。这四种信号及含义如下表所示。监听程序通过协程的方式监听退出信号，一旦监听到退出信号，调用cancle方法即会向ctx对象的Done通道发送消息。Done通道一旦接收到消息运行Server的线程就会退出。接着就会打印退出信息返回退出码。

信号	值	动作	说明
SIGHUP	1	Term	终端控制进程结束(终端连接断开)
SIGHINT	2	Term	用户发送INTR字符(Ctrl+C)触发
SIGTERM	15	Term	结束程序(可以被捕获、阻塞或忽略)
SIGQUIT	3	Core	用户发送QUIT字符(Ctrl+/)触发

创建 PD Server：

ctx, cancel := context.WithCancel(context.Background())
serviceBuilders := []server.HandlerBuilder{api.NewHandler, swaggerserver.NewHandler, autoscaling.NewHandler}
serviceBuilders = append(serviceBuilders, dashboard.GetServiceBuilders()...)
svr, err := server.CreateServer(ctx, cfg, serviceBuilders...)
if err != nil {log.Fatal("create server failed", errs.ZapError(err))
}

总的来说，PD节点的启动会经历读取配置、设置logger、启动prometheus监控、join集群、启动server、监听退出命令后退出等步骤。

我们今天主要了解了PD节点启动的基本步骤，也了解到PD对zap logger和Prometheus等中间件的集成使用。最后学习了使用协程监听退出命令。

整个PD的启动流程用下面流程图表示一下：

本篇文章只是对PD节点启动做的一个粗略的解读，有些地方可能存在错误希望有真知灼见的大神能不吝赐教，指出我的问题，多多交流。