feat(monitor): track validator power and status (#1821)

Adds basic validator monitoring. Specifically adds a
`monitor_validator_power` metric that maps operator eth address to
consensus eth address and consensus cometBFT address which is required
to map built-in cometBFT validator metrics to our xchain attest
validator metrics.

issue: #1811

monitor/app/app.go

@@ -21,6 +21,7 @@ import (
 	"github.com/omni-network/omni/monitor/avs"
 	"github.com/omni-network/omni/monitor/loadgen"
 	"github.com/omni-network/omni/monitor/routerecon"
+	"github.com/omni-network/omni/monitor/validator"
 	"github.com/omni-network/omni/monitor/xfeemngr"
 	"github.com/omni-network/omni/monitor/xmonitor"
 	"github.com/omni-network/omni/monitor/xmonitor/indexer"
@@ -100,6 +101,7 @@ func Run(ctx context.Context, cfg Config) error {
 	startMonitoringSyncDiff(ctx, network, ethClients)
 	go runHistoricalBaselineForever(ctx, network, cprov)
 	go routerecon.ReconForever(ctx, network, xprov, ethClients)
+	go validator.MonitorForever(ctx, cprov)
 
 	select {
 	case <-ctx.Done():

monitor/validator/metrics.go

@@ -0,0 +1,30 @@
+package validator
+
+import (
+	"github.com/omni-network/omni/lib/promutil"
+
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+var (
+	powerGauge = promutil.NewResetGaugeVec(prometheus.GaugeOpts{
+		Namespace: "monitor",
+		Subsystem: "validator",
+		Name:      "power",
+		Help:      "Current power by validator",
+	}, []string{"validator", "validator_address", "operator"}) // Main metric with additional label identifiers
+
+	jailedGauge = promutil.NewResetGaugeVec(prometheus.GaugeOpts{
+		Namespace: "monitor",
+		Subsystem: "validator",
+		Name:      "jailed",
+		Help:      "Constant gauge set to 1 if the validator is jailed otherwise 0",
+	}, []string{"validator"})
+
+	bondedGauge = promutil.NewResetGaugeVec(prometheus.GaugeOpts{
+		Namespace: "monitor",
+		Subsystem: "validator",
+		Name:      "bonded",
+		Help:      "Constant gauge set to 1 if the validator is bonded otherwise 0",
+	}, []string{"validator"})
+)

monitor/validator/validator.go

@@ -0,0 +1,85 @@
+package validator
+
+import (
+	"context"
+	"time"
+
+	"github.com/omni-network/omni/lib/cchain"
+	"github.com/omni-network/omni/lib/errors"
+	"github.com/omni-network/omni/lib/log"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/crypto"
+
+	cosmosk1 "github.com/cosmos/cosmos-sdk/crypto/keys/secp256k1"
+	sdk "github.com/cosmos/cosmos-sdk/types"
+	"github.com/cosmos/gogoproto/proto"
+)
+
+func MonitorForever(ctx context.Context, cprov cchain.Provider) {
+	ticker := time.NewTicker(time.Second * 30)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			err := monitorOnce(ctx, cprov)
+			if err != nil {
+				log.Warn(ctx, "Monitoring validator failed (will retry)", err)
+			}
+		}
+	}
+}
+
+func monitorOnce(ctx context.Context, cprov cchain.Provider) error {
+	vals, err := cprov.Validators(ctx)
+	if err != nil {
+		return errors.Wrap(err, "query validators")
+	}
+
+	// Reset existing time-series since validator may be removed
+	powerGauge.Reset()
+	jailedGauge.Reset()
+	bondedGauge.Reset()
+
+	for _, val := range vals {
+		pk := new(cosmosk1.PubKey)
+		err := proto.Unmarshal(val.ConsensusPubkey.Value, pk)
+		if err != nil {
+			return errors.Wrap(err, "unmarshal consensus pubkey")
+		}
+
+		pubkey, err := crypto.DecompressPubkey(pk.Bytes())
+		if err != nil {
+			return errors.Wrap(err, "decompress pubkey")
+		}
+
+		opAddr, err := sdk.ValAddressFromBech32(val.OperatorAddress)
+		if err != nil {
+			return errors.Wrap(err, "parse operator address")
+		}
+
+		opAddrEth := common.BytesToAddress(opAddr)
+		consAddrEth := crypto.PubkeyToAddress(*pubkey)
+		consAddrCmt := pk.Address()
+		power := val.ConsensusPower(sdk.DefaultPowerReduction)
+		jailed := val.IsJailed()
+		bonded := val.IsBonded()
+
+		powerGauge.WithLabelValues(consAddrEth.String(), consAddrCmt.String(), opAddrEth.String()).Set(float64(power))
+		jailedGauge.WithLabelValues(consAddrEth.String()).Set(boolToFloat(jailed))
+		bondedGauge.WithLabelValues(consAddrEth.String()).Set(boolToFloat(bonded))
+	}
+
+	return nil
+}
+
+func boolToFloat(b bool) float64 {
+	if !b {
+		return 0
+	}
+
+	return 1
+}

monitor/xmonitor/indexer/metrics.go

@@ -56,11 +56,15 @@ type sample struct {
 }
 
 func instrumentSample(s sample) {
+	// Initialize success/revert counters so both exist
+	revertCounter.WithLabelValues(s.Stream, s.XDApp).Add(0)
+	successCounter.WithLabelValues(s.Stream, s.XDApp).Add(0)
 	if s.Success {
 		successCounter.WithLabelValues(s.Stream, s.XDApp).Inc()
 	} else {
 		revertCounter.WithLabelValues(s.Stream, s.XDApp).Inc()
 	}
+
 	if s.FuzzyOverride {
 		fuzzyOverrideCounter.WithLabelValues(s.Stream, s.XDApp).Inc()
 	}