Repository: cosmos/relayer
Pull request #963 (2022-08-30 13:59:01)

Additional Metrics
Adding metrics for:
-    CurrentBlockHeight
-    CurrentRelayerBalance
 -   CurrentFeesPaid

jackzampolin: This looks great but the build is broken. Great PR @joeabbey !
jackzampolin: Looking at the issues
- https://github.com/cosmos/relayer/issues/263
- https://github.com/cosmos/relayer/issues/877

The only one we don't have is the fees paid which could be derived from wallet balance but would be nice to have seperately.

Patch:
diff --git a/relayer/chains/cosmos/cosmos_chain_processor.go b/relayer/chains/cosmos/cosmos_chain_processor.go
index eafbaea..6e77791 100644
--- a/relayer/chains/cosmos/cosmos_chain_processor.go
+++ b/relayer/chains/cosmos/cosmos_chain_processor.go
@@ -4,9 +4,11 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"math/big"
 	"time"
 
 	"github.com/avast/retry-go/v4"
+	sdk "github.com/cosmos/cosmos-sdk/types"
 	clienttypes "github.com/cosmos/ibc-go/v5/modules/core/02-client/types"
 	conntypes "github.com/cosmos/ibc-go/v5/modules/core/03-connection/types"
 	chantypes "github.com/cosmos/ibc-go/v5/modules/core/04-channel/types"
@@ -44,9 +46,18 @@ type CosmosChainProcessor struct {
 
 	// map of channel ID to connection ID
 	channelConnections map[string]string
+
+	// metrics to monitor lifetime of processor
+	metrics *processor.PrometheusMetrics
+
+	// parsed gas prices accepted by the chain (only used for metrics)
+	parsedGasPrices *sdk.DecCoins
 }
 
-func NewCosmosChainProcessor(log *zap.Logger, provider *CosmosProvider) *CosmosChainProcessor {
+func NewCosmosChainProcessor(log *zap.Logger, provider *CosmosProvider, metrics *processor.PrometheusMetrics) *CosmosChainProcessor {
+
+	provider.SetMetrics(metrics)
+
 	return &CosmosChainProcessor{
 		log:                  log.With(zap.String("chain_name", provider.ChainName()), zap.String("chain_id", provider.ChainId())),
 		chainProvider:        provider,
@@ -55,6 +66,7 @@ func NewCosmosChainProcessor(log *zap.Logger, provider *CosmosProvider) *CosmosC
 		channelStateCache:    make(processor.ChannelStateCache),
 		connectionClients:    make(map[string]string),
 		channelConnections:   make(map[string]string),
+		metrics:              metrics,
 	}
 }
 
@@ -64,8 +76,9 @@ const (
 	latestHeightQueryRetryDelay = 1 * time.Second
 	latestHeightQueryRetries    = 5
 
-	defaultMinQueryLoopDuration = 1 * time.Second
-	inSyncNumBlocksThreshold    = 2
+	defaultMinQueryLoopDuration      = 1 * time.Second
+	defaultBalanceUpdateWaitDuration = 60 * time.Second
+	inSyncNumBlocksThreshold         = 2
 )
 
 // latestClientState is a map of clientID to the latest clientInfo for that client.
@@ -132,9 +145,11 @@ func (ccp *CosmosChainProcessor) clientState(ctx context.Context, clientID strin
 
 // queryCyclePersistence hold the variables that should be retained across queryCycles.
 type queryCyclePersistence struct {
-	latestHeight         int64
-	latestQueriedBlock   int64
-	minQueryLoopDuration time.Duration
+	latestHeight              int64
+	latestQueriedBlock        int64
+	minQueryLoopDuration      time.Duration
+	lastBalanceUpdate         time.Time
+	balanceUpdateWaitDuration time.Duration
 }
 
 // Run starts the query loop for the chain which will gather applicable ibc messages and push events out to the relevant PathProcessors.
@@ -143,7 +158,9 @@ type queryCyclePersistence struct {
 func (ccp *CosmosChainProcessor) Run(ctx context.Context, initialBlockHistory uint64) error {
 	// this will be used for persistence across query cycle loop executions
 	persistence := queryCyclePersistence{
-		minQueryLoopDuration: defaultMinQueryLoopDuration,
+		minQueryLoopDuration:      defaultMinQueryLoopDuration,
+		lastBalanceUpdate:         time.Unix(0, 0),
+		balanceUpdateWaitDuration: defaultBalanceUpdateWaitDuration,
 	}
 
 	// Infinite retry to get initial latest height
@@ -267,6 +284,10 @@ func (ccp *CosmosChainProcessor) queryCycle(ctx context.Context, persistence *qu
 		zap.Int64("latest_height", persistence.latestHeight),
 	)
 
+	if ccp.metrics != nil {
+		ccp.CollectMetrics(ctx, persistence)
+	}
+
 	// used at the end of the cycle to send signal to path processors to start processing if both chains are in sync and no new messages came in this cycle
 	firstTimeInSync := false
 
@@ -390,3 +411,51 @@ func (ccp *CosmosChainProcessor) queryCycle(ctx context.Context, persistence *qu
 
 	return nil
 }
+
+func (ccp *CosmosChainProcessor) CollectMetrics(ctx context.Context, persistence *queryCyclePersistence) {
+	ccp.CurrentBlockHeight(ctx, persistence)
+
+	// Wait a while before updating the balance
+	if time.Since(persistence.lastBalanceUpdate) > persistence.balanceUpdateWaitDuration {
+		ccp.CurrentRelayerBalance(ctx)
+		persistence.lastBalanceUpdate = time.Now()
+	}
+}
+
+func (ccp *CosmosChainProcessor) CurrentBlockHeight(ctx context.Context, persistence *queryCyclePersistence) {
+	ccp.metrics.SetLatestHeight(ccp.chainProvider.ChainId(), persistence.latestHeight)
+}
+
+func (ccp *CosmosChainProcessor) CurrentRelayerBalance(ctx context.Context) {
+	// memoize the current gas prices to only show metrics for "interesting" denoms
+	if ccp.parsedGasPrices == nil {
+		gp, err := sdk.ParseDecCoins(ccp.chainProvider.Config.GasPrices)
+		if err != nil {
+			ccp.log.Error(
+				"Failed to parse gas prices",
+				zap.Error(err),
+			)
+		}
+		ccp.parsedGasPrices = &gp
+	}
+
+	// Get the balance for the chain provider's key
+	relayerWalletBalance, err := ccp.chainProvider.QueryBalance(ctx, ccp.chainProvider.Key())
+	if err != nil {
+		ccp.log.Error(
+			"Failed to query relayer balance",
+			zap.Error(err),
+		)
+	}
+
+	// Print the relevant gas prices
+	for _, gasDenom := range *ccp.parsedGasPrices {
+		for _, balance := range relayerWalletBalance {
+			if balance.Denom == gasDenom.Denom {
+				// Convert to a big float to get a float64 for metrics
+				f, _ := big.NewFloat(0.0).SetInt(balance.Amount.BigInt()).Float64()
+				ccp.metrics.SetWalletBalance(ccp.chainProvider.ChainId(), ccp.chainProvider.Key(), balance.Denom, f)
+			}
+		}
+	}
+}
diff --git a/relayer/chains/cosmos/provider.go b/relayer/chains/cosmos/provider.go
index 0d821fa..113e767 100644
--- a/relayer/chains/cosmos/provider.go
+++ b/relayer/chains/cosmos/provider.go
@@ -6,10 +6,12 @@ import (
 	"os"
 	"time"
 
+	sdk "github.com/cosmos/cosmos-sdk/types"
 	"github.com/cosmos/cosmos-sdk/types/module"
 	commitmenttypes "github.com/cosmos/ibc-go/v5/modules/core/23-commitment/types"
 	ibcexported "github.com/cosmos/ibc-go/v5/modules/core/exported"
 	tmclient "github.com/cosmos/ibc-go/v5/modules/light-clients/07-tendermint/types"
+	"github.com/cosmos/relayer/v2/relayer/processor"
 	"github.com/cosmos/relayer/v2/relayer/provider"
 	"github.com/gogo/protobuf/proto"
 	lens "github.com/strangelove-ventures/lens/client"
@@ -62,9 +64,9 @@ func (pc CosmosProviderConfig) NewProvider(log *zap.Logger, homepath string, deb
 		return nil, err
 	}
 	pc.ChainName = chainName
-	return &CosmosProvider{
-		log: log,
 
+	return &CosmosProvider{
+		log:         log,
 		ChainClient: *cc,
 		PCfg:        pc,
 	}, nil
@@ -95,6 +97,10 @@ type CosmosProvider struct {
 
 	lens.ChainClient
 	PCfg CosmosProviderConfig
+
+	// metrics to monitor the provider
+	TotalFees sdk.Coins
+	metrics   *processor.PrometheusMetrics
 }
 
 type CosmosIBCHeader struct {
@@ -237,3 +243,7 @@ func (cc *CosmosProvider) BlockTime(ctx context.Context, height int64) (time.Tim
 	}
 	return resultBlock.Block.Time, nil
 }
+
+func (cc *CosmosProvider) SetMetrics(m *processor.PrometheusMetrics) {
+	cc.metrics = m
+}
diff --git a/relayer/chains/cosmos/tx.go b/relayer/chains/cosmos/tx.go
index d4a4ffe..9b17b69 100644
--- a/relayer/chains/cosmos/tx.go
+++ b/relayer/chains/cosmos/tx.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"math/big"
 	"strings"
 	"time"
 
@@ -71,9 +72,11 @@ func (cc *CosmosProvider) SendMessage(ctx context.Context, msg provider.RelayerM
 // sent and executed successfully is returned.
 func (cc *CosmosProvider) SendMessages(ctx context.Context, msgs []provider.RelayerMessage, memo string) (*provider.RelayerTxResponse, bool, error) {
 	var resp *sdk.TxResponse
+	var fees sdk.Coins
 
 	if err := retry.Do(func() error {
-		txBytes, err := cc.buildMessages(ctx, msgs, memo)
+		txBytes, f, err := cc.buildMessages(ctx, msgs, memo)
+		fees = f
 		if err != nil {
 			errMsg := err.Error()
 
@@ -179,10 +182,13 @@ func (cc *CosmosProvider) SendMessages(ctx context.Context, msgs []provider.Rela
 	// transaction was successfully executed.
 	if rlyResp.Code != 0 {
 		cc.LogFailedTx(rlyResp, nil, msgs)
+		cc.UpdateFeesSpent(cc.ChainId(), cc.Key(), fees)
 		return rlyResp, false, fmt.Errorf("transaction failed with code: %d", resp.Code)
 	}
 
 	cc.LogSuccessTx(resp, msgs)
+	cc.UpdateFeesSpent(cc.ChainId(), cc.Key(), fees)
+
 	return rlyResp, true, nil
 }
 
@@ -208,11 +214,11 @@ func parseEventsFromTxResponse(resp *sdk.TxResponse) []provider.RelayerEvent {
 	return events
 }
 
-func (cc *CosmosProvider) buildMessages(ctx context.Context, msgs []provider.RelayerMessage, memo string) ([]byte, error) {
+func (cc *CosmosProvider) buildMessages(ctx context.Context, msgs []provider.RelayerMessage, memo string) ([]byte, sdk.Coins, error) {
 	// Query account details
 	txf, err := cc.PrepareFactory(cc.TxFactory())
 	if err != nil {
-		return nil, err
+		return nil, sdk.Coins{}, err
 	}
 
 	if memo != "" {
@@ -225,7 +231,7 @@ func (cc *CosmosProvider) buildMessages(ctx context.Context, msgs []provider.Rel
 	// If users pass gas adjustment, then calculate gas
 	_, adjusted, err := cc.CalculateGas(ctx, txf, CosmosMsgs(msgs...)...)
 	if err != nil {
-		return nil, err
+		return nil, sdk.Coins{}, err
 	}
 
 	// Set the gas amount on the transaction factory
@@ -240,7 +246,7 @@ func (cc *CosmosProvider) buildMessages(ctx context.Context, msgs []provider.Rel
 		}
 		return nil
 	}, retry.Context(ctx), rtyAtt, rtyDel, rtyErr); err != nil {
-		return nil, err
+		return nil, sdk.Coins{}, err
 	}
 
 	// Attach the signature to the transaction
@@ -257,11 +263,13 @@ func (cc *CosmosProvider) buildMessages(ctx context.Context, msgs []provider.Rel
 		}
 		return nil
 	}, retry.Context(ctx), rtyAtt, rtyDel, rtyErr); err != nil {
-		return nil, err
+		return nil, sdk.Coins{}, err
 	}
 
 	done()
 
+	fees := txb.GetTx().GetFee()
+
 	var txBytes []byte
 	// Generate the transaction bytes
 	if err := retry.Do(func() error {
@@ -272,10 +280,10 @@ func (cc *CosmosProvider) buildMessages(ctx context.Context, msgs []provider.Rel
 		}
 		return nil
 	}, retry.Context(ctx), rtyAtt, rtyDel, rtyErr); err != nil {
-		return nil, err
+		return nil, sdk.Coins{}, err
 	}
 
-	return txBytes, nil
+	return txBytes, fees, nil
 }
 
 // MsgCreateClient creates an sdk.Msg to update the client on src with consensus state from dst
@@ -1075,3 +1083,18 @@ func (cc *CosmosProvider) NewClientState(
 		AllowUpdateAfterMisbehaviour: allowUpdateAfterMisbehaviour,
 	}, nil
 }
+
+func (cc *CosmosProvider) UpdateFeesSpent(chain, key string, fees sdk.Coins) {
+	// Don't set the metrics in testing
+	if cc.metrics == nil {
+		return
+	}
+
+	cc.TotalFees = cc.TotalFees.Add(fees...)
+
+	for _, fee := range cc.TotalFees {
+		// Convert to a big float to get a float64 for metrics
+		f, _ := big.NewFloat(0.0).SetInt(fee.Amount.BigInt()).Float64()
+		cc.metrics.SetFeesSpent(chain, key, fee.GetDenom(), f)
+	}
+}
diff --git a/relayer/channel.go b/relayer/channel.go
index 9b1b9c7..23a0169 100644
--- a/relayer/channel.go
+++ b/relayer/channel.go
@@ -77,8 +77,8 @@ func (c *Chain) CreateOpenChannels(
 
 	return processor.NewEventProcessor().
 		WithChainProcessors(
-			srcPathChain.chainProcessor(c.log),
-			dstPathChain.chainProcessor(c.log),
+			srcPathChain.chainProcessor(c.log, nil),
+			dstPathChain.chainProcessor(c.log, nil),
 		).
 		WithPathProcessors(pp).
 		WithInitialBlockHistory(0).
@@ -135,8 +135,8 @@ func (c *Chain) CloseChannel(
 
 	return processor.NewEventProcessor().
 		WithChainProcessors(
-			srcPathChain.chainProcessor(c.log),
-			dstPathChain.chainProcessor(c.log),
+			srcPathChain.chainProcessor(c.log, nil),
+			dstPathChain.chainProcessor(c.log, nil),
 		).
 		WithPathProcessors(processor.NewPathProcessor(
 			c.log,
diff --git a/relayer/connection.go b/relayer/connection.go
index 38d875f..e188a68 100644
--- a/relayer/connection.go
+++ b/relayer/connection.go
@@ -64,8 +64,8 @@ func (c *Chain) CreateOpenConnections(
 
 	return modified, processor.NewEventProcessor().
 		WithChainProcessors(
-			srcpathChain.chainProcessor(c.log),
-			dstpathChain.chainProcessor(c.log),
+			srcpathChain.chainProcessor(c.log, nil),
+			dstpathChain.chainProcessor(c.log, nil),
 		).
 		WithPathProcessors(pp).
 		WithInitialBlockHistory(initialBlockHistory).
diff --git a/relayer/processor/metrics.go b/relayer/processor/metrics.go
new file mode 100644
index 0000000..0e267cf
--- /dev/null
+++ b/relayer/processor/metrics.go
@@ -0,0 +1,62 @@
+package processor
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+)
+
+type PrometheusMetrics struct {
+	PacketObservedCounter *prometheus.CounterVec
+	PacketRelayedCounter  *prometheus.CounterVec
+	LatestHeightGauge     *prometheus.GaugeVec
+	WalletBalance         *prometheus.GaugeVec
+	FeesSpent             *prometheus.GaugeVec
+}
+
+func (m *PrometheusMetrics) AddPacketsObserved(path, chain, channel, port, eventType string, count int) {
+	m.PacketObservedCounter.WithLabelValues(path, chain, channel, port, eventType).Add(float64(count))
+}
+
+func (m *PrometheusMetrics) IncPacketsRelayed(path, chain, channel, port, eventType string) {
+	m.PacketRelayedCounter.WithLabelValues(path, chain, channel, port, eventType).Inc()
+}
+
+func (m *PrometheusMetrics) SetLatestHeight(chain string, height int64) {
+	m.LatestHeightGauge.WithLabelValues(chain).Set(float64(height))
+}
+
+func (m *PrometheusMetrics) SetWalletBalance(chain, key, denom string, balance float64) {
+	m.WalletBalance.WithLabelValues(chain, key, denom).Set(balance)
+}
+
+func (m *PrometheusMetrics) SetFeesSpent(chain, key, denom string, amount float64) {
+	m.FeesSpent.WithLabelValues(chain, key, denom).Set(amount)
+}
+
+func NewPrometheusMetrics() *PrometheusMetrics {
+	packetLabels := []string{"path", "chain", "channel", "port", "type"}
+	heightLabels := []string{"chain"}
+	walletLabels := []string{"chain", "key", "denom"}
+	return &PrometheusMetrics{
+		PacketObservedCounter: promauto.NewCounterVec(prometheus.CounterOpts{
+			Name: "cosmos_relayer_observed_packets",
+			Help: "The total number of observed packets",
+		}, packetLabels),
+		PacketRelayedCounter: promauto.NewCounterVec(prometheus.CounterOpts{
+			Name: "cosmos_relayer_relayed_packets",
+			Help: "The total number of relayed packets",
+		}, packetLabels),
+		LatestHeightGauge: promauto.NewGaugeVec(prometheus.GaugeOpts{
+			Name: "cosmos_relayer_chain_latest_height",
+			Help: "The current height of the chain",
+		}, heightLabels),
+		WalletBalance: promauto.NewGaugeVec(prometheus.GaugeOpts{
+			Name: "cosmos_relayer_wallet_balance",
+			Help: "The current balance for the relayer's wallet",
+		}, walletLabels),
+		FeesSpent: promauto.NewGaugeVec(prometheus.GaugeOpts{
+			Name: "cosmos_relayer_fees_spent",
+			Help: "The amount of fees spent from the relayer's wallet",
+		}, walletLabels),
+	}
+}
diff --git a/relayer/processor/types.go b/relayer/processor/types.go
index 84d7f7f..ab75503 100644
--- a/relayer/processor/types.go
+++ b/relayer/processor/types.go
@@ -6,8 +6,6 @@ import (
 
 	chantypes "github.com/cosmos/ibc-go/v5/modules/core/04-channel/types"
 	"github.com/cosmos/relayer/v2/relayer/provider"
-	"github.com/prometheus/client_golang/prometheus"
-	"github.com/prometheus/client_golang/prometheus/promauto"
 	"go.uber.org/zap/zapcore"
 )
 
@@ -453,30 +451,3 @@ func ConnectionInfoConnectionKey(info provider.ConnectionInfo) ConnectionKey {
 		CounterpartyConnID:   info.CounterpartyConnID,
 	}
 }
-
-type PrometheusMetrics struct {
-	PacketObservedCounter *prometheus.CounterVec
-	PacketRelayedCounter  *prometheus.CounterVec
-}
-
-func (m *PrometheusMetrics) AddPacketsObserved(path, chain, channel, port, eventType string, count int) {
-	m.PacketObservedCounter.WithLabelValues(path, chain, channel, port, eventType).Add(float64(count))
-}
-
-func (m *PrometheusMetrics) IncPacketsRelayed(path, chain, channel, port, eventType string) {
-	m.PacketRelayedCounter.WithLabelValues(path, chain, channel, port, eventType).Inc()
-}
-
-func NewPrometheusMetrics() *PrometheusMetrics {
-	packetLabels := []string{"path", "chain", "channel", "port", "type"}
-	return &PrometheusMetrics{
-		PacketObservedCounter: promauto.NewCounterVec(prometheus.CounterOpts{
-			Name: "observed_packets",
-			Help: "The total number of observed packets",
-		}, packetLabels),
-		PacketRelayedCounter: promauto.NewCounterVec(prometheus.CounterOpts{
-			Name: "relayed_packets",
-			Help: "The total number of relayed packets",
-		}, packetLabels),
-	}
-}
diff --git a/relayer/strategies.go b/relayer/strategies.go
index 06c9548..f214890 100644
--- a/relayer/strategies.go
+++ b/relayer/strategies.go
@@ -86,11 +86,11 @@ type pathChain struct {
 }
 
 // chainProcessor returns the corresponding ChainProcessor implementation instance for a pathChain.
-func (chain pathChain) chainProcessor(log *zap.Logger) processor.ChainProcessor {
+func (chain pathChain) chainProcessor(log *zap.Logger, metrics *processor.PrometheusMetrics) processor.ChainProcessor {
 	// Handle new ChainProcessor implementations as cases here
 	switch p := chain.provider.(type) {
 	case *cosmos.CosmosProvider:
-		return cosmos.NewCosmosChainProcessor(log, p)
+		return cosmos.NewCosmosChainProcessor(log, p, metrics)
 	default:
 		panic(fmt.Errorf("unsupported chain provider type: %T", chain.provider))
 	}
@@ -115,8 +115,8 @@ func relayerStartEventProcessor(
 	for _, p := range paths {
 		epb = epb.
 			WithChainProcessors(
-				p.src.chainProcessor(log),
-				p.dst.chainProcessor(log),
+				p.src.chainProcessor(log, metrics),
+				p.dst.chainProcessor(log, metrics),
 			).
 			WithPathProcessors(processor.NewPathProcessor(
 				log,


Test patch:
diff --git a/relayer/chains/cosmos/message_handlers_test.go b/relayer/chains/cosmos/message_handlers_test.go
index 1fa617a..d5e5e76 100644
--- a/relayer/chains/cosmos/message_handlers_test.go
+++ b/relayer/chains/cosmos/message_handlers_test.go
@@ -34,7 +34,7 @@ func TestConnectionStateCache(t *testing.T) {
 		// will populate the connectionStateCache with a key that has an empty counterparty connection ID.
 		// The MsgConnectionOpenTry needs to replace this key with a key that has the counterparty connection ID.
 
-		ccp := NewCosmosChainProcessor(zap.NewNop(), &CosmosProvider{})
+		ccp := NewCosmosChainProcessor(zap.NewNop(), &CosmosProvider{}, nil)
 		c := processor.NewIBCMessagesCache()
 
 		// Observe MsgConnectionOpenInit, which does not have counterparty connection ID.
@@ -67,7 +67,7 @@ func TestConnectionStateCache(t *testing.T) {
 		// We need to make sure that the connectionStateCache does not have two keys for the same connection,
 		// i.e. one key with the counterparty connection ID, and one without.
 
-		ccp := NewCosmosChainProcessor(zap.NewNop(), &CosmosProvider{})
+		ccp := NewCosmosChainProcessor(zap.NewNop(), &CosmosProvider{}, nil)
 		c := processor.NewIBCMessagesCache()
 
 		// Initialize connectionStateCache with populated connection ID and counterparty connection ID.
@@ -118,7 +118,7 @@ func TestChannelStateCache(t *testing.T) {
 		// will populate the channelStateCache with a key that has an empty counterparty channel ID.
 		// The MsgChannelOpenTry needs to replace this key with a key that has the counterparty channel ID.
 
-		ccp := NewCosmosChainProcessor(zap.NewNop(), &CosmosProvider{})
+		ccp := NewCosmosChainProcessor(zap.NewNop(), &CosmosProvider{}, nil)
 		c := processor.NewIBCMessagesCache()
 
 		// Observe MsgChannelOpenInit, which does not have counterparty channel ID.
@@ -151,7 +151,7 @@ func TestChannelStateCache(t *testing.T) {
 		// We need to make sure that the channelStateCache does not have two keys for the same channel,
 		// i.e. one key with the counterparty channel ID, and one without.
 
-		ccp := NewCosmosChainProcessor(zap.NewNop(), &CosmosProvider{})
+		ccp := NewCosmosChainProcessor(zap.NewNop(), &CosmosProvider{}, nil)
 		c := processor.NewIBCMessagesCache()
 
 		// Initialize channelStateCache with populated channel ID and counterparty channel ID.