前一篇讲完通道创建，接下来马上趁热打铁来看下加入通道这部分的实现，先看下命令。
peer channel join -b mychannel.block
可以看到，这里用上了上一步所生成的通道的genesisblock。
peer
executeJoin
func executeJoin(cf *ChannelCmdFactory) (err error) {
spec, err := getJoinCCSpec()
if err != nil {
return err
}
// Build the ChaincodeInvocationSpec message
invocation := &pb.ChaincodeInvocationSpec{ChaincodeSpec: spec}
creator, err := cf.Signer.Serialize()
if err != nil {
return fmt.Errorf("Error serializing identity for %s: %s", cf.Signer.GetIdentifier(), err)
}
var prop *pb.Proposal
prop, _, err = putils.CreateProposalFromCIS(pcommon.HeaderType_CONFIG, "", invocation, creator)
if err != nil {
return fmt.Errorf("Error creating proposal for join %s", err)
}
var signedProp *pb.SignedProposal
signedProp, err = putils.GetSignedProposal(prop, cf.Signer)
if err != nil {
return fmt.Errorf("Error creating signed proposal %s", err)
}
var proposalResp *pb.ProposalResponse
proposalResp, err = cf.EndorserClient.ProcessProposal(context.Background(), signedProp)
if err != nil {
return ProposalFailedErr(err.Error())
}
if proposalResp == nil {
return ProposalFailedErr("nil proposal response")
}
if proposalResp.Response.Status != 0 && proposalResp.Response.Status != 200 {
return ProposalFailedErr(fmt.Sprintf("bad proposal response %d: %s", proposalResp.Response.Status, proposalResp.Response.Message))
}
logger.Info("Successfully submitted proposal to join channel")
return nil
}
这里主要做几件事情
生成cis，也就是ChaincodeInvocationSpec，这里跟chaincode invoke篇高度相似，这里就不再赘述。而这里cis的重点是里面最终调用的是什么地方。
- 可以看到最终是调用cscc的JoinChain
func getJoinCCSpec() (*pb.ChaincodeSpec, error) { if genesisBlockPath == common.UndefinedParamValue { return nil, errors.New("Must supply genesis block file") } gb, err := ioutil.ReadFile(genesisBlockPath) if err != nil { return nil, GBFileNotFoundErr(err.Error()) } // Build the spec input := &pb.ChaincodeInput{Args: [][]byte{[]byte(cscc.JoinChain), gb}} spec := &pb.ChaincodeSpec{ Type: pb.ChaincodeSpec_Type(pb.ChaincodeSpec_Type_value["GOLANG"]), ChaincodeId: &pb.ChaincodeID{Name: "cscc"}, Input: input, } return spec, nil }
- 最后包装成HeaderType_CONFIG的Proposal
- 注意，这里会去生成数字签名，当然是用的signidentity，而签名身份是跟admin是一致的。
- 接着就是开始处理proposal了
cscc
case JoinChain:
if args[1] == nil {
return shim.Error("Cannot join the channel <nil> configuration block provided")
}
block, err := utils.GetBlockFromBlockBytes(args[1])
if err != nil {
return shim.Error(fmt.Sprintf("Failed to reconstruct the genesis block, %s", err))
}
cid, err := utils.GetChainIDFromBlock(block)
if err != nil {
return shim.Error(fmt.Sprintf("\"JoinChain\" request failed to extract "+
"channel id from the block due to [%s]", err))
}
if err := validateConfigBlock(block); err != nil {
return shim.Error(fmt.Sprintf("\"JoinChain\" for chainID = %s failed because of validation "+
"of configuration block, because of %s", cid, err))
}
// 2. check local MSP Admins policy
// TODO: move to ACLProvider once it will support chainless ACLs
if err = e.policyChecker.CheckPolicyNoChannel(mgmt.Admins, sp); err != nil {
return shim.Error(fmt.Sprintf("access denied for [%s][%s]: [%s]", fname, cid, err))
}
// Initialize txsFilter if it does not yet exist. We can do this safely since
// it's the genesis block anyway
txsFilter := util.TxValidationFlags(block.Metadata.Metadata[common.BlockMetadataIndex_TRANSACTIONS_FILTER])
if len(txsFilter) == 0 {
// add array of validation code hardcoded to valid
txsFilter = util.NewTxValidationFlagsSetValue(len(block.Data.Data), pb.TxValidationCode_VALID)
block.Metadata.Metadata[common.BlockMetadataIndex_TRANSACTIONS_FILTER] = txsFilter
}
return joinChain(cid, block, e.ccp, e.sccp)
前面就是block的各种校验，跳过
CheckPolicyNoChannel(mgmt.Admins, sp)，为什么前面要强调签名身份跟admin一致，说明这个proposal包是admin签署的。而这里就是校验规则是否是admin。截取片段证明我所言非虚。
case m.MSPRole_ADMIN: mspLogger.Debugf("Checking if identity satisfies ADMIN role for %s", msp.name) // in the case of admin, we check that the // id is exactly one of our admins for _, admincert := range msp.admins { if bytes.Equal(id.(*identity).cert.Raw, admincert.(*identity).cert.Raw) { // we do not need to check whether the admin is a valid identity // according to this MSP, since we already check this at Setup time // if there is a match, we can just return return nil } }
- 可以看到这里就是比对证书是否跟本地保存的该组织的admin是否一致。
joinChain
func joinChain(chainID string, block *common.Block, ccp ccprovider.ChaincodeProvider, sccp sysccprovider.SystemChaincodeProvider) pb.Response {
if err := peer.CreateChainFromBlock(block, ccp, sccp); err != nil {
return shim.Error(err.Error())
}
peer.InitChain(chainID)
return shim.Success(nil)
}
这里分两步，CreateChainFromBlock和InitChain下面会讲到。处理完后给peer返回success
CreateChainFromBlock
func CreateChainFromBlock(cb *common.Block, ccp ccprovider.ChaincodeProvider, sccp sysccprovider.SystemChaincodeProvider) error {
cid, err := utils.GetChainIDFromBlock(cb)
if err != nil {
return err
}
var l ledger.PeerLedger
if l, err = ledgermgmt.CreateLedger(cb); err != nil {
return errors.WithMessage(err, "cannot create ledger from genesis block")
}
return createChain(cid, l, cb, ccp, sccp, pluginMapper)
}
这里做两件事情
- 创建本地帐本
- createChain，回忆下上篇创建通道，那里会初始化orderer端的chain，并启动。而这里是peer端
createChain
func createChain(cid string, ledger ledger.PeerLedger, cb *common.Block, ccp ccprovider.ChaincodeProvider, sccp sysccprovider.SystemChaincodeProvider, pm txvalidator.PluginMapper) error {
chanConf, err := retrievePersistedChannelConfig(ledger)
if err != nil {
return err
}
var bundle *channelconfig.Bundle
if chanConf != nil {
bundle, err = channelconfig.NewBundle(cid, chanConf)
if err != nil {
return err
}
} else {
// Config was only stored in the statedb starting with v1.1 binaries
// so if the config is not found there, extract it manually from the config block
envelopeConfig, err := utils.ExtractEnvelope(cb, 0)
if err != nil {
return err
}
bundle, err = channelconfig.NewBundleFromEnvelope(envelopeConfig)
if err != nil {
return err
}
}
capabilitiesSupportedOrPanic(bundle)
channelconfig.LogSanityChecks(bundle)
gossipEventer := service.GetGossipService().NewConfigEventer()
gossipCallbackWrapper := func(bundle *channelconfig.Bundle) {
ac, ok := bundle.ApplicationConfig()
if !ok {
// TODO, handle a missing ApplicationConfig more gracefully
ac = nil
}
gossipEventer.ProcessConfigUpdate(&gossipSupport{
Validator: bundle.ConfigtxValidator(),
Application: ac,
Channel: bundle.ChannelConfig(),
})
service.GetGossipService().SuspectPeers(func(identity api.PeerIdentityType) bool {
// TODO: this is a place-holder that would somehow make the MSP layer suspect
// that a given certificate is revoked, or its intermediate CA is revoked.
// In the meantime, before we have such an ability, we return true in order
// to suspect ALL identities in order to validate all of them.
return true
})
}
trustedRootsCallbackWrapper := func(bundle *channelconfig.Bundle) {
updateTrustedRoots(bundle)
}
mspCallback := func(bundle *channelconfig.Bundle) {
// TODO remove once all references to mspmgmt are gone from peer code
mspmgmt.XXXSetMSPManager(cid, bundle.MSPManager())
}
ac, ok := bundle.ApplicationConfig()
if !ok {
ac = nil
}
cs := &chainSupport{
Application: ac, // TODO, refactor as this is accessible through Manager
ledger: ledger,
}
peerSingletonCallback := func(bundle *channelconfig.Bundle) {
ac, ok := bundle.ApplicationConfig()
if !ok {
ac = nil
}
cs.Application = ac
cs.Resources = bundle
}
cs.bundleSource = channelconfig.NewBundleSource(
bundle,
gossipCallbackWrapper,
trustedRootsCallbackWrapper,
mspCallback,
peerSingletonCallback,
)
vcs := struct {
*chainSupport
*semaphore.Weighted
}{cs, validationWorkersSemaphore}
validator := txvalidator.NewTxValidator(cid, vcs, sccp, pm)
c := committer.NewLedgerCommitterReactive(ledger, func(block *common.Block) error {
chainID, err := utils.GetChainIDFromBlock(block)
if err != nil {
return err
}
return SetCurrConfigBlock(block, chainID)
})
ordererAddresses := bundle.ChannelConfig().OrdererAddresses()
if len(ordererAddresses) == 0 {
return errors.New("no ordering service endpoint provided in configuration block")
}
// TODO: does someone need to call Close() on the transientStoreFactory at shutdown of the peer?
store, err := TransientStoreFactory.OpenStore(bundle.ConfigtxValidator().ChainID())
if err != nil {
return errors.Wrapf(err, "[channel %s] failed opening transient store", bundle.ConfigtxValidator().ChainID())
}
csStoreSupport := &CollectionSupport{
PeerLedger: ledger,
}
simpleCollectionStore := privdata.NewSimpleCollectionStore(csStoreSupport)
service.GetGossipService().InitializeChannel(bundle.ConfigtxValidator().ChainID(), ordererAddresses, service.Support{
Validator: validator,
Committer: c,
Store: store,
Cs: simpleCollectionStore,
IdDeserializeFactory: csStoreSupport,
})
chains.Lock()
defer chains.Unlock()
chains.list[cid] = &chain{
cs: cs,
cb: cb,
committer: c,
}
return nil
}
首先根据传入的genesisblock中拿到envelope，进而转换成bundle，又来了，看过上篇的，应该很熟悉了。
拿到bundle后，第一时间就是组装chainsupport，可以看到后面的wrapper和callback都是用来处理当bundle更新的场景，也就是当配置变更的时候会回调这些逻辑，这里就不展开了。自行分析吧。
cs := &chainSupport{ Application: ac, // TODO, refactor as this is accessible through Manager ledger: ledger, } cs.bundleSource = channelconfig.NewBundleSource( bundle, gossipCallbackWrapper, trustedRootsCallbackWrapper, mspCallback, peerSingletonCallback, )
然后就是根据账本去初始化LedgerCommitter，这是账本提交相关的。
最后最最重要的是初始化gossip服务，回忆下gossip会做什么？
- 去orderer拉取block
- peer节点间选举
- peer成员间主动被动同步block
- peer成员状态同步
而这些都会在这里面去初始化，service.GetGossipService().InitializeChannel感兴趣的可以进去看看，可以找到前面所讲的deliveryService，leaderelection，GossipStateProvider
InitChain
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")
}
cceventmgmt.GetMgr().Register(cid, sub)
}
当peer node start的时候也会调用sccp.DeploySysCCs(cid, ccp)，只不过cid为空，说明是系统级的scc，而这里部署的是通道相关的scc。也就是说以后在通道的基础上调用的scc就是该通道独有的，跟其他通道以及系统区分开。
好奇，到底有什么区别，都用系统级的不好么？进去看看
deploySysCC
func deploySysCC(chainID string, ccprov ccprovider.ChaincodeProvider, syscc SelfDescribingSysCC) error {
...
if chainID != "" {
lgr := peer.GetLedger(chainID)
if lgr == nil {
panic(fmt.Sprintf("syschain %s start up failure - unexpected nil ledger for channel %s", syscc.Name(), chainID))
}
txsim, err := lgr.NewTxSimulator(txid)
if err != nil {
return err
}
txParams.TXSimulator = txsim
defer txsim.Done()
}
...
}
进来可以看到，如果绑定通道id的结果是生成独有的事件模拟器。
小结
至此，整个join处理完毕，很多部分前面已经讲过了，显得这里很粗糙，建议搞清楚了再来，抱歉。
网友评论