feat: refactor consensus querying (#35)

* chore: add config example

* chore: removed unnecessary config option

* chore: added disclaimer on how to get consensus-address

config.example.toml

@@ -43,12 +43,13 @@ denom-coefficient = 1000000
 # Bech32 prefix for a wallet address (example: "cosmos" for a Cosmos wallet). If omitted,
 # the self-delegation metric will not be present.
 bech-wallet-prefix = "bitsong"
-# Bech32 prefix for a consensus address (example: "cosmosvalcons" for a Cosmos network).
-# If omitted, the missed blocks counter will not be present.
-bech-consensus-prefix = "bitsongvalcons"
 # List of validators to monitor.
+# Address is required, consensus-address is optional but will result in omitting
+# signing-infos metrics (like missed blocks counter).
+# You can get your consensus-address by running "<appd> tendermint show-address" on your validator node,
+# if you are not using KMS solutions.
 validators = [
-    "bitsongvaloper14rvn7anf22e00vj5x3al4w50ns78s7n42rc0ge"
+    { address = "bitsongvaloper14rvn7anf22e00vj5x3al4w50ns78s7n42rc0ge", consensus-address = "bitsongvalcons16ktzzs4ra4kcw668demahue52zh2xjllwdd6n3" }
 ]
 # List of queries to enable/disable.
 # If the list is not provided, or the value for query is not specified,
@@ -87,7 +88,6 @@ coingecko-currency = "e-money"
 base-denom = "ungm"
 denom = "ngm"
 bech-wallet-prefix = "emoney"
-bech-consensus-prefix = "emoneyvalcons"
 validators = [
-    "emoneyvaloper1jk4n79c5gv36972ptnly3rvx5nvn3hl3hgly9g"
+    { address = "emoneyvaloper1jk4n79c5gv36972ptnly3rvx5nvn3hl3hgly9g" }
 ]

pkg/app.go

@@ -50,6 +50,7 @@ func NewApp(configPath string) *App {
 		queriersPkg.NewSlashingParamsQuerier(log, appConfig),
 		queriersPkg.NewValidatorQuerier(log, appConfig),
 		queriersPkg.NewDenomCoefficientsQuerier(log, appConfig),
+		queriersPkg.NewSigningInfoQuerier(log, appConfig),
 	}
 
 	return &App{

pkg/config/config.go

@@ -8,19 +8,31 @@ import (
 	"github.com/mcuadros/go-defaults"
 )
 
+type Validator struct {
+	Address          string `toml:"address"`
+	ConsensusAddress string `toml:"consensus-address"`
+}
+
+func (v *Validator) Validate() error {
+	if v.Address == "" {
+		return fmt.Errorf("validator address is expected!")
+	}
+
+	return nil
+}
+
 type Chain struct {
-	Name                string          `toml:"name"`
-	LCDEndpoint         string          `toml:"lcd-endpoint"`
-	CoingeckoCurrency   string          `toml:"coingecko-currency"`
-	DexScreenerChainID  string          `toml:"dex-screener-chain-id"`
-	DexScreenerPair     string          `toml:"dex-screener-pair"`
-	BaseDenom           string          `toml:"base-denom"`
-	Denom               string          `toml:"denom"`
-	DenomCoefficient    int64           `toml:"denom-coefficient" default:"1000000"`
-	BechWalletPrefix    string          `toml:"bech-wallet-prefix"`
-	BechConsensusPrefix string          `toml:"bech-consensus-prefix"`
-	Validators          []string        `toml:"validators"`
-	Queries             map[string]bool `toml:"queries"`
+	Name               string          `toml:"name"`
+	LCDEndpoint        string          `toml:"lcd-endpoint"`
+	CoingeckoCurrency  string          `toml:"coingecko-currency"`
+	DexScreenerChainID string          `toml:"dex-screener-chain-id"`
+	DexScreenerPair    string          `toml:"dex-screener-pair"`
+	BaseDenom          string          `toml:"base-denom"`
+	Denom              string          `toml:"denom"`
+	DenomCoefficient   int64           `toml:"denom-coefficient" default:"1000000"`
+	BechWalletPrefix   string          `toml:"bech-wallet-prefix"`
+	Validators         []Validator     `toml:"validators"`
+	Queries            map[string]bool `toml:"queries"`
 }
 
 func (c *Chain) Validate() error {
@@ -36,6 +48,12 @@ func (c *Chain) Validate() error {
 		return fmt.Errorf("no validators provided")
 	}
 
+	for index, validator := range c.Validators {
+		if err := validator.Validate(); err != nil {
+			return fmt.Errorf("error in validator #%d: %s", index, err)
+		}
+	}
+
 	return nil
 }
 

pkg/queriers/commission.go

@@ -72,7 +72,7 @@ func (q *CommissionQuerier) GetMetrics() ([]prometheus.Collector, []*types.Query
 						"denom":   balance.Denom,
 					}).Set(balance.Amount)
 				}
-			}(validator, rpc, chain)
+			}(validator.Address, rpc, chain)
 		}
 	}
 

pkg/queriers/delegations.go

@@ -70,7 +70,7 @@ func (q *DelegationsQuerier) GetMetrics() ([]prometheus.Collector, []*types.Quer
 					"chain":   chain.Name,
 					"address": validator,
 				}).Set(float64(utils.StrToInt64(delegatorsResponse.Pagination.Total)))
-			}(validator, rpc, chain)
+			}(validator.Address, rpc, chain)
 		}
 	}
 

pkg/queriers/rewards.go

@@ -88,7 +88,7 @@ func (q *RewardsQuerier) GetMetrics() ([]prometheus.Collector, []*types.QueryInf
 						"denom":   balance.Denom,
 					}).Set(balance.Amount)
 				}
-			}(validator, rpc, chain)
+			}(validator.Address, rpc, chain)
 		}
 	}
 

pkg/queriers/self_delegations.go

@@ -88,7 +88,7 @@ func (q *SelfDelegationsQuerier) GetMetrics() ([]prometheus.Collector, []*types.
 						"denom":   balance.Denom,
 					}).Set(balance.Amount)
 				}
-			}(validator, rpc, chain)
+			}(validator.Address, rpc, chain)
 		}
 	}
 

pkg/queriers/signing_info.go

@@ -0,0 +1,84 @@
+package queriers
+
+import (
+	"main/pkg/config"
+	"main/pkg/tendermint"
+	"main/pkg/types"
+	"main/pkg/utils"
+	"sync"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/rs/zerolog"
+)
+
+type SigningInfoQuerier struct {
+	Logger zerolog.Logger
+	Config *config.Config
+}
+
+func NewSigningInfoQuerier(logger *zerolog.Logger, config *config.Config) *SigningInfoQuerier {
+	return &SigningInfoQuerier{
+		Logger: logger.With().Str("component", "rewards_querier").Logger(),
+		Config: config,
+	}
+}
+
+func (q *SigningInfoQuerier) GetMetrics() ([]prometheus.Collector, []*types.QueryInfo) {
+	var queryInfos []*types.QueryInfo
+
+	missedBlocksGauge := prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Name: "cosmos_validators_exporter_missed_blocks",
+			Help: "Validator's missed blocks",
+		},
+		[]string{"chain", "address"},
+	)
+
+	var wg sync.WaitGroup
+	var mutex sync.Mutex
+
+	for _, chain := range q.Config.Chains {
+		rpc := tendermint.NewRPC(chain, q.Config.Timeout, q.Logger)
+
+		for _, validator := range chain.Validators {
+			wg.Add(1)
+			go func(validator config.Validator, rpc *tendermint.RPC, chain config.Chain) {
+				defer wg.Done()
+
+				if validator.ConsensusAddress == "" {
+					return
+				}
+
+				signingInfo, signingInfoQuery, err := rpc.GetSigningInfo(validator.ConsensusAddress)
+
+				mutex.Lock()
+				defer mutex.Unlock()
+
+				queryInfos = append(queryInfos, signingInfoQuery)
+
+				if err != nil {
+					q.Logger.Error().
+						Err(err).
+						Str("chain", chain.Name).
+						Str("address", validator.Address).
+						Msg("Error getting validator signing info")
+					return
+				}
+
+				missedBlocksCounter := utils.StrToInt64(signingInfo.ValSigningInfo.MissedBlocksCounter)
+				if missedBlocksCounter >= 0 {
+					missedBlocksGauge.With(prometheus.Labels{
+						"chain":   chain.Name,
+						"address": validator.Address,
+					}).Set(float64(missedBlocksCounter))
+				}
+			}(validator, rpc, chain)
+		}
+	}
+
+	wg.Wait()
+
+	return []prometheus.Collector{
+		missedBlocksGauge,
+	}, queryInfos
+}

pkg/queriers/unbonds.go

@@ -70,7 +70,7 @@ func (q *UnbondsQuerier) GetMetrics() ([]prometheus.Collector, []*types.QueryInf
 					"chain":   chain.Name,
 					"address": validator,
 				}).Set(float64(utils.StrToInt64(unbondsResponse.Pagination.Total)))
-			}(validator, rpc, chain)
+			}(validator.Address, rpc, chain)
 		}
 	}
 

pkg/queriers/validator.go

@@ -94,14 +94,6 @@ func (q *ValidatorQuerier) GetMetrics() ([]prometheus.Collector, []*types.QueryI
 		[]string{"chain", "address"},
 	)
 
-	missedBlocksGauge := prometheus.NewGaugeVec(
-		prometheus.GaugeOpts{
-			Name: "cosmos_validators_exporter_missed_blocks",
-			Help: "Validator's missed blocks",
-		},
-		[]string{"chain", "address"},
-	)
-
 	activeSetSizeGauge := prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Name: "cosmos_validators_exporter_active_set_size",
@@ -157,10 +149,6 @@ func (q *ValidatorQuerier) GetMetrics() ([]prometheus.Collector, []*types.QueryI
 					allValidatorsQueryInfo  *types.QueryInfo
 					allValidatorsQueryError error
 
-					signingInfo           *types.SigningInfoResponse
-					signingInfoQuery      *types.QueryInfo
-					signingInfoQueryError error
-
 					stakingParams           *types.StakingParamsResponse
 					stakingParamsQuery      *types.QueryInfo
 					stakingParamsQueryError error
@@ -179,31 +167,6 @@ func (q *ValidatorQuerier) GetMetrics() ([]prometheus.Collector, []*types.QueryI
 							Str("chain", chain.Name).
 							Str("address", validator).
 							Msg("Error querying for validator info")
-						return
-					}
-
-					if chain.BechConsensusPrefix == "" || validatorInfo == nil {
-						return
-					}
-
-					valConsAddress, err := validatorInfo.Validator.ConsensusPubkey.GetValConsAddress(chain.BechConsensusPrefix)
-					if err != nil {
-						q.Logger.Error().
-							Err(validatorQueryError).
-							Str("chain", chain.Name).
-							Str("address", validator).
-							Msg("Error getting validator consensus address")
-						signingInfoQueryError = err
-					} else {
-						signingInfo, signingInfoQuery, signingInfoQueryError = rpc.GetSigningInfo(valConsAddress)
-
-						if signingInfoQueryError != nil {
-							q.Logger.Error().
-								Err(validatorQueryError).
-								Str("chain", chain.Name).
-								Str("address", validator).
-								Msg("Error getting validator signing info")
-						}
 					}
 				}()
 
@@ -244,9 +207,6 @@ func (q *ValidatorQuerier) GetMetrics() ([]prometheus.Collector, []*types.QueryI
 				if validatorQueryInfo != nil {
 					queryInfos = append(queryInfos, validatorQueryInfo)
 				}
-				if signingInfoQuery != nil {
-					queryInfos = append(queryInfos, signingInfoQuery)
-				}
 				if stakingParamsQuery != nil {
 					queryInfos = append(queryInfos, stakingParamsQuery)
 				}
@@ -297,16 +257,6 @@ func (q *ValidatorQuerier) GetMetrics() ([]prometheus.Collector, []*types.QueryI
 					}).Set(utils.StrToFloat64(validatorInfo.Validator.DelegatorShares))
 				}
 
-				if signingInfo != nil {
-					missedBlocksCounter := utils.StrToInt64(signingInfo.ValSigningInfo.MissedBlocksCounter)
-					if missedBlocksCounter >= 0 {
-						missedBlocksGauge.With(prometheus.Labels{
-							"chain":   chain.Name,
-							"address": validator,
-						}).Set(float64(missedBlocksCounter))
-					}
-				}
-
 				if stakingParams != nil {
 					maxValidators := int64(stakingParams.StakingParams.MaxValidators)
 					if maxValidators >= 0 {
@@ -364,7 +314,7 @@ func (q *ValidatorQuerier) GetMetrics() ([]prometheus.Collector, []*types.QueryI
 				}
 
 				wg.Done()
-			}(validator, rpc, chain)
+			}(validator.Address, rpc, chain)
 		}
 	}
 
@@ -378,7 +328,6 @@ func (q *ValidatorQuerier) GetMetrics() ([]prometheus.Collector, []*types.QueryI
 		commissionMaxGauge,
 		commissionMaxChangeGauge,
 		delegationsGauge,
-		missedBlocksGauge,
 		activeSetSizeGauge,
 		validatorRankGauge,
 		validatorsCountGauge,

pkg/queriers/wallet.go

@@ -88,7 +88,7 @@ func (q *WalletQuerier) GetMetrics() ([]prometheus.Collector, []*types.QueryInfo
 						"denom":   balance.Denom,
 					}).Set(balance.Amount)
 				}
-			}(validator, rpc, chain)
+			}(validator.Address, rpc, chain)
 		}
 	}
 

pkg/tendermint/tendermint.go

@@ -233,6 +233,10 @@ func (rpc *RPC) GetSigningInfo(valcons string) (*types.SigningInfoResponse, *typ
 		return nil, &info, err
 	}
 
+	if response.Code != 0 {
+		return &types.SigningInfoResponse{}, &info, fmt.Errorf("expected code 0, but got %d", response.Code)
+	}
+
 	return response, &info, nil
 }
 

pkg/types/types.go

@@ -199,6 +199,7 @@ type BalanceInResponse struct {
 }
 
 type SigningInfoResponse struct {
+	Code           int                  `json:"code"`
 	ValSigningInfo ValidatorSigningInfo `json:"val_signing_info"`
 }