fabric1.0学习笔记

Posted 2021-10-09 这是谁的部将？

fabric1.0学习笔记（1）
fabric1.0学习笔记（2）

一、fabric1.0目录结构（只列出了主要的文件夹）

• bccsp 密码学相关：加密签名及证书等，将相关函数抽象成了一组接口
• bddtests 一种新的软件开发模式：行为驱动开发 需求->开发
• common 公共库：包括错误处理、日志处理、账本存储及各种工具
• core 核心库：组件核心逻辑
• devenv 开发环境：用的是vagrant
• docs 文档：文档相关的内容
• events 事件监听：事件监听机制，例如确定某一笔交易已经包括到区块中
• example 例子：里面包括了一些fabric网络的例子
• gossip gossip协议：最终一致性共识算法，用于组织内部的区块同步
• images 镜像：用于docker镜像打包
• msp 成员服务管理：member service provider证书管理
• orderer orderer模块：orderer（排序）节点入口
• peer peer模块：peer（记账）节点入口
• proposals 提案：新功能提案
• protos 数据定义：包括了几乎所有fabric的数据结构、数据服务的定义。

阅读源码需最好用ide方便查看代码之间的引用，这里选用的是goland，把fabric文件放到gopath下，编译器import首先寻找的是gopath

因为orderer节点相当于fabric网络中的一个中心节点，所以从orderer的代码开始看
代码注释括号里的内容是个人理解可能认识的有问题

• orderer的main函数

func main() {

	kingpin.Version("0.0.1")
	switch kingpin.MustParse(app.Parse(os.Args[1:])) {

	// "start" command
	case start.FullCommand():
		logger.Infof("Starting %s", metadata.GetVersionInfo())
	
        '载入配置信息'
		conf := config.Load()
		
		'初始化日志级别'
		initializeLoggingLevel(conf)
		
		'启动profile服务（里面涉及的pprof是性能分析服务）'
		initializeProfilingService(conf)
		
		'初始化grpc服务，（让系统监听某个端口）'
		grpcServer := initializeGrpcServer(conf)
		
		'加载本地msp服务，需要LocalMspDir,BCCSP,LocalMspID三个参数'
		initializeLocalMsp(conf)
		
		'msp证书用于签名者实例化(实例化了一个接口)'
		signer := localmsp.NewSigner()

		'初始化多链管理者，这里的manager需要着重看一下！！！'
		manager := initializeMultiChainManager(conf, signer)
		server := NewServer(manager, signer)
		ab.RegisterAtomicBroadcastServer(grpcServer.Server(), server)
		logger.Info("Beginning to serve requests")
		grpcServer.Start()
	// "version" command
	case version.FullCommand():
		fmt.Println(metadata.GetVersionInfo())
	}

}

• manager包

// Manager coordinates the creation and access of chains
type Manager interface {
	// GetChain retrieves the chain support for a chain (and whether it exists)
	'根据链的名称获取一个链对象（chainsupport是个接口，并没有对链直接操作）'
	GetChain(chainID string) (ChainSupport, bool)

	// SystemChannelID returns the channel ID for the system channel
	'获取系统链的ID（系统链是一个空链，用于引导产生新的链，因为交易都要属于一个链，所以创建链时的交易就需要一个初始链即系统链）'
	SystemChannelID() string

	// NewChannelConfig returns a bare bones configuration ready for channel
	// creation request to be applied on top of it
	'生成或更新链的配置'
	NewChannelConfig(envConfigUpdate *cb.Envelope) (configtxapi.Manager, error)
}

'这是一个配置资源，每个链的配置都会打包在这个类里'
type configResources struct {
	configtxapi.Manager
}

'获取orderer节点的配置'
func (cr *configResources) SharedConfig() config.Orderer {
	oc, ok := cr.OrdererConfig()
	if !ok {
		logger.Panicf("[channel %s] has no orderer configuration", cr.ChainID())
	}
	return oc
}
'账本资源类，包括了账本的配置资源和账本的读写对象，可看作对账本进行操作的入口'
type ledgerResources struct {
	*configResources
	ledger ledger.ReadWriter
}
'manager实现类'
type multiLedger struct {
	chains          map[string]*chainSupport]   多个链的对象
	consenters      map[string]Consenter        目前支持的共识机制
	ledgerFactory   ledger.Factory              账本工厂
	signer          crypto.LocalSigner          签名对象（方法），一般是msp，msp实现signer的接口
	systemChannelID string                      系统链名称
	systemChannel   *chainSupport				系统链对象
}
'获取区块里最新的配置交易解码后打印出来，reader是用来操作账本的'
func getConfigTx(reader ledger.Reader) *cb.Envelope {
	lastBlock := ledger.GetBlock(reader, reader.Height()-1)
	index, err := utils.GetLastConfigIndexFromBlock(lastBlock)
	if err != nil {
		logger.Panicf("Chain did not have appropriately encoded last config in its latest block: %s", err)
	}
	configBlock := ledger.GetBlock(reader, index)
	if configBlock == nil {
		logger.Panicf("Config block does not exist")
	}

	return utils.ExtractEnvelopeOrPanic(configBlock, 0)
}

// NewManagerImpl produces an instance of a Manager

'输入账本工厂、支持的共识机制、签名对象返回一个manager实例'
func NewManagerImpl(ledgerFactory ledger.Factory, consenters map[string]Consenter, signer crypto.LocalSigner) Manager {
	ml := &multiLedger{
		chains:        make(map[string]*chainSupport),
		ledgerFactory: ledgerFactory,
		consenters:    consenters,
		signer:        signer,
	}
	'读取本地存储链的ID'
	existingChains := ledgerFactory.ChainIDs()
	for _, chainID := range existingChains {
		//通过账本工厂实例化一个账本读的对象，read ledger
		rl, err := ledgerFactory.GetOrCreate(chainID)
		if err != nil {
			logger.Panicf("Ledger factory reported chainID %s but could not retrieve it: %s", chainID, err)
		}
		'获取链最新的配置交易'
		configTx := getConfigTx(rl)
		if configTx == nil {
			logger.Panic("Programming error, configTx should never be nil here")
		}
		'将配置交易和ledger对象绑定起来'
		ledgerResources := ml.newLedgerResources(configTx)
		chainID := ledgerResources.ChainID()
		//判断是否有创建其他链的权限
		if _, ok := ledgerResources.ConsortiumsConfig(); ok {
			if ml.systemChannelID != "" {
				logger.Panicf("There appear to be two system chains %s and %s", ml.systemChannelID, chainID)
			}
			chain := newChainSupport(createSystemChainFilters(ml, ledgerResources),
				ledgerResources,
				consenters,
				signer)
			logger.Infof("Starting with system channel %s and orderer type %s", chainID, chain.SharedConfig().ConsensusType())
			ml.chains[chainID] = chain
			ml.systemChannelID = chainID
			ml.systemChannel = chain
			// We delay starting this chain, as it might try to copy and replace the chains map via newChain before the map is fully built
			defer chain.start()//延迟启动该链
		} else {
			logger.Debugf("Starting chain: %s", chainID)
			chain := newChainSupport(createStandardFilters(ledgerResources),
				ledgerResources,
				consenters,
				signer)
			ml.chains[chainID] = chain
			chain.start()
		}

	}
	'最后检验'
	if ml.systemChannelID == "" {
		logger.Panicf("No system chain found.  If bootstrapping, does your system channel contain a consortiums group definition?")
	}

	return ml
}
'获取系统链（通道）名称'
func (ml *multiLedger) SystemChannelID() string {
	return ml.systemChannelID
}
'(返回orderer节点中的一条链，以链对象chainsupport形式存储)'
// GetChain retrieves the chain support for a chain (and whether it exists)
func (ml *multiLedger) GetChain(chainID string) (ChainSupport, bool) {
	cs, ok := ml.chains[chainID]
	return cs, ok
}

'新建账本资源，即新建一个账本并返回新账本的资源类作为账本的入口'
func (ml *multiLedger) newLedgerResources(configTx *cb.Envelope) *ledgerResources {
	initializer := configtx.NewInitializer()
	configManager, err := configtx.NewManagerImpl(configTx, initializer, nil)
	if err != nil {
		logger.Panicf("Error creating configtx manager and handlers: %s", err)
	}
		'（这里的configManager是单条链上的manager，本文件是orderer上的manager是multiManager）'
	chainID := configManager.ChainID()

	ledger, err := ml.ledgerFactory.GetOrCreate(chainID)
	if err != nil {
		logger.Panicf("Error getting ledger for %s", chainID)
	}

	return &ledgerResources{
		configResources: &configResources{Manager: configManager},
		ledger:          ledger,
	}
}

'创建新链（通道），会把配置信息写入新的账本中，向multiLedger的链的map中增加新链的内容'
func (ml *multiLedger) newChain(configtx *cb.Envelope) {
	ledgerResources := ml.newLedgerResources(configtx)
	ledgerResources.ledger.Append(ledger.CreateNextBlock(ledgerResources.ledger, []*cb.Envelope{configtx}))      配置信息写入新链的账本

	// Copy the map to allow concurrent reads from broadcast/deliver while the new chainSupport is
	newChains := make(map[string]*chainSupport)
	for key, value := range ml.chains {
		newChains[key] = value
	}

	cs := newChainSupport(createStandardFilters(ledgerResources), ledgerResources, ml.consenters, ml.signer)
	chainID := ledgerResources.ChainID()

	logger.Infof("Created and starting new chain %s", chainID)

	newChains[string(chainID)] = cs
	cs.start()

	ml.chains = newChains
}
'返回当前orderer节点上的链数（通道数）'
func (ml *multiLedger) channelsCount() int {
	return len(ml.chains)
}

'生成新链的配置，主要是一些检查性的工作 （新建通道配置文件，也用到了单链上的manager，应该是有区别）'
func (ml *multiLedger) NewChannelConfig(envConfigUpdate *cb.Envelope) (configtxapi.Manager, error) {
	configUpdatePayload, err := utils.UnmarshalPayload(envConfigUpdate.Payload)
	if err != nil {
		return nil, fmt.Errorf("Failing initial channel config creation because of payload unmarshaling error: %s", err)
	}

	configUpdateEnv, err := configtx.UnmarshalConfigUpdateEnvelope(configUpdatePayload.Data)
	if err != nil {
		return nil, fmt.Errorf("Failing initial channel config creation because of config update envelope unmarshaling error: %s", err)
	}

	if configUpdatePayload.Header == nil {
		return nil, fmt.Errorf("Failed initial channel config creation because config update header was missing")
	}
	channelHeader, err := utils.UnmarshalChannelHeader(configUpdatePayload.Header.ChannelHeader)

	configUpdate, err := configtx.UnmarshalConfigUpdate(configUpdateEnv.ConfigUpdate)
	if err != nil {
		return nil, fmt.Errorf("Failing initial channel config creation because of config update unmarshaling error: %s", err)
	}

	if configUpdate.ChannelId != channelHeader.ChannelId {
		return nil, fmt.Errorf("Failing initial channel config creation: mismatched channel IDs: '%s' != '%s'", configUpdate.ChannelId, channelHeader.ChannelId)
	}

	if configUpdate.WriteSet == nil {
		return nil, fmt.Errorf("Config update has an empty writeset")
	}

	if configUpdate.WriteSet.Groups == nil || configUpdate.WriteSet.Groups[config.ApplicationGroupKey] == nil {
		return nil, fmt.Errorf("Config update has missing application group")
	}

	if uv := configUpdate.WriteSet.Groups[config.ApplicationGroupKey].Version; uv != 1 {
		return nil, fmt.Errorf("Config update for channel creation does not set application group version to 1, was %d", uv)
	}

	consortiumConfigValue, ok := configUpdate.WriteSet.Values[config.ConsortiumKey]
	if !ok {
		return nil, fmt.Errorf("Consortium config value missing")
	}

	consortium := &cb.Consortium{}
	err = proto.Unmarshal(consortiumConfigValue.Value, consortium)
	if err != nil {
		return nil, fmt.Errorf("Error reading unmarshaling consortium name: %s", err)
	}

	applicationGroup := cb.NewConfigGroup()
	consortiumsConfig, ok := ml.systemChannel.ConsortiumsConfig()
	if !ok {
		return nil, fmt.Errorf("The ordering system channel does not appear to support creating channels")
	}

	consortiumConf, ok := consortiumsConfig.Consortiums()[consortium.Name]
	if !ok {
		return nil, fmt.Errorf("Unknown consortium name: %s", consortium.Name)
	}

	applicationGroup.Policies[config.ChannelCreationPolicyKey] = &cb.ConfigPolicy{
		Policy: consortiumConf.ChannelCreationPolicy(),
	}
	applicationGroup.ModPolicy = config.ChannelCreationPolicyKey

	// Get the current system channel config
	systemChannelGroup := ml.systemChannel.ConfigEnvelope().Config.ChannelGroup

	// If the consortium group has no members, allow the source request to have no members.  However,
	// if the consortium group has any members, there must be at least one member in the source request
	if len(systemChannelGroup.Groups[config.ConsortiumsGroupKey].Groups[consortium.Name].Groups) > 0 &&
		len(configUpdate.WriteSet.Groups[config.ApplicationGroupKey].Groups) == 0 {
		return nil, fmt.Errorf("Proposed configuration has no application group members, but consortium contains members")
	}

	// If the consortium has no members, allow the source request to contain arbitrary members
	// Otherwise, require that the supplied members are a subset of the consortium members
	if len(systemChannelGroup.Groups[config.ConsortiumsGroupKey].Groups[consortium.Name].Groups) > 0 {
		for orgName := range configUpdate.WriteSet.Groups[config.ApplicationGroupKey].Groups {
			consortiumGroup, ok := systemChannelGroup.Groups[config.ConsortiumsGroupKey].Groups[consortium.Name].Groups[orgName]
			if !ok {
				return nil, fmt.Errorf("Attempted to include a member which is not in the consortium")
			}
			applicationGroup.Groups[orgName] = consortiumGroup
		}
	}

	channelGroup := cb.NewConfigGroup()

	// Copy the system channel Channel level config to the new config
	for key, value := range systemChannelGroup.Values {
		channelGroup.Values[key] = value
		if key == config.ConsortiumKey {
			// Do not set the consortium name, we do this later
			continue
		}
	}

	for key, policy := range systemChannelGroup.Policies {
		channelGroup.Policies[key] = policy
	}

	// Set the new config orderer group to the system channel orderer group and the application group to the new application group
	channelGroup.Groups[config.OrdererGroupKey] = systemChannelGroup.Groups[config.OrdererGroupKey]
	channelGroup.Groups[config.ApplicationGroupKey] = applicationGroup
	channelGroup.Values[config.ConsortiumKey] = config.TemplateConsortium(consortium.Name).Values[config.ConsortiumKey]

	templateConfig, _ := utils.CreateSignedEnvelope(cb.HeaderType_CONFIG, configUpdate.ChannelId, ml.signer, &cb.ConfigEnvelope{
		Config: &cb.Config{
			ChannelGroup: channelGroup,
		},
	}, msgVersion, epoch)

	initializer := configtx.NewInitializer()

	// This is a very hacky way to disable the sanity check logging in the policy manager
	// for the template configuration, but it is the least invasive near a release
	pm, ok := initializer.PolicyManager().(*policies.ManagerImpl)
	if ok {
		pm.SuppressSanityLogMessages = true
		defer func() {
			pm.SuppressSanityLogMessages = false
		}()
	}

	return configtx.NewManagerImpl(templateConfig, initializer, nil)
}