Heartbeats chain simulator

cmd/node/config/external.toml

@@ -12,8 +12,8 @@
     Username          = ""
     Password          = ""
     # EnabledIndexes represents a slice of indexes that will be enabled for indexing. Full list is:
-    # ["rating", "transactions", "blocks", "validators", "miniblocks", "rounds", "accounts", "accountshistory", "receipts", "scresults", "accountsesdt", "accountsesdthistory", "epochinfo", "scdeploys", "tokens", "tags", "logs", "delegators", "operations", "esdts"]
-    EnabledIndexes    = ["rating", "transactions", "blocks", "validators", "miniblocks", "rounds", "accounts", "accountshistory", "receipts", "scresults", "accountsesdt", "accountsesdthistory", "epochinfo", "scdeploys", "tokens", "tags", "logs", "delegators", "operations", "esdts"]
+    # ["rating", "transactions", "blocks", "validators", "miniblocks", "rounds", "accounts", "accountshistory", "receipts", "scresults", "accountsesdt", "accountsesdthistory", "epochinfo", "scdeploys", "tokens", "tags", "logs", "delegators", "operations", "esdts", "events"]
+    EnabledIndexes    = ["rating", "transactions", "blocks", "validators", "miniblocks", "rounds", "accounts", "accountshistory", "receipts", "scresults", "accountsesdt", "accountsesdthistory", "epochinfo", "scdeploys", "tokens", "tags", "logs", "delegators", "operations", "esdts", "events"]
 
 # EventNotifierConnector defines settings needed to configure and launch the event notifier component
 # HTTP event notifier connector integration will be DEPRECATED in the following iterations

node/chainSimulator/chainSimulator_test.go

@@ -83,6 +83,12 @@ func TestChainSimulator_GenerateBlocksShouldWork(t *testing.T) {
 	require.Nil(t, err)
 	require.NotNil(t, chainSimulator)
 
+	time.Sleep(time.Second)
+
+	heartBeats, err := chainSimulator.GetNodeHandler(0).GetFacadeHandler().GetHeartbeats()
+	require.Nil(t, err)
+	require.Equal(t, 1, len(heartBeats))
+
 	defer chainSimulator.Close()
 
 	time.Sleep(time.Second)

node/chainSimulator/components/nodeFacade.go

@@ -13,6 +13,7 @@ import (
 	"github.com/multiversx/mx-chain-go/config"
 	"github.com/multiversx/mx-chain-go/facade"
 	apiComp "github.com/multiversx/mx-chain-go/factory/api"
+	"github.com/multiversx/mx-chain-go/factory/heartbeat"
 	nodePack "github.com/multiversx/mx-chain-go/node"
 	"github.com/multiversx/mx-chain-go/node/metrics"
 	"github.com/multiversx/mx-chain-go/process/mock"
@@ -73,6 +74,33 @@ func (node *testOnlyProcessingNode) createFacade(configs config.Configs, apiInte
 
 	flagsConfig := configs.FlagsConfig
 
+	heartBeatComponentsFactory, err := heartbeat.NewHeartbeatV2ComponentsFactory(heartbeat.ArgHeartbeatV2ComponentsFactory{
+		Config:               *configs.GeneralConfig,
+		Prefs:                *configs.PreferencesConfig,
+		BaseVersion:          "1",
+		AppVersion:           "1",
+		BootstrapComponents:  node.BootstrapComponentsHolder,
+		CoreComponents:       node.CoreComponentsHolder,
+		DataComponents:       node.DataComponentsHolder,
+		NetworkComponents:    node.NetworkComponentsHolder,
+		CryptoComponents:     node.CryptoComponentsHolder,
+		ProcessComponents:    node.ProcessComponentsHolder,
+		StatusCoreComponents: node.StatusCoreComponents,
+	})
+	if err != nil {
+		return err
+	}
+
+	managedHeartbeatV2Components, err := heartbeat.NewManagedHeartbeatV2Components(heartBeatComponentsFactory)
+	if err != nil {
+		return err
+	}
+
+	err = managedHeartbeatV2Components.Create()
+	if err != nil {
+		return err
+	}
+
 	nd, err := nodePack.NewNode(
 		nodePack.WithStatusCoreComponents(node.StatusCoreComponents),
 		nodePack.WithCoreComponents(node.CoreComponentsHolder),
@@ -95,6 +123,7 @@ func (node *testOnlyProcessingNode) createFacade(configs config.Configs, apiInte
 		nodePack.WithNodeStopChannel(node.CoreComponentsHolder.ChanStopNodeProcess()),
 		nodePack.WithImportMode(configs.ImportDbConfig.IsImportDBMode),
 		nodePack.WithESDTNFTStorageHandler(node.ProcessComponentsHolder.ESDTDataStorageHandlerForAPI()),
+		nodePack.WithHeartbeatV2Components(managedHeartbeatV2Components),
 	)
 	if err != nil {
 		return errors.New("error creating node: " + err.Error())

node/chainSimulator/components/statusComponents.go

@@ -9,16 +9,22 @@ import (
 	"github.com/multiversx/mx-chain-core-go/core"
 	"github.com/multiversx/mx-chain-core-go/core/appStatusPolling"
 	"github.com/multiversx/mx-chain-core-go/core/check"
+	nodeData "github.com/multiversx/mx-chain-core-go/data"
+	outportCore "github.com/multiversx/mx-chain-core-go/data/outport"
 	factoryMarshalizer "github.com/multiversx/mx-chain-core-go/marshal/factory"
 	indexerFactory "github.com/multiversx/mx-chain-es-indexer-go/process/factory"
 	"github.com/multiversx/mx-chain-go/common"
 	"github.com/multiversx/mx-chain-go/common/statistics"
 	"github.com/multiversx/mx-chain-go/config"
+	"github.com/multiversx/mx-chain-go/epochStart"
+	"github.com/multiversx/mx-chain-go/epochStart/notifier"
 	"github.com/multiversx/mx-chain-go/errors"
+	"github.com/multiversx/mx-chain-go/factory"
 	"github.com/multiversx/mx-chain-go/integrationTests/mock"
 	"github.com/multiversx/mx-chain-go/outport"
-	"github.com/multiversx/mx-chain-go/outport/factory"
+	outportFactory "github.com/multiversx/mx-chain-go/outport/factory"
 	"github.com/multiversx/mx-chain-go/process"
+	"github.com/multiversx/mx-chain-go/sharding/nodesCoordinator"
 	"github.com/multiversx/mx-chain-go/testscommon"
 )
 
@@ -32,10 +38,11 @@ type statusComponentsHolder struct {
 	statusPollingIntervalSec int
 	cancelFunc               func()
 	mutex                    sync.RWMutex
+	nodesCoordinator         nodesCoordinator.NodesCoordinator
 }
 
 // CreateStatusComponents will create a new instance of status components holder
-func CreateStatusComponents(shardID uint32, appStatusHandler core.AppStatusHandler, statusPollingIntervalSec int, external config.ExternalConfig, coreComponents process.CoreComponentsHolder) (*statusComponentsHolder, error) {
+func CreateStatusComponents(shardID uint32, appStatusHandler core.AppStatusHandler, statusPollingIntervalSec int, external config.ExternalConfig, coreComponents factory.CoreComponentsHandler) (*statusComponentsHolder, error) {
 	if check.IfNil(appStatusHandler) {
 		return nil, core.ErrNilAppStatusHandler
 	}
@@ -51,12 +58,12 @@ func CreateStatusComponents(shardID uint32, appStatusHandler core.AppStatusHandl
 	if err != nil {
 		return nil, err
 	}
-	instance.outportHandler, err = factory.CreateOutport(&factory.OutportFactoryArgs{
+	instance.outportHandler, err = outportFactory.CreateOutport(&outportFactory.OutportFactoryArgs{
 		IsImportDB:                false,
 		ShardID:                   shardID,
 		RetrialInterval:           time.Second,
 		HostDriversArgs:           hostDriverArgs,
-		EventNotifierFactoryArgs:  &factory.EventNotifierFactoryArgs{},
+		EventNotifierFactoryArgs:  &outportFactory.EventNotifierFactoryArgs{},
 		ElasticIndexerFactoryArgs: makeElasticIndexerArgs(external, coreComponents),
 	})
 	if err != nil {
@@ -65,13 +72,45 @@ func CreateStatusComponents(shardID uint32, appStatusHandler core.AppStatusHandl
 	instance.softwareVersionChecker = &mock.SoftwareVersionCheckerMock{}
 	instance.managedPeerMonitor = &testscommon.ManagedPeersMonitorStub{}
 
+	if shardID == core.MetachainShardId {
+		coreComponents.EpochStartNotifierWithConfirm().RegisterHandler(instance.epochStartEventHandler())
+	}
+
 	instance.collectClosableComponents()
 
 	return instance, nil
 }
 
-func makeHostDriversArgs(external config.ExternalConfig) ([]factory.ArgsHostDriverFactory, error) {
-	argsHostDriverFactorySlice := make([]factory.ArgsHostDriverFactory, 0, len(external.HostDriversConfig))
+// SetNodesCoordinator will set the nodes coordinator
+func (s *statusComponentsHolder) SetNodesCoordinator(nodesCoordinator nodesCoordinator.NodesCoordinator) {
+	s.mutex.Lock()
+	s.nodesCoordinator = nodesCoordinator
+	s.mutex.Unlock()
+}
+
+func (s *statusComponentsHolder) epochStartEventHandler() epochStart.ActionHandler {
+	subscribeHandler := notifier.NewHandlerForEpochStart(func(hdr nodeData.HeaderHandler) {
+		currentEpoch := hdr.GetEpoch()
+		validatorsPubKeys, err := s.nodesCoordinator.GetAllEligibleValidatorsPublicKeys(currentEpoch)
+		if err != nil {
+			log.Warn("s.nodesCoordinator.GetAllEligibleValidatorPublicKeys for current epoch failed",
+				"epoch", currentEpoch,
+				"error", err.Error())
+		}
+
+		s.outportHandler.SaveValidatorsPubKeys(&outportCore.ValidatorsPubKeys{
+			ShardID:                hdr.GetShardID(),
+			ShardValidatorsPubKeys: outportCore.ConvertPubKeys(validatorsPubKeys),
+			Epoch:                  currentEpoch,
+		})
+
+	}, func(_ nodeData.HeaderHandler) {}, common.IndexerOrder)
+
+	return subscribeHandler
+}
+
+func makeHostDriversArgs(external config.ExternalConfig) ([]outportFactory.ArgsHostDriverFactory, error) {
+	argsHostDriverFactorySlice := make([]outportFactory.ArgsHostDriverFactory, 0, len(external.HostDriversConfig))
 	for idx := 0; idx < len(external.HostDriversConfig); idx++ {
 		hostConfig := external.HostDriversConfig[idx]
 		if !hostConfig.Enabled {
@@ -83,7 +122,7 @@ func makeHostDriversArgs(external config.ExternalConfig) ([]factory.ArgsHostDriv
 			return argsHostDriverFactorySlice, err
 		}
 
-		argsHostDriverFactorySlice = append(argsHostDriverFactorySlice, factory.ArgsHostDriverFactory{
+		argsHostDriverFactorySlice = append(argsHostDriverFactorySlice, outportFactory.ArgsHostDriverFactory{
 			Marshaller: marshaller,
 			HostConfig: hostConfig,
 		})

node/chainSimulator/components/testOnlyProcessingNode.go

@@ -148,7 +148,8 @@ func NewTestOnlyProcessingNode(args ArgsTestOnlyProcessingNode) (*testOnlyProces
 	}
 
 	selfShardID := instance.GetShardCoordinator().SelfId()
-	instance.StatusComponentsHolder, err = CreateStatusComponents(
+
+	statusComponentsH, err := CreateStatusComponents(
 		selfShardID,
 		instance.StatusCoreComponents.AppStatusHandler(),
 		args.Configs.GeneralConfig.GeneralSettings.StatusPollingIntervalSec,
@@ -159,6 +160,8 @@ func NewTestOnlyProcessingNode(args ArgsTestOnlyProcessingNode) (*testOnlyProces
 		return nil, err
 	}
 
+	instance.StatusComponentsHolder = statusComponentsH
+
 	err = instance.createBlockChain(selfShardID)
 	if err != nil {
 		return nil, err
@@ -184,6 +187,8 @@ func NewTestOnlyProcessingNode(args ArgsTestOnlyProcessingNode) (*testOnlyProces
 		return nil, err
 	}
 
+	statusComponentsH.SetNodesCoordinator(instance.NodesCoordinator)
+
 	instance.DataComponentsHolder, err = CreateDataComponents(ArgsDataComponentsHolder{
 		Chain:              instance.ChainHandler,
 		StorageService:     instance.StoreService,