[FAB-4155] Make deliveryClient failover on 503 error

This change set addresses the client-side of the peer <---> orderer
connection in spite of internal errors in the ordering service, namely
kafka connection issues.

This commit:
- Exposes a method of the DeliveryClient - Disconnect()
  that makes the client re-connect to some ordering service node
  (can be the same one, it is randomly chosen)
- Adds logic in the blocks provider to invoke Disconnect()
  upon a reception of ServiceUnavailable(503) status from the orderer node
- Fixes a small bug - the seeding of time was missing from the connection
  producer and as a result it was always chosing orderer endpoints
  deterministically.

Change-Id: I949453237c88cb27f9a4fa96f7acd81974f29712
Signed-off-by: Yacov Manevich <[email protected]>

Author: yacovm

core/comm/producer.go

@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"math/rand"
 	"sync"
+	"time"
 
 	"github.com/hyperledger/fabric/common/flogging"
 	"google.golang.org/grpc"
@@ -91,22 +92,10 @@ func (cp *connProducer) UpdateEndpoints(endpoints []string) {
 func shuffle(a []string) []string {
 	n := len(a)
 	returnedSlice := make([]string, n)
+	rand.Seed(time.Now().UnixNano())
 	indices := rand.Perm(n)
 	for i, idx := range indices {
 		returnedSlice[i] = a[idx]
 	}
 	return returnedSlice
 }
-
-// filterOut receives a slice of strings and a string to filter out
-// and returns the slice without the string
-func filterOut(a []string, filteredOut string) []string {
-	var slice2Return []string
-	for _, s := range a {
-		if s == filteredOut {
-			continue
-		}
-		slice2Return = append(slice2Return, s)
-	}
-	return slice2Return
-}

core/deliverservice/blocksprovider/blocksprovider.go

@@ -79,6 +79,9 @@ type streamClient interface {
 
 	// Close closes the stream and its underlying connection
 	Close()
+
+	// Disconnect disconnects from the remote node
+	Disconnect()
 }
 
 // blocksProviderImpl the actual implementation for BlocksProvider interface
@@ -127,6 +130,10 @@ func (b *blocksProviderImpl) DeliverBlocks() {
 				return
 			}
 			logger.Warningf("[%s] Got error %v", b.chainID, t)
+			if t.Status == common.Status_SERVICE_UNAVAILABLE {
+				b.client.Disconnect()
+				continue
+			}
 		case *orderer.DeliverResponse_Block:
 			seqNum := t.Block.Header.Number
 

core/deliverservice/blocksprovider/blocksprovider_test.go

@@ -127,7 +127,7 @@ func TestBlocksProvider_CheckTerminationDeliveryResponseStatus(t *testing.T) {
 	tmp := struct{ mocks.MockBlocksDeliverer }{}
 
 	// Making mocked Recv() function to return DeliverResponse_Status to force block
-	// provider to fail and exit, cheking that in that case to block was actually
+	// provider to fail and exit, checking that in that case to block was actually
 	// delivered.
 	tmp.MockRecv = func(mock *mocks.MockBlocksDeliverer) (*orderer.DeliverResponse, error) {
 		return &orderer.DeliverResponse{
@@ -179,6 +179,71 @@ func TestBlocksProvider_CheckTerminationDeliveryResponseStatus(t *testing.T) {
 	}
 }
 
+func TestBlocksProvider_DeliveryServiceUnavailable(t *testing.T) {
+	sendBlock := func(seqNum uint64) *orderer.DeliverResponse {
+		return &orderer.DeliverResponse{
+			Type: &orderer.DeliverResponse_Block{
+				Block: &common.Block{
+					Header: &common.BlockHeader{
+						Number:       seqNum,
+						DataHash:     []byte{},
+						PreviousHash: []byte{},
+					},
+					Data: &common.BlockData{
+						Data: [][]byte{},
+					},
+				}},
+		}
+	}
+	sendStatus := func(status common.Status) *orderer.DeliverResponse {
+		return &orderer.DeliverResponse{
+			Type: &orderer.DeliverResponse_Status{
+				Status: status,
+			},
+		}
+	}
+
+	bd := mocks.MockBlocksDeliverer{DisconnectCalled: make(chan struct{}, 1)}
+	mcs := &mockMCS{}
+	mcs.On("VerifyBlock", mock.Anything).Return(nil)
+	gossipServiceAdapter := &mocks.MockGossipServiceAdapter{GossipBlockDisseminations: make(chan uint64, 2)}
+	provider := &blocksProviderImpl{
+		chainID: "***TEST_CHAINID***",
+		gossip:  gossipServiceAdapter,
+		client:  &bd,
+		mcs:     mcs,
+	}
+
+	attempts := int32(0)
+	bd.MockRecv = func(mock *mocks.MockBlocksDeliverer) (*orderer.DeliverResponse, error) {
+		atomic.AddInt32(&attempts, 1)
+		switch atomic.LoadInt32(&attempts) {
+		case int32(1):
+			return sendBlock(0), nil
+		case int32(2):
+			return sendStatus(common.Status_SERVICE_UNAVAILABLE), nil
+		case int32(3):
+			return sendBlock(1), nil
+		default:
+			provider.Stop()
+			return nil, errors.New("Stopping")
+		}
+	}
+
+	go provider.DeliverBlocks()
+	assert.Len(t, bd.DisconnectCalled, 0)
+	for i := 0; i < 2; i++ {
+		select {
+		case seq := <-gossipServiceAdapter.GossipBlockDisseminations:
+			assert.Equal(t, uint64(i), seq)
+		case <-time.After(time.Second * 3):
+			assert.Fail(t, "Didn't receive a block within a timely manner")
+		}
+	}
+	// Make sure disconnect was called in between the deliveries
+	assert.Len(t, bd.DisconnectCalled, 1)
+}
+
 func TestBlockFetchFailure(t *testing.T) {
 	rcvr := func(mock *mocks.MockBlocksDeliverer) (*orderer.DeliverResponse, error) {
 		return nil, errors.New("Failed fetching block")

core/deliverservice/client.go

@@ -120,7 +120,7 @@ func (bc *broadcastClient) doAction(action func() (interface{}, error)) (interfa
 	}
 	resp, err := action()
 	if err != nil {
-		bc.disconnect()
+		bc.Disconnect()
 		return nil, err
 	}
 	return resp, nil
@@ -151,7 +151,7 @@ func (bc *broadcastClient) connect() error {
 	}
 	// If we reached here, lets make sure connection is closed
 	// and nullified before we return
-	bc.disconnect()
+	bc.Disconnect()
 	return err
 }
 
@@ -191,6 +191,7 @@ func (bc *broadcastClient) shouldStop() bool {
 	return atomic.LoadInt32(&bc.stopFlag) == int32(1)
 }
 
+// Close makes the client close its connection and shut down
 func (bc *broadcastClient) Close() {
 	bc.Lock()
 	defer bc.Unlock()
@@ -205,7 +206,8 @@ func (bc *broadcastClient) Close() {
 	bc.conn.Close()
 }
 
-func (bc *broadcastClient) disconnect() {
+// Disconnect makes the client close the existing connection
+func (bc *broadcastClient) Disconnect() {
 	bc.Lock()
 	defer bc.Unlock()
 	if bc.conn == nil {

core/deliverservice/client_test.go

@@ -490,11 +490,14 @@ func TestCloseWhileRecv(t *testing.T) {
 	}
 	bc := NewBroadcastClient(cp, clFactory, setup, backoffStrategy)
 	var flag int32
-	time.AfterFunc(time.Second, func() {
+	go func() {
+		for fakeOrderer.ConnCount() == 0 {
+			time.Sleep(time.Second)
+		}
 		atomic.StoreInt32(&flag, int32(1))
 		bc.Close()
 		bc.Close() // Try to close a second time
-	})
+	}()
 	resp, err := bc.Recv()
 	// Ensure we returned because bc.Close() was called and not because some other reason
 	assert.Equal(t, int32(1), atomic.LoadInt32(&flag), "Recv returned before bc.Close() was called")
@@ -595,6 +598,72 @@ func TestProductionUsage(t *testing.T) {
 	cl.Close()
 }
 
+func TestDisconnect(t *testing.T) {
+	// Scenario: spawn 2 ordering service instances
+	// and a client.
+	// Have the client try to Recv() from one of them,
+	// and disconnect the client until it tries connecting
+	// to the other instance.
+
+	defer ensureNoGoroutineLeak(t)()
+	os1 := mocks.NewOrderer(5613, t)
+	os1.SetNextExpectedSeek(5)
+	os2 := mocks.NewOrderer(5614, t)
+	os2.SetNextExpectedSeek(5)
+
+	defer os1.Shutdown()
+	defer os2.Shutdown()
+
+	waitForConnectionToSomeOSN := func() {
+		for {
+			if os1.ConnCount() > 0 || os2.ConnCount() > 0 {
+				return
+			}
+			time.Sleep(time.Millisecond * 100)
+		}
+	}
+
+	connFact := func(endpoint string) (*grpc.ClientConn, error) {
+		return grpc.Dial(endpoint, grpc.WithInsecure(), grpc.WithBlock())
+	}
+	prod := comm.NewConnectionProducer(connFact, []string{"localhost:5613", "localhost:5614"})
+	clFact := func(cc *grpc.ClientConn) orderer.AtomicBroadcastClient {
+		return orderer.NewAtomicBroadcastClient(cc)
+	}
+	onConnect := func(bd blocksprovider.BlocksDeliverer) error {
+		return nil
+	}
+	retryPol := func(attemptNum int, elapsedTime time.Duration) (time.Duration, bool) {
+		return time.Millisecond * 10, attemptNum < 100
+	}
+
+	cl := NewBroadcastClient(prod, clFact, onConnect, retryPol)
+	stopChan := make(chan struct{})
+	go func() {
+		cl.Recv()
+		stopChan <- struct{}{}
+	}()
+	waitForConnectionToSomeOSN()
+	cl.Disconnect()
+
+	i := 0
+	for (os1.ConnCount() == 0 || os2.ConnCount() == 0) && i < 100 {
+		t.Log("Attempt", i, "os1ConnCount()=", os1.ConnCount(), "os2ConnCount()=", os2.ConnCount())
+		i++
+		if i == 100 {
+			assert.Fail(t, "Didn't switch to other instance after many attempts")
+		}
+		cl.Disconnect()
+		time.Sleep(time.Millisecond * 500)
+	}
+	cl.Close()
+	select {
+	case <-stopChan:
+	case <-time.After(time.Second * 20):
+		assert.Fail(t, "Didn't stop within a timely manner")
+	}
+}
+
 func newTestSeekInfo() *orderer.SeekInfo {
 	return &orderer.SeekInfo{Start: &orderer.SeekPosition{Type: &orderer.SeekPosition_Specified{Specified: &orderer.SeekSpecified{Number: 5}}},
 		Stop:     &orderer.SeekPosition{Type: &orderer.SeekPosition_Specified{Specified: &orderer.SeekSpecified{Number: math.MaxUint64}}},

core/deliverservice/deliveryclient_test.go

@@ -245,6 +245,67 @@ func TestDeliverServiceFailover(t *testing.T) {
 	service.Stop()
 }
 
+func TestDeliverServiceServiceUnavailable(t *testing.T) {
+	defer ensureNoGoroutineLeak(t)()
+	// Scenario: bring up 2 ordering service instances,
+	// Make the instance the client connects to fail after a delivery of a block and send SERVICE_UNAVAILABLE
+	// whenever subsequent seeks are sent to it.
+	// The client is expected to connect to the other instance, and to ask for a block sequence that is the next block
+	// after the last block it got from the first ordering service node.
+
+	os1 := mocks.NewOrderer(5615, t)
+	os2 := mocks.NewOrderer(5616, t)
+
+	time.Sleep(time.Second)
+	gossipServiceAdapter := &mocks.MockGossipServiceAdapter{GossipBlockDisseminations: make(chan uint64)}
+
+	service, err := NewDeliverService(&Config{
+		Endpoints:   []string{"localhost:5615", "localhost:5616"},
+		Gossip:      gossipServiceAdapter,
+		CryptoSvc:   &mockMCS{},
+		ABCFactory:  DefaultABCFactory,
+		ConnFactory: DefaultConnectionFactory,
+	})
+	assert.NoError(t, err)
+	li := &mocks.MockLedgerInfo{Height: 100}
+	os1.SetNextExpectedSeek(100)
+	os2.SetNextExpectedSeek(100)
+
+	err = service.StartDeliverForChannel("TEST_CHAINID", li)
+	assert.NoError(t, err, "can't start delivery")
+	// We need to discover to which instance the client connected to
+	go os1.SendBlock(100)
+	// Is it the first instance?
+	instance2fail := os1
+	nextBlockSeek := uint64(100)
+	select {
+	case seq := <-gossipServiceAdapter.GossipBlockDisseminations:
+		// Just for sanity check, ensure we got block seq 100
+		assert.Equal(t, uint64(100), seq)
+		// Connected to the first instance
+		// Advance ledger's height by 1
+		atomic.StoreUint64(&li.Height, 101)
+		// Backup instance should expect a seek of 101 since we got 100
+		os2.SetNextExpectedSeek(uint64(101))
+		nextBlockSeek = uint64(101)
+		// Have backup instance prepare to send a block
+		os2.SendBlock(101)
+	case <-time.After(time.Second * 5):
+		// We didn't get a block on time, so seems like we're connected to the 2nd instance
+		// and not to the first.
+		instance2fail = os2
+	}
+
+	instance2fail.Fail()
+	// Ensure the client asks blocks from the other ordering service node
+	assertBlockDissemination(nextBlockSeek, gossipServiceAdapter.GossipBlockDisseminations, t)
+
+	// Cleanup
+	os1.Shutdown()
+	os2.Shutdown()
+	service.Stop()
+}
+
 func TestDeliverServiceShutdown(t *testing.T) {
 	defer ensureNoGoroutineLeak(t)()
 	// Scenario: Launch an ordering service node and let the client pull some blocks.

core/deliverservice/mocks/blocksprovider.go

@@ -69,7 +69,8 @@ func (mock *MockGossipServiceAdapter) Gossip(msg *gossip_proto.GossipMessage) {
 // MockBlocksDeliverer mocking structure of BlocksDeliverer interface to initialize
 // the blocks provider implementation
 type MockBlocksDeliverer struct {
-	Pos uint64
+	DisconnectCalled chan struct{}
+	Pos              uint64
 	grpc.ClientStream
 	RecvCnt  int32
 	MockRecv func(mock *MockBlocksDeliverer) (*orderer.DeliverResponse, error)
@@ -121,6 +122,10 @@ func (mock *MockBlocksDeliverer) Send(env *common.Envelope) error {
 	return nil
 }
 
+func (mock *MockBlocksDeliverer) Disconnect() {
+	mock.DisconnectCalled <- struct{}{}
+}
+
 func (mock *MockBlocksDeliverer) Close() {}
 
 // MockLedgerInfo mocking implementation of LedgerInfo interface, needed