Implement bridge stats (#82)

* feat: implement bridgestats logics * chore: Add necessary environment variables to trigger netstats and add send error messages to bridge's stats into `task.go` * nit: remove unused init function in bridge stats * chore: correct readLoop and reportLoop to make them return if quitCh is closed * Add pipeline on pull request * fix compile error * remove unused code * Remove unused code * Rename from pass to secret * Disable pushing nightly from PR * Feat/implement bridge stats (#83) * fix compile error * Remove unused code * Rename from pass to secret * Add log for tracking * Add for tracking report process * Add read loop * Rework ronin stats logic * Remove lock * Rework readLoop * Add ping handler from server * Reduce ticker from 10 to 5s * Remove unused locking, add handle signal quit, and move functionality to a separate function * Fix typo and create a constant * Remove lock in Service struct --------- Co-authored-by: Huy Ngo <[email protected]>
axieinfinity · Aug 7, 2023 · 4f21bd6 · 4f21bd6
1 parent 1ef9f11
commit 4f21bd6
Show file tree

Hide file tree

Showing 9 changed files with 434 additions and 3 deletions.
diff --git a/Dockerfile b/Dockerfile
@@ -45,6 +45,10 @@ ENV DB_CONN_MAX_LIFE_TIME ''
 ENV DB_MAX_IDLE_CONNS ''
 ENV DB_MAX_OPEN_CONNS ''
 
+ENV BRIDGE_STATS_NODE_NAME ''
+ENV BRIDGE_STATS_URL ''
+ENV BRIDGE_STATS_SECRET ''
+
 COPY --from=builder /go/bin/bridge /usr/local/bin/bridge
 COPY --from=builder /opt/bridge/config/ ./
 COPY --from=builder /opt/bridge/docker/entrypoint.sh ./

diff --git a/cmd/bridge/main.go b/cmd/bridge/main.go
@@ -3,6 +3,8 @@ package main
 import (
 	"encoding/json"
 	"fmt"
+	"github.com/axieinfinity/bridge-v2/stats"
+	"gorm.io/gorm"
 	"io/ioutil"
 	"os"
 	"os/signal"
@@ -85,6 +87,10 @@ const (
 	RoninNetwork    = "Ronin"
 
 	fromBlock = "FROM_BLOCK"
+
+	bridgeStatsNodeName = "BRIDGE_STATS_NODE_NAME"
+	bridgeStatsUrl      = "BRIDGE_STATS_URL"
+	bridgeStatsSecret   = "BRIDGE_STATS_SECRET"
 )
 
 var (
@@ -401,13 +407,40 @@ func createPgDb(cfg *bridgeCore.Config) {
 	}
 }
 
+func setupStats(cfg *bridgeCore.Config, db *gorm.DB) {
+	// check if bridge stats is enabled
+	if os.Getenv(bridgeStatsNodeName) != "" && os.Getenv(bridgeStatsUrl) != "" {
+		// NOTE: only get chainId, operator from ronin
+		var (
+			chainId, operator string
+			node              = os.Getenv(bridgeStatsNodeName)
+			host              = os.Getenv(bridgeStatsUrl)
+			pass              = os.Getenv(bridgeStatsSecret)
+		)
+		if _, ok := cfg.Listeners[RoninNetwork]; ok {
+			chainId = cfg.Listeners[RoninNetwork].ChainId
+			if cfg.Listeners[RoninNetwork].Secret != nil && cfg.Listeners[RoninNetwork].Secret.BridgeOperator != nil {
+				signMethod, err := bridgeCoreUtils.NewSignMethod(cfg.Listeners[RoninNetwork].Secret.BridgeOperator)
+				if err != nil {
+					panic(err)
+				}
+				operator = signMethod.GetAddress().Hex()
+			}
+		}
+		stats.NewService(node, chainId, operator, host, pass, db)
+		go stats.BridgeStats.Start()
+	}
+}
+
 func bridge(ctx *cli.Context) {
 	cfg := prepare(ctx)
 	// init db
 	db, err := bridgeCoreStore.MustConnectDatabase(cfg.DB, false)
 	if err != nil {
 		panic(err)
 	}
+	// setup stats
+	setupStats(cfg, db)
 	// start migration
 	if err = migration.Migrate(db, cfg); err != nil {
 		panic(err)
@@ -427,6 +460,9 @@ func bridge(ctx *cli.Context) {
 	select {
 	case <-sigc:
 		controller.Close()
+		if stats.BridgeStats != nil {
+			stats.BridgeStats.Stop()
+		}
 	}
 }
 

diff --git a/listener/ethereum.go b/listener/ethereum.go
@@ -9,6 +9,8 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/axieinfinity/bridge-v2/stats"
+
 	bridgeCore "github.com/axieinfinity/bridge-core"
 	"github.com/axieinfinity/bridge-core/metrics"
 	bridgeCoreModels "github.com/axieinfinity/bridge-core/models"
@@ -290,7 +292,9 @@ func (e *EthereumListener) SaveCurrentBlockToDB() error {
 	if err := e.store.GetProcessedBlockStore().Save(hexutil.EncodeBig(chainId), int64(e.GetCurrentBlock().GetHeight())); err != nil {
 		return err
 	}
-
+	if stats.BridgeStats != nil {
+		stats.BridgeStats.SendProcessedBlock(e.GetName(), e.GetCurrentBlock().GetHeight())
+	}
 	metrics.Pusher.IncrCounter(fmt.Sprintf(metrics.ListenerProcessedBlockMetric, e.GetName()), 1)
 	return nil
 }

diff --git a/listener/types.go b/listener/types.go
@@ -2,6 +2,7 @@ package listener
 
 import (
 	"context"
+	"github.com/axieinfinity/bridge-v2/stats"
 	"math/big"
 	"time"
 
@@ -216,6 +217,12 @@ func NewEthListenJob(jobType int, listener bridgeCore.Listener, subscriptionName
 	}
 }
 
+func (e *EthListenJob) Process() ([]byte, error) {
+	data, err := e.BaseJob.Process()
+	stats.SendErrorToStats(e.GetListener(), err)
+	return data, err
+}
+
 type EthCallbackJob struct {
 	*bridgeCore.BaseJob
 	result interface{}
@@ -248,10 +255,12 @@ func (e *EthCallbackJob) Process() ([]byte, error) {
 	log.Info("[EthCallbackJob] Start Process", "method", e.method, "jobId", e.GetID())
 	val, err := e.Utils().Invoke(e.GetListener(), e.method, e.FromChainID(), e.GetTransaction(), e.GetData())
 	if err != nil {
+		stats.SendErrorToStats(e.GetListener(), err)
 		return nil, err
 	}
 	invokeErr, ok := val.Interface().(error)
 	if ok {
+		stats.SendErrorToStats(e.GetListener(), invokeErr)
 		return nil, invokeErr
 	}
 	return nil, nil