[FAB-1279] Add dynamic chain creation path

This changeset introduces the runtime  chain creation logic into the
orderer.  It's presently only exercised via unit tests, pending the
creation of bdd and sample clients to exercise this behavior.

The chain creation checking is still somewhat limited, and does not
enforce the sorts of keys which can be created, or do real policy
evaluation at this point, but the plug points are there.

Change-Id: I44b904884d81f1e6e279c7214238950b8c6b059f
Signed-off-by: Jason Yellick <[email protected]>

Author: Jason Yellick

orderer/common/broadcast/broadcast.go

@@ -39,7 +39,13 @@ type Handler interface {
 
 // SupportManager provides a way for the Handler to look up the Support for a chain
 type SupportManager interface {
+	// GetChain gets the chain support for a given ChainID
 	GetChain(chainID string) (Support, bool)
+
+	// ProposeChain accepts a configuration transaction for a chain which does not already exists
+	// The status returned is whether the proposal is accepted for consideration, only after consensus
+	// occurs will the proposal be committed or rejected
+	ProposeChain(env *cb.Envelope) cb.Status
 }
 
 // Support provides the backing resources needed to support broadcast on a chain
@@ -125,21 +131,27 @@ func (b *broadcaster) queueEnvelopes(srv ab.AtomicBroadcast_BroadcastServer) err
 
 		support, ok := b.bs.sm.GetChain(payload.Header.ChainHeader.ChainID)
 		if !ok {
-			// XXX Hook in chain creation logic here
-			panic("Unimplemented")
-		}
-
-		_, filterErr := support.Filters().Apply(msg)
-
-		if filterErr != nil {
-			logger.Debugf("Rejecting broadcast message")
-			err = srv.Send(&ab.BroadcastResponse{Status: cb.Status_BAD_REQUEST})
+			// Chain not found, maybe create one?
+			if payload.Header.ChainHeader.Type != int32(cb.HeaderType_CONFIGURATION_TRANSACTION) {
+				err = srv.Send(&ab.BroadcastResponse{Status: cb.Status_NOT_FOUND})
+			} else {
+				logger.Debugf("Proposing new chain")
+				err = srv.Send(&ab.BroadcastResponse{Status: b.bs.sm.ProposeChain(msg)})
+			}
 		} else {
-			select {
-			case b.queue <- &msgAndSupport{msg: msg, support: support}:
-				err = srv.Send(&ab.BroadcastResponse{Status: cb.Status_SUCCESS})
-			default:
-				err = srv.Send(&ab.BroadcastResponse{Status: cb.Status_SERVICE_UNAVAILABLE})
+			// Normal transaction for existing chain
+			_, filterErr := support.Filters().Apply(msg)
+
+			if filterErr != nil {
+				logger.Debugf("Rejecting broadcast message")
+				err = srv.Send(&ab.BroadcastResponse{Status: cb.Status_BAD_REQUEST})
+			} else {
+				select {
+				case b.queue <- &msgAndSupport{msg: msg, support: support}:
+					err = srv.Send(&ab.BroadcastResponse{Status: cb.Status_SUCCESS})
+				default:
+					err = srv.Send(&ab.BroadcastResponse{Status: cb.Status_SERVICE_UNAVAILABLE})
+				}
 			}
 		}
 

orderer/common/broadcast/broadcast_test.go

@@ -20,10 +20,10 @@ import (
 	"fmt"
 	"testing"
 
-	"github.com/golang/protobuf/proto"
 	"github.com/hyperledger/fabric/orderer/common/filter"
 	cb "github.com/hyperledger/fabric/protos/common"
 	ab "github.com/hyperledger/fabric/protos/orderer"
+	"github.com/hyperledger/fabric/protos/utils"
 
 	"google.golang.org/grpc"
 )
@@ -65,6 +65,19 @@ func (mm *mockSupportManager) GetChain(chainID string) (Support, bool) {
 	return chain, ok
 }
 
+func (mm *mockSupportManager) ProposeChain(configTx *cb.Envelope) cb.Status {
+	payload := utils.ExtractPayloadOrPanic(configTx)
+
+	mm.chains[string(payload.Header.ChainHeader.ChainID)] = &mockSupport{
+		filters: filter.NewRuleSet([]filter.Rule{
+			filter.EmptyRejectRule,
+			filter.AcceptRule,
+		}),
+		queue: make(chan *cb.Envelope),
+	}
+	return cb.Status_SUCCESS
+}
+
 func (mm *mockSupportManager) halt() {
 	for _, chain := range mm.chains {
 		chain.halt()
@@ -95,6 +108,21 @@ func (ms *mockSupport) halt() {
 	}
 }
 
+func makeConfigMessage(chainID string) *cb.Envelope {
+	payload := &cb.Payload{
+		Data: utils.MarshalOrPanic(&cb.ConfigurationEnvelope{}),
+		Header: &cb.Header{
+			ChainHeader: &cb.ChainHeader{
+				ChainID: chainID,
+				Type:    int32(cb.HeaderType_CONFIGURATION_TRANSACTION),
+			},
+		},
+	}
+	return &cb.Envelope{
+		Payload: utils.MarshalOrPanic(payload),
+	}
+}
+
 func makeMessage(chainID string, data []byte) *cb.Envelope {
 	payload := &cb.Payload{
 		Data: data,
@@ -104,14 +132,9 @@ func makeMessage(chainID string, data []byte) *cb.Envelope {
 			},
 		},
 	}
-	data, err := proto.Marshal(payload)
-	if err != nil {
-		panic(err)
+	return &cb.Envelope{
+		Payload: utils.MarshalOrPanic(payload),
 	}
-	env := &cb.Envelope{
-		Payload: data,
-	}
-	return env
 }
 
 func getMultichainManager() *mockSupportManager {
@@ -200,3 +223,44 @@ func TestEmptyEnvelope(t *testing.T) {
 	}
 
 }
+
+func TestBadChainID(t *testing.T) {
+	mm := getMultichainManager()
+	defer mm.halt()
+	bh := NewHandlerImpl(mm, 2)
+	m := newMockB()
+	defer close(m.recvChan)
+	go bh.Handle(m)
+
+	m.recvChan <- makeMessage("Wrong chain", []byte("Some bytes"))
+	reply := <-m.sendChan
+	if reply.Status != cb.Status_NOT_FOUND {
+		t.Fatalf("Should have rejected message to a chain which does not exist")
+	}
+
+}
+
+func TestNewChainID(t *testing.T) {
+	mm := getMultichainManager()
+	defer mm.halt()
+	bh := NewHandlerImpl(mm, 2)
+	m := newMockB()
+	defer close(m.recvChan)
+	go bh.Handle(m)
+
+	m.recvChan <- makeConfigMessage("New chain")
+	reply := <-m.sendChan
+	if reply.Status != cb.Status_SUCCESS {
+		t.Fatalf("Should have created a new chain, got %d", reply.Status)
+	}
+
+	if len(mm.chains) != 2 {
+		t.Fatalf("Should have created a new chain")
+	}
+
+	m.recvChan <- makeMessage("New chain", []byte("Some bytes"))
+	reply = <-m.sendChan
+	if reply.Status != cb.Status_SUCCESS {
+		t.Fatalf("Should have successfully sent message to new chain")
+	}
+}

orderer/multichain/chainsupport.go

@@ -67,6 +67,12 @@ type ChainSupport interface {
 	broadcast.Support
 	deliver.Support
 	ConsenterSupport
+
+	// ChainID returns the ChainID for this chain support
+	ChainID() string
+
+	// ConfigTxManager returns the corresponding configtx.Manager for this chain
+	ConfigTxManager() configtx.Manager
 }
 
 type chainSupport struct {
@@ -79,9 +85,16 @@ type chainSupport struct {
 	filters             *filter.RuleSet
 }
 
-func newChainSupport(configManager configtx.Manager, policyManager policies.Manager, backing rawledger.ReadWriter, sharedConfigManager sharedconfig.Manager, consenters map[string]Consenter) *chainSupport {
+func newChainSupport(
+	filters *filter.RuleSet,
+	configManager configtx.Manager,
+	policyManager policies.Manager,
+	backing rawledger.ReadWriter,
+	sharedConfigManager sharedconfig.Manager,
+	consenters map[string]Consenter,
+) *chainSupport {
+
 	batchSize := sharedConfigManager.BatchSize() // XXX this needs to be pushed deeper so that the blockcutter queries it after each write for reconfiguration support
-	filters := createBroadcastRuleset(configManager)
 	cutter := blockcutter.NewReceiverImpl(batchSize, filters)
 	consenterType := sharedConfigManager.ConsensusType()
 	consenter, ok := consenters[consenterType]
@@ -107,12 +120,24 @@ func newChainSupport(configManager configtx.Manager, policyManager policies.Mana
 	return cs
 }
 
-func createBroadcastRuleset(configManager configtx.Manager) *filter.RuleSet {
+// createStandardFilters creates the set of filters for a normal (non-system) chain
+func createStandardFilters(configManager configtx.Manager) *filter.RuleSet {
 	return filter.NewRuleSet([]filter.Rule{
 		filter.EmptyRejectRule,
 		configtx.NewFilter(configManager),
 		filter.AcceptRule,
 	})
+
+}
+
+// createSystemChainFilters creates the set of filters for the ordering system chain
+func createSystemChainFilters(ml *multiLedger, configManager configtx.Manager) *filter.RuleSet {
+	return filter.NewRuleSet([]filter.Rule{
+		filter.EmptyRejectRule,
+		newSystemChainFilter(ml),
+		configtx.NewFilter(configManager),
+		filter.AcceptRule,
+	})
 }
 
 func (cs *chainSupport) start() {
@@ -123,10 +148,14 @@ func (cs *chainSupport) SharedConfig() sharedconfig.Manager {
 	return cs.sharedConfigManager
 }
 
-func (cs *chainSupport) ConfigManager() configtx.Manager {
+func (cs *chainSupport) ConfigTxManager() configtx.Manager {
 	return cs.configManager
 }
 
+func (cs *chainSupport) ChainID() string {
+	return cs.configManager.ChainID()
+}
+
 func (cs *chainSupport) PolicyManager() policies.Manager {
 	return cs.policyManager
 }

orderer/multichain/manager.go

@@ -17,7 +17,7 @@ limitations under the License.
 package multichain
 
 import (
-	"sync"
+	"fmt"
 
 	"github.com/hyperledger/fabric/orderer/common/configtx"
 	"github.com/hyperledger/fabric/orderer/common/policies"
@@ -48,13 +48,18 @@ func init() {
 type Manager interface {
 	// GetChain retrieves the chain support for a chain (and whether it exists)
 	GetChain(chainID string) (ChainSupport, bool)
+
+	// ProposeChain accepts a configuration transaction for a chain which does not already exists
+	// The status returned is whether the proposal is accepted for consideration, only after consensus
+	// occurs will the proposal be committed or rejected
+	ProposeChain(env *cb.Envelope) cb.Status
 }
 
 type multiLedger struct {
 	chains        map[string]*chainSupport
 	consenters    map[string]Consenter
 	ledgerFactory rawledger.Factory
-	mutex         sync.Mutex
+	sysChain      *systemChain
 }
 
 // getConfigTx, this should ultimately be done more intelligently, but for now, we search the whole chain for txs and pick the last config one
@@ -104,6 +109,7 @@ func NewManagerImpl(ledgerFactory rawledger.Factory, consenters map[string]Conse
 	ml := &multiLedger{
 		chains:        make(map[string]*chainSupport),
 		ledgerFactory: ledgerFactory,
+		consenters:    consenters,
 	}
 
 	existingChains := ledgerFactory.ChainIDs()
@@ -118,23 +124,43 @@ func NewManagerImpl(ledgerFactory rawledger.Factory, consenters map[string]Conse
 		}
 		configManager, policyManager, backingLedger, sharedConfigManager := ml.newResources(configTx)
 		chainID := configManager.ChainID()
-		ml.chains[chainID] = newChainSupport(configManager, policyManager, backingLedger, sharedConfigManager, consenters)
-	}
 
-	for _, cs := range ml.chains {
-		cs.start()
+		if sharedConfigManager.ChainCreators() != nil {
+			if ml.sysChain != nil {
+				logger.Fatalf("There appear to be two system chains %x and %x", ml.sysChain.support.ChainID(), chainID)
+			}
+			logger.Debugf("Starting with system chain: %x", chainID)
+			chain := newChainSupport(createSystemChainFilters(ml, configManager), configManager, policyManager, backingLedger, sharedConfigManager, consenters)
+			ml.chains[string(chainID)] = chain
+			ml.sysChain = newSystemChain(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: %x", chainID)
+			chain := newChainSupport(createStandardFilters(configManager), configManager, policyManager, backingLedger, sharedConfigManager, consenters)
+			ml.chains[string(chainID)] = chain
+			chain.start()
+		}
+
 	}
 
 	return ml
 }
 
+// ProposeChain accepts a configuration transaction for a chain which does not already exists
+// The status returned is whether the proposal is accepted for consideration, only after consensus
+// occurs will the proposal be committed or rejected
+func (ml *multiLedger) ProposeChain(env *cb.Envelope) cb.Status {
+	return ml.sysChain.proposeChain(env)
+}
+
 // 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) newResources(configTx *cb.Envelope) (configtx.Manager, policies.Manager, rawledger.ReadWriter, sharedconfig.Manager) {
+func newConfigTxManagerAndHandlers(configEnvelope *cb.ConfigurationEnvelope) (configtx.Manager, policies.Manager, sharedconfig.Manager, error) {
 	policyManager := policies.NewManagerImpl(xxxCryptoHelper{})
 	sharedConfigManager := sharedconfig.NewManagerImpl()
 	configHandlerMap := make(map[cb.ConfigurationItem_ConfigurationType]configtx.Handler)
@@ -150,6 +176,15 @@ func (ml *multiLedger) newResources(configTx *cb.Envelope) (configtx.Manager, po
 		}
 	}
 
+	configManager, err := configtx.NewConfigurationManager(configEnvelope, policyManager, configHandlerMap)
+	if err != nil {
+		return nil, nil, nil, fmt.Errorf("Error unpacking configuration transaction: %s", err)
+	}
+
+	return configManager, policyManager, sharedConfigManager, nil
+}
+
+func (ml *multiLedger) newResources(configTx *cb.Envelope) (configtx.Manager, policies.Manager, rawledger.ReadWriter, sharedconfig.Manager) {
 	payload := &cb.Payload{}
 	err := proto.Unmarshal(configTx.Payload, payload)
 	if err != nil {
@@ -162,9 +197,10 @@ func (ml *multiLedger) newResources(configTx *cb.Envelope) (configtx.Manager, po
 		logger.Fatalf("Error unmarshaling a config transaction to config envelope: %s", err)
 	}
 
-	configManager, err := configtx.NewConfigurationManager(configEnvelope, policyManager, configHandlerMap)
+	configManager, policyManager, sharedConfigManager, err := newConfigTxManagerAndHandlers(configEnvelope)
+
 	if err != nil {
-		logger.Fatalf("Error unpacking configuration transaction: %s", err)
+		logger.Fatalf("Error creating configtx manager and handlers: %s", err)
 	}
 
 	chainID := configManager.ChainID()
@@ -176,3 +212,28 @@ func (ml *multiLedger) newResources(configTx *cb.Envelope) (configtx.Manager, po
 
 	return configManager, policyManager, ledger, sharedConfigManager
 }
+
+func (ml *multiLedger) systemChain() *systemChain {
+	return ml.sysChain
+}
+
+func (ml *multiLedger) newChain(configtx *cb.Envelope) {
+	configManager, policyManager, backingLedger, sharedConfig := ml.newResources(configtx)
+	backingLedger.Append([]*cb.Envelope{configtx}, nil)
+
+	// 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(configManager), configManager, policyManager, backingLedger, sharedConfig, ml.consenters)
+	chainID := configManager.ChainID()
+
+	logger.Debugf("Created and starting new chain %s", chainID)
+
+	newChains[string(chainID)] = cs
+	cs.start()
+
+	ml.chains = newChains
+}