feat: new export for stuck packets

cmd/relayer_exporter/relayer_exporter.go

@@ -6,11 +6,12 @@ import (
 	"net/http"
 	"os"
 
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+
 	"github.com/archway-network/relayer_exporter/pkg/collector"
 	"github.com/archway-network/relayer_exporter/pkg/config"
 	log "github.com/archway-network/relayer_exporter/pkg/logger"
-	"github.com/prometheus/client_golang/prometheus"
-	"github.com/prometheus/client_golang/prometheus/promhttp"
 )
 
 var (
@@ -42,7 +43,14 @@ func main() {
 		log.Fatal(err.Error())
 	}
 
-	log.Info(fmt.Sprintf("Getting IBC paths from %s/%s/%s on GitHub", cfg.GitHub.Org, cfg.GitHub.Repo, cfg.GitHub.IBCDir))
+	log.Info(
+		fmt.Sprintf(
+			"Getting IBC paths from %s/%s/%s on GitHub",
+			cfg.GitHub.Org,
+			cfg.GitHub.Repo,
+			cfg.GitHub.IBCDir,
+		),
+	)
 
 	// TODO: Add a feature to refresh paths at configured interval
 	paths, err := cfg.IBCPaths()
@@ -52,7 +60,7 @@ func main() {
 
 	rpcs := cfg.GetRPCsMap()
 
-	clientsCollector := collector.IBCClientsCollector{
+	ibcCollector := collector.IBCCollector{
 		RPCs:  rpcs,
 		Paths: paths,
 	}
@@ -62,7 +70,7 @@ func main() {
 		Accounts: cfg.Accounts,
 	}
 
-	prometheus.MustRegister(clientsCollector)
+	prometheus.MustRegister(ibcCollector)
 	prometheus.MustRegister(balancesCollector)
 
 	http.Handle("/metrics", promhttp.Handler())

config.yaml

@@ -35,7 +35,7 @@ rpc:
     url: https://noble-rpc.polkachu.com:443
   - chainName: nois
     chainId: nois-1
-    url: https://rpc.cosmos.directory/nois:443
+    url: https://nois.rpc.kjnodes.com:443
   - chainName: osmosistestnet
     chainId: osmo-test-5
     url: https://rpc.osmotest5.osmosis.zone:443

pkg/chain/chain.go

@@ -5,6 +5,7 @@ import (
 
 	"github.com/cosmos/relayer/v2/relayer"
 	"github.com/cosmos/relayer/v2/relayer/chains/cosmos"
+	"go.uber.org/zap"
 )
 
 const (
@@ -19,7 +20,7 @@ type Info struct {
 }
 
 func PrepChain(info Info) (*relayer.Chain, error) {
-	chain := relayer.Chain{}
+	logger := zap.NewNop()
 	providerConfig := cosmos.CosmosProviderConfig{
 		ChainID:        info.ChainID,
 		Timeout:        rpcTimeout,
@@ -37,12 +38,12 @@ func PrepChain(info Info) (*relayer.Chain, error) {
 		return nil, err
 	}
 
-	chain.ChainProvider = provider
+	chain := relayer.NewChain(logger, provider, false)
 
 	err = chain.SetPath(&relayer.PathEnd{ClientID: info.ClientID})
 	if err != nil {
 		return nil, err
 	}
 
-	return &chain, nil
+	return chain, nil
 }

pkg/collector/collector.go

@@ -1,22 +1,26 @@
 package collector
 
 import (
+	"fmt"
 	"math/big"
+	"reflect"
 	"sync"
 
-	"github.com/archway-network/relayer_exporter/pkg/config"
-	"github.com/archway-network/relayer_exporter/pkg/ibc"
-	log "github.com/archway-network/relayer_exporter/pkg/logger"
 	"github.com/cosmos/relayer/v2/relayer"
 	"github.com/prometheus/client_golang/prometheus"
 	"go.uber.org/zap"
+
+	"github.com/archway-network/relayer_exporter/pkg/config"
+	"github.com/archway-network/relayer_exporter/pkg/ibc"
+	log "github.com/archway-network/relayer_exporter/pkg/logger"
 )
 
 const (
-	successStatus           = "success"
-	errorStatus             = "error"
-	clientExpiryMetricName  = "cosmos_ibc_client_expiry"
-	walletBalanceMetricName = "cosmos_wallet_balance"
+	successStatus                 = "success"
+	errorStatus                   = "error"
+	clientExpiryMetricName        = "cosmos_ibc_client_expiry"
+	walletBalanceMetricName       = "cosmos_wallet_balance"
+	channelStuckPacketsMetricName = "cosmos_ibc_stuck_packets"
 )
 
 var (
@@ -25,14 +29,26 @@ var (
 		"Returns light client expiry in unixtime.",
 		[]string{"host_chain_id", "client_id", "target_chain_id", "status"}, nil,
 	)
+	channelStuckPackets = prometheus.NewDesc(
+		channelStuckPacketsMetricName,
+		"Returns stuck packets for a channel.",
+		[]string{
+			"src_channel_id",
+			"dst_channel_id",
+			"src_chain_id",
+			"dst_chain_id",
+			"status",
+		},
+		nil,
+	)
 	walletBalance = prometheus.NewDesc(
 		walletBalanceMetricName,
 		"Returns wallet balance for an address on a chain.",
 		[]string{"account", "chain_id", "denom", "status"}, nil,
 	)
 )
 
-type IBCClientsCollector struct {
+type IBCCollector struct {
 	RPCs  *map[string]config.RPC
 	Paths []*relayer.IBCdata
 }
@@ -42,12 +58,19 @@ type WalletBalanceCollector struct {
 	Accounts []config.Account
 }
 
-func (cc IBCClientsCollector) Describe(ch chan<- *prometheus.Desc) {
+func (cc IBCCollector) Describe(ch chan<- *prometheus.Desc) {
 	ch <- clientExpiry
+	ch <- channelStuckPackets
 }
 
-func (cc IBCClientsCollector) Collect(ch chan<- prometheus.Metric) {
-	log.Debug("Start collecting", zap.String("metric", clientExpiryMetricName))
+func (cc IBCCollector) Collect(ch chan<- prometheus.Metric) {
+	log.Debug(
+		"Start collecting",
+		zap.String(
+			"metrics",
+			fmt.Sprintf("%s, %s", clientExpiryMetricName, channelStuckPacketsMetricName),
+		),
+	)
 
 	var wg sync.WaitGroup
 
@@ -57,6 +80,7 @@ func (cc IBCClientsCollector) Collect(ch chan<- prometheus.Metric) {
 		go func(path *relayer.IBCdata) {
 			defer wg.Done()
 
+			// Client info
 			ci, err := ibc.GetClientsInfo(path, cc.RPCs)
 			status := successStatus
 
@@ -79,6 +103,46 @@ func (cc IBCClientsCollector) Collect(ch chan<- prometheus.Metric) {
 				float64(ci.ChainBClientExpiration.Unix()),
 				[]string{(*cc.RPCs)[path.Chain2.ChainName].ChainID, path.Chain2.ClientID, (*cc.RPCs)[path.Chain1.ChainName].ChainID, status}...,
 			)
+
+			// Stuck packets
+			status = successStatus
+
+			stuckPackets, err := ibc.GetChannelsInfo(path, cc.RPCs)
+			if err != nil {
+				status = errorStatus
+
+				log.Error(err.Error())
+			}
+
+			if !reflect.DeepEqual(stuckPackets, ibc.ChannelsInfo{}) {
+				for _, sp := range stuckPackets.Channels {
+					ch <- prometheus.MustNewConstMetric(
+						channelStuckPackets,
+						prometheus.GaugeValue,
+						float64(sp.StuckPackets.Source),
+						[]string{
+							sp.Source,
+							sp.Destination,
+							(*cc.RPCs)[path.Chain1.ChainName].ChainID,
+							(*cc.RPCs)[path.Chain2.ChainName].ChainID,
+							status,
+						}...,
+					)
+
+					ch <- prometheus.MustNewConstMetric(
+						channelStuckPackets,
+						prometheus.GaugeValue,
+						float64(sp.StuckPackets.Destination),
+						[]string{
+							sp.Destination,
+							sp.Source,
+							(*cc.RPCs)[path.Chain2.ChainName].ChainID,
+							(*cc.RPCs)[path.Chain1.ChainName].ChainID,
+							status,
+						}...,
+					)
+				}
+			}
 		}(p)
 	}
 

pkg/ibc/ibc.go

@@ -5,11 +5,15 @@ import (
 	"fmt"
 	"time"
 
+	chantypes "github.com/cosmos/ibc-go/v7/modules/core/04-channel/types"
+	"github.com/cosmos/relayer/v2/relayer"
+
 	"github.com/archway-network/relayer_exporter/pkg/chain"
 	"github.com/archway-network/relayer_exporter/pkg/config"
-	"github.com/cosmos/relayer/v2/relayer"
 )
 
+const stateOpen = 3
+
 type ClientsInfo struct {
 	ChainA                 *relayer.Chain
 	ChainAClientInfo       relayer.ClientStateInfo
@@ -19,6 +23,22 @@ type ClientsInfo struct {
 	ChainBClientExpiration time.Time
 }
 
+type ChannelsInfo struct {
+	Channels []Channel
+}
+
+type Channel struct {
+	Source          string
+	Destination     string
+	SourcePort      string
+	DestinationPort string
+	Ordering        string
+	StuckPackets    struct {
+		Source      int
+		Destination int
+	}
+}
+
 func GetClientsInfo(ibc *relayer.IBCdata, rpcs *map[string]config.RPC) (ClientsInfo, error) {
 	clientsInfo := ClientsInfo{}
 
@@ -50,15 +70,106 @@ func GetClientsInfo(ibc *relayer.IBCdata, rpcs *map[string]config.RPC) (ClientsI
 
 	ctx := context.Background()
 
-	clientsInfo.ChainAClientExpiration, clientsInfo.ChainAClientInfo, err = relayer.QueryClientExpiration(ctx, chainA, chainB)
+	clientsInfo.ChainAClientExpiration, clientsInfo.ChainAClientInfo, err = relayer.QueryClientExpiration(
+		ctx,
+		chainA,
+		chainB,
+	)
 	if err != nil {
 		return ClientsInfo{}, fmt.Errorf("Error: %w path %v <-> %v", err, cdA, cdB)
 	}
 
-	clientsInfo.ChainBClientExpiration, clientsInfo.ChainBClientInfo, err = relayer.QueryClientExpiration(ctx, chainB, chainA)
+	clientsInfo.ChainBClientExpiration, clientsInfo.ChainBClientInfo, err = relayer.QueryClientExpiration(
+		ctx,
+		chainB,
+		chainA,
+	)
 	if err != nil {
 		return ClientsInfo{}, fmt.Errorf("Error: %w path %v <-> %v", err, cdB, cdA)
 	}
 
 	return clientsInfo, nil
 }
+
+func GetChannelsInfo(ibc *relayer.IBCdata, rpcs *map[string]config.RPC) (ChannelsInfo, error) {
+	ctx := context.Background()
+	channelInfo := ChannelsInfo{}
+
+	// Init channel data
+	for _, c := range ibc.Channels {
+		var channel Channel
+		channel.Source = c.Chain1.ChannelID
+		channel.Destination = c.Chain2.ChannelID
+		channel.SourcePort = c.Chain1.PortID
+		channel.DestinationPort = c.Chain2.PortID
+		channel.Ordering = c.Ordering
+		channelInfo.Channels = append(channelInfo.Channels, channel)
+	}
+
+	if (*rpcs)[ibc.Chain1.ChainName].ChainID == "" || (*rpcs)[ibc.Chain2.ChainName].ChainID == "" {
+		return channelInfo, fmt.Errorf(
+			"Error: RPC data is missing, cannot retrieve channel data: %v",
+			ibc.Channels,
+		)
+	}
+
+	cdA := chain.Info{
+		ChainID:  (*rpcs)[ibc.Chain1.ChainName].ChainID,
+		RPCAddr:  (*rpcs)[ibc.Chain1.ChainName].URL,
+		ClientID: ibc.Chain1.ClientID,
+	}
+
+	chainA, err := chain.PrepChain(cdA)
+	if err != nil {
+		return ChannelsInfo{}, fmt.Errorf("Error: %w for %v", err, cdA)
+	}
+
+	cdB := chain.Info{
+		ChainID:  (*rpcs)[ibc.Chain2.ChainName].ChainID,
+		RPCAddr:  (*rpcs)[ibc.Chain2.ChainName].URL,
+		ClientID: ibc.Chain2.ClientID,
+	}
+
+	chainB, err := chain.PrepChain(cdB)
+	if err != nil {
+		return ChannelsInfo{}, fmt.Errorf("Error: %w for %v", err, cdB)
+	}
+
+	// test that RPC endpoints are working
+	if _, _, err := relayer.QueryLatestHeights(
+		ctx, chainA, chainB,
+	); err != nil {
+		return channelInfo, fmt.Errorf("Error: %w for %v", err, cdA)
+	}
+
+	for i, c := range channelInfo.Channels {
+		var order chantypes.Order
+
+		switch c.Ordering {
+		case "none":
+			order = chantypes.NONE
+		case "unordered":
+			order = chantypes.UNORDERED
+		case "ordered":
+			order = chantypes.ORDERED
+		}
+
+		ch := chantypes.IdentifiedChannel{
+			State:    stateOpen,
+			Ordering: order,
+			Counterparty: chantypes.Counterparty{
+				PortId:    c.DestinationPort,
+				ChannelId: c.Destination,
+			},
+			PortId:    c.SourcePort,
+			ChannelId: c.Source,
+		}
+
+		unrelayedSequences := relayer.UnrelayedSequences(ctx, chainA, chainB, &ch)
+
+		channelInfo.Channels[i].StuckPackets.Source += len(unrelayedSequences.Src)
+		channelInfo.Channels[i].StuckPackets.Destination += len(unrelayedSequences.Dst)
+	}
+
+	return channelInfo, nil
+}