[FAB-6832] Add peer resource config bundlesource

This CR mimics the channel config bundle, and adds a peer resource
config bundle source which provides access to an underlying set of
channel config bundle and peer resource config bundle.

The underlying data structures are updateable via an atomic pointer
assignment, and stable views of the current configuration (totaling both
the channel configuration and the peer resource configuration) may be
obtained.

Change-Id: I84646c526b256aa140d23f4d7982c91dd67b0aa3
Signed-off-by: Jason Yellick <jyellick@us.ibm.com>

Author: Jason Yellick

common/resourcesconfig/bundle.go

@@ -10,13 +10,15 @@ import (
 	"fmt"
 
 	"github.com/hyperledger/fabric/common/cauthdsl"
+	"github.com/hyperledger/fabric/common/channelconfig"
 	"github.com/hyperledger/fabric/common/configtx"
 	configtxapi "github.com/hyperledger/fabric/common/configtx/api"
 	"github.com/hyperledger/fabric/common/flogging"
 	"github.com/hyperledger/fabric/common/policies"
-	"github.com/hyperledger/fabric/msp"
 	cb "github.com/hyperledger/fabric/protos/common"
 	"github.com/hyperledger/fabric/protos/utils"
+
+	"github.com/pkg/errors"
 )
 
 // RootGroupKey is the namespace in the config tree for this set of config
@@ -26,29 +28,18 @@ var logger = flogging.MustGetLogger("common/config/resource")
 
 // Bundle stores an immutable group of resources configuration
 type Bundle struct {
-	rg  *resourceGroup
-	cm  configtxapi.Manager
-	pm  policies.Manager
-	rpm policies.Manager
+	channelID string
+	resConf   *cb.Config
+	chanConf  channelconfig.Resources
+	rg        *resourceGroup
+	cm        configtxapi.Manager
+	pm        *policyRouter
 }
 
-// New creates a new resources config bundle
-// TODO, change interface to take config and not an envelope
+// NewBundleFromEnvelope creates a new resources config bundle.
+// TODO, this method should probably be removed, as the resourcesconfig is never in an Envelope naturally.
 // TODO, add an atomic BundleSource
-func New(envConfig *cb.Envelope, mspManager msp.MSPManager, channelPolicyManager policies.Manager) (*Bundle, error) {
-	policyProviderMap := make(map[int32]policies.Provider)
-	for pType := range cb.Policy_PolicyType_name {
-		rtype := cb.Policy_PolicyType(pType)
-		switch rtype {
-		case cb.Policy_UNKNOWN:
-			// Do not register a handler
-		case cb.Policy_SIGNATURE:
-			policyProviderMap[pType] = cauthdsl.NewPolicyProvider(mspManager)
-		case cb.Policy_MSP:
-			// Add hook for MSP Handler here
-		}
-	}
-
+func NewBundleFromEnvelope(envConfig *cb.Envelope, chanConf channelconfig.Resources) (*Bundle, error) {
 	payload, err := utils.UnmarshalPayload(envConfig.Payload)
 	if err != nil {
 		return nil, err
@@ -63,31 +54,28 @@ func New(envConfig *cb.Envelope, mspManager msp.MSPManager, channelPolicyManager
 		return nil, fmt.Errorf("config is nil")
 	}
 
-	resourcesPolicyManager, err := policies.NewManagerImpl(RootGroupKey, policyProviderMap, configEnvelope.Config.ChannelGroup)
+	chdr, err := utils.UnmarshalChannelHeader(payload.Header.ChannelHeader)
 	if err != nil {
-		return nil, err
+		return nil, errors.Wrap(err, "failed to unmarshal channel header")
 	}
 
-	resourceGroup, err := newResourceGroup(configEnvelope.Config.ChannelGroup)
+	return NewBundle(chdr.ChannelId, configEnvelope.Config, chanConf)
+}
+
+// NewBundle creates a resources config bundle which implements the Resources interface.
+func NewBundle(channelID string, config *cb.Config, chanConf channelconfig.Resources) (*Bundle, error) {
+	resourceGroup, err := newResourceGroup(config.ChannelGroup)
 	if err != nil {
 		return nil, err
 	}
 
 	b := &Bundle{
-		rpm: resourcesPolicyManager,
-		rg:  resourceGroup,
-		pm: &policyRouter{
-			channelPolicyManager:   channelPolicyManager,
-			resourcesPolicyManager: resourcesPolicyManager,
-		},
-	}
-
-	b.cm, err = configtx.NewManagerImpl(envConfig, b)
-	if err != nil {
-		return nil, err
+		channelID: channelID,
+		rg:        resourceGroup,
+		resConf:   config,
 	}
 
-	return b, nil
+	return b.NewFromChannelConfig(chanConf)
 }
 
 // RootGroupKey returns the name of the key for the root group (the namespace for this config).
@@ -114,3 +102,66 @@ func (b *Bundle) APIPolicyMapper() PolicyMapper {
 func (b *Bundle) ChaincodeRegistry() ChaincodeRegistry {
 	return b.rg.chaincodesGroup
 }
+
+// ChannelConfig returns the channel config which this resources config depends on.
+// Note, consumers of the resourcesconfig should almost never refer to the PolicyManager
+// within the channel config, and should instead refer to the PolicyManager exposed by
+// this Bundle.
+func (b *Bundle) ChannelConfig() channelconfig.Resources {
+	return b.chanConf
+}
+
+// ValidateNew is currently a no-op.  The idea is that there may be additional checking which needs to be done
+// between an old resource configuration and a new one.  In the channel resource configuration case, we add some
+// checks to make sure that the MSP IDs have not changed, and that the consensus type has not changed.  There is
+// currently no such check required in the peer resources, but, in the interest of following the pattern established
+// by the channel configuration, it is included here.
+func (b *Bundle) ValidateNew(resources Resources) error {
+	return nil
+}
+
+// NewFromChannelConfig builds a new Bundle, based on this Bundle, but with a different underlying
+// channel config.  This is usually invoked when a channel config update is processed.
+func (b *Bundle) NewFromChannelConfig(chanConf channelconfig.Resources) (*Bundle, error) {
+	policyProviderMap := make(map[int32]policies.Provider)
+	for pType := range cb.Policy_PolicyType_name {
+		rtype := cb.Policy_PolicyType(pType)
+		switch rtype {
+		case cb.Policy_UNKNOWN:
+			// Do not register a handler
+		case cb.Policy_SIGNATURE:
+			policyProviderMap[pType] = cauthdsl.NewPolicyProvider(chanConf.MSPManager())
+		case cb.Policy_MSP:
+			// Add hook for MSP Handler here
+		}
+	}
+
+	resourcesPolicyManager, err := policies.NewManagerImpl(RootGroupKey, policyProviderMap, b.resConf.ChannelGroup)
+
+	if err != nil {
+		return nil, err
+	}
+
+	result := &Bundle{
+		channelID: b.channelID,
+		chanConf:  chanConf,
+		pm: &policyRouter{
+			channelPolicyManager:   chanConf.PolicyManager(),
+			resourcesPolicyManager: resourcesPolicyManager,
+		},
+		rg:      b.rg,
+		resConf: b.resConf,
+	}
+
+	env, err := utils.CreateSignedEnvelope(cb.HeaderType_CONFIG, b.channelID, nil, &cb.ConfigEnvelope{Config: b.resConf}, 0, 0)
+	if err != nil {
+		return nil, errors.Wrap(err, "creating envelope for configtx manager failed")
+	}
+
+	result.cm, err = configtx.NewManagerImpl(env, b)
+	if err != nil {
+		return nil, err
+	}
+
+	return result, nil
+}

common/resourcesconfig/bundle_test.go

@@ -9,6 +9,7 @@ package resourcesconfig
 import (
 	"testing"
 
+	mockchannelconfig "github.com/hyperledger/fabric/common/mocks/config"
 	cb "github.com/hyperledger/fabric/protos/common"
 	"github.com/hyperledger/fabric/protos/utils"
 
@@ -32,15 +33,17 @@ func TestBundle(t *testing.T) {
 	}, 0, 0)
 	assert.NoError(t, err)
 
-	b, err := New(env, nil, nil)
+	b, err := NewBundleFromEnvelope(env, &mockchannelconfig.Resources{})
 	assert.NoError(t, err)
 	assert.NotNil(t, b)
 
 	assert.Equal(t, b.RootGroupKey(), RootGroupKey)
 	assert.NotNil(t, b.ConfigtxManager())
 	assert.NotNil(t, b.PolicyManager())
 	assert.NotNil(t, b.APIPolicyMapper())
+	assert.NotNil(t, b.ChannelConfig())
 	assert.NotNil(t, b.ChaincodeRegistry())
+	assert.Nil(t, b.ValidateNew(nil))
 }
 
 func TestBundleFailure(t *testing.T) {
@@ -55,7 +58,7 @@ func TestBundleFailure(t *testing.T) {
 	}, 0, 0)
 	assert.NoError(t, err)
 
-	b, err := New(env, nil, nil)
+	b, err := NewBundleFromEnvelope(env, &mockchannelconfig.Resources{})
 	assert.Error(t, err)
 	assert.Nil(t, b)
 }

common/resourcesconfig/bundlesource.go

@@ -0,0 +1,90 @@
+/*
+Copyright IBM Corp. All Rights Reserved.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package resourcesconfig
+
+import (
+	"sync/atomic"
+
+	"github.com/hyperledger/fabric/common/channelconfig"
+	configtxapi "github.com/hyperledger/fabric/common/configtx/api"
+	"github.com/hyperledger/fabric/common/policies"
+)
+
+// BundleSource stores a reference to the current peer resource configuration bundle
+// It also provides a method to update this bundle.  The assorted methods of BundleSource
+// largely pass through to the underlying bundle, but do so through an atomic pointer
+// so that cross go-routine reads are not vulnerable to out-of-order execution memory
+// type bugs.
+type BundleSource struct {
+	bundle    atomic.Value
+	callbacks []func(*Bundle)
+}
+
+// NewBundleSource creates a new BundleSource with an initial Bundle value
+// The callbacks will be invoked whenever the Update method is called for the
+// BundleSource.  Note, these callbacks are called immediately before this function
+// returns.
+func NewBundleSource(bundle *Bundle, callbacks ...func(*Bundle)) *BundleSource {
+	bs := &BundleSource{
+		callbacks: callbacks,
+	}
+	bs.Update(bundle)
+	return bs
+}
+
+// Update sets a new bundle as the bundle source and calls any registered callbacks
+func (bs *BundleSource) Update(newBundle *Bundle) {
+	bs.bundle.Store(newBundle)
+	for _, callback := range bs.callbacks {
+		callback(newBundle)
+	}
+}
+
+// StableBundle returns a pointer to a stable Bundle.
+// It is stable because calls to its assorted methods will always return the same
+// result, as the underlying data structures are immutable.  For instance, calling
+// BundleSource.PolicyManager() and BundleSource.APIPolicyMapper() to get first the
+// the policy manager and then references into that policy manager could result in a
+// bug because an update might replace the underlying Bundle in between.  Therefore,
+// for operations which require consistency between the Bundle calls, the caller
+// should first retrieve a StableBundle, then operate on it.
+func (bs *BundleSource) StableBundle() *Bundle {
+	return bs.bundle.Load().(*Bundle)
+}
+
+// ConfigtxManager returns a reference to a configtx.Manager which can process updates to this config.
+func (bs *BundleSource) ConfigtxManager() configtxapi.Manager {
+	return bs.StableBundle().ConfigtxManager()
+}
+
+// PolicyManager returns a policy manager which can resolve names both in the /Channel and /Resources namespaces.
+func (bs *BundleSource) PolicyManager() policies.Manager {
+	return bs.StableBundle().PolicyManager()
+}
+
+// APIPolicyMapper returns a way to map API names to policies governing their invocation.
+func (bs *BundleSource) APIPolicyMapper() PolicyMapper {
+	return bs.StableBundle().APIPolicyMapper()
+}
+
+// ChaincodeRegistery returns a way to query for chaincodes defined in this channel.
+func (bs *BundleSource) ChaincodeRegistry() ChaincodeRegistry {
+	return bs.StableBundle().ChaincodeRegistry()
+}
+
+// ChannelConfig returns the channel config which this resources config depends on.
+// Note, consumers of the resourcesconfig should almost never refer to the PolicyManager
+// within the channel config, and should instead refer to the PolicyManager exposed by
+// this Bundle.
+func (bs *BundleSource) ChannelConfig() channelconfig.Resources {
+	return bs.StableBundle().ChannelConfig()
+}
+
+// ValidateNew passes through to the current bundle
+func (bs *BundleSource) ValidateNew(resources Resources) error {
+	return bs.StableBundle().ValidateNew(resources)
+}

common/resourcesconfig/bundlesource_test.go

@@ -0,0 +1,54 @@
+/*
+Copyright IBM Corp. All Rights Reserved.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package resourcesconfig
+
+import (
+	"testing"
+
+	mockchannelconfig "github.com/hyperledger/fabric/common/mocks/config"
+	cb "github.com/hyperledger/fabric/protos/common"
+	"github.com/hyperledger/fabric/protos/utils"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestBundleSource(t *testing.T) {
+	env, err := utils.CreateSignedEnvelope(cb.HeaderType_CONFIG, "foo", nil, &cb.ConfigEnvelope{
+		Config: &cb.Config{
+			ChannelGroup: sampleResourceGroup,
+		},
+	}, 0, 0)
+	assert.NoError(t, err)
+
+	b, err := NewBundleFromEnvelope(env, &mockchannelconfig.Resources{})
+	assert.NoError(t, err)
+	assert.NotNil(t, b)
+
+	calledBack := false
+	var calledBackWith *Bundle
+
+	bs := NewBundleSource(b, func(ib *Bundle) {
+		calledBack = true
+		calledBackWith = ib
+	})
+
+	assert.True(t, calledBack)
+	assert.Equal(t, b, calledBackWith)
+
+	assert.Equal(t, b.ConfigtxManager(), bs.ConfigtxManager())
+	assert.Equal(t, b.PolicyManager(), bs.PolicyManager())
+	assert.Equal(t, b.APIPolicyMapper(), bs.APIPolicyMapper())
+	assert.Equal(t, b.ChannelConfig(), bs.ChannelConfig())
+	assert.Equal(t, b.ChaincodeRegistry(), bs.ChaincodeRegistry())
+	assert.Equal(t, b.ValidateNew(nil), bs.ValidateNew(nil))
+
+	calledBack = false
+	nb := &Bundle{}
+	bs.Update(nb)
+	assert.True(t, calledBack)
+	assert.Equal(t, nb, calledBackWith)
+}

common/resourcesconfig/resourcesconfig.go

@@ -7,6 +7,7 @@ SPDX-License-Identifier: Apache-2.0
 package resourcesconfig
 
 import (
+	"github.com/hyperledger/fabric/common/channelconfig"
 	configtxapi "github.com/hyperledger/fabric/common/configtx/api"
 	"github.com/hyperledger/fabric/common/policies"
 )
@@ -53,11 +54,16 @@ type Resources interface {
 	ConfigtxManager() configtxapi.Manager
 
 	// PolicyManager returns a policy manager which can resolve names both in the /Channel and /Resources namespaces.
+	// Note, the result of this method is almost definitely the one you want.  Calling ChannelConfig().PolicyManager()
+	// will return a policy manager which can only resolve policies in the /Channel namespace.
 	PolicyManager() policies.Manager
 
 	// APIPolicyMapper returns a way to map API names to policies governing their invocation.
 	APIPolicyMapper() PolicyMapper
 
 	// ChaincodeRegistery returns a way to query for chaincodes defined in this channel.
 	ChaincodeRegistry() ChaincodeRegistry
+
+	// ChannelConfig returns the channelconfig.Resources which this config depends on.
+	ChannelConfig() channelconfig.Resources
 }

core/peer/peer.go

@@ -411,6 +411,17 @@ func GetLedger(cid string) ledger.PeerLedger {
 	return nil
 }
 
+// GetChannelConfig returns the channel configuration of the chain with channel ID. Note that this
+// call returns nil if chain cid has not been created.
+func GetChannelConfig(cid string) channelconfig.Resources {
+	chains.RLock()
+	defer chains.RUnlock()
+	if c, ok := chains.list[cid]; ok {
+		return c.cs
+	}
+	return nil
+}
+
 // GetPolicyManager returns the policy manager of the chain with chain ID. Note that this
 // call returns nil if chain cid has not been created.
 func GetPolicyManager(cid string) policies.Manager {