historical_uptime: modularize metrics tallying logic

Signed-off-by: bingyuyap <bingyu.yap.21@gmail.com>

Author: bingyuyap

fly/cmd/historical_uptime/main.go

@@ -153,51 +153,23 @@ func initObservationScraper(db *db.Database, logger *zap.Logger) {
 			case <-ctx.Done():
 				return nil
 			case <-t.C:
-				messages, err := db.QueryMessagesByIndex(false, common.ExpiryDuration)
+				messages, err := db.QueryMessagesByIndex(false, 15 * time.Second)
 				if err != nil {
 					logger.Error("QueryMessagesByIndex error", zap.Error(err))
 					continue
 				}
 
 				// Tally the number of messages for each chain
-				messagesPerChain := make(map[string]int)
-				for _, message := range messages {
-					chainID := message.MessageID.ChainID()
-					messagesPerChain[chainID]++
-				}
+				messagesPerChain := historical_uptime.TallyMessagesPerChain(messages)
 
 				// Initialize the missing observations count for each guardian for each chain
-				guardianMissingObservations := make(map[string]map[string]int)
-				for _, message := range messages {
-					chainID := message.MessageID.ChainID()
-					for _, guardianName := range common.GetGuardianIndexToNameMap() {
-						if guardianMissingObservations[guardianName] == nil {
-							guardianMissingObservations[guardianName] = make(map[string]int)
-						}
-						// Initialize the count for this chain for each guardian with the number of messages for that chain
-						guardianMissingObservations[guardianName][chainID] = messagesPerChain[chainID]
-					}
-				}
+				guardianMissingObservations := historical_uptime.InitializeMissingObservationsCount(messages, messagesPerChain)
 
 				// Decrement the missing observations count for each observed message
-				for _, message := range messages {
-					chainID := message.MessageID.ChainID()
-					for _, observation := range message.Observations {
-						guardianName, ok := common.GetGuardianName(observation.GuardianID)
-						if !ok {
-							logger.Error("Unknown guardian address", zap.String("guardianAddr", observation.GuardianID))
-							continue
-						}
-						guardianMissingObservations[guardianName][chainID]--
-					}
-				}
+				historical_uptime.DecrementMissingObservationsCount(logger, guardianMissingObservations, messages)
 
 				// Update the metrics with the final count of missing observations
-				for guardianName, chains := range guardianMissingObservations {
-					for chainID, missingCount := range chains {
-						guardianMissedObservations.WithLabelValues(guardianName, chainID).Add(float64(missingCount))
-					}
-				}
+				historical_uptime.UpdateMetrics(guardianMissedObservations, guardianMissingObservations)
 			}
 		}
 	})

fly/pkg/historical_uptime/metrics.go

@@ -0,0 +1,55 @@
+package historical_uptime
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/wormhole-foundation/wormhole-monitor/fly/common"
+	"github.com/wormhole-foundation/wormhole-monitor/fly/pkg/db"
+	"go.uber.org/zap"
+)
+
+func TallyMessagesPerChain(messages []*db.Message) map[string]int {
+	messagesPerChain := make(map[string]int)
+	for _, message := range messages {
+		chainID := message.MessageID.ChainID()
+		messagesPerChain[chainID]++
+	}
+	return messagesPerChain
+}
+
+func InitializeMissingObservationsCount(messages []*db.Message, messagesPerChain map[string]int) map[string]map[string]int {
+	guardianMissingObservations := make(map[string]map[string]int)
+	for _, message := range messages {
+		chainID := message.MessageID.ChainID()
+		for _, guardianName := range common.GetGuardianIndexToNameMap() {
+			if guardianMissingObservations[guardianName] == nil {
+				guardianMissingObservations[guardianName] = make(map[string]int)
+			}
+			// Initialize the count for this chain for each guardian with the number of messages for that chain
+			guardianMissingObservations[guardianName][chainID] = messagesPerChain[chainID]
+		}
+	}
+
+	return guardianMissingObservations
+}
+
+func DecrementMissingObservationsCount(logger *zap.Logger, guardianMissingObservations map[string]map[string]int, messages []*db.Message) {
+	for _, message := range messages {
+		chainID := message.MessageID.ChainID()
+		for _, observation := range message.Observations {
+			guardianName, ok := common.GetGuardianName(observation.GuardianID)
+			if !ok {
+				logger.Error("Unknown guardian address", zap.String("guardianAddr", observation.GuardianID))
+				continue
+			}
+			guardianMissingObservations[guardianName][chainID]--
+		}
+	}
+}
+
+func UpdateMetrics(guardianMissedObservations *prometheus.CounterVec, guardianMissingObservations map[string]map[string]int) {
+	for guardianName, chains := range guardianMissingObservations {
+		for chainID, missingCount := range chains {
+			guardianMissedObservations.WithLabelValues(guardianName, chainID).Add(float64(missingCount))
+		}
+	}
+}