Skip to content

Commit 08b456e

Browse files
yacovmyacovm
committed
[FAB-2759] DeliveryService peer<->OS high availability
In previous commits, I added a delivery client: https://gerrit.hyperledger.org/r/#/c/7343 and a connection producer: https://gerrit.hyperledger.org/r/#/c/7337 In this commit, I integrate the two into the existing delivery service, and: - Decouple the blocks request action from the blocks provider and instead- put it into requester.go, which is used by the delivery client (client.go) in the following way- the client has a function that is invoked upon each successful (re)connection to the ordering service. This function utilizes the BlockRequester (requester.go) and makes it send a seekInfo message to the ordering service. - Instead of the BlocksDeliverer to be created once at startup- we have the broadcastClient (client.go) from a previous change set that implements the BlocksDeliverer, and it does reconnection logic upon demand and the delivery service is oblivious of this. This change set makes the delivery service automatically failover/reconnect to backup ordering service endpoints once it disconnects from the ordering service. I added the following test cases: - Peer reconnect upon restart of the ordering service - Peer failover to other ordering service node - Peer is disconnected from ordering service upon close of the delivery service Total code coverage of deliveryclient.go went up to 97% Signed-off-by: Yacov Manevich <[email protected]> Change-Id: I2b4687cc3b5fc767150fa0de607890a68fd38449
1 parent c0f8d75 commit 08b456e

9 files changed

+530
-200
lines changed

‎core/deliverservice/blocksprovider/blocksprovider.go

+18-71
Original file line numberDiff line numberDiff line change
@@ -17,19 +17,16 @@ limitations under the License.
1717
package blocksprovider
1818

1919
import (
20-
"math"
2120
"sync/atomic"
2221

2322
"github.com/golang/protobuf/proto"
2423
gossipcommon "github.com/hyperledger/fabric/gossip/common"
2524
"github.com/hyperledger/fabric/gossip/discovery"
2625

27-
"github.com/hyperledger/fabric/common/localmsp"
2826
"github.com/hyperledger/fabric/gossip/api"
2927
"github.com/hyperledger/fabric/protos/common"
3028
gossip_proto "github.com/hyperledger/fabric/protos/gossip"
3129
"github.com/hyperledger/fabric/protos/orderer"
32-
"github.com/hyperledger/fabric/protos/utils"
3330
"github.com/op/go-logging"
3431
)
3532

@@ -56,10 +53,6 @@ type GossipServiceAdapter interface {
5653
// BlocksProvider used to read blocks from the ordering service
5754
// for specified chain it subscribed to
5855
type BlocksProvider interface {
59-
// RequestBlock acquire new blocks from ordering service based on
60-
// information provided by ledger info instance
61-
RequestBlocks(ledgerInfoProvider LedgerInfo) error
62-
6356
// DeliverBlocks starts delivering and disseminating blocks
6457
DeliverBlocks()
6558

@@ -69,21 +62,29 @@ type BlocksProvider interface {
6962

7063
// BlocksDeliverer defines interface which actually helps
7164
// to abstract the AtomicBroadcast_DeliverClient with only
72-
// required method for blocks provider. This also help to
73-
// build up mocking facilities for testing purposes
65+
// required method for blocks provider.
66+
// This also decouples the production implementation of the gRPC stream
67+
// from the code in order for the code to be more modular and testable.
7468
type BlocksDeliverer interface {
75-
// Recv capable to bring new blocks from the ordering service
69+
// Recv retrieves a response from the ordering service
7670
Recv() (*orderer.DeliverResponse, error)
7771

78-
// Send used to send request to the ordering service to obtain new blocks
72+
// Send sends an envelope to the ordering service
7973
Send(*common.Envelope) error
8074
}
8175

76+
type streamClient interface {
77+
BlocksDeliverer
78+
79+
// Close closes the stream and its underlying connection
80+
Close()
81+
}
82+
8283
// blocksProviderImpl the actual implementation for BlocksProvider interface
8384
type blocksProviderImpl struct {
8485
chainID string
8586

86-
client BlocksDeliverer
87+
client streamClient
8788

8889
gossip GossipServiceAdapter
8990

@@ -98,8 +99,8 @@ func init() {
9899
logger = logging.MustGetLogger("blocksProvider")
99100
}
100101

101-
// NewBlocksProvider constructor function to creare blocks deliverer instance
102-
func NewBlocksProvider(chainID string, client BlocksDeliverer, gossip GossipServiceAdapter, mcs api.MessageCryptoService) BlocksProvider {
102+
// NewBlocksProvider constructor function to create blocks deliverer instance
103+
func NewBlocksProvider(chainID string, client streamClient, gossip GossipServiceAdapter, mcs api.MessageCryptoService) BlocksProvider {
103104
return &blocksProviderImpl{
104105
chainID: chainID,
105106
client: client,
@@ -111,6 +112,7 @@ func NewBlocksProvider(chainID string, client BlocksDeliverer, gossip GossipServ
111112
// DeliverBlocks used to pull out blocks from the ordering service to
112113
// distributed them across peers
113114
func (b *blocksProviderImpl) DeliverBlocks() {
115+
defer b.client.Close()
114116
for !b.isDone() {
115117
msg, err := b.client.Recv()
116118
if err != nil {
@@ -157,72 +159,17 @@ func (b *blocksProviderImpl) DeliverBlocks() {
157159
}
158160
}
159161

160-
// Stops blocks delivery provider
162+
// Stop stops blocks delivery provider
161163
func (b *blocksProviderImpl) Stop() {
162164
atomic.StoreInt32(&b.done, 1)
165+
b.client.Close()
163166
}
164167

165168
// Check whenever provider is stopped
166169
func (b *blocksProviderImpl) isDone() bool {
167170
return atomic.LoadInt32(&b.done) == 1
168171
}
169172

170-
func (b *blocksProviderImpl) RequestBlocks(ledgerInfoProvider LedgerInfo) error {
171-
height, err := ledgerInfoProvider.LedgerHeight()
172-
if err != nil {
173-
logger.Errorf("Can't get legder height from committer [%s]", err)
174-
return err
175-
}
176-
177-
if height > 0 {
178-
logger.Debugf("Starting deliver with block [%d]", height)
179-
if err := b.seekLatestFromCommitter(height); err != nil {
180-
return err
181-
}
182-
} else {
183-
logger.Debug("Starting deliver with olders block")
184-
if err := b.seekOldest(); err != nil {
185-
return err
186-
}
187-
}
188-
189-
return nil
190-
}
191-
192-
func (b *blocksProviderImpl) seekOldest() error {
193-
seekInfo := &orderer.SeekInfo{
194-
Start: &orderer.SeekPosition{Type: &orderer.SeekPosition_Oldest{Oldest: &orderer.SeekOldest{}}},
195-
Stop: &orderer.SeekPosition{Type: &orderer.SeekPosition_Specified{Specified: &orderer.SeekSpecified{Number: math.MaxUint64}}},
196-
Behavior: orderer.SeekInfo_BLOCK_UNTIL_READY,
197-
}
198-
199-
//TODO- epoch and msgVersion may need to be obtained for nowfollowing usage in orderer/configupdate/configupdate.go
200-
msgVersion := int32(0)
201-
epoch := uint64(0)
202-
env, err := utils.CreateSignedEnvelope(common.HeaderType_CONFIG_UPDATE, b.chainID, localmsp.NewSigner(), seekInfo, msgVersion, epoch)
203-
if err != nil {
204-
return err
205-
}
206-
return b.client.Send(env)
207-
}
208-
209-
func (b *blocksProviderImpl) seekLatestFromCommitter(height uint64) error {
210-
seekInfo := &orderer.SeekInfo{
211-
Start: &orderer.SeekPosition{Type: &orderer.SeekPosition_Specified{Specified: &orderer.SeekSpecified{Number: height}}},
212-
Stop: &orderer.SeekPosition{Type: &orderer.SeekPosition_Specified{Specified: &orderer.SeekSpecified{Number: math.MaxUint64}}},
213-
Behavior: orderer.SeekInfo_BLOCK_UNTIL_READY,
214-
}
215-
216-
//TODO- epoch and msgVersion may need to be obtained for nowfollowing usage in orderer/configupdate/configupdate.go
217-
msgVersion := int32(0)
218-
epoch := uint64(0)
219-
env, err := utils.CreateSignedEnvelope(common.HeaderType_CONFIG_UPDATE, b.chainID, localmsp.NewSigner(), seekInfo, msgVersion, epoch)
220-
if err != nil {
221-
return err
222-
}
223-
return b.client.Send(env)
224-
}
225-
226173
func createGossipMsg(chainID string, payload *gossip_proto.Payload) *gossip_proto.GossipMessage {
227174
gossipMsg := &gossip_proto.GossipMessage{
228175
Nonce: 0,

‎core/deliverservice/blocksprovider/blocksprovider_test.go

+12-12
Original file line numberDiff line numberDiff line change
@@ -59,7 +59,7 @@ func (*mockMCS) ValidateIdentity(peerIdentity api.PeerIdentityType) error {
5959
// from given block sequence number.
6060
func makeTestCase(ledgerHeight uint64) func(*testing.T) {
6161
return func(t *testing.T) {
62-
gossipServiceAdapter := &mocks.MockGossipServiceAdapter{}
62+
gossipServiceAdapter := &mocks.MockGossipServiceAdapter{GossipBlockDisseminations: make(chan uint64)}
6363
deliverer := &mocks.MockBlocksDeliverer{Pos: ledgerHeight}
6464
deliverer.MockRecv = mocks.MockRecv
6565

@@ -70,15 +70,9 @@ func makeTestCase(ledgerHeight uint64) func(*testing.T) {
7070
mcs: &mockMCS{},
7171
}
7272

73-
provider.RequestBlocks(&mocks.MockLedgerInfo{ledgerHeight})
74-
75-
var wg sync.WaitGroup
76-
wg.Add(1)
77-
7873
ready := make(chan struct{})
7974
go func() {
80-
provider.DeliverBlocks()
81-
wg.Done()
75+
go provider.DeliverBlocks()
8276
// Send notification
8377
ready <- struct{}{}
8478
}()
@@ -91,7 +85,11 @@ func makeTestCase(ledgerHeight uint64) func(*testing.T) {
9185
{
9286
// Check that all blocks received eventually get gossiped and locally committed
9387
assert.True(t, deliverer.RecvCnt == gossipServiceAdapter.AddPayloadsCnt)
94-
assert.True(t, deliverer.RecvCnt == gossipServiceAdapter.GossipCallsCnt)
88+
select {
89+
case <-gossipServiceAdapter.GossipBlockDisseminations:
90+
case <-time.After(time.Second):
91+
assert.Fail(t, "Didn't gossip a block within a timely manner")
92+
}
9593
return
9694
}
9795
case <-time.After(time.Duration(1) * time.Second):
@@ -140,8 +138,6 @@ func TestBlocksProvider_CheckTerminationDeliveryResponseStatus(t *testing.T) {
140138
client: &tmp,
141139
}
142140

143-
provider.RequestBlocks(&mocks.MockLedgerInfo{0})
144-
145141
var wg sync.WaitGroup
146142
wg.Add(1)
147143

@@ -163,7 +159,11 @@ func TestBlocksProvider_CheckTerminationDeliveryResponseStatus(t *testing.T) {
163159
// No payload should commit locally
164160
assert.Equal(t, int32(0), gossipServiceAdapter.AddPayloadsCnt)
165161
// No payload should be transfered to other peers
166-
assert.Equal(t, int32(0), gossipServiceAdapter.GossipCallsCnt)
162+
select {
163+
case <-gossipServiceAdapter.GossipBlockDisseminations:
164+
assert.Fail(t, "Gossiped block but shouldn't have")
165+
case <-time.After(time.Second):
166+
}
167167
return
168168
}
169169
case <-time.After(time.Duration(1) * time.Second):

‎core/deliverservice/client_test.go

+4-1
Original file line numberDiff line numberDiff line change
@@ -565,11 +565,12 @@ func (s *signerMock) Sign(message []byte) ([]byte, error) {
565565
}
566566

567567
func TestProductionUsage(t *testing.T) {
568+
defer ensureNoGoroutineLeak(t)()
568569
// This test configures the client in a similar fashion as will be
569570
// in production, and tests against a live gRPC server.
570571
os := mocks.NewOrderer(5612, t)
571572
os.SetNextExpectedSeek(5)
572-
defer os.Shutdown()
573+
573574
connFact := func(endpoint string) (*grpc.ClientConn, error) {
574575
return grpc.Dial(endpoint, grpc.WithInsecure(), grpc.WithBlock())
575576
}
@@ -593,6 +594,8 @@ func TestProductionUsage(t *testing.T) {
593594
assert.NoError(t, err)
594595
assert.NotNil(t, resp)
595596
assert.Equal(t, uint64(5), resp.GetBlock().Header.Number)
597+
os.Shutdown()
598+
cl.Close()
596599
}
597600

598601
func newTestSeekInfo() *orderer.SeekInfo {

0 commit comments

Comments
 (0)