区块链教程Fabric1.0源代码分析Peer peer node start命令实现,2018年下半年,区块链行业正逐渐褪去发展之初的浮躁、回归理性,表面上看相关人才需求与身价似乎正在回落。但事实上,正是初期泡沫的渐退,让人们更多的关注点放在了区块链真正的技术之上。

Fabric 1.0源代码笔记 之 Peer #peer node start命令实现

区块链教程Fabric1.0源代码分析Peer peer node start命令实现

1、peer node加载子命令start和status

peer node加载子命令start和status,代码如下:

func Cmd() *cobra.Command {
    nodeCmd.AddCommand(startCmd()) //加载子命令start
    nodeCmd.AddCommand(statusCmd()) //加载子命令status
    return nodeCmd
}

var nodeCmd = &cobra.Command{
    Use:   nodeFuncName,
    Short: fmt.Sprint(shortDes),
    Long:  fmt.Sprint(longDes),
}
//代码在peer/node/node.go

startCmd()代码如下:
其中serve(args)为peer node start的实现代码,比较复杂,本文将重点讲解。
另statusCmd()代码与startCmd()相近,暂略。

func startCmd() *cobra.Command {
    flags := nodeStartCmd.Flags()
    flags.BoolVarP(&chaincodeDevMode, "peer-chaincodedev", "", false, "Whether peer in chaincode development mode")
    flags.BoolVarP(&peerDefaultChain, "peer-defaultchain", "", false, "Whether to start peer with chain testchainid")
    flags.StringVarP(&orderingEndpoint, "orderer", "o", "orderer:7050", "Ordering service endpoint") //orderer
    return nodeStartCmd
}

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 {
        return serve(args) //serve(args)为peer node start的实现代码
    },
}
//代码在peer/node/start.go

注:如下内容均为serve(args)的代码,即peer node start命令执行流程。

2、初始化Ledger(账本)

初始化账本,即如下一条代码:

ledgermgmt.Initialize()
//代码在peer/node/start.go

ledgermgmt.Initialize()代码展开如下:

func initialize() {
    openedLedgers = make(map[string]ledger.PeerLedger) //openedLedgers为全局变量,存储目前使用的账本列表
    provider, err := kvledger.NewProvider() //创建账本Provider实例
    ledgerProvider = provider
}
//代码在core/ledger/ledgermgmt/ledger_mgmt.go

kvledger.NewProvider()代码如下:

func NewProvider() (ledger.PeerLedgerProvider, error) {
    idStore := openIDStore(ledgerconfig.GetLedgerProviderPath()) //打开idStore

    //创建并初始化blkstorage
    attrsToIndex := []blkstorage.IndexableAttr{
        blkstorage.IndexableAttrBlockHash,
        blkstorage.IndexableAttrBlockNum,
        blkstorage.IndexableAttrTxID,
        blkstorage.IndexableAttrBlockNumTranNum,
        blkstorage.IndexableAttrBlockTxID,
        blkstorage.IndexableAttrTxValidationCode,
    }
    indexConfig := &blkstorage.IndexConfig{AttrsToIndex: attrsToIndex}
    blockStoreProvider := fsblkstorage.NewProvider(
        fsblkstorage.NewConf(ledgerconfig.GetBlockStorePath(), ledgerconfig.GetMaxBlockfileSize()),
        indexConfig)

    //创建并初始化statedb
    var vdbProvider statedb.VersionedDBProvider
    if !ledgerconfig.IsCouchDBEnabled() {
        vdbProvider = stateleveldb.NewVersionedDBProvider()
    } else {
        vdbProvider, err = statecouchdb.NewVersionedDBProvider()
    }

    //创建并初始化historydb
    var historydbProvider historydb.HistoryDBProvider
    historydbProvider = historyleveldb.NewHistoryDBProvider()
    //构造Provider
    provider := &Provider{idStore, blockStoreProvider, vdbProvider, historydbProvider}
    provider.recoverUnderConstructionLedger()
    return provider, nil
}
//代码在core/ledger/kvledger/kv_ledger_provider.go

Ledger更详细内容,参考:Fabric 1.0源代码笔记 之 Ledger(账本)

3、配置及创建PeerServer、创建EventHubServer(事件中心服务器)

//初始化全局变量localAddress和peerEndpoint
err := peer.CacheConfiguration() 
peerEndpoint, err := peer.GetPeerEndpoint() //获取peerEndpoint
listenAddr := viper.GetString("peer.listenAddress") //PeerServer监听地址
secureConfig, err := peer.GetSecureConfig() //获取PeerServer安全配置,是否启用TLS、公钥、私钥、根证书
//以监听地址和安全配置,创建Peer GRPC Server
peerServer, err := peer.CreatePeerServer(listenAddr, secureConfig)
//创建EventHubServer(事件中心服务器)
ehubGrpcServer, err := createEventHubServer(secureConfig)
//代码在peer/node/start.go

func createEventHubServer(secureConfig comm.SecureServerConfig) (comm.GRPCServer, error)代码如下:

var lis net.Listener
var err error
lis, err = net.Listen("tcp", viper.GetString("peer.events.address")) //创建Listen
grpcServer, err := comm.NewGRPCServerFromListener(lis, secureConfig) //从Listen创建GRPCServer
ehServer := producer.NewEventsServer(
    uint(viper.GetInt("peer.events.buffersize")), //最大进行缓冲的消息数
    viper.GetDuration("peer.events.timeout")) //缓冲已满的情况下,往缓冲中发送消息的超时
pb.RegisterEventsServer(grpcServer.Server(), ehServer) //EventHubServer注册至grpcServer
return grpcServer, nil
//代码在peer/node/start.go

pb.RegisterEventsServer(grpcServer.Server(), ehServer)代码如下:

func RegisterEventsServer(s *grpc.Server, srv EventsServer) {
    s.RegisterService(&_Events_serviceDesc, srv) 
}
//代码在protos/peer/events.pb.go

events(事件服务)更详细内容,参考:Fabric 1.0源代码笔记 之 events(事件服务)

4、创建并启动Chaincode Server,并注册系统链码

代码如下:

ccprovider.EnableCCInfoCache() //ccInfoCacheEnabled = true
//创建Chaincode Server
//如果peer.chaincodeListenAddress,没有定义或者定义和peerListenAddress相同,均直接使用peerServer
//否则另行创建NewGRPCServer用于Chaincode Server
ccSrv, ccEpFunc := createChaincodeServer(peerServer, listenAddr)
//Chaincode service注册到grpcServer,并注册系统链码
registerChaincodeSupport(ccSrv.Server(), ccEpFunc)
go ccSrv.Start() //启动grpcServer
//代码在peer/node/start.go

ccSrv, ccEpFunc := createChaincodeServer(peerServer, listenAddr)代码如下:创建Chaincode Server。

func createChaincodeServer(peerServer comm.GRPCServer, peerListenAddress string) (comm.GRPCServer, ccEndpointFunc) {
    //peer.chaincodeListenAddress,链码容器连接时的监听地址
    cclistenAddress := viper.GetString("peer.chaincodeListenAddress")

    var srv comm.GRPCServer
    var ccEpFunc ccEndpointFunc
    
    //如果peer.chaincodeListenAddress,没有定义或者定义和peerListenAddress相同,均直接使用peerServer
    //否则另行创建NewGRPCServer用于Chaincode Server
    if cclistenAddress == "" {
        ccEpFunc = peer.GetPeerEndpoint
        srv = peerServer
    } else if cclistenAddress == peerListenAddress {
        ccEpFunc = peer.GetPeerEndpoint
        srv = peerServer
    } else {
        config, err := peer.GetSecureConfig()
        srv, err = comm.NewGRPCServer(cclistenAddress, config)
        ccEpFunc = getChaincodeAddressEndpoint
    }

    return srv, ccEpFunc
}
//代码在peer/node/start.go

registerChaincodeSupport(ccSrv.Server(), ccEpFunc)代码如下:Chaincode service注册到grpcServer。

func registerChaincodeSupport(grpcServer *grpc.Server, ccEpFunc ccEndpointFunc) {
    userRunsCC := chaincode.IsDevMode() //是否开发模式
    ccStartupTimeout := viper.GetDuration("chaincode.startuptimeout") //启动链码容器的超时
    if ccStartupTimeout < time.Duration(5)*time.Second { //至少5秒
        ccStartupTimeout = time.Duration(5) * time.Second
    } else {
    }
    //构造ChaincodeSupport
    ccSrv := chaincode.NewChaincodeSupport(ccEpFunc, userRunsCC, ccStartupTimeout)
    scc.RegisterSysCCs() //注册系统链码
    pb.RegisterChaincodeSupportServer(grpcServer, ccSrv) //service注册到grpcServer
}
//代码在peer/node/start.go

ccSrv := chaincode.NewChaincodeSupport(ccEpFunc, userRunsCC, ccStartupTimeout)代码如下:构造ChaincodeSupport。

var theChaincodeSupport *ChaincodeSupport

func NewChaincodeSupport(getCCEndpoint func() (*pb.PeerEndpoint, error), userrunsCC bool, ccstartuptimeout time.Duration) *ChaincodeSupport {
    //即/var/hyperledger/production/chaincodes
    ccprovider.SetChaincodesPath(config.GetPath("peer.fileSystemPath") + string(filepath.Separator) + "chaincodes")

    pnid := viper.GetString("peer.networkId") //网络ID
    pid := viper.GetString("peer.id") //节点ID

    //构造ChaincodeSupport
    theChaincodeSupport = &ChaincodeSupport{runningChaincodes: &runningChaincodes{chaincodeMap: make(map[string]*chaincodeRTEnv), launchStarted: make(map[string]bool)}, peerNetworkID: 
pnid, peerID: pid}

    ccEndpoint, err := getCCEndpoint() //此处传入ccEpFunc
    if err != nil {
        theChaincodeSupport.peerAddress = viper.GetString("chaincode.peerAddress")
    } else {
        theChaincodeSupport.peerAddress = ccEndpoint.Address
    }
    if theChaincodeSupport.peerAddress == "" {
        theChaincodeSupport.peerAddress = peerAddressDefault
    }

    theChaincodeSupport.userRunsCC = userrunsCC //是否开发模式
    theChaincodeSupport.ccStartupTimeout = ccstartuptimeout //启动链码容器的超时

    theChaincodeSupport.peerTLS = viper.GetBool("peer.tls.enabled") //是否启用TLS
    if theChaincodeSupport.peerTLS {
        theChaincodeSupport.peerTLSCertFile = config.GetPath("peer.tls.cert.file")
        theChaincodeSupport.peerTLSKeyFile = config.GetPath("peer.tls.key.file")
        theChaincodeSupport.peerTLSSvrHostOrd = viper.GetString("peer.tls.serverhostoverride")
    }

    kadef := 0
    // Peer和链码之间的心跳超时,小于或等于0意味着关闭
    if ka := viper.GetString("chaincode.keepalive"); ka == "" {
        theChaincodeSupport.keepalive = time.Duration(kadef) * time.Second //0
    } else {
        t, terr := strconv.Atoi(ka)
        if terr != nil {
            t = kadef //0
        } else if t <= 0 {
            t = kadef //0
        }
        theChaincodeSupport.keepalive = time.Duration(t) * time.Second //非0
    }

    execto := time.Duration(30) * time.Second
    //invoke和initialize命令执行超时
    if eto := viper.GetDuration("chaincode.executetimeout"); eto <= time.Duration(1)*time.Second {
        //小于1秒时,默认30秒
    } else {
        execto = eto
    }
    theChaincodeSupport.executetimeout = execto

    viper.SetEnvPrefix("CORE")
    viper.AutomaticEnv()
    replacer := strings.NewReplacer(".", "_")
    viper.SetEnvKeyReplacer(replacer)

    theChaincodeSupport.chaincodeLogLevel = getLogLevelFromViper("level")
    theChaincodeSupport.shimLogLevel = getLogLevelFromViper("shim")
    theChaincodeSupport.logFormat = viper.GetString("chaincode.logging.format")

    return theChaincodeSupport
}

//代码在core/chaincode/chaincode_support.go

scc.RegisterSysCCs()代码如下:注册系统链码。

func RegisterSysCCs() {
    //cscc、lscc、escc、vscc、qscc
    for _, sysCC := range systemChaincodes {
        RegisterSysCC(sysCC)
    }
}
代码在core/scc/importsysccs.go

5、注册Admin server和Endorser server

代码如下:

//s.RegisterService(&_Admin_serviceDesc, srv)
//var _Admin_serviceDesc = grpc.ServiceDesc{...}
//core.NewAdminServer()构造ServerAdmin结构体,ServerAdmin结构体实现type AdminServer interface接口
pb.RegisterAdminServer(peerServer.Server(), core.NewAdminServer())
//构造结构体Endorser,Endorser结构体实现type EndorserServer interface接口
serverEndorser := endorser.NewEndorserServer()
//s.RegisterService(&_Endorser_serviceDesc, srv)
//var _Endorser_serviceDesc = grpc.ServiceDesc{...}
pb.RegisterEndorserServer(peerServer.Server(), serverEndorser)
//代码在peer/node/start.go

附type AdminServer interface接口定义:

type AdminServer interface {
    GetStatus(context.Context, *google_protobuf.Empty) (*ServerStatus, error)
    StartServer(context.Context, *google_protobuf.Empty) (*ServerStatus, error)
    GetModuleLogLevel(context.Context, *LogLevelRequest) (*LogLevelResponse, error)
    SetModuleLogLevel(context.Context, *LogLevelRequest) (*LogLevelResponse, error)
    RevertLogLevels(context.Context, *google_protobuf.Empty) (*google_protobuf.Empty, error)
}
//代码在protos/peer/admin.pb.go

附type EndorserServer interface接口定义:

type EndorserServer interface {
    ProcessProposal(context.Context, *SignedProposal) (*ProposalResponse, error)
}
//代码在protos/peer/peer.pb.go

6、初始化Gossip服务

bootstrap := viper.GetStringSlice("peer.gossip.bootstrap") //启动节点后gossip连接的初始节点
serializedIdentity, err := mgmt.GetLocalSigningIdentityOrPanic().Serialize() //获取签名身份
messageCryptoService := peergossip.NewMCS( //构造mspMessageCryptoService(消息加密服务)
    peer.NewChannelPolicyManagerGetter(), //构造type channelPolicyManagerGetter struct{}
    localmsp.NewSigner(), //构造type mspSigner struct {}
    mgmt.NewDeserializersManager()) //构造type mspDeserializersManager struct{}
secAdv := peergossip.NewSecurityAdvisor(mgmt.NewDeserializersManager()) //构造mspSecurityAdvisor(安全顾问)

secureDialOpts := func() []grpc.DialOption {
    var dialOpts []grpc.DialOption
    dialOpts = append(dialOpts, grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(comm.MaxRecvMsgSize()), //MaxRecvMsgSize
        grpc.MaxCallSendMsgSize(comm.MaxSendMsgSize()))) //MaxSendMsgSize
    dialOpts = append(dialOpts, comm.ClientKeepaliveOptions()...) //ClientKeepaliveOptions
        
    if comm.TLSEnabled() {
        tlsCert := peerServer.ServerCertificate()
        dialOpts = append(dialOpts, grpc.WithTransportCredentials(comm.GetCASupport().GetPeerCredentials(tlsCert)))
    } else {
        dialOpts = append(dialOpts, grpc.WithInsecure())
    }
    return dialOpts
}

err = service.InitGossipService(serializedIdentity, peerEndpoint.Address, peerServer.Server(),
    messageCryptoService, secAdv, secureDialOpts, bootstrap...) //构造gossipServiceImpl
defer service.GetGossipService().Stop()
//代码在peer/node/start.go

Gossip更详细内容参考:Fabric 1.0源代码笔记 之 gossip(流言算法)

7、初始化、部署并执行系统链码(scc)

代码如下:

initSysCCs() //初始化系统链码,调用scc.DeploySysCCs("")
peer.Initialize(func(cid string) { //初始化所有链
    scc.DeploySysCCs(cid) //按chain id部署并运行系统链码
})
//代码在peer/node/start.go

func Initialize(init func(string))代码如下:

func Initialize(init func(string)) {
    chainInitializer = init

    var cb *common.Block
    var ledger ledger.PeerLedger
    ledgermgmt.Initialize()
    ledgerIds, err := ledgermgmt.GetLedgerIDs()
    for _, cid := range ledgerIds {
        ledger, err = ledgermgmt.OpenLedger(cid)
        cb, err = getCurrConfigBlockFromLedger(ledger) //获取最新的配置块
        err = createChain(cid, ledger, cb)
        InitChain(cid) //即调用chainInitializer(cid),即scc.DeploySysCCs(cid)
    }
}
//代码在core/peer/peer.go

createChain(cid, ledger, cb)代码如下:

func createChain(cid string, ledger ledger.PeerLedger, cb *common.Block) error {
    envelopeConfig, err := utils.ExtractEnvelope(cb, 0) //获取配置交易
    configtxInitializer := configtx.NewInitializer() //构造initializer
    gossipEventer := service.GetGossipService().NewConfigEventer() //获取gossipServiceInstance,并构造configEventer

    gossipCallbackWrapper := func(cm configtxapi.Manager) {
        ac, ok := configtxInitializer.ApplicationConfig()
        gossipEventer.ProcessConfigUpdate(&chainSupport{
            Manager:     cm,
            Application: ac,
        })
        //验证可疑节点身份,并关闭无效链接
        service.GetGossipService().SuspectPeers(func(identity api.PeerIdentityType) bool {
            return true
        })
    }

    trustedRootsCallbackWrapper := func(cm configtxapi.Manager) {
        updateTrustedRoots(cm)
    }

    configtxManager, err := configtx.NewManagerImpl(
        envelopeConfig,
        configtxInitializer,
        []func(cm configtxapi.Manager){gossipCallbackWrapper, trustedRootsCallbackWrapper},
    )
    mspmgmt.XXXSetMSPManager(cid, configtxManager.MSPManager())

    ac, ok := configtxInitializer.ApplicationConfig()
    cs := &chainSupport{
        Manager:     configtxManager,
        Application: ac, // TODO, refactor as this is accessible through Manager
        ledger:      ledger,
    }

    c := committer.NewLedgerCommitterReactive(ledger, txvalidator.NewTxValidator(cs), func(block *common.Block) error {
        chainID, err := utils.GetChainIDFromBlock(block)
        if err != nil {
            return err
        }
        return SetCurrConfigBlock(block, chainID)
    })

    ordererAddresses := configtxManager.ChannelConfig().OrdererAddresses()
    service.GetGossipService().InitializeChannel(cs.ChainID(), c, ordererAddresses)

    chains.Lock()
    defer chains.Unlock()
    chains.list[cid] = &chain{
        cs:        cs,
        cb:        cb,
        committer: c,
    }
    return nil
}
//代码在core/peer/peer.go

scc更详细内容参考:Fabric 1.0源代码笔记 之 scc(系统链码)

8、启动peerServer和ehubGrpcServer,并监控系统信号,以及启动Go自带的profiling支持进行调试

代码如下:

serve := make(chan error)
sigs := make(chan os.Signal, 1)
signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM)

go func() {
    sig := <-sigs //接收系统信号
    serve <- nil
}

go func() {
    var grpcErr error
    grpcErr = peerServer.Start() //启动peerServer
    serve <- grpcErr
}

err := writePid(config.GetPath("peer.fileSystemPath")+"/peer.pid", os.Getpid()) //写入pid

go ehubGrpcServer.Start() //启动ehubGrpcServer

if viper.GetBool("peer.profile.enabled") {
    go func() { //启动Go自带的profiling支持进行调试
        profileListenAddress := viper.GetString("peer.profile.listenAddress")
        profileErr := http.ListenAndServe(profileListenAddress, nil)
    }
}

return <-serve //等待serve
//代码在peer/node/start.go

9、按配置文件重新更新模块日志级别

overrideLogModules := []string{"msp", "gossip", "ledger", "cauthdsl", "policies", "grpc"}
for _, module := range overrideLogModules {
    err = common.SetLogLevelFromViper(module)
}
flogging.SetPeerStartupModulesMap()
//代码在peer/node/start.go