[FAB-13180] Orderer: auto-join existing inactive chains

This change set makes cluster type OSNs autonomously detect channels
that exist and that they should be part of (the channel configuration
has their public credentials as a consenter for the channel),
but that they do not run chains for, or have the blocks in their ledger.

This can happen from several reasons:
- The OSN is added to an existing chain, and since it didn't participate
  in the chain so far, it didn't get the blocks that tell it is now
  part of the channel.
- The OSN tried to detect whether it is part of a channel, but it
  wasn't able, because all OSNs of the system channel returned
  service-unavailable. This can happen if:
  - a leader election takes place
  - the network is acting up so the leadership was lost
  - a channel has been deserted (all OSNs left it).

To take care of such use cases, all OSNs now:
- Track inactive chains that they know of, but they do not participate in
- Periodically(*) probe the system channel OSNs to see if they are now
  part of these chains or not.
- If so, then they replicate the chains, and create instances of them,
  and replace the instances of the inactive chains in the registrar
  with the new instances of type etcdraft.

(*) - 10 seconds after boot, then after 20 seconds,
      then after 40 seconds, etc. etc. eventually- every 5 minutes.

Change-Id: I3c2a84e6f4f402e011e7a895345b3d3982247083
Signed-off-by: yacovm <yacovm@il.ibm.com>

Author: yacovm

integration/e2e/etcdraft_reconfig_test.go

@@ -316,8 +316,30 @@ var _ = Describe("EndToEnd reconfiguration and onboarding", func() {
 			assertInvoke(network, peer, o4, mycc3.Name, "testchannel3", "channel testchannel3 is not serviced by me", 1)
 			Expect(string(orderer4Runner.Err().Contents())).To(ContainSubstring("I do not belong to channel testchannel2 or am forbidden pulling it (not in the channel), skipping chain retrieval"))
 			Expect(string(orderer4Runner.Err().Contents())).To(ContainSubstring("I do not belong to channel testchannel3 or am forbidden pulling it (forbidden), skipping chain retrieval"))
+
+			By("Adding orderer4 to testchannel2")
+			nwo.AddConsenter(network, peer, o1, "testchannel2", etcdraft.Consenter{
+				ServerTlsCert: orderer4Certificate,
+				ClientTlsCert: orderer4Certificate,
+				Host:          "127.0.0.1",
+				Port:          uint32(network.OrdererPort(o4, nwo.ListenPort)),
+			})
+
+			By("Waiting for orderer4 and to replicate testchannel2")
+			assertBlockReception(map[string]int{
+				"testchannel2": 2,
+			}, []*nwo.Orderer{o4}, peer, network)
+
 			By("Ensuring that all orderers don't log errors to the log")
 			assertNoErrorsAreLogged(ordererRunners)
+
+			By("Submitting a transaction through orderer4")
+			assertInvoke(network, peer, o4, mycc2.Name, "testchannel2", "Chaincode invoke successful. result: status:200", 0)
+
+			By("And ensuring it is propagated amongst all orderers")
+			assertBlockReception(map[string]int{
+				"testchannel2": 3,
+			}, orderers, peer, network)
 		})
 	})
 })
@@ -467,14 +489,17 @@ func waitForBlockReception(o *nwo.Orderer, submitter *nwo.Peer, network *nwo.Net
 	Eventually(func() string {
 		sess, err := network.OrdererAdminSession(o, submitter, c)
 		Expect(err).NotTo(HaveOccurred())
-		Eventually(sess, network.EventuallyTimeout).Should(gexec.Exit(0))
+		Eventually(sess, network.EventuallyTimeout).Should(gexec.Exit())
+		if sess.ExitCode() != 0 {
+			return fmt.Sprintf("exit code is %d: %s", sess.ExitCode(), string(sess.Err.Contents()))
+		}
 		sessErr := string(sess.Err.Contents())
 		expected := fmt.Sprintf("Received block: %d", blockSeq)
 		if strings.Contains(sessErr, expected) {
 			return ""
 		}
 		return sessErr
-	}, network.EventuallyTimeout).Should(BeEmpty())
+	}, time.Minute, time.Second).Should(BeEmpty())
 }
 
 func assertNoErrorsAreLogged(ordererRunners []*ginkgomon.Runner) {

orderer/common/cluster/replication.go

@@ -77,14 +77,15 @@ type ChannelLister interface {
 
 // Replicator replicates chains
 type Replicator struct {
-	Filter           func(string) bool
-	SystemChannel    string
-	ChannelLister    ChannelLister
-	Logger           *flogging.FabricLogger
-	Puller           *BlockPuller
-	BootBlock        *common.Block
-	AmIPartOfChannel selfMembershipPredicate
-	LedgerFactory    LedgerFactory
+	DoNotPanicIfClusterNotReachable bool
+	Filter                          func(string) bool
+	SystemChannel                   string
+	ChannelLister                   ChannelLister
+	Logger                          *flogging.FabricLogger
+	Puller                          *BlockPuller
+	BootBlock                       *common.Block
+	AmIPartOfChannel                selfMembershipPredicate
+	LedgerFactory                   LedgerFactory
 }
 
 // IsReplicationNeeded returns whether replication is needed,
@@ -112,13 +113,20 @@ func (r *Replicator) IsReplicationNeeded() (bool, error) {
 }
 
 // ReplicateChains pulls chains and commits them.
-func (r *Replicator) ReplicateChains() {
+// Returns the names of the chains replicated successfully.
+func (r *Replicator) ReplicateChains() []string {
+	var replicatedChains []string
 	channels := r.discoverChannels()
 	pullHints := r.channelsToPull(channels)
 	totalChannelCount := len(pullHints.channelsToPull) + len(pullHints.channelsNotToPull)
 	r.Logger.Info("Found myself in", len(pullHints.channelsToPull), "channels out of", totalChannelCount, ":", pullHints)
 	for _, channel := range pullHints.channelsToPull {
-		r.PullChannel(channel.ChannelName)
+		err := r.PullChannel(channel.ChannelName)
+		if err == nil {
+			replicatedChains = append(replicatedChains, channel.ChannelName)
+		} else {
+			r.Logger.Warningf("Failed pulling channel %s: %v", channel.ChannelName, err)
+		}
 	}
 	// Next, just commit the genesis blocks of the channels we shouldn't pull.
 	for _, channel := range pullHints.channelsNotToPull {
@@ -137,9 +145,10 @@ func (r *Replicator) ReplicateChains() {
 	}
 
 	// Last, pull the system chain
-	if err := r.PullChannel(r.SystemChannel); err != nil {
+	if err := r.PullChannel(r.SystemChannel); err != nil && err != ErrSkipped {
 		r.Logger.Panicf("Failed pulling system channel: %v", err)
 	}
+	return replicatedChains
 }
 
 func (r *Replicator) discoverChannels() []ChannelGenesisBlock {
@@ -156,7 +165,7 @@ func (r *Replicator) discoverChannels() []ChannelGenesisBlock {
 func (r *Replicator) PullChannel(channel string) error {
 	if !r.Filter(channel) {
 		r.Logger.Info("Channel", channel, "shouldn't be pulled. Skipping it")
-		return nil
+		return ErrSkipped
 	}
 	r.Logger.Info("Pulling channel", channel)
 	puller := r.Puller.Clone()
@@ -239,7 +248,7 @@ type channelPullHints struct {
 }
 
 func (r *Replicator) channelsToPull(channels GenesisBlocks) channelPullHints {
-	r.Logger.Info("Will now pull channels:", channels.Names())
+	r.Logger.Info("Will now attempt to pull channels:", channels.Names())
 	var channelsNotToPull []ChannelGenesisBlock
 	var channelsToPull []ChannelGenesisBlock
 	for _, channel := range channels {
@@ -266,9 +275,12 @@ func (r *Replicator) channelsToPull(channels GenesisBlocks) channelPullHints {
 			continue
 		}
 		if err != nil {
-			r.Logger.Panicf("Failed classifying whether I belong to channel %s: %v, skipping chain retrieval", channel.ChannelName, err)
+			if !r.DoNotPanicIfClusterNotReachable {
+				r.Logger.Panicf("Failed classifying whether I belong to channel %s: %v, skipping chain retrieval", channel.ChannelName, err)
+			}
 			continue
 		}
+		r.Logger.Infof("I need to pull channel %s", channel.ChannelName)
 		channelsToPull = append(channelsToPull, channel)
 	}
 	return channelPullHints{
@@ -360,6 +372,9 @@ type ChainInspector struct {
 	LastConfigBlock *common.Block
 }
 
+// ErrSkipped denotes that replicating a chain was skipped
+var ErrSkipped = errors.New("skipped")
+
 // ErrForbidden denotes that an ordering node refuses sending blocks due to access control.
 var ErrForbidden = errors.New("forbidden")
 
@@ -520,6 +535,7 @@ func ChannelCreationBlockToGenesisBlock(block *common.Block) (*common.Block, err
 		return nil, err
 	}
 	block.Data.Data = [][]byte{payload.Data}
+	block.Header.DataHash = block.Data.Hash()
 	block.Header.Number = 0
 	block.Header.PreviousHash = nil
 	metadata := &common.BlockMetadata{

orderer/common/cluster/replication_test.go

@@ -1399,5 +1399,5 @@ func TestFilter(t *testing.T) {
 		},
 		Logger: logger,
 	}
-	assert.Nil(t, r.PullChannel("foo"))
+	assert.Equal(t, cluster.ErrSkipped, r.PullChannel("foo"))
 }

orderer/common/localconfig/config.go

@@ -58,18 +58,19 @@ type General struct {
 }
 
 type Cluster struct {
-	ListenAddress           string
-	ListenPort              uint16
-	ServerCertificate       string
-	ServerPrivateKey        string
-	ClientCertificate       string
-	ClientPrivateKey        string
-	RootCAs                 []string
-	DialTimeout             time.Duration
-	RPCTimeout              time.Duration
-	ReplicationBufferSize   int
-	ReplicationPullTimeout  time.Duration
-	ReplicationRetryTimeout time.Duration
+	ListenAddress                        string
+	ListenPort                           uint16
+	ServerCertificate                    string
+	ServerPrivateKey                     string
+	ClientCertificate                    string
+	ClientPrivateKey                     string
+	RootCAs                              []string
+	DialTimeout                          time.Duration
+	RPCTimeout                           time.Duration
+	ReplicationBufferSize                int
+	ReplicationPullTimeout               time.Duration
+	ReplicationRetryTimeout              time.Duration
+	ReplicationBackgroundRefreshInterval time.Duration
 }
 
 // Keepalive contains configuration for gRPC servers.

orderer/common/multichannel/registrar.go

@@ -11,6 +11,7 @@ package multichannel
 
 import (
 	"fmt"
+	"reflect"
 	"sync"
 
 	"github.com/hyperledger/fabric/common/channelconfig"
@@ -104,7 +105,9 @@ type Registrar struct {
 	callbacks          []func(bundle *channelconfig.Bundle)
 }
 
-func getConfigTx(reader blockledger.Reader) *cb.Envelope {
+// ConfigBlock retrieves the last configuration block from the given ledger.
+// Panics on failure.
+func ConfigBlock(reader blockledger.Reader) *cb.Block {
 	lastBlock := blockledger.GetBlock(reader, reader.Height()-1)
 	index, err := utils.GetLastConfigIndexFromBlock(lastBlock)
 	if err != nil {
@@ -115,7 +118,11 @@ func getConfigTx(reader blockledger.Reader) *cb.Envelope {
 		logger.Panicf("Config block does not exist")
 	}
 
-	return utils.ExtractEnvelopeOrPanic(configBlock, 0)
+	return configBlock
+}
+
+func configTx(reader blockledger.Reader) *cb.Envelope {
+	return utils.ExtractEnvelopeOrPanic(ConfigBlock(reader), 0)
 }
 
 // NewRegistrar produces an instance of a *Registrar.
@@ -140,7 +147,7 @@ func (r *Registrar) Initialize(consenters map[string]consensus.Consenter) {
 		if err != nil {
 			logger.Panicf("Ledger factory reported chainID %s but could not retrieve it: %s", chainID, err)
 		}
-		configTx := getConfigTx(rl)
+		configTx := configTx(rl)
 		if configTx == nil {
 			logger.Panic("Programming error, configTx should never be nil here")
 		}
@@ -275,12 +282,30 @@ func (r *Registrar) newLedgerResources(configTx *cb.Envelope) *ledgerResources {
 	}
 }
 
+// CreateChain makes the Registrar create a chain with the given name.
+func (r *Registrar) CreateChain(chainName string) {
+	lf, err := r.ledgerFactory.GetOrCreate(chainName)
+	if err != nil {
+		logger.Panicf("Failed obtaining ledger factory for %s: %v", chainName, err)
+	}
+	chain := r.GetChain(chainName)
+	if chain != nil {
+		logger.Infof("A chain of type %v for channel %s already exists. "+
+			"Halting it.", reflect.TypeOf(chain.Chain), chainName)
+		chain.Halt()
+	}
+	r.newChain(configTx(lf))
+}
+
 func (r *Registrar) newChain(configtx *cb.Envelope) {
 	r.lock.Lock()
 	defer r.lock.Unlock()
 
 	ledgerResources := r.newLedgerResources(configtx)
-	ledgerResources.Append(blockledger.CreateNextBlock(ledgerResources, []*cb.Envelope{configtx}))
+	// If we have no blocks, we need to create the genesis block ourselves.
+	if ledgerResources.Height() == 0 {
+		ledgerResources.Append(blockledger.CreateNextBlock(ledgerResources, []*cb.Envelope{configtx}))
+	}
 
 	// Copy the map to allow concurrent reads from broadcast/deliver while the new chainSupport is
 	newChains := make(map[string]*ChainSupport)

orderer/common/multichannel/registrar_test.go

@@ -80,7 +80,7 @@ func TestGetConfigTx(t *testing.T) {
 	block.Metadata.Metadata[cb.BlockMetadataIndex_LAST_CONFIG] = utils.MarshalOrPanic(&cb.Metadata{Value: utils.MarshalOrPanic(&cb.LastConfig{Index: 7})})
 	rl.Append(block)
 
-	pctx := getConfigTx(rl)
+	pctx := configTx(rl)
 	assert.True(t, proto.Equal(pctx, ctx), "Did not select most recent config transaction")
 }
 
@@ -94,11 +94,11 @@ func TestGetConfigTxFailure(t *testing.T) {
 		}))
 	}
 	rl.Append(blockledger.CreateNextBlock(rl, []*cb.Envelope{makeNormalTx(genesisconfig.TestChainID, 11)}))
-	assert.Panics(t, func() { getConfigTx(rl) }, "Should have panicked because there was no config tx")
+	assert.Panics(t, func() { configTx(rl) }, "Should have panicked because there was no config tx")
 
 	block := blockledger.CreateNextBlock(rl, []*cb.Envelope{makeNormalTx(genesisconfig.TestChainID, 12)})
 	block.Metadata.Metadata[cb.BlockMetadataIndex_LAST_CONFIG] = []byte("bad metadata")
-	assert.Panics(t, func() { getConfigTx(rl) }, "Should have panicked because of bad last config metadata")
+	assert.Panics(t, func() { configTx(rl) }, "Should have panicked because of bad last config metadata")
 }
 
 // This test checks to make sure the orderer refuses to come up if it cannot find a system channel
@@ -167,6 +167,41 @@ func TestManagerImpl(t *testing.T) {
 	}
 }
 
+func TestCreateChain(t *testing.T) {
+	lf, _ := NewRAMLedgerAndFactory(10)
+
+	consenters := make(map[string]consensus.Consenter)
+	consenters[conf.Orderer.OrdererType] = &mockConsenter{}
+
+	manager := NewRegistrar(lf, mockCrypto(), &disabled.Provider{})
+	manager.Initialize(consenters)
+
+	ledger, err := lf.GetOrCreate("mychannel")
+	assert.NoError(t, err)
+
+	genesisBlock := encoder.New(conf).GenesisBlockForChannel("mychannel")
+	ledger.Append(genesisBlock)
+
+	// Before creating the chain, it doesn't exist
+	assert.Nil(t, manager.GetChain("mychannel"))
+	// After creating the chain, it exists
+	manager.CreateChain("mychannel")
+	chain := manager.GetChain("mychannel")
+	assert.NotNil(t, chain)
+	// A subsequent creation, replaces the chain.
+	manager.CreateChain("mychannel")
+	chain2 := manager.GetChain("mychannel")
+	assert.NotNil(t, chain2)
+	// They are not the same
+	assert.NotEqual(t, chain, chain2)
+	// The old chain is halted
+	_, ok := <-chain.Chain.(*mockChain).queue
+	assert.False(t, ok)
+	// The new chain is not halted: Close the channel to prove that.
+	close(chain2.Chain.(*mockChain).queue)
+
+}
+
 // This test brings up the entire system, with the mock consenter, including the broadcasters etc. and creates a new chain
 func TestNewChain(t *testing.T) {
 	expectedLastConfigBlockNumber := uint64(0)