feat: add consumer validators count

pkg/app.go

@@ -83,6 +83,7 @@ func NewApp(configPath string, version string) *App {
 		fetchersPkg.NewBalanceFetcher(logger, appConfig, rpcs, tracer),
 		fetchersPkg.NewSelfDelegationFetcher(logger, appConfig, rpcs, tracer),
 		fetchersPkg.NewValidatorsFetcher(logger, appConfig, rpcs, tracer),
+		fetchersPkg.NewConsumerValidatorsFetcher(logger, appConfig, rpcs, tracer),
 		fetchersPkg.NewStakingParamsFetcher(logger, appConfig, rpcs, tracer),
 		fetchersPkg.NewPriceFetcher(logger, appConfig, tracer, coingecko, dexScreener),
 		fetchersPkg.NewNodeInfoFetcher(logger, appConfig, rpcs, tracer),

pkg/constants/constants.go

@@ -13,6 +13,7 @@ const (
 	FetcherNameBalance             FetcherName = "balance"
 	FetcherNameSelfDelegation      FetcherName = "self-delegation"
 	FetcherNameValidators          FetcherName = "validators"
+	FetcherNameConsumerValidators  FetcherName = "consumer-validators"
 	FetcherNameStakingParams       FetcherName = "staking_params"
 	FetcherNamePrice               FetcherName = "price"
 	FetcherNameNodeInfo            FetcherName = "node_info"

pkg/fetchers/consumer_validators.go

@@ -0,0 +1,96 @@
+package fetchers
+
+import (
+	"context"
+	"main/pkg/config"
+	"main/pkg/constants"
+	"main/pkg/tendermint"
+	"main/pkg/types"
+	"sync"
+
+	"github.com/rs/zerolog"
+	"go.opentelemetry.io/otel/trace"
+)
+
+type ConsumerValidatorsFetcher struct {
+	Logger zerolog.Logger
+	Config *config.Config
+	RPCs   map[string]*tendermint.RPCWithConsumers
+	Tracer trace.Tracer
+
+	wg    sync.WaitGroup
+	mutex sync.Mutex
+
+	queryInfos    []*types.QueryInfo
+	allValidators map[string]*types.ConsumerValidatorsResponse
+}
+
+type ConsumerValidatorsData struct {
+	Validators map[string]*types.ConsumerValidatorsResponse
+}
+
+func NewConsumerValidatorsFetcher(
+	logger *zerolog.Logger,
+	config *config.Config,
+	rpcs map[string]*tendermint.RPCWithConsumers,
+	tracer trace.Tracer,
+) *ConsumerValidatorsFetcher {
+	return &ConsumerValidatorsFetcher{
+		Logger: logger.With().Str("component", "validators_fetcher").Logger(),
+		Config: config,
+		RPCs:   rpcs,
+		Tracer: tracer,
+	}
+}
+
+func (f *ConsumerValidatorsFetcher) Fetch(
+	ctx context.Context,
+) (interface{}, []*types.QueryInfo) {
+	f.queryInfos = []*types.QueryInfo{}
+	f.allValidators = map[string]*types.ConsumerValidatorsResponse{}
+
+	for _, chain := range f.Config.Chains {
+		f.wg.Add(len(chain.ConsumerChains))
+
+		rpc, _ := f.RPCs[chain.Name]
+
+		for _, consumerChain := range chain.ConsumerChains {
+			go f.processChain(ctx, rpc.RPC, consumerChain)
+		}
+	}
+
+	f.wg.Wait()
+
+	return ConsumerValidatorsData{Validators: f.allValidators}, f.queryInfos
+}
+
+func (f *ConsumerValidatorsFetcher) Name() constants.FetcherName {
+	return constants.FetcherNameConsumerValidators
+}
+
+func (f *ConsumerValidatorsFetcher) processChain(
+	ctx context.Context,
+	rpc *tendermint.RPC,
+	chain *config.ConsumerChain,
+) {
+	defer f.wg.Done()
+
+	allValidatorsList, queryInfo, err := rpc.GetConsumerValidators(ctx, chain.ChainID)
+
+	f.mutex.Lock()
+	defer f.mutex.Unlock()
+
+	if queryInfo != nil {
+		f.queryInfos = append(f.queryInfos, queryInfo)
+	}
+
+	if err != nil {
+		f.Logger.Error().
+			Err(err).
+			Str("chain", chain.Name).
+			Msg("Error querying consumer validators")
+		return
+	}
+
+	f.allValidators[chain.Name] = allValidatorsList
+}

pkg/generators/validators_info.go

@@ -24,6 +24,11 @@ func (g *ValidatorsInfoGenerator) Generate(state *statePkg.State) []prometheus.C
 		return []prometheus.Collector{}
 	}
 
+	consumersDataRaw, ok := state.Get(constants.FetcherNameConsumerValidators)
+	if !ok {
+		return []prometheus.Collector{}
+	}
+
 	validatorsCountGauge := prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Name: constants.MetricsPrefix + "validators_count",
@@ -41,6 +46,7 @@ func (g *ValidatorsInfoGenerator) Generate(state *statePkg.State) []prometheus.C
 	)
 
 	data, _ := dataRaw.(fetchersPkg.ValidatorsData)
+	consumersData, _ := consumersDataRaw.(fetchersPkg.ConsumerValidatorsData)
 
 	for chain, validators := range data.Validators {
 		activeValidators := utils.Filter(validators.Validators, func(v types.Validator) bool {
@@ -66,5 +72,11 @@ func (g *ValidatorsInfoGenerator) Generate(state *statePkg.State) []prometheus.C
 		}).Set(totalStake)
 	}
 
+	for chain, validators := range consumersData.Validators {
+		validatorsCountGauge.With(prometheus.Labels{
+			"chain": chain,
+		}).Set(float64(len(validators.Validators)))
+	}
+
 	return []prometheus.Collector{validatorsCountGauge, totalBondedTokensGauge}
 }

pkg/tendermint/rpc.go

@@ -190,6 +190,36 @@ func (rpc *RPC) GetAllValidators(
 	return response, &info, nil
 }
 
+func (rpc *RPC) GetConsumerValidators(
+	ctx context.Context,
+	chainId string,
+) (*types.ConsumerValidatorsResponse, *types.QueryInfo, error) {
+	if !rpc.ChainQueries.Enabled("consumer-validators") {
+		return nil, nil, nil
+	}
+
+	childQuerierCtx, span := rpc.Tracer.Start(
+		ctx,
+		"Fetching consumer validators list",
+	)
+	defer span.End()
+
+	url := rpc.ChainHost + "/interchain_security/ccv/provider/consumer_validators/" + chainId
+
+	var response *types.ConsumerValidatorsResponse
+	info, err := rpc.Get(url, &response, childQuerierCtx)
+	if err != nil {
+		return nil, &info, err
+	}
+
+	if response.Code != 0 {
+		info.Success = false
+		return &types.ConsumerValidatorsResponse{}, &info, fmt.Errorf("expected code 0, but got %d", response.Code)
+	}
+
+	return response, &info, nil
+}
+
 func (rpc *RPC) GetValidatorCommission(
 	address string,
 	ctx context.Context,

pkg/types/tendermint.go

@@ -1,15 +1,9 @@
 package types
 
 import (
-	b64 "encoding/base64"
 	"main/pkg/constants"
 	"main/pkg/utils"
 	"time"
-
-	codecTypes "github.com/cosmos/cosmos-sdk/codec/types"
-	cryptoTypes "github.com/cosmos/cosmos-sdk/crypto/types"
-	"github.com/cosmos/cosmos-sdk/simapp"
-	"github.com/cosmos/cosmos-sdk/types"
 )
 
 type ValidatorResponse struct {
@@ -53,30 +47,6 @@ type ConsensusPubkey struct {
 	Key  string `json:"key"`
 }
 
-func (key *ConsensusPubkey) GetValConsAddress(prefix string) (string, error) {
-	encCfg := simapp.MakeTestEncodingConfig()
-	interfaceRegistry := encCfg.InterfaceRegistry
-
-	sDec, _ := b64.StdEncoding.DecodeString(key.Key)
-	pk := codecTypes.Any{
-		TypeUrl: key.Type,
-		Value:   append([]byte{10, 32}, sDec...),
-	}
-
-	var pkProto cryptoTypes.PubKey
-	if err := interfaceRegistry.UnpackAny(&pk, &pkProto); err != nil {
-		return "", err
-	}
-
-	cosmosValCons := types.ConsAddress(pkProto.Address()).String()
-	properValCons, err := utils.ChangeBech32Prefix(cosmosValCons, prefix)
-	if err != nil {
-		return "", err
-	}
-
-	return properValCons, nil
-}
-
 type PaginationResponse struct {
 	Code       int        `json:"code"`
 	Pagination Pagination `json:"pagination"`
@@ -182,3 +152,12 @@ type NodeInfoResponse struct {
 		CosmosSDKVersion string `json:"cosmos_sdk_version"`
 	} `json:"application_version"`
 }
+
+type ConsumerValidator struct {
+	ProviderAddress string `json:"provider_address"`
+}
+
+type ConsumerValidatorsResponse struct {
+	Code       int                 `json:"code"`
+	Validators []ConsumerValidator `json:"validators"`
+}