Fabric 1.4 源码分析peer节点启动
peer模块采用cobra库来实现cli命令。
Cobra提供简单的接口来创建强大的现代化CLI接口，比如git与go工具。Cobra同时也是一个程序, 用于创建CLI程序
peer支持的命令如下所示：
Usage:
peer [command]
Available Commands:
chaincode Operate a chaincode: install|instantiate|invoke|package|query|signpackage|upgrade|list.
channel Operate a channel: create|fetch|join|list|update|signconfigtx|getinfo.
help Help about any command
logging Log levels: getlevel|setlevel|revertlevels.
node Operate a peer node: start|status.
version Print fabric peer version.
Flags:
-h, --help help for peer
--logging-level string Default logging level and overrides, see core.yaml for full syntax通过peer 的docker-compose文件可知，peer启动命令为peer node start。从下列代码可知，peer启动时调用serve()接口。
var nodeStartCmd = &cobra.Command{
Use: "start",
Short: "Starts the node.",
Long: `Starts a node that interacts with the network.`,
RunE: func(cmd *cobra.Command, args []string) error {
if len(args) != 0 {
return fmt.Errorf("trailing args detected")
}
// Parsing of the command line is done so silence cmd usage
cmd.SilenceUsage = true
return serve(args)
},
}接下来深入分析serve()接口。
func serve(args []string) error {
// currently the peer only works with the standard MSP
// because in certain scenarios the MSP has to make sure
// that from a single credential you only have a single 'identity'.
// Idemix does not support this *YET* but it can be easily
// fixed to support it. For now, we just make sure that
// the peer only comes up with the standard MSP
// 当前peer启动时只支持标准MSP即Fabric。
mspType := mgmt.GetLocalMSP().GetType()
if mspType != msp.FABRIC {
panic("Unsupported msp type " + msp.ProviderTypeToString(mspType))
}
// Trace RPCs with the golang.org/x/net/trace package. This was moved out of
// the deliver service connection factory as it has process wide implications
// and was racy with respect to initialization of gRPC clients and servers.
grpc.EnableTracing = true
logger.Infof("Starting %s", version.GetInfo())
//startup aclmgmt with default ACL providers (resource based and default 1.0 policies based).
//Users can pass in their own ACLProvider to RegisterACLProvider (currently unit tests do this)
// 创建ACL提供者 ACL访问控制列表
aclProvider := aclmgmt.NewACLProvider(
aclmgmt.ResourceGetter(peer.GetStableChannelConfig),
)
// 平台注册
pr := platforms.NewRegistry(
&golang.Platform{},
&node.Platform{},
&java.Platform{},
&car.Platform{},
)
// 定义部署链码提供者
deployedCCInfoProvider := &lscc.DeployedCCInfoProvider{}DeployedCCInfoProvider实现了DeployedChaincodeInfoProvider。
DeployedChaincodeInfoProvider是ledger用于构建集合配置历史记录的依赖项
LSCC模块应该为这个依赖项提供实现
type DeployedChaincodeInfoProvider interface {
Namespaces() []string //命名空间
UpdatedChaincodes(stateUpdates map[string][]*kvrwset.KVWrite) ([]*ChaincodeLifecycleInfo, error) // 保存更新的链码
ChaincodeInfo(chaincodeName string, qe SimpleQueryExecutor) (*DeployedChaincodeInfo, error) // 保存链码信息
CollectionInfo(chaincodeName, collectionName string, qe SimpleQueryExecutor) (*common.StaticCollectionConfig, error) // 链码集合信息
}初始化账本资源ledgermgmt.Initialize
// 获取通道MSP管理员。如果不存在则创建
identityDeserializerFactory := func(chainID string) msp.IdentityDeserializer {
return mgmt.GetManagerForChain(chainID)
}
// peer 初始化
// 保存 peer 一些基本信息 ListenAddress TLS
opsSystem := newOperationsSystem()
// 监听 ListenAddress
err := opsSystem.Start()
if err != nil {
return errors.WithMessage(err, "failed to initialize operations subystems")
}
defer opsSystem.Stop()
metricsProvider := opsSystem.Provider
logObserver := floggingmetrics.NewObserver(metricsProvider)
flogging.Global.SetObserver(logObserver)
// 实例化私密数据成员membershipInfoProvider 用来判断peer是否在某个私密数据的集合中
membershipInfoProvider := privdata.NewMembershipInfoProvider(createSelfSignedData(), identityDeserializerFactory)
//initialize resource management exit
// 初始化账本资源 将前面实例化的对象都进行赋值
ledgermgmt.Initialize(
&ledgermgmt.Initializer{
CustomTxProcessors: peer.ConfigTxProcessors,
PlatformRegistry: pr,
DeployedChaincodeInfoProvider: deployedCCInfoProvider,
MembershipInfoProvider: membershipInfoProvider,
MetricsProvider: metricsProvider,
HealthCheckRegistry: opsSystem,
},
)初始化peer GRPC服务配置
// Parameter overrides must be processed before any parameters are
// cached. Failures to cache cause the server to terminate immediately.
// 判断链码是否时开发者模式
if chaincodeDevMode {
logger.Info("Running in chaincode development mode")
logger.Info("Disable loading validity system chaincode")
viper.Set("chaincode.mode", chaincode.DevModeUserRunsChaincode)
}
// 缓存peer地址getLocalAddress address:port
if err := peer.CacheConfiguration(); err != nil {
return err
}
// 获取peer endpoint，没有则调用CacheConfiguration接口
peerEndpoint, err := peer.GetPeerEndpoint()
if err != nil {
err = fmt.Errorf("Failed to get Peer Endpoint: %s", err)
return err
}
// 获取peer Host
peerHost, _, err := net.SplitHostPort(peerEndpoint.Address)
if err != nil {
return fmt.Errorf("peer address is not in the format of host:port: %v", err)
}
listenAddr := viper.GetString("peer.listenAddress")
// 获取peer grpc相关配置 主要是TLS设置和心跳设置
serverConfig, err := peer.GetServerConfig()
if err != nil {
logger.Fatalf("Error loading secure config for peer (%s)", err)
}
// 设置GRPC最大并发2500
throttle := comm.NewThrottle(grpcMaxConcurrency)
// GRPC server的一些配置
serverConfig.Logger = flogging.MustGetLogger("core.comm").With("server", "PeerServer")
serverConfig.MetricsProvider = metricsProvider
serverConfig.UnaryInterceptors = append(
serverConfig.UnaryInterceptors,
grpcmetrics.UnaryServerInterceptor(grpcmetrics.NewUnaryMetrics(metricsProvider)),
grpclogging.UnaryServerInterceptor(flogging.MustGetLogger("comm.grpc.server").Zap()),
throttle.UnaryServerIntercptor,
)
serverConfig.StreamInterceptors = append(
serverConfig.StreamInterceptors,
grpcmetrics.StreamServerInterceptor(grpcmetrics.NewStreamMetrics(metricsProvider)),
grpclogging.StreamServerInterceptor(flogging.MustGetLogger("comm.grpc.server").Zap()),
throttle.StreamServerInterceptor,
)将GRPC相关配置及Address传入创建GRPC服务器
peerServer, err := peer.NewPeerServer(listenAddr, serverConfig)
if err != nil {
logger.Fatalf("Failed to create peer server (%s)", err)
}TLS及策略相关
// TLS相关配置
if serverConfig.SecOpts.UseTLS {
logger.Info("Starting peer with TLS enabled")
// set up credential support
cs := comm.GetCredentialSupport()
roots, err := peer.GetServerRootCAs()
if err != nil {
logger.Fatalf("Failed to set TLS server root CAs: %s", err)
}
cs.ServerRootCAs = roots
// set the cert to use if client auth is requested by remote endpoints
clientCert, err := peer.GetClientCertificate()
if err != nil {
logger.Fatalf("Failed to set TLS client certificate: %s", err)
}
comm.GetCredentialSupport().SetClientCertificate(clientCert)
}
mutualTLS := serverConfig.SecOpts.UseTLS && serverConfig.SecOpts.RequireClientCert
// 策略校验
policyCheckerProvider := func(resourceName string) deliver.PolicyCheckerFunc {
return func(env *cb.Envelope, channelID string) error {
return aclProvider.CheckACL(resourceName, channelID, env)
}
}创建deliver server 传输区块及过滤区块
abServer := peer.NewDeliverEventsServer(mutualTLS, policyCheckerProvider, &peer.DeliverChainManager{}, metricsProvider)
pb.RegisterDeliverServer(peerServer.Server(), abServer)初始化链码服务
startChaincodeServer将完成与链代码相关的初始化，包括：
1）设置本地链代码安装路径
2）创建特定链代码的CA
3）启动特定链代码的gRPC监听服务
// Initialize chaincode service
chaincodeSupport, ccp, sccp, packageProvider := startChaincodeServer(peerHost, aclProvider, pr, opsSystem)
logger.Debugf("Running peer")注册背书服务，gossip组件初始化等操作
// Start the Admin server
startAdminServer(listenAddr, peerServer.Server(), metricsProvider)
privDataDist := func(channel string, txID string, privateData *transientstore.TxPvtReadWriteSetWithConfigInfo, blkHt uint64) error {
// 分发私有数据到其他节点
return service.GetGossipService().DistributePrivateData(channel, txID, privateData, blkHt)
}
// 获取本地签名
signingIdentity := mgmt.GetLocalSigningIdentityOrPanic()
serializedIdentity, err := signingIdentity.Serialize()
if err != nil {
logger.Panicf("Failed serializing self identity: %v", err)
}
libConf := library.Config{}
if err = viperutil.EnhancedExactUnmarshalKey("peer.handlers", &libConf); err != nil {
return errors.WithMessage(err, "could not load YAML config")
}
reg := library.InitRegistry(libConf)
// 背书 验证相关配置
authFilters := reg.Lookup(library.Auth).([]authHandler.Filter)
endorserSupport := &endorser.SupportImpl{
SignerSupport: signingIdentity,
Peer: peer.Default,
PeerSupport: peer.DefaultSupport,
ChaincodeSupport: chaincodeSupport,
SysCCProvider: sccp,
ACLProvider: aclProvider,
}
endorsementPluginsByName := reg.Lookup(library.Endorsement).(map[string]endorsement2.PluginFactory)
validationPluginsByName := reg.Lookup(library.Validation).(map[string]validation.PluginFactory)
signingIdentityFetcher := (endorsement3.SigningIdentityFetcher)(endorserSupport)
channelStateRetriever := endorser.ChannelStateRetriever(endorserSupport)
pluginMapper := endorser.MapBasedPluginMapper(endorsementPluginsByName)
pluginEndorser := endorser.NewPluginEndorser(&endorser.PluginSupport{
ChannelStateRetriever: channelStateRetriever,
TransientStoreRetriever: peer.TransientStoreFactory,
PluginMapper: pluginMapper,
SigningIdentityFetcher: signingIdentityFetcher,
})
endorserSupport.PluginEndorser = pluginEndorser
serverEndorser := endorser.NewEndorserServer(privDataDist, endorserSupport, pr, metricsProvider)
auth := authHandler.ChainFilters(serverEndorser, authFilters...)
// Register the Endorser server
pb.RegisterEndorserServer(peerServer.Server(), auth)
policyMgr := peer.NewChannelPolicyManagerGetter()
// Initialize gossip component
err = initGossipService(policyMgr, metricsProvider, peerServer, serializedIdentity, peerEndpoint.Address)
if err != nil {
return err
}
defer service.GetGossipService().Stop()
// register prover grpc service
// FAB-12971 disable prover service before v1.4 cut. Will uncomment after v1.4 cut
// err = registerProverService(peerServer, aclProvider, signingIdentity)
// if err != nil {
// return err
// }初始化系统链码。
// initialize system chaincodes
// deploy system chaincodes
// 部署系统链码
sccp.DeploySysCCs("", ccp)
logger.Infof("Deployed system chaincodes")
// 查看已经安装等链码
installedCCs := func() ([]ccdef.InstalledChaincode, error) {
return packageProvider.ListInstalledChaincodes()
}
// 创建链码的生命周期
lifecycle, err := cc.NewLifeCycle(cc.Enumerate(installedCCs))
if err != nil {
logger.Panicf("Failed creating lifecycle: +%v", err)
}
// HandleMetadataUpdate在链代码生命周期更改发生变化时触发
onUpdate := cc.HandleMetadataUpdate(func(channel string, chaincodes ccdef.MetadataSet) {
// 更新链码
service.GetGossipService().UpdateChaincodes(chaincodes.AsChaincodes(), gossipcommon.ChainID(channel))
})
// 监听器 监听链码更新
lifecycle.AddListener(onUpdate)通道相关配置
// this brings up all the channels
peer.Initialize(func(cid string) {
logger.Debugf("Deploying system CC, for channel <%s>", cid)
sccp.DeploySysCCs(cid, ccp)
sub, err := lifecycle.NewChannelSubscription(cid, cc.QueryCreatorFunc(func() (cc.Query, error) {
// 返回通道的查询器
return peer.GetLedger(cid).NewQueryExecutor()
}))
if err != nil {
logger.Panicf("Failed subscribing to chaincode lifecycle updates")
}
// 注册该通道ChaincodeLifecycleEventListener
cceventmgmt.GetMgr().Register(cid, sub)
}, ccp, sccp, txvalidator.MapBasedPluginMapper(validationPluginsByName),
pr, deployedCCInfoProvider, membershipInfoProvider, metricsProvider)
// 获取peer一些配置
if viper.GetBool("peer.discovery.enabled") {
registerDiscoveryService(peerServer, policyMgr, lifecycle)
}
networkID := viper.GetString("peer.networkId")
logger.Infof("Starting peer with ID=[%s], network ID=[%s], address=[%s]", peerEndpoint.Id, networkID, peerEndpoint.Address)
// Get configuration before starting go routines to avoid
// racing in tests
profileEnabled := viper.GetBool("peer.profile.enabled")
profileListenAddress := viper.GetString("peer.profile.listenAddress")
// Start the grpc server. Done in a goroutine so we can deploy the
// genesis block if needed.
serve := make(chan error)
// 开启peer grpc服务
go func() {
var grpcErr error
if grpcErr = peerServer.Start(); grpcErr != nil {
grpcErr = fmt.Errorf("grpc server exited with error: %s", grpcErr)
} else {
logger.Info("peer server exited")
}
serve <- grpcErr
}()
// Start profiling http endpoint if enabled
if profileEnabled {
go func() {
logger.Infof("Starting profiling server with listenAddress = %s", profileListenAddress)
if profileErr := http.ListenAndServe(profileListenAddress, nil); profileErr != nil {
logger.Errorf("Error starting profiler: %s", profileErr)
}
}()
}
go handleSignals(addPlatformSignals(map[os.Signal]func(){
syscall.SIGINT: func() { serve <- nil },
syscall.SIGTERM: func() { serve <- nil },
}))
// peer启动区块归档任务
if ledgerconfig.IsDataDumpEnabled() {
logger.Debugf("DataDump:{DumpDir:%s,LoadDir:%s,MaxFileLimit:%d,DumpCron:%v,DumpInterval:%d,LoadRetryTimes:%d}",
ledgerconfig.GetDataDumpPath(), ledgerconfig.GetDataLoadPath(), ledgerconfig.GetDataDumpFileLimit(),
ledgerconfig.GetDataDumpCron(), ledgerconfig.GetDataDumpInterval(), ledgerconfig.GetDataLoadRetryTimes())
go func() {
cronList := ledgerconfig.GetDataDumpCron()
if cronList != nil && len(cronList) > 0 {
cronTask := cron.New()
cronTask.Start()
for _, crontab := range cronList {
logger.Debugf("Crontab addFunc for %s", crontab)
err := cronTask.AddFunc(crontab, func() {
chainInfoArray := peer.GetChannelsInfo()
for _, chainInfo := range chainInfoArray {
chainId := chainInfo.ChannelId
l := peer.GetLedger(chainId)
if err := l.DataDump(datadump.DumpForCronTab); err != nil {
logger.Errorf("Failed to datadump for [%s]", err)
}
}
})
if err != nil {
logger.Errorf("Failed to add crontab task for %s", err)
}
}
}
}()
}
logger.Infof("Started peer with ID=[%s], network ID=[%s], address=[%s]", peerEndpoint.Id, networkID, peerEndpoint.Address)
if viper.GetBool("peer.enBlkrouter") {
go func() {
startBlockServer()
}()
}
// Block until grpc server exits
// 阻塞 直到grpc服务退出
return <-serve
}到这里Peer节点已经启动完成了，过程还是很复杂的，这里总结一下整体的过程：
-
首先就是读取配置信息，创建Cache结构，以及检测其他Peer节点的信息。
CacheConfiguration(),主要保存其他Peer节点的相关信息。 -
创建
PeerServer。peerServer, err := peer.NewPeerServer(listenAddr, serverConfig) -
创建
DeliverEventsServerabServer := peer.NewDeliverEventsServer(mutualTLS, policyCheckerProvider, &peer.DeliverChainManager{}, metricsProvider)pb.RegisterDeliverServer(peerServer.Server(), abServer)-
fabric/core/peer/deliverevents.go,该服务主要用于区块的交付与过滤,主要方法：Deliver(),DeliverFiltered()
-
启动
ChaincodeServerchaincodeSupport, ccp, sccp, packageProvider := startChaincodeServer(peerHost, aclProvider, pr, opsSystem)-
core/chaincode/chaincode_support.go,返回了ChaincodeSupport：为Peer提供执行链码的接口，主要功能有Launch():启动一个停止运行的链码,Stop():停止链码的运行,HandleChaincodeStream():处理链码流信息,Register():将链码注册到当前Peer节点 ,createCCMessage():创建一个交易,ExecuteLegacyInit():链码的实例化,Execute():执行链码并返回回原始的响应,processChaincodeExecutionResult():处理链码的执行结果,InvokeInit():调用链码的Init方法,Invoke():调用链码,execute():执行一个交易
-
启动
AdminServerstartAdminServer(listenAddr, peerServer.Server(), metricsProvider)-
core/protos/peer/admin.go文件,具有GetStatus(),StartServer(),GetModuleLogLevel(),SetModuleLogLevel()等方法
-
创建
EndorserServerpb.RegisterEndorserServer(peerServer.Server(), auth)-
core/endorser/endorser.go文件,注册背书服务器，提供了一个很重要的方法：ProcessProposal()，这个方法值得看一下。
-
创建
GossipServiceerr = initGossipService(policyMgr, metricsProvider, peerServer, serializedIdentity, peerEndpoint.Address)-
gossip/service/gossip_service.go,具有InitializeChannel(),createSelfSignedData(),updateAnchors(),AddPayload()等方法
部署系统链码。
初始化通道。
启动gRPC服务。
如果启用了profile，还会启动监听服务。