feat: add metrics for relayer (#24)

* add metrics for relayer

* re-format

* renaming

app/app.go

@@ -13,13 +13,15 @@ import (
 	"github.com/bnb-chain/greenfield-relayer/db/model"
 	"github.com/bnb-chain/greenfield-relayer/executor"
 	"github.com/bnb-chain/greenfield-relayer/listener"
+	"github.com/bnb-chain/greenfield-relayer/metric"
 	"github.com/bnb-chain/greenfield-relayer/relayer"
 	"github.com/bnb-chain/greenfield-relayer/vote"
 )
 
 type App struct {
-	BSCRelayer  *relayer.BSCRelayer
-	GnfdRelayer *relayer.GreenfieldRelayer
+	BSCRelayer    *relayer.BSCRelayer
+	GnfdRelayer   *relayer.GreenfieldRelayer
+	metricService *metric.MetricService
 }
 
 func NewApp(cfg *config.Config) *App {
@@ -44,9 +46,11 @@ func NewApp(cfg *config.Config) *App {
 	greenfieldExecutor.SetBSCExecutor(bscExecutor)
 	bscExecutor.SetGreenfieldExecutor(greenfieldExecutor)
 
+	metricService := metric.NewMetricService(cfg)
+
 	// listeners
-	greenfieldListener := listener.NewGreenfieldListener(cfg, greenfieldExecutor, bscExecutor, daoManager)
-	bscListener := listener.NewBSCListener(cfg, bscExecutor, greenfieldExecutor, daoManager)
+	greenfieldListener := listener.NewGreenfieldListener(cfg, greenfieldExecutor, bscExecutor, daoManager, metricService)
+	bscListener := listener.NewBSCListener(cfg, bscExecutor, greenfieldExecutor, daoManager, metricService)
 
 	// vote signer
 	signer := vote.NewVoteSigner(ethcommon.Hex2Bytes(cfg.GreenfieldConfig.BlsPrivateKey))
@@ -56,20 +60,22 @@ func NewApp(cfg *config.Config) *App {
 	bscVoteProcessor := vote.NewBSCVoteProcessor(cfg, daoManager, signer, bscExecutor)
 
 	// assemblers
-	greenfieldAssembler := assembler.NewGreenfieldAssembler(cfg, greenfieldExecutor, daoManager, bscExecutor)
-	bscAssembler := assembler.NewBSCAssembler(cfg, bscExecutor, daoManager, greenfieldExecutor)
+	greenfieldAssembler := assembler.NewGreenfieldAssembler(cfg, greenfieldExecutor, daoManager, bscExecutor, metricService)
+	bscAssembler := assembler.NewBSCAssembler(cfg, bscExecutor, daoManager, greenfieldExecutor, metricService)
 
 	// relayers
 	gnfdRelayer := relayer.NewGreenfieldRelayer(greenfieldListener, greenfieldExecutor, bscExecutor, greenfieldVoteProcessor, greenfieldAssembler)
 	bscRelayer := relayer.NewBSCRelayer(bscListener, greenfieldExecutor, bscExecutor, bscVoteProcessor, bscAssembler)
 
 	return &App{
-		BSCRelayer:  bscRelayer,
-		GnfdRelayer: gnfdRelayer,
+		BSCRelayer:    bscRelayer,
+		GnfdRelayer:   gnfdRelayer,
+		metricService: metricService,
 	}
 }
 
 func (a *App) Start() {
 	a.GnfdRelayer.Start()
 	a.BSCRelayer.Start()
+	a.metricService.Start()
 }

assembler/bsc_assembler.go

@@ -3,6 +3,7 @@ package assembler
 import (
 	"encoding/hex"
 	"fmt"
+	"github.com/bnb-chain/greenfield-relayer/metric"
 	"time"
 
 	"github.com/bnb-chain/greenfield-relayer/common"
@@ -23,15 +24,17 @@ type BSCAssembler struct {
 	bscExecutor        *executor.BSCExecutor
 	daoManager         *dao.DaoManager
 	blsPubKey          string
+	metricService      *metric.MetricService
 }
 
-func NewBSCAssembler(cfg *config.Config, executor *executor.BSCExecutor, dao *dao.DaoManager, greenfieldExecutor *executor.GreenfieldExecutor) *BSCAssembler {
+func NewBSCAssembler(cfg *config.Config, executor *executor.BSCExecutor, dao *dao.DaoManager, greenfieldExecutor *executor.GreenfieldExecutor, ms *metric.MetricService) *BSCAssembler {
 	return &BSCAssembler{
 		config:             cfg,
 		bscExecutor:        executor,
 		daoManager:         dao,
 		greenfieldExecutor: greenfieldExecutor,
 		blsPubKey:          hex.EncodeToString(util.BlsPubKeyFromPrivKeyStr(cfg.GreenfieldConfig.BlsPrivateKey)),
+		metricService:      ms,
 	}
 }
 
@@ -55,6 +58,7 @@ func (a *BSCAssembler) process(channelId types.ChannelId) error {
 		return err
 	}
 	isInturnRelyer := inturnRelayer.BlsPubKey == a.blsPubKey
+	a.metricService.SetGnfdInturnRelayerMetrics(isInturnRelyer, inturnRelayer.RelayInterval.Start, inturnRelayer.RelayInterval.End)
 	var startSequence uint64
 	if isInturnRelyer {
 		seq, err := a.daoManager.SequenceDao.GetByChannelId(uint8(channelId))
@@ -81,6 +85,12 @@ func (a *BSCAssembler) process(channelId types.ChannelId) error {
 			}
 		}
 		logging.Logger.Debug("bsc relay as in-turn relayer")
+		a.metricService.SetNextSequenceForChannelFromDB(uint8(channelId), startSequence)
+		seqFromChain, err := a.bscExecutor.GetNextDeliveryOracleSequenceWithRetry()
+		if err != nil {
+			return err
+		}
+		a.metricService.SetNextSequenceForChannelFromChain(uint8(channelId), seqFromChain)
 	} else {
 		// non-inturn relayer retries every 10 second, gets the sequence from chain
 		time.Sleep(time.Duration(a.config.RelayConfig.GreenfieldSequenceUpdateLatency) * time.Second)
@@ -162,6 +172,7 @@ func (a *BSCAssembler) processPkgs(pkgs []*model.BscRelayPackage, channelId uint
 	for _, p := range pkgs {
 		pkgIds = append(pkgIds, p.Id)
 	}
+	a.metricService.SetBSCProcessedBlockHeight(pkgs[0].Height)
 	if !isInturnRelyer {
 		if err = a.daoManager.BSCDao.UpdateBatchPackagesClaimedTxHash(pkgIds, txHash); err != nil {
 			return err
@@ -176,6 +187,5 @@ func (a *BSCAssembler) processPkgs(pkgs []*model.BscRelayPackage, channelId uint
 	if err = a.daoManager.SequenceDao.Upsert(channelId, sequence+1); err != nil {
 		return err
 	}
-
 	return nil
 }

assembler/greenfield_assembler.go

@@ -12,6 +12,7 @@ import (
 	"github.com/bnb-chain/greenfield-relayer/db/model"
 	"github.com/bnb-chain/greenfield-relayer/executor"
 	"github.com/bnb-chain/greenfield-relayer/logging"
+	"github.com/bnb-chain/greenfield-relayer/metric"
 	"github.com/bnb-chain/greenfield-relayer/types"
 	"github.com/bnb-chain/greenfield-relayer/util"
 	"github.com/bnb-chain/greenfield-relayer/vote"
@@ -23,15 +24,18 @@ type GreenfieldAssembler struct {
 	greenfieldExecutor *executor.GreenfieldExecutor
 	daoManager         *dao.DaoManager
 	blsPubKey          string
+	metricService      *metric.MetricService
 }
 
-func NewGreenfieldAssembler(cfg *config.Config, executor *executor.GreenfieldExecutor, dao *dao.DaoManager, bscExecutor *executor.BSCExecutor) *GreenfieldAssembler {
+func NewGreenfieldAssembler(cfg *config.Config, executor *executor.GreenfieldExecutor, dao *dao.DaoManager, bscExecutor *executor.BSCExecutor,
+	ms *metric.MetricService) *GreenfieldAssembler {
 	return &GreenfieldAssembler{
 		config:             cfg,
 		greenfieldExecutor: executor,
 		daoManager:         dao,
 		bscExecutor:        bscExecutor,
 		blsPubKey:          hex.EncodeToString(util.BlsPubKeyFromPrivKeyStr(cfg.GreenfieldConfig.BlsPrivateKey)),
+		metricService:      ms,
 	}
 }
 
@@ -58,6 +62,7 @@ func (a *GreenfieldAssembler) process(channelId types.ChannelId) error {
 		return err
 	}
 	isInturnRelyer := inturnRelayer.BlsPublicKey == a.blsPubKey
+	a.metricService.SetBSCInturnRelayerMetrics(isInturnRelyer, inturnRelayer.Start, inturnRelayer.End)
 	var startSequence uint64
 	if isInturnRelyer {
 		// get next delivered sequence from DB
@@ -85,7 +90,12 @@ func (a *GreenfieldAssembler) process(channelId types.ChannelId) error {
 				return err
 			}
 		}
-		logging.Logger.Debug("gnfd relay as in-turn relayer")
+		a.metricService.SetNextSequenceForChannelFromDB(uint8(channelId), startSequence)
+		seqFromChain, err := a.greenfieldExecutor.GetNextDeliverySequenceForChannelWithRetry(channelId)
+		if err != nil {
+			return err
+		}
+		a.metricService.SetNextSequenceForChannelFromChain(uint8(channelId), seqFromChain)
 	} else {
 		time.Sleep(time.Duration(a.config.RelayConfig.BSCSequenceUpdateLatency) * time.Second)
 		startSequence, err = a.greenfieldExecutor.GetNextDeliverySequenceForChannelWithRetry(channelId)
@@ -131,6 +141,7 @@ func (a *GreenfieldAssembler) process(channelId types.ChannelId) error {
 			return err
 		}
 		logging.Logger.Infof("relayed tx with channel id %d and sequence %d ", tx.ChannelId, tx.Sequence)
+
 		nonce++
 	}
 	return nil
@@ -157,6 +168,7 @@ func (a *GreenfieldAssembler) processTx(tx *model.GreenfieldRelayTransaction, no
 		return err
 	}
 	logging.Logger.Infof("relayed transaction with channel id %d and sequence %d, get txHash %s", tx.ChannelId, tx.Sequence, txHash)
+	a.metricService.SetGnfdProcessedBlockHeight(tx.Height)
 
 	// update next delivery sequence in DB for inturn relayer, for non-inturn relayer, there is enough time for
 	// sequence update, so they can track next start seq from chain
@@ -173,7 +185,6 @@ func (a *GreenfieldAssembler) processTx(tx *model.GreenfieldRelayTransaction, no
 	if err = a.daoManager.SequenceDao.Upsert(tx.ChannelId, tx.Sequence+1); err != nil {
 		return err
 	}
-
 	return nil
 }
 

config/config.go

@@ -18,12 +18,12 @@ type Config struct {
 }
 
 type AdminConfig struct {
-	ListenAddr string `json:"listen_addr"`
+	Port uint16 `json:"port"`
 }
 
 func (cfg *AdminConfig) Validate() {
-	if cfg.ListenAddr == "" {
-		panic("listen address should not be empty")
+	if cfg.Port <= 0 || cfg.Port > 65535 {
+		panic("port should be within (0, 65535]")
 	}
 }
 

integrationtest/config/config_test.json

@@ -54,7 +54,7 @@
     "compress": false
   },
   "admin_config": {
-    "listen_addr": "0.0.0.0:8080"
+    "port": 8080
   },
   "db_config": {
     "dialect": "mysql",

listener/bsc_listener.go

@@ -18,6 +18,7 @@ import (
 	"github.com/bnb-chain/greenfield-relayer/executor"
 	"github.com/bnb-chain/greenfield-relayer/executor/crosschain"
 	"github.com/bnb-chain/greenfield-relayer/logging"
+	"github.com/bnb-chain/greenfield-relayer/metric"
 	rtypes "github.com/bnb-chain/greenfield-relayer/types"
 )
 
@@ -27,9 +28,10 @@ type BSCListener struct {
 	greenfieldExecutor *executor.GreenfieldExecutor
 	DaoManager         *dao.DaoManager
 	crossChainAbi      abi.ABI
+	monitorService     *metric.MetricService
 }
 
-func NewBSCListener(cfg *config.Config, bscExecutor *executor.BSCExecutor, gnfdExecutor *executor.GreenfieldExecutor, dao *dao.DaoManager) *BSCListener {
+func NewBSCListener(cfg *config.Config, bscExecutor *executor.BSCExecutor, gnfdExecutor *executor.GreenfieldExecutor, dao *dao.DaoManager, ms *metric.MetricService) *BSCListener {
 	crossChainAbi, err := abi.JSON(strings.NewReader(crosschain.CrosschainMetaData.ABI))
 	if err != nil {
 		panic("marshal abi error")
@@ -40,6 +42,7 @@ func NewBSCListener(cfg *config.Config, bscExecutor *executor.BSCExecutor, gnfdE
 		greenfieldExecutor: gnfdExecutor,
 		DaoManager:         dao,
 		crossChainAbi:      crossChainAbi,
+		monitorService:     ms,
 	}
 }
 
@@ -130,14 +133,17 @@ func (l *BSCListener) monitorCrossChainPkgAt(nextHeight uint64, latestPolledBloc
 		relayPkgs = append(relayPkgs, relayPkg)
 	}
 
-	return l.DaoManager.BSCDao.SaveBlockAndBatchPackages(
+	if err := l.DaoManager.BSCDao.SaveBlockAndBatchPackages(
 		&model.BscBlock{
 			BlockHash:  nextHeightBlockHeader.Hash().String(),
 			ParentHash: nextHeightBlockHeader.ParentHash.String(),
 			Height:     nextHeight,
 			BlockTime:  int64(nextHeightBlockHeader.Time),
-		},
-		relayPkgs)
+		}, relayPkgs); err != nil {
+		return err
+	}
+	l.monitorService.SetBSCSavedBlockHeight(nextHeight)
+	return nil
 }
 
 func (l *BSCListener) queryCrossChainLogs(blockHash ethcommon.Hash) ([]types.Log, error) {

listener/greenfield_listener.go

@@ -17,6 +17,7 @@ import (
 	"github.com/bnb-chain/greenfield-relayer/db/model"
 	"github.com/bnb-chain/greenfield-relayer/executor"
 	"github.com/bnb-chain/greenfield-relayer/logging"
+	"github.com/bnb-chain/greenfield-relayer/metric"
 	"github.com/bnb-chain/greenfield-relayer/util"
 )
 
@@ -25,14 +26,17 @@ type GreenfieldListener struct {
 	greenfieldExecutor *executor.GreenfieldExecutor
 	bscExecutor        *executor.BSCExecutor
 	DaoManager         *dao.DaoManager
+	metricService      *metric.MetricService
 }
 
-func NewGreenfieldListener(cfg *config.Config, gnfdExecutor *executor.GreenfieldExecutor, bscExecutor *executor.BSCExecutor, dao *dao.DaoManager) *GreenfieldListener {
+func NewGreenfieldListener(cfg *config.Config, gnfdExecutor *executor.GreenfieldExecutor, bscExecutor *executor.BSCExecutor,
+	dao *dao.DaoManager, ms *metric.MetricService) *GreenfieldListener {
 	return &GreenfieldListener{
 		config:             cfg,
 		greenfieldExecutor: gnfdExecutor,
 		bscExecutor:        bscExecutor,
 		DaoManager:         dao,
+		metricService:      ms,
 	}
 }
 
@@ -83,7 +87,11 @@ func (l *GreenfieldListener) poll() error {
 				Height:    uint64(block.Height),
 				BlockTime: block.Time.Unix(),
 			}
-			return l.DaoManager.GreenfieldDao.SaveBlockAndBatchTransactions(b, txs)
+			if err := l.DaoManager.GreenfieldDao.SaveBlockAndBatchTransactions(b, txs); err != nil {
+				return err
+			}
+			l.metricService.SetGnfdSavedBlockHeight(uint64(block.Height))
+			return nil
 		}
 	}
 }