Merge "[FAB-1524] Reinitialize chains on orderer restart"

orderer/main.go

@@ -117,8 +117,6 @@ func main() {
 		}
 	} else {
 		logger.Infof("Not bootstrapping because of existing chains")
-		logger.Warningf("XXXXXXX RESTART IS NOT CURRENTLY SUPPORTED XXXXXXXXX")
-		// XXX Remove this once restart is supported
 	}
 
 	if conf.Kafka.Verbose {

orderer/multichain/chainsupport_test.go

@@ -26,8 +26,6 @@ import (
 	cb "github.com/hyperledger/fabric/protos/common"
 	ab "github.com/hyperledger/fabric/protos/orderer"
 	"github.com/hyperledger/fabric/protos/utils"
-
-	"github.com/golang/protobuf/proto"
 )
 
 type mockLedgerReadWriter struct {
@@ -94,17 +92,11 @@ func TestWriteLastConfiguration(t *testing.T) {
 	cs := &chainSupport{ledger: ml, configManager: cm}
 
 	lastConfig := func(block *cb.Block) uint64 {
-		md := &cb.Metadata{}
-		err := proto.Unmarshal(block.Metadata.Metadata[cb.BlockMetadataIndex_LAST_CONFIGURATION], md)
-		if err != nil {
-			panic(err)
-		}
-		lc := &cb.LastConfiguration{}
-		err = proto.Unmarshal(md.Value, lc)
+		index, err := utils.GetLastConfigurationIndexFromBlock(block)
 		if err != nil {
 			panic(err)
 		}
-		return lc.Index
+		return index
 	}
 
 	expected := uint64(0)

orderer/multichain/manager.go

@@ -25,7 +25,7 @@ import (
 	"github.com/hyperledger/fabric/orderer/common/sharedconfig"
 	"github.com/hyperledger/fabric/orderer/rawledger"
 	cb "github.com/hyperledger/fabric/protos/common"
-	ab "github.com/hyperledger/fabric/protos/orderer"
+	"github.com/hyperledger/fabric/protos/utils"
 	"github.com/op/go-logging"
 
 	"github.com/golang/protobuf/proto"
@@ -59,48 +59,18 @@ type multiLedger struct {
 	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
 func getConfigTx(reader rawledger.Reader) *cb.Envelope {
-	var lastConfigTx *cb.Envelope
-
-	it, _ := reader.Iterator(&ab.SeekPosition{Type: &ab.SeekPosition_Oldest{}})
-	// Iterate over the blockchain, looking for config transactions, track the most recent one encountered
-	// this will be the transaction which is returned
-	for {
-		select {
-		case <-it.ReadyChan():
-			block, status := it.Next()
-			if status != cb.Status_SUCCESS {
-				logger.Fatalf("Error parsing blockchain at startup: %v", status)
-			}
-			// ConfigTxs should always be by themselves
-			if len(block.Data.Data) != 1 {
-				continue
-			}
-
-			maybeConfigTx := &cb.Envelope{}
-
-			err := proto.Unmarshal(block.Data.Data[0], maybeConfigTx)
-
-			if err != nil {
-				logger.Fatalf("Found data which was not an envelope: %s", err)
-			}
-
-			payload := &cb.Payload{}
-			if err = proto.Unmarshal(maybeConfigTx.Payload, payload); err != nil {
-				logger.Fatalf("Unable to unmarshal transaction payload: %s", err)
-			}
-
-			if payload.Header.ChainHeader.Type != int32(cb.HeaderType_CONFIGURATION_TRANSACTION) {
-				continue
-			}
-
-			logger.Debugf("Found configuration transaction for chain %x at block %d", payload.Header.ChainHeader.ChainID, block.Header.Number)
-			lastConfigTx = maybeConfigTx
-		default:
-			return lastConfigTx
-		}
+	lastBlock := rawledger.GetBlock(reader, reader.Height()-1)
+	index, err := utils.GetLastConfigurationIndexFromBlock(lastBlock)
+	if err != nil {
+		logger.Panicf("Chain did not have appropriately encoded last configuration in its latest block: %s", err)
 	}
+	configBlock := rawledger.GetBlock(reader, index)
+	if configBlock == nil {
+		logger.Panicf("Configuration block does not exist")
+	}
+
+	return utils.ExtractEnvelopeOrPanic(configBlock, 0)
 }
 
 // NewManagerImpl produces an instance of a Manager
@@ -126,16 +96,16 @@ func NewManagerImpl(ledgerFactory rawledger.Factory, consenters map[string]Conse
 
 		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.Fatalf("There appear to be two system chains %s and %s", ml.sysChain.support.ChainID(), chainID)
 			}
-			logger.Debugf("Starting with system chain: %x", chainID)
+			logger.Debugf("Starting with system chain: %s", 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)
+			logger.Debugf("Starting chain: %s", chainID)
 			chain := newChainSupport(createStandardFilters(configManager), configManager, policyManager, backingLedger, sharedConfigManager, consenters)
 			ml.chains[string(chainID)] = chain
 			chain.start()

orderer/multichain/manager_test.go

@@ -85,7 +85,10 @@ func TestGetConfigTx(t *testing.T) {
 	rl.Append(rawledger.CreateNextBlock(rl, []*cb.Envelope{makeConfigTx(provisional.TestChainID, 5)}))
 	ctx := makeConfigTx(provisional.TestChainID, 6)
 	rl.Append(rawledger.CreateNextBlock(rl, []*cb.Envelope{ctx}))
-	rl.Append(rawledger.CreateNextBlock(rl, []*cb.Envelope{makeNormalTx(provisional.TestChainID, 7)}))
+
+	block := rawledger.CreateNextBlock(rl, []*cb.Envelope{makeNormalTx(provisional.TestChainID, 7)})
+	block.Metadata.Metadata[cb.BlockMetadataIndex_LAST_CONFIGURATION] = utils.MarshalOrPanic(&cb.Metadata{Value: utils.MarshalOrPanic(&cb.LastConfiguration{Index: 7})})
+	rl.Append(block)
 
 	pctx := getConfigTx(rl)
 
@@ -104,11 +107,13 @@ func TestGetConfigTxFailure(t *testing.T) {
 		}))
 	}
 	rl.Append(rawledger.CreateNextBlock(rl, []*cb.Envelope{makeNormalTx(provisional.TestChainID, 11)}))
-	pctx := getConfigTx(rl)
+	defer func() {
+		if recover() == nil {
+			t.Fatalf("Should have panic-ed because there was no configuration tx")
+		}
+	}()
+	getConfigTx(rl)
 
-	if pctx != nil {
-		t.Fatalf("Should not have found a configuration tx")
-	}
 }
 
 // This test essentially brings the entire system up and is ultimately what main.go will replicate

orderer/rawledger/blackbox_test.go

@@ -39,20 +39,6 @@ type ledgerTestFactory interface {
 
 var testables []ledgerTestable
 
-func getBlock(number uint64, li ReadWriter) *cb.Block {
-	i, _ := li.Iterator(&ab.SeekPosition{Type: &ab.SeekPosition_Specified{Specified: &ab.SeekSpecified{Number: number}}})
-	select {
-	case <-i.ReadyChan():
-		block, status := i.Next()
-		if status != cb.Status_SUCCESS {
-			return nil
-		}
-		return block
-	default:
-		return nil
-	}
-}
-
 func allTest(t *testing.T, test func(ledgerTestFactory, *testing.T)) {
 	for _, lt := range testables {
 
@@ -83,7 +69,7 @@ func testInitialization(lf ledgerTestFactory, t *testing.T) {
 	if li.Height() != 1 {
 		t.Fatalf("Block height should be 1")
 	}
-	block := getBlock(0, li)
+	block := GetBlock(li, 0)
 	if block == nil {
 		t.Fatalf("Error retrieving genesis block")
 	}
@@ -109,7 +95,7 @@ func testReinitialization(lf ledgerTestFactory, t *testing.T) {
 	if li.Height() != 2 {
 		t.Fatalf("Block height should be 2")
 	}
-	block := getBlock(1, li)
+	block := GetBlock(li, 1)
 	if block == nil {
 		t.Fatalf("Error retrieving block 1")
 	}
@@ -124,7 +110,7 @@ func TestAddition(t *testing.T) {
 
 func testAddition(lf ledgerTestFactory, t *testing.T) {
 	_, li := lf.New()
-	genesis := getBlock(0, li)
+	genesis := GetBlock(li, 0)
 	if genesis == nil {
 		t.Fatalf("Could not retrieve genesis block")
 	}
@@ -134,7 +120,7 @@ func testAddition(lf ledgerTestFactory, t *testing.T) {
 	if li.Height() != 2 {
 		t.Fatalf("Block height should be 2")
 	}
-	block := getBlock(1, li)
+	block := GetBlock(li, 1)
 	if block == nil {
 		t.Fatalf("Error retrieving genesis block")
 	}
@@ -255,7 +241,7 @@ func testMultichain(lf ledgerTestFactory, t *testing.T) {
 		t.Fatalf("Error retrieving chain1: %s", err)
 	}
 
-	if b := getBlock(1, c1); !reflect.DeepEqual(c1b1, b) {
+	if b := GetBlock(c1, 1); !reflect.DeepEqual(c1b1, b) {
 		t.Fatalf("Did not properly store block 1 on chain 1:")
 	}
 
@@ -264,7 +250,7 @@ func testMultichain(lf ledgerTestFactory, t *testing.T) {
 		t.Fatalf("Error retrieving chain2: %s", err)
 	}
 
-	if b := getBlock(0, c2); reflect.DeepEqual(c2b0, b) {
+	if b := GetBlock(c2, 0); reflect.DeepEqual(c2b0, b) {
 		t.Fatalf("Did not properly store block 1 on chain 1")
 	}
 }

orderer/rawledger/rawledger.go

@@ -95,3 +95,18 @@ func CreateNextBlock(rl Reader, messages []*cb.Envelope) *cb.Block {
 
 	return block
 }
+
+// GetBlock is a utility method for retrieving a single block
+func GetBlock(rl Reader, index uint64) *cb.Block {
+	i, _ := rl.Iterator(&ab.SeekPosition{Type: &ab.SeekPosition_Specified{Specified: &ab.SeekSpecified{Number: index}}})
+	select {
+	case <-i.ReadyChan():
+		block, status := i.Next()
+		if status != cb.Status_SUCCESS {
+			return nil
+		}
+		return block
+	default:
+		return nil
+	}
+}

protos/utils/blockutils.go

@@ -48,6 +48,21 @@ func GetChainIDFromBlock(block *cb.Block) (string, error) {
 	return payload.Header.ChainHeader.ChainID, nil
 }
 
+// GetLastConfigurationIndexFromBlock retrieves the index of the last configuration block as encoded in the block metadata
+func GetLastConfigurationIndexFromBlock(block *cb.Block) (uint64, error) {
+	md := &cb.Metadata{}
+	err := proto.Unmarshal(block.Metadata.Metadata[cb.BlockMetadataIndex_LAST_CONFIGURATION], md)
+	if err != nil {
+		return 0, err
+	}
+	lc := &cb.LastConfiguration{}
+	err = proto.Unmarshal(md.Value, lc)
+	if err != nil {
+		return 0, err
+	}
+	return lc.Index, nil
+}
+
 // GetBlockFromBlockBytes marshals the bytes into Block
 func GetBlockFromBlockBytes(blockBytes []byte) (*cb.Block, error) {
 	block := &cb.Block{}

protos/utils/configuration.go

@@ -22,24 +22,32 @@ import (
 	ab "github.com/hyperledger/fabric/protos/orderer"
 )
 
-const CreationPolicyKey = "CreationPolicy"
+const (
+	CreationPolicyKey = "CreationPolicy"
+	ChainCreatorsKey  = "ChainCreators"
+)
 
 // ChainCreationConfiguration creates a new chain creation configuration envelope from
 // the supplied creationPolicy, new chainID, and a template configuration envelope
 // The template configuration envelope will have the correct chainID set on all items,
 // and the first item will be a CreationPolicy which is ready for the signatures as
-// required by the policy
+// required by the policy, it also strips out the ChainCreators item as this is invalid
+// for the ordering system chain
 func ChainCreationConfiguration(creationPolicy, newChainID string, template *cb.ConfigurationEnvelope) *cb.ConfigurationEnvelope {
-	newConfigItems := make([]*cb.SignedConfigurationItem, len(template.Items))
+	var newConfigItems []*cb.SignedConfigurationItem
 	var hashBytes []byte
 
-	for i, item := range template.Items {
+	for _, item := range template.Items {
 		configItem := UnmarshalConfigurationItemOrPanic(item.ConfigurationItem)
+		if configItem.Type == cb.ConfigurationItem_Orderer && configItem.Key == ChainCreatorsKey {
+			continue
+		}
 		configItem.Header.ChainID = newChainID
-		newConfigItems[i] = &cb.SignedConfigurationItem{
+		newConfigItem := &cb.SignedConfigurationItem{
 			ConfigurationItem: MarshalOrPanic(configItem),
 		}
-		hashBytes = append(hashBytes, newConfigItems[i].ConfigurationItem...)
+		newConfigItems = append(newConfigItems, newConfigItem)
+		hashBytes = append(hashBytes, newConfigItem.ConfigurationItem...)
 	}
 
 	digest := cu.ComputeCryptoHash(hashBytes)