add it all and start over

cmd/relayer_exporter/relayer_exporter.go

@@ -68,8 +68,9 @@ func main() {
 	}
 
 	ibcCollector := collector.IBCCollector{
-		RPCs:  rpcs,
-		Paths: paths,
+		RPCs:          rpcs,
+		Paths:         paths,
+		AckProcessors: map[string]collector.AckProcessor{},
 	}
 
 	balancesCollector := collector.WalletBalanceCollector{

config.yaml

@@ -2,7 +2,7 @@ github:
   org: archway-network
   repo: networks
   dir: _IBC
-  testnetsDir: testnets/_IBC
+  testnetsDir: devnets/_IBC
 rpc:
   # mainnets
   - chainName: archway
@@ -22,7 +22,7 @@ rpc:
     url: https://cosmoshub-rpc.stakely.io:443
   - chainName: decentr
     chainId: mainnet-3
-    url: https://poseidon.mainnet.decentr.xyz:443
+    url: https://rpc.decentr.chaintools.tech:443
   - chainName: jackal
     chainId: jackal-1
     url: https://jackal-rpc.polkachu.com:443
@@ -43,7 +43,7 @@ rpc:
     url: https://omniflix.kingnodes.com:443
   - chainName: osmosis
     chainId: osmosis-1
-    url: https://osmosis-rpc.stakely.io:443
+    url: https://osmosis-rpc.onivalidator.com:443
   - chainName: quicksilver
     chainId: quicksilver-2
     url: https://rpc.quicksilver.zone:443
@@ -64,28 +64,35 @@ rpc:
     url: https://rpc.comdex.one:443
   - chainName: neutron
     chainId: neutron-1
-    url: https://rpc-kralum.neutron-1.neutron.org:443
+    url: https://rpc.novel.remedy.tm.p2p.org:443
+  - chainName: qwoyn
+    chainId: qwoyn-1
+    url: https://qwoyn-rpc.staketab.org:443
   - chainName: stargaze
     chainId: stargaze-1
     url: https://rpc.stargaze-apis.com:443
+  - chainName: andromeda
+    chainId: andromeda-1
+    url: https://andromeda.rpc.kjnodes.com:443
 
   # testnets
-  - chainName: archwaytestnet
-    chainId: constantine-3
-    url: https://rpc.constantine.archway.tech:443
-  - chainName: axelartestnet
-    chainId: axelar-testnet-lisbon-3
-    url: https://axelar-testnet-rpc.qubelabs.io:443
+  - chainName: archwaydevnet
+    chainId: titus-3
+    url: https://rpc.titus.archway.io:443
   - chainName: osmosistestnet
     chainId: osmo-test-5
-    url: https://rpc.osmotest5.osmosis.zone:443
+    url: https://rpc-1.testnet.osmosis.nodes.guru:443
 
 accounts:
-  # Foundation
+  # foundation-feegrant-astrovault
   - address: archway1gpyqzc0aerc85cpk2cm8ec6zkc95x5yqrakskv
     chainName: archway
     denom: aarch
+  # foundation-feegrant-ojo
+  - address: archway1c2cu99uzjauaj5hg45edhzgsk5saz43y6d3zxp
+    chainName: archway
+    denom: aarch
   # PhiLabs
   - address: archway1ktka5q3cnsy3ar7qwj2huzz6qj9q4ys7h74l9y
     chainName: archway
-    denom: aarch
+    denom: aarch

pkg/collector/ibc_collector.go

@@ -53,13 +53,12 @@ var (
 	)
 	channelNewAckSinceStuck = prometheus.NewDesc(
 		channelNewAckSinceStuckMetricName,
-		"Returns 1 if new IBC ack was observed since last stuck packet detection, else returns 0.",
+		"Returns block height of new observed IBC Ack since last stuck packet detection, else returns 0.",
 		[]string{
 			"src_channel_id",
 			"dst_channel_id",
 			"src_chain_id",
 			"dst_chain_id",
-			"src_chain_height",
 			"src_chain_name",
 			"dst_chain_name",
 			"status",
@@ -78,16 +77,16 @@ type AckProcessor struct {
 	ChainID      string
 	ChannelID    string
 	StartHeight  int64
-	NewAckHeight chan<- int64
+	NewAckHeight chan uint64
 }
 
-func (cc IBCCollector) Describe(ch chan<- *prometheus.Desc) {
-	ch <- clientExpiry
-	ch <- channelStuckPackets
-	ch <- channelNewAckSinceStuck
+func (cc IBCCollector) Describe(metricDesc chan<- *prometheus.Desc) {
+	metricDesc <- clientExpiry
+	metricDesc <- channelStuckPackets
+	metricDesc <- channelNewAckSinceStuck
 }
 
-func (cc IBCCollector) Collect(ch chan<- prometheus.Metric) {
+func (cc IBCCollector) Collect(metric chan<- prometheus.Metric) {
 	log.Debug(
 		"Start collecting",
 		zap.String(
@@ -116,7 +115,7 @@ func (cc IBCCollector) Collect(ch chan<- prometheus.Metric) {
 				log.Error(err.Error())
 			}
 
-			ch <- prometheus.MustNewConstMetric(
+			metric <- prometheus.MustNewConstMetric(
 				clientExpiry,
 				prometheus.GaugeValue,
 				float64(ci.ChainAClientExpiration.Unix()),
@@ -129,7 +128,7 @@ func (cc IBCCollector) Collect(ch chan<- prometheus.Metric) {
 				}...,
 			)
 
-			ch <- prometheus.MustNewConstMetric(
+			metric <- prometheus.MustNewConstMetric(
 				clientExpiry,
 				prometheus.GaugeValue,
 				float64(ci.ChainBClientExpiration.Unix()),
@@ -154,7 +153,7 @@ func (cc IBCCollector) Collect(ch chan<- prometheus.Metric) {
 
 			if !reflect.DeepEqual(channelsInfo, ibc.ChannelsInfo{}) {
 				for _, chInfo := range channelsInfo.Channels {
-					ch <- prometheus.MustNewConstMetric(
+					metric <- prometheus.MustNewConstMetric(
 						channelStuckPackets,
 						prometheus.GaugeValue,
 						float64(len(chInfo.StuckPackets.Src)),
@@ -172,7 +171,7 @@ func (cc IBCCollector) Collect(ch chan<- prometheus.Metric) {
 						}...,
 					)
 
-					ch <- prometheus.MustNewConstMetric(
+					metric <- prometheus.MustNewConstMetric(
 						channelStuckPackets,
 						prometheus.GaugeValue,
 						float64(len(chInfo.StuckPackets.Dst)),
@@ -191,7 +190,7 @@ func (cc IBCCollector) Collect(ch chan<- prometheus.Metric) {
 					)
 
 					// cosmos_ibc_new_ack_since_stuck metric collection
-					err := cc.SetNewAckSinceStuckMetric(ctx, ch, chInfo, path)
+					err := cc.SetNewAckSinceStuckMetric(ctx, chInfo, path, metric)
 
 					if err != nil {
 						log.Error(err.Error())
@@ -208,53 +207,154 @@ func (cc IBCCollector) Collect(ch chan<- prometheus.Metric) {
 
 func (cc IBCCollector) SetNewAckSinceStuckMetric(
 	ctx context.Context,
-	ch chan<- prometheus.Metric,
 	chInfo ibc.Channel,
-	path *config.IBCData) error {
-	// start packet processor if stuck packets exist on src or dst
-	// - if a packet process is already active on a chain for a channel, do not start another one.
-	// - channel collector (cc) must keep track of active packet processors
-	// - add new packet processor to cc
-	// packet processor
-	// - start at height of stuck packet
-	// - check for latest height
-	// - if latest height is greater than last_queried_block_height, increment last_queried_block_height and query block_results api for that height
-	// - if block_results has new ack, publish metric, stop packet processor and remove from cc
+	path *config.IBCData,
+	metric chan<- prometheus.Metric) error {
+
+	// Note: cosmos sdk block height is uint64 but prometheus metric type expects float64
+	var newAckHeight uint64
 
+	// Hash key to search for AckProcessor in cc (<chainName_channelID>)
+	srcHashKey := fmt.Sprintf("%s_%s", path.Chain1.ChainName, chInfo.Source)
+	dstHashKey := fmt.Sprintf("%s_%s", path.Chain2.ChainName, chInfo.Destination)
+
+	// Publish source chain metric
 	// Start Src chain AckProcessors if stuck packets exist and no processor is running
-	if _, ok := cc.AckProcessors[path.Chain1.ChainName]; len(chInfo.StuckPackets.Src) > 0 && !ok {
-		log.Info("Starting packet processor", zap.String("ChainName", path.Chain1.ChainName), zap.String("ChannelID", chInfo.Source))
-		cc.AckProcessors[path.Chain1.ChainName] = AckProcessor{
+	if _, ok := cc.AckProcessors[srcHashKey]; len(chInfo.StuckPackets.Src) > 0 && !ok {
+		log.Info("Creating packet processor", zap.String("ChainName", path.Chain1.ChainName), zap.String("ChannelID", chInfo.Source))
+		cc.AckProcessors[srcHashKey] = AckProcessor{
 			ChainID:      (*cc.RPCs)[path.Chain1.ChainName].ChainID,
 			ChannelID:    chInfo.Source,
 			StartHeight:  chInfo.StuckPackets.SrcHeight,
-			NewAckHeight: make(chan<- int64),
+			NewAckHeight: make(chan uint64),
+		}
+		// set default for newAckHeight
+		newAckHeight = 0
+
+		//TODO:start scan for new Acks from start height
+
+		// read from NewAckHeight from AckProcessor
+		select {
+		case newAckHeight = <-cc.AckProcessors[srcHashKey].NewAckHeight:
+			log.Debug("new ack found", zap.String("ChainName", path.Chain1.ChainName), zap.String("ChannelID", chInfo.Source),
+				zap.Uint64("newAckHeight", newAckHeight))
+		default:
+			log.Debug("no new acks found", zap.String("ChainName", path.Chain1.ChainName), zap.String("ChannelID", chInfo.Source))
+			newAckHeight = 0
 		}
 
-		go ibc.ScanForIBCAcksEvents(ctx, chInfo.StuckPackets.SrcHeight, (*cc.RPCs)[path.Chain1.ChainName])
-
-		// start packet processor
-		return nil
+	} else if _, ok := cc.AckProcessors[srcHashKey]; ok {
+		// AckProcessor already running, read from NewAckHeight
+		// set default for newAckHeight
+		newAckHeight = 0
+
+		//TODO:start scan for new Acks from start height
+
+		// read from NewAckHeight from AckProcessor
+		select {
+		case newAckHeight = <-cc.AckProcessors[srcHashKey].NewAckHeight:
+			log.Debug("new ack found", zap.String("ChainName", path.Chain1.ChainName), zap.String("ChannelID", chInfo.Source),
+				zap.Uint64("newAckHeight", newAckHeight))
+		default:
+			log.Debug("no new acks found", zap.String("ChainName", path.Chain1.ChainName), zap.String("ChannelID", chInfo.Source))
+			newAckHeight = 0
+		}
+	} else {
+		// no stuck packets, no processor running, no new acks
+		newAckHeight = 0
 	}
 
-	// Start Dst chain AckProcessors if stuck packets exist and no processor is running
-	if _, ok := cc.AckProcessors[path.Chain2.ChainName]; len(chInfo.StuckPackets.Dst) > 0 && !ok {
-		log.Info("Starting packet processor", zap.String("ChainName", path.Chain2.ChainName), zap.String("ChannelID", chInfo.Source))
-		cc.AckProcessors[path.Chain1.ChainName] = AckProcessor{
-			ChainID:     (*cc.RPCs)[path.Chain2.ChainName].ChainID,
-			ChannelID:   chInfo.Source,
-			StartHeight: chInfo.StuckPackets.SrcHeight,
+	// Publish source chain metric
+	// channelNewAckSinceStuck = prometheus.NewDesc(
+	// 	channelNewAckSinceStuckMetricName,
+	// 	"Returns 1 if new IBC ack was observed since last stuck packet detection, else returns 0.",
+	// 	[]string{
+	// 		"src_channel_id",
+	// 		"dst_channel_id",
+	// 		"src_chain_id",
+	// 		"dst_chain_id",
+	// 		"src_chain_height",
+	// 		"src_chain_name",
+	// 		"dst_chain_name",
+	// 		"status",
+	// 	},
+	// 	nil,
+	// )
+	metric <- prometheus.MustNewConstMetric(
+		channelNewAckSinceStuck,
+		prometheus.GaugeValue,
+		float64(newAckHeight),
+		[]string{
+			chInfo.Source,
+			chInfo.Destination,
+			(*cc.RPCs)[path.Chain1.ChainName].ChainID,
+			(*cc.RPCs)[path.Chain2.ChainName].ChainID,
+			path.Chain1.ChainName,
+			path.Chain2.ChainName,
+			"success",
+		}...,
+	)
+
+	// Publish destination chain metric
+	if _, ok := cc.AckProcessors[dstHashKey]; len(chInfo.StuckPackets.Dst) > 0 && !ok {
+		// Start Dst chain AckProcessors if stuck packets exist and no processor is running
+		log.Info("Creating packet processor", zap.String("ChainName", path.Chain2.ChainName), zap.String("ChannelID", chInfo.Destination))
+		cc.AckProcessors[dstHashKey] = AckProcessor{
+			ChainID:      (*cc.RPCs)[path.Chain2.ChainName].ChainID,
+			ChannelID:    chInfo.Source,
+			StartHeight:  chInfo.StuckPackets.DstHeight,
+			NewAckHeight: make(chan uint64),
+		}
+		//start scan for new Acks from start height
+
+		// read from NewAckHeight from AckProcessor
+		log.Info("creating packet processor", zap.String("ChainName", path.Chain2.ChainName), zap.String("ChannelID", chInfo.Destination))
+
+		select {
+		case newAckHeight = <-cc.AckProcessors[dstHashKey].NewAckHeight:
+			log.Debug("new ack found", zap.String("ChainName", path.Chain2.ChainName), zap.String("ChannelID", chInfo.Destination),
+				zap.Uint64("newAckHeight", newAckHeight))
+		default:
+			log.Debug("no new acks found", zap.String("ChainName", path.Chain2.ChainName), zap.String("ChannelID", chInfo.Destination))
+			newAckHeight = 0
 		}
-		return nil
-	}
 
-	if _, ok := cc.AckProcessors[path.Chain2.ChainName]; ok {
-		log.Info("Packet processor already running", zap.String("ChainName", path.Chain1.ChainName), zap.String("ChannelID", chInfo.Source))
+		newAckHeight = <-cc.AckProcessors[path.Chain2.ChainName].NewAckHeight
+	} else if _, ok := cc.AckProcessors[dstHashKey]; ok {
+		// AckProcessor already running, read from NewAckHeight
+		// set default for newAckHeight
+		newAckHeight = 0
+
+		//TODO:start scan for new Acks from start height
+
+		// read from NewAckHeight from AckProcessor
+		select {
+		case newAckHeight = <-cc.AckProcessors[dstHashKey].NewAckHeight:
+			log.Debug("new ack found", zap.String("ChainName", path.Chain2.ChainName), zap.String("ChannelID", chInfo.Source),
+				zap.Uint64("newAckHeight", newAckHeight))
+		default:
+			log.Debug("no new acks found", zap.String("ChainName", path.Chain2.ChainName), zap.String("ChannelID", chInfo.Source))
+			newAckHeight = 0
+		}
+	} else {
+		// no stuck packets, no processor running, no new acks
+		newAckHeight = 0
 	}
 
-	if _, ok := cc.AckProcessors[path.Chain2.ChainName]; ok {
-		log.Info("Packet processor already running", zap.String("ChainName", path.Chain2.ChainName), zap.String("ChannelID", chInfo.Source))
-	}
+	metric <- prometheus.MustNewConstMetric(
+		channelNewAckSinceStuck,
+		prometheus.GaugeValue,
+		float64(newAckHeight),
+		[]string{
+			chInfo.Source,
+			chInfo.Destination,
+			(*cc.RPCs)[path.Chain1.ChainName].ChainID,
+			(*cc.RPCs)[path.Chain2.ChainName].ChainID,
+			path.Chain1.ChainName,
+			path.Chain2.ChainName,
+			"fail",
+		}...,
+	)
 
 	return nil
 }