[FAB-15488] Refactor viper in gossip/state

Refactor viper usage in gossip/state package. Added stateConfig
structure to hold configuration values.
Configuration variables names changed accordinly.

FAB-15488 #done

Change-Id: I0f241c90492a858e39eab710e5b8bc040661af43
Signed-off-by: Chongxin Luo <Chongxin.Luo@ibm.com>

Author: DereckLuo

gossip/privdata/config.go

@@ -13,7 +13,7 @@ import (
 )
 
 const (
-	reconcileSleepIntervalDefault = time.Minute * 1
+	reconcileSleepIntervalDefault = time.Minute
 	reconcileBatchSizeDefault     = 10
 )
 

gossip/service/gossip_service.go

@@ -343,8 +343,14 @@ func (g *GossipService) InitializeChannel(channelID string, endpoints []string,
 	g.privateHandlers[channelID].reconciler.Start()
 
 	blockingMode := !g.serviceConfig.NonBlockingCommitMode
-	g.chains[channelID] = state.NewGossipStateProvider(channelID, servicesAdapter, coordinator,
-		g.metrics.StateMetrics, blockingMode)
+	stateConfig := state.GlobalConfig()
+	g.chains[channelID] = state.NewGossipStateProvider(
+		channelID,
+		servicesAdapter,
+		coordinator,
+		g.metrics.StateMetrics,
+		blockingMode,
+		stateConfig)
 	if g.deliveryService[channelID] == nil {
 		var err error
 		g.deliveryService[channelID], err = g.deliveryFactory.Service(g, endpoints, g.mcs)

gossip/state/config.go

@@ -0,0 +1,71 @@
+/*
+Copyright IBM Corp. All Rights Reserved.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package state
+
+import (
+	"time"
+
+	"github.com/spf13/viper"
+)
+
+const (
+	DefStateCheckInterval   = 10 * time.Second
+	DefStateResponseTimeout = 3 * time.Second
+	DefStateBatchSize       = 10
+	DefStateMaxRetries      = 3
+	DefStateBlockBufferSize = 100
+	DefStateChannelSize     = 100
+	DefStateEnabled         = true
+)
+
+type StateConfig struct {
+	StateCheckInterval   time.Duration
+	StateResponseTimeout time.Duration
+	StateBatchSize       uint64
+	StateMaxRetries      int
+	StateBlockBufferSize int
+	StateChannelSize     int
+	StateEnabled         bool
+}
+
+func GlobalConfig() *StateConfig {
+	c := &StateConfig{}
+	c.loadStateConfig()
+	return c
+}
+
+func (c *StateConfig) loadStateConfig() {
+
+	c.StateCheckInterval = DefStateCheckInterval
+	if viper.IsSet("peer.gossip.state.checkInterval") {
+		c.StateCheckInterval = viper.GetDuration("peer.gossip.state.checkInterval")
+	}
+	c.StateResponseTimeout = DefStateResponseTimeout
+	if viper.IsSet("peer.gossip.state.responseTimeout") {
+		c.StateResponseTimeout = viper.GetDuration("peer.gossip.state.responseTimeout")
+	}
+	c.StateBatchSize = DefStateBatchSize
+	if viper.IsSet("peer.gossip.state.batchSize") {
+		c.StateBatchSize = uint64(viper.GetInt("peer.gossip.state.batchSize"))
+	}
+	c.StateMaxRetries = DefStateMaxRetries
+	if viper.IsSet("peer.gossip.state.maxRetries") {
+		c.StateMaxRetries = viper.GetInt("peer.gossip.state.maxRetries")
+	}
+	c.StateBlockBufferSize = DefStateBlockBufferSize
+	if viper.IsSet("peer.gossip.state.blockBufferSize") {
+		c.StateBlockBufferSize = viper.GetInt("peer.gossip.state.blockBufferSize")
+	}
+	c.StateChannelSize = DefStateChannelSize
+	if viper.IsSet("peer.gossip.state.channelSize") {
+		c.StateChannelSize = viper.GetInt("peer.gossip.state.channelSize")
+	}
+	c.StateEnabled = DefStateEnabled
+	if viper.IsSet("peer.gossip.state.enabled") {
+		c.StateEnabled = viper.GetBool("peer.gossip.state.enabled")
+	}
+}

gossip/state/config_test.go

@@ -0,0 +1,60 @@
+/*
+Copyright IBM Corp. All Rights Reserved.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package state_test
+
+import (
+	"testing"
+	"time"
+
+	"github.com/hyperledger/fabric/gossip/state"
+	"github.com/stretchr/testify/assert"
+
+	"github.com/spf13/viper"
+)
+
+func TestGlobalConfig(t *testing.T) {
+	viper.Reset()
+	viper.Set("peer.gossip.state.checkInterval", "1s")
+	viper.Set("peer.gossip.state.responseTimeout", "2s")
+	viper.Set("peer.gossip.state.batchSize", 3)
+	viper.Set("peer.gossip.state.maxRetries", 4)
+	viper.Set("peer.gossip.state.blockBufferSize", 5)
+	viper.Set("peer.gossip.state.channelSize", 6)
+	viper.Set("peer.gossip.state.enabled", false)
+
+	coreConfig := state.GlobalConfig()
+
+	expectedConfig := &state.StateConfig{
+		StateCheckInterval:   time.Second,
+		StateResponseTimeout: 2 * time.Second,
+		StateBatchSize:       uint64(3),
+		StateMaxRetries:      4,
+		StateBlockBufferSize: 5,
+		StateChannelSize:     6,
+		StateEnabled:         false,
+	}
+
+	assert.Equal(t, expectedConfig, coreConfig)
+}
+
+func TestGlobalConfigDefaults(t *testing.T) {
+	viper.Reset()
+
+	coreConfig := state.GlobalConfig()
+
+	expectedConfig := &state.StateConfig{
+		StateCheckInterval:   10 * time.Second,
+		StateResponseTimeout: 3 * time.Second,
+		StateBatchSize:       uint64(10),
+		StateMaxRetries:      3,
+		StateBlockBufferSize: 100,
+		StateChannelSize:     100,
+		StateEnabled:         true,
+	}
+
+	assert.Equal(t, expectedConfig, coreConfig)
+}

gossip/state/state.go

@@ -27,7 +27,6 @@ import (
 	"github.com/hyperledger/fabric/protos/transientstore"
 	"github.com/hyperledger/fabric/protoutil"
 	"github.com/pkg/errors"
-	"github.com/spf13/viper"
 )
 
 // GossipStateProvider is the interface to acquire sequences of the ledger blocks
@@ -166,7 +165,7 @@ type GossipStateProviderImpl struct {
 
 	blockingMode bool
 
-	config *Configuration
+	config *StateConfig
 }
 
 var logger = util.GetLogger(util.StateLogger, "")
@@ -188,53 +187,16 @@ func (v *stateRequestValidator) validate(request *proto.RemoteStateRequest, batc
 	return nil
 }
 
-// readConfiguration reading state configuration
-func readConfiguration() *Configuration {
-	config := &Configuration{
-		AntiEntropyInterval:             defAntiEntropyInterval,
-		AntiEntropyStateResponseTimeout: defAntiEntropyStateResponseTimeout,
-		AntiEntropyBatchSize:            defAntiEntropyBatchSize,
-		MaxBlockDistance:                defMaxBlockDistance,
-		AntiEntropyMaxRetries:           defAntiEntropyMaxRetries,
-		ChannelBufferSize:               defChannelBufferSize,
-		EnableStateTransfer:             true,
-	}
-
-	if viper.IsSet("peer.gossip.state.checkInterval") {
-		config.AntiEntropyInterval = viper.GetDuration("peer.gossip.state.checkInterval")
-	}
-
-	if viper.IsSet("peer.gossip.state.responseTimeout") {
-		config.AntiEntropyStateResponseTimeout = viper.GetDuration("peer.gossip.state.responseTimeout")
-	}
-
-	if viper.IsSet("peer.gossip.state.batchSize") {
-		config.AntiEntropyBatchSize = uint64(viper.GetInt("peer.gossip.state.batchSize"))
-	}
-
-	if viper.IsSet("peer.gossip.state.blockBufferSize") {
-		config.MaxBlockDistance = viper.GetInt("peer.gossip.state.blockBufferSize")
-	}
-
-	if viper.IsSet("peer.gossip.state.maxRetries") {
-		config.AntiEntropyMaxRetries = viper.GetInt("peer.gossip.state.maxRetries")
-	}
-
-	if viper.IsSet("peer.gossip.state.channelSize") {
-		config.ChannelBufferSize = viper.GetInt("peer.gossip.state.channelSize")
-	}
-
-	if viper.IsSet("peer.gossip.state.enabled") {
-		config.EnableStateTransfer = viper.GetBool("peer.gossip.state.enabled")
-	}
-
-	return config
-}
-
 // NewGossipStateProvider creates state provider with coordinator instance
 // to orchestrate arrival of private rwsets and blocks before committing them into the ledger.
-func NewGossipStateProvider(chainID string, services *ServicesMediator, ledger ledgerResources,
-	stateMetrics *metrics.StateMetrics, blockingMode bool) GossipStateProvider {
+func NewGossipStateProvider(
+	chainID string,
+	services *ServicesMediator,
+	ledger ledgerResources,
+	stateMetrics *metrics.StateMetrics,
+	blockingMode bool,
+	config *StateConfig,
+) GossipStateProvider {
 
 	gossipChan, _ := services.Accept(func(message interface{}) bool {
 		// Get only data messages
@@ -278,48 +240,31 @@ func NewGossipStateProvider(chainID string, services *ServicesMediator, ledger l
 		return nil
 	}
 
-	// Reading state configuration
-	config := readConfiguration()
-
 	s := &GossipStateProviderImpl{
 		// MessageCryptoService
 		mediator: services,
-
 		// Chain ID
 		chainID: chainID,
-
 		// Channel to read new messages from
 		gossipChan: gossipChan,
-
 		// Channel to read direct messages from other peers
 		commChan: commChan,
-
 		// Create a queue for payloads, wrapped in a metrics buffer
 		payloads: &metricsBuffer{
 			PayloadsBuffer: NewPayloadsBuffer(height),
 			sizeMetrics:    stateMetrics.PayloadBufferSize,
 			chainID:        chainID,
 		},
-
-		ledger: ledger,
-
-		stateResponseCh: make(chan protoext.ReceivedMessage, config.ChannelBufferSize),
-
-		stateRequestCh: make(chan protoext.ReceivedMessage, config.ChannelBufferSize),
-
-		stopCh: make(chan struct{}),
-
+		ledger:              ledger,
+		stateResponseCh:     make(chan protoext.ReceivedMessage, config.StateChannelSize),
+		stateRequestCh:      make(chan protoext.ReceivedMessage, config.StateChannelSize),
+		stopCh:              make(chan struct{}),
 		stateTransferActive: 0,
-
-		once: sync.Once{},
-
-		stateMetrics: stateMetrics,
-
-		requestValidator: &stateRequestValidator{},
-
-		blockingMode: blockingMode,
-
-		config: config,
+		once:                sync.Once{},
+		stateMetrics:        stateMetrics,
+		requestValidator:    &stateRequestValidator{},
+		blockingMode:        blockingMode,
+		config:              config,
 	}
 
 	logger.Infof("Updating metadata information, "+
@@ -331,7 +276,7 @@ func NewGossipStateProvider(chainID string, services *ServicesMediator, ledger l
 	go s.listen()
 	// Deliver in order messages into the incoming channel
 	go s.deliverPayloads()
-	if s.config.EnableStateTransfer {
+	if s.config.StateEnabled {
 		// Execute anti entropy to fill missing gaps
 		go s.antiEntropy()
 	}
@@ -438,7 +383,7 @@ func (s *GossipStateProviderImpl) directMessage(msg protoext.ReceivedMessage) {
 	incoming := msg.GetGossipMessage()
 
 	if incoming.GetStateRequest() != nil {
-		if len(s.stateRequestCh) < s.config.ChannelBufferSize {
+		if len(s.stateRequestCh) < s.config.StateChannelSize {
 			// Forward state request to the channel, if there are too
 			// many message of state request ignore to avoid flooding.
 			s.stateRequestCh <- msg
@@ -471,7 +416,7 @@ func (s *GossipStateProviderImpl) handleStateRequest(msg protoext.ReceivedMessag
 	}
 	request := msg.GetGossipMessage().GetStateRequest()
 
-	if err := s.requestValidator.validate(request, s.config.AntiEntropyBatchSize); err != nil {
+	if err := s.requestValidator.validate(request, s.config.StateBatchSize); err != nil {
 		logger.Errorf("State request validation failed, %s. Ignoring request...", err)
 		return
 	}
@@ -654,7 +599,7 @@ func (s *GossipStateProviderImpl) antiEntropy() {
 		select {
 		case <-s.stopCh:
 			return
-		case <-time.After(s.config.AntiEntropyInterval):
+		case <-time.After(s.config.StateCheckInterval):
 			ourHeight, err := s.ledger.LedgerHeight()
 			if err != nil {
 				// Unable to read from ledger continue to the next round
@@ -699,15 +644,15 @@ func (s *GossipStateProviderImpl) requestBlocksInRange(start uint64, end uint64)
 	defer atomic.StoreInt32(&s.stateTransferActive, 0)
 
 	for prev := start; prev <= end; {
-		next := min(end, prev+s.config.AntiEntropyBatchSize)
+		next := min(end, prev+s.config.StateBatchSize)
 
 		gossipMsg := s.stateRequestMessage(prev, next)
 
 		responseReceived := false
 		tryCounts := 0
 
 		for !responseReceived {
-			if tryCounts > s.config.AntiEntropyMaxRetries {
+			if tryCounts > s.config.StateMaxRetries {
 				logger.Warningf("Wasn't  able to get blocks in range [%d...%d), after %d retries",
 					prev, next, tryCounts)
 				return
@@ -745,7 +690,7 @@ func (s *GossipStateProviderImpl) requestBlocksInRange(start uint64, end uint64)
 				}
 				prev = index + 1
 				responseReceived = true
-			case <-time.After(s.config.AntiEntropyStateResponseTimeout):
+			case <-time.After(s.config.StateResponseTimeout):
 			}
 		}
 	}
@@ -824,11 +769,11 @@ func (s *GossipStateProviderImpl) addPayload(payload *proto.Payload, blockingMod
 		return errors.Wrap(err, "Failed obtaining ledger height")
 	}
 
-	if !blockingMode && payload.SeqNum-height >= uint64(s.config.MaxBlockDistance) {
+	if !blockingMode && payload.SeqNum-height >= uint64(s.config.StateBlockBufferSize) {
 		return errors.Errorf("Ledger height is at %d, cannot enqueue block with sequence of %d", height, payload.SeqNum)
 	}
 
-	for blockingMode && s.payloads.Size() > s.config.MaxBlockDistance*2 {
+	for blockingMode && s.payloads.Size() > s.config.StateBlockBufferSize*2 {
 		time.Sleep(enqueueRetryInterval)
 	}