Merge pull request #463 from multiversx/get_number_of_shards_from_obs…

…ervers

Get number of shards from observers

cmd/proxy/config/config.toml

@@ -39,15 +39,18 @@
    # With this flag disabled, /transaction/pool route will return an error
    AllowEntireTxPoolFetch = false
 
-   # NumberOfShards represents the total number of shards from the network (excluding metachain)
-   NumberOfShards = 3
+   # NumShardsTimeoutInSec represents the maximum number of seconds to wait for at least one observer online until throwing an error
+   NumShardsTimeoutInSec = 90
+
+   # TimeBetweenNodesRequestsInSec represents time to wait before retry to get the number of shards from observers
+   TimeBetweenNodesRequestsInSec = 2
 
 [AddressPubkeyConverter]
-    #Length specifies the length in bytes of an address
-    Length = 32
+   #Length specifies the length in bytes of an address
+   Length = 32
 
-    # Type specifies the type of public keys: hex or bech32
-    Type = "bech32"
+   # Type specifies the type of public keys: hex or bech32
+   Type = "bech32"
 
 [Marshalizer]
    Type = "gogo protobuf"

cmd/proxy/main.go

@@ -328,7 +328,6 @@ func createVersionsRegistryTestOrProduction(
 				ValStatsCacheValidityDurationSec:         60,
 				EconomicsMetricsCacheValidityDurationSec: 6,
 				FaucetValue:                              "10000000000",
-				NumberOfShards:                           3,
 			},
 			ApiLogging: config.ApiLoggingConfig{
 				LoggingEnabled:          true,
@@ -409,12 +408,12 @@ func createVersionsRegistry(
 		return nil, err
 	}
 
-	shardCoord, err := sharding.NewMultiShardCoordinator(cfg.GeneralSettings.NumberOfShards, 0)
+	numShards, err := getNumOfShards(cfg)
 	if err != nil {
 		return nil, err
 	}
 
-	nodesProviderFactory, err := observer.NewNodesProviderFactory(*cfg, configurationFilePath)
+	nodesProviderFactory, err := observer.NewNodesProviderFactory(*cfg, configurationFilePath, numShards)
 	if err != nil {
 		return nil, err
 	}
@@ -431,6 +430,11 @@ func createVersionsRegistry(
 		}
 	}
 
+	shardCoord, err := sharding.NewMultiShardCoordinator(numShards, 0)
+	if err != nil {
+		return nil, err
+	}
+
 	bp, err := process.NewBaseProcessor(
 		cfg.GeneralSettings.RequestTimeoutSec,
 		shardCoord,
@@ -612,6 +616,32 @@ func waitForServerShutdown(httpServer *http.Server, closableComponents *data.Clo
 	_ = httpServer.Close()
 }
 
+// getNumOfShards will delay the start of proxy until it successfully gets the number of shards
+func getNumOfShards(cfg *config.Config) (uint32, error) {
+	httpClient := &http.Client{}
+	httpClient.Timeout = time.Duration(cfg.GeneralSettings.RequestTimeoutSec) * time.Second
+	observersList := make([]string, 0, len(cfg.Observers))
+	for _, node := range cfg.Observers {
+		observersList = append(observersList, node.Address)
+	}
+	argsNumShardsProcessor := process.ArgNumShardsProcessor{
+		HttpClient:                    httpClient,
+		Observers:                     observersList,
+		TimeBetweenNodesRequestsInSec: cfg.GeneralSettings.TimeBetweenNodesRequestsInSec,
+		NumShardsTimeoutInSec:         cfg.GeneralSettings.NumShardsTimeoutInSec,
+		RequestTimeoutInSec:           cfg.GeneralSettings.RequestTimeoutSec,
+	}
+	numShardsProcessor, err := process.NewNumShardsProcessor(argsNumShardsProcessor)
+	if err != nil {
+		return 0, err
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	return numShardsProcessor.GetNetworkNumShards(ctx)
+}
+
 func removeLogColors() {
 	err := logger.RemoveLogObserver(os.Stdout)
 	if err != nil {

config/config.go

@@ -16,7 +16,8 @@ type GeneralSettingsConfig struct {
 	BalancedObservers                        bool
 	BalancedFullHistoryNodes                 bool
 	AllowEntireTxPoolFetch                   bool
-	NumberOfShards                           uint32
+	NumShardsTimeoutInSec                    int
+	TimeBetweenNodesRequestsInSec            int
 }
 
 // Config will hold the whole config file's data

observer/baseNodeProvider.go

@@ -139,16 +139,6 @@ func (bnp *baseNodeProvider) ReloadNodes(nodesType data.NodeType) data.NodesRelo
 		}
 	}
 
-	numOldShards := bnp.numOfShards
-	numNewShards := newConfig.GeneralSettings.NumberOfShards
-	if numOldShards != numNewShards {
-		return data.NodesReloadResponse{
-			OkRequest:   false,
-			Description: "not reloaded",
-			Error:       fmt.Sprintf("different number of shards. before: %d, now: %d", numOldShards, numNewShards),
-		}
-	}
-
 	nodes := newConfig.Observers
 	if nodesType == data.FullHistoryNode {
 		nodes = newConfig.FullHistoryNodes

observer/baseNodeProvider_test.go

@@ -73,18 +73,6 @@ func TestBaseNodeProvider_InvalidShardForObserver(t *testing.T) {
 	require.True(t, strings.Contains(err.Error(), "addr1"))
 }
 
-func TestBaseNodeProvider_ReloadNodesDifferentNumberOfNewShard(t *testing.T) {
-	bnp := &baseNodeProvider{
-		configurationFilePath: configurationPath,
-		shardIds:              []uint32{0, 1},
-		numOfShards:           2,
-	}
-
-	response := bnp.ReloadNodes(data.Observer)
-	require.False(t, response.OkRequest)
-	require.Contains(t, response.Error, "different number of shards")
-}
-
 func TestBaseNodeProvider_ReloadNodesConfigurationFileNotFound(t *testing.T) {
 	bnp := &baseNodeProvider{
 		configurationFilePath: "wrong config path",

observer/circularQueueNodesProvider_test.go

@@ -23,9 +23,6 @@ func getDummyConfig() config.Config {
 				ShardId: 1,
 			},
 		},
-		GeneralSettings: config.GeneralSettingsConfig{
-			NumberOfShards: 2,
-		},
 	}
 }
 

observer/nodesProviderFactory.go

@@ -11,13 +11,15 @@ var log = logger.GetOrCreate("observer")
 type nodesProviderFactory struct {
 	cfg                   config.Config
 	configurationFilePath string
+	numberOfShards        uint32
 }
 
 // NewNodesProviderFactory returns a new instance of nodesProviderFactory
-func NewNodesProviderFactory(cfg config.Config, configurationFilePath string) (*nodesProviderFactory, error) {
+func NewNodesProviderFactory(cfg config.Config, configurationFilePath string, numberOfShards uint32) (*nodesProviderFactory, error) {
 	return &nodesProviderFactory{
 		cfg:                   cfg,
 		configurationFilePath: configurationFilePath,
+		numberOfShards:        numberOfShards,
 	}, nil
 }
 
@@ -27,13 +29,13 @@ func (npf *nodesProviderFactory) CreateObservers() (NodesProviderHandler, error)
 		return NewCircularQueueNodesProvider(
 			npf.cfg.Observers,
 			npf.configurationFilePath,
-			npf.cfg.GeneralSettings.NumberOfShards)
+			npf.numberOfShards)
 	}
 
 	return NewSimpleNodesProvider(
 		npf.cfg.Observers,
 		npf.configurationFilePath,
-		npf.cfg.GeneralSettings.NumberOfShards)
+		npf.numberOfShards)
 }
 
 // CreateFullHistoryNodes will create and return an object of type NodesProviderHandler based on a flag
@@ -42,7 +44,7 @@ func (npf *nodesProviderFactory) CreateFullHistoryNodes() (NodesProviderHandler,
 		nodesProviderHandler, err := NewCircularQueueNodesProvider(
 			npf.cfg.FullHistoryNodes,
 			npf.configurationFilePath,
-			npf.cfg.GeneralSettings.NumberOfShards)
+			npf.numberOfShards)
 		if err != nil {
 			return getDisabledFullHistoryNodesProviderIfNeeded(err)
 		}
@@ -53,7 +55,7 @@ func (npf *nodesProviderFactory) CreateFullHistoryNodes() (NodesProviderHandler,
 	nodesProviderHandler, err := NewSimpleNodesProvider(
 		npf.cfg.FullHistoryNodes,
 		npf.configurationFilePath,
-		npf.cfg.GeneralSettings.NumberOfShards)
+		npf.numberOfShards)
 	if err != nil {
 		return getDisabledFullHistoryNodesProviderIfNeeded(err)
 	}

observer/nodesProviderFactory_test.go

@@ -10,7 +10,7 @@ import (
 func TestNewObserversProviderFactory_ShouldWork(t *testing.T) {
 	t.Parallel()
 
-	opf, err := NewNodesProviderFactory(config.Config{}, "path")
+	opf, err := NewNodesProviderFactory(config.Config{}, "path", 2)
 	assert.Nil(t, err)
 	assert.NotNil(t, opf)
 }
@@ -21,7 +21,7 @@ func TestObserversProviderFactory_CreateShouldReturnSimple(t *testing.T) {
 	cfg := getDummyConfig()
 	cfg.GeneralSettings.BalancedObservers = false
 
-	opf, _ := NewNodesProviderFactory(cfg, "path")
+	opf, _ := NewNodesProviderFactory(cfg, "path", 2)
 	op, err := opf.CreateObservers()
 	assert.Nil(t, err)
 	_, ok := op.(*simpleNodesProvider)
@@ -34,7 +34,7 @@ func TestObserversProviderFactory_CreateShouldReturnCircularQueue(t *testing.T)
 	cfg := getDummyConfig()
 	cfg.GeneralSettings.BalancedObservers = true
 
-	opf, _ := NewNodesProviderFactory(cfg, "path")
+	opf, _ := NewNodesProviderFactory(cfg, "path", 2)
 	op, err := opf.CreateObservers()
 	assert.Nil(t, err)
 	_, ok := op.(*circularQueueNodesProvider)

process/errors.go

@@ -56,18 +56,12 @@ var ErrNoFaucetAccountForGivenShard = errors.New("no faucet account found for th
 // ErrNilNodesProvider signals that a nil observers provider has been provided
 var ErrNilNodesProvider = errors.New("nil nodes provider")
 
-// ErrInvalidShardId signals that a invalid shard id has been provided
-var ErrInvalidShardId = errors.New("invalid shard id")
-
 // ErrNilPubKeyConverter signals that a nil pub key converter has been provided
 var ErrNilPubKeyConverter = errors.New("nil pub key converter provided")
 
 // ErrNoValidTransactionToSend signals that no valid transaction were received
 var ErrNoValidTransactionToSend = errors.New("no valid transaction to send")
 
-// ErrNilDatabaseConnector signals that a nil database connector was provided
-var ErrNilDatabaseConnector = errors.New("not valid database connector")
-
 // ErrCannotParseNodeStatusMetrics signals that the node status metrics cannot be parsed
 var ErrCannotParseNodeStatusMetrics = errors.New("cannot parse node status metrics")
 
@@ -115,3 +109,6 @@ var ErrEmptyCommitString = errors.New("empty commit id string")
 
 // ErrEmptyPubKey signals that an empty public key has been provided
 var ErrEmptyPubKey = errors.New("public key is empty")
+
+// ErrNilHttpClient signals that a nil http client has been provided
+var ErrNilHttpClient = errors.New("nil http client")

process/interface.go

@@ -1,6 +1,8 @@
 package process
 
 import (
+	"net/http"
+
 	"github.com/multiversx/mx-chain-core-go/core"
 	"github.com/multiversx/mx-chain-core-go/data/transaction"
 	"github.com/multiversx/mx-chain-core-go/data/vm"
@@ -78,3 +80,8 @@ type StatusMetricsProvider interface {
 	GetMetricsForPrometheus() string
 	IsInterfaceNil() bool
 }
+
+// HttpClient defines an interface for the http client
+type HttpClient interface {
+	Do(req *http.Request) (*http.Response, error)
+}

process/mock/httpClientMock.go

@@ -0,0 +1,16 @@
+package mock
+
+import "net/http"
+
+// HttpClientMock -
+type HttpClientMock struct {
+	DoCalled func(req *http.Request) (*http.Response, error)
+}
+
+// Do -
+func (mock *HttpClientMock) Do(req *http.Request) (*http.Response, error) {
+	if mock.DoCalled != nil {
+		return mock.DoCalled(req)
+	}
+	return &http.Response{}, nil
+}