fabric中的plugin机制

之前是研究1.0的实现，发现endorse的过程中去escc签名的流程变化很大。下面以escc为例，看下新版的fabric是怎么实现的。

首先怎么自定义一个escc的plugin

handlers:
    ...
    endorsers:
      escc:
        name: DefaultEndorsement
      custom:
        name: customEndorsement
        library: /etc/hyperledger/fabric/plugins/customEndorsement.so
    ...

core.yaml，可以看到/etc/hyperledger/fabric/plugins/customEndorsement.so，这里是使用了golang的一个plugin的特性，来实现endorse策略的热插拔，比1.0的实现灵活很多。配置好之后peer启动的时候会自动将这部分加载进来。在endorse的流程中会用到。

实现

定义

type Config struct {
   AuthFilters []*HandlerConfig `mapstructure:"authFilters" yaml:"authFilters"`
   Decorators  []*HandlerConfig `mapstructure:"decorators" yaml:"decorators"`
   Endorsers   PluginMapping    `mapstructure:"endorsers" yaml:"endorsers"`
   Validators  PluginMapping    `mapstructure:"validators" yaml:"validators"`
}

type PluginMapping map[string]*HandlerConfig

type HandlerConfig struct {
    Name    string `mapstructure:"name" yaml:"name"`
    Library string `mapstructure:"library" yaml:"library"`
}

type registry struct {
   filters    []auth.Filter
   decorators []decoration.Decorator
   endorsers  map[string]endorsement2.PluginFactory
   validators map[string]validation.PluginFactory
}

在解析配置的时候会将上面的配置映射到config里面，然后在加载完毕后会存在registry里面。

加载

func InitRegistry(c Config) Registry {
    once.Do(func() {
        reg = registry{
            endorsers:  make(map[string]endorsement2.PluginFactory),
            validators: make(map[string]validation.PluginFactory),
        }
        reg.loadHandlers(c)
    })
    return &reg
}

func (r *registry) evaluateModeAndLoad(c *HandlerConfig, handlerType HandlerType, extraArgs ...string) {
   if c.Library != "" {
      r.loadPlugin(c.Library, handlerType, extraArgs...)
   } else {
      r.loadCompiled(c.Name, handlerType, extraArgs...)
   }
}

当然加载的时机就是peer节点启动的时候，会去初始化registry，接下来就是加载到运行时了。

这里escc的handler配置分两种，一种是系统内建的，另外一种就是热插拔的方式。

从配置的区别是有没有设置Library。

下面我们分别看下什么区别

loadCompiled

func (r *registry) loadCompiled(handlerFactory string, handlerType HandlerType, extraArgs ...string) {
   registryMD := reflect.ValueOf(&HandlerLibrary{})

   o := registryMD.MethodByName(handlerFactory)
   if !o.IsValid() {
      logger.Panicf(fmt.Sprintf("Method %s isn't a method of HandlerLibrary", handlerFactory))
   }

   inst := o.Call(nil)[0].Interface()

   if handlerType == Auth {
    ...
   } else if handlerType == Endorsement {
      if len(extraArgs) != 1 {
         logger.Panicf("expected 1 argument in extraArgs")
      }
      r.endorsers[extraArgs[0]] = inst.(endorsement2.PluginFactory)
   } 
   ...
}

意思是调用HandlerLibrary的DefaultEndorsement方法，用builtin.DefaultEndorsementFactory来作为escc的plugin

loadPlugin

func (r *registry) loadPlugin(pluginPath string, handlerType HandlerType, extraArgs ...string) {
   if _, err := os.Stat(pluginPath); err != nil {
      logger.Panicf(fmt.Sprintf("Could not find plugin at path %s: %s", pluginPath, err))
   }
   p, err := plugin.Open(pluginPath)
   if err != nil {
      logger.Panicf(fmt.Sprintf("Error opening plugin at path %s: %s", pluginPath, err))
   }

   ...
   } else if handlerType == Endorsement {
      r.initEndorsementPlugin(p, extraArgs...)
   } 
   ...
}


func (r *registry) initEndorsementPlugin(p *plugin.Plugin, extraArgs ...string) {
    if len(extraArgs) != 1 {
        logger.Panicf("expected 1 argument in extraArgs")
    }
    factorySymbol, err := p.Lookup(pluginFactory)
    if err != nil {
        panicWithLookupError(pluginFactory, err)
    }

    constructor, ok := factorySymbol.(func() endorsement2.PluginFactory)
    if !ok {
        panicWithDefinitionError(pluginFactory)
    }
    factory := constructor()
    if factory == nil {
        logger.Panicf("factory instance returned nil")
    }
    r.endorsers[extraArgs[0]] = factory
}

无非就是热加载golang的plugin并初始化

使用

...
endorsementPluginsByName := reg.Lookup(library.Endorsement).(map[string]endorsement2.PluginFactory)
...
pluginMapper := endorser.MapBasedPluginMapper(endorsementPluginsByName)
pluginEndorser := endorser.NewPluginEndorser(&endorser.PluginSupport{
    ChannelStateRetriever:   channelStateRetriever,
    TransientStoreRetriever: peer.TransientStoreFactory,
    PluginMapper:            pluginMapper,
    SigningIdentityFetcher:  signingIdentityFetcher,
})
endorserSupport.PluginEndorser = pluginEndorser...
...

去已经初始化完成的registry里面查找endorsements, 并通过endorserSupport去对外提供服务。里面的关键是初始化了MapBasedPluginMapper

func (e *Endorser) endorseProposal(...) (*pb.ProposalResponse, error) {
 
    ...
   if isSysCC {
      escc = "escc"
   } else {
      escc = cd.Endorsement()
   }
    ...
   return e.s.EndorseWithPlugin(ctx)
}

func (pe *PluginEndorser) EndorseWithPlugin(ctx Context) (*pb.ProposalResponse, error) {
    ...
    plugin, err := pe.getOrCreatePlugin(PluginName(ctx.PluginName), ctx.Channel)
    ...
    endorsement, prpBytes, err := plugin.Endorse(prpBytes, ctx.SignedProposal)
    ...
}

可以看到在endorse的流程中会去查找escc的endorsehandler来切入进来做背书，在getOrCreatePlugin里面会去MapBasedPluginMapper拿到escc对应的plugin，并最终调用plugin.Endorse。

func (e *DefaultEndorsement) Endorse(prpBytes []byte, sp *peer.SignedProposal) (*peer.Endorsement, []byte, error) {
   signer, err := e.SigningIdentityForRequest(sp)
   if err != nil {
      return nil, nil, errors.New(fmt.Sprintf("failed fetching signing identity: %v", err))
   }
   // serialize the signing identity
   identityBytes, err := signer.Serialize()
   if err != nil {
      return nil, nil, errors.New(fmt.Sprintf("could not serialize the signing identity: %v", err))
   }

   // sign the concatenation of the proposal response and the serialized endorser identity with this endorser's key
   signature, err := signer.Sign(append(prpBytes, identityBytes...))
   if err != nil {
      return nil, nil, errors.New(fmt.Sprintf("could not sign the proposal response payload: %v", err))
   }
   endorsement := &peer.Endorsement{Signature: signature, Endorser: identityBytes}
   return endorsement, prpBytes, nil
}

这里引用了系统内建的Endorser，实现很清楚就是用本地签名人身份进行数字签名。