feat: add supply metric (#78)

* feat: add supply metric

* chore: fixed flaky test

assets/supply.json

@@ -0,0 +1,12 @@
+{
+  "supply": [
+    {
+      "denom": "uatom",
+      "amount": "123456"
+    }
+  ],
+  "pagination": {
+    "next_key": null,
+    "total": "1"
+  }
+}

pkg/app.go

@@ -96,6 +96,7 @@ func NewApp(configPath string, filesystem fs.FS, version string) *App {
 		fetchersPkg.NewValidatorConsumersFetcher(logger, appConfig.Chains, rpcs, tracer),
 		fetchersPkg.NewConsumerCommissionFetcher(logger, appConfig.Chains, rpcs, tracer),
 		fetchersPkg.NewInflationFetcher(logger, appConfig.Chains, rpcs, tracer),
+		fetchersPkg.NewSupplyFetcher(logger, appConfig.Chains, rpcs, tracer),
 	}
 
 	generators := []generatorsPkg.Generator{
@@ -122,6 +123,7 @@ func NewApp(configPath string, filesystem fs.FS, version string) *App {
 		generatorsPkg.NewValidatorActiveGenerator(appConfig.Chains, logger),
 		generatorsPkg.NewValidatorCommissionRateGenerator(appConfig.Chains, logger),
 		generatorsPkg.NewInflationGenerator(),
+		generatorsPkg.NewSupplyGenerator(appConfig.Chains),
 	}
 
 	server := &http.Server{Addr: appConfig.ListenAddress, Handler: nil}

pkg/app_test.go

@@ -73,7 +73,9 @@ func TestAppLoadConfigOk(t *testing.T) {
 
 	for {
 		request, err := http.Get("http://localhost:9560/healthcheck")
-		_ = request.Body.Close()
+		if request.Body != nil {
+			_ = request.Body.Close()
+		}
 		if err == nil {
 			break
 		}

pkg/constants/constants.go

@@ -22,6 +22,7 @@ const (
 	FetcherNamePrice              FetcherName = "price"
 	FetcherNameNodeInfo           FetcherName = "node_info"
 	FetcherNameInflation          FetcherName = "inflation"
+	FetcherNameSupply             FetcherName = "supply"
 
 	MetricsPrefix string = "cosmos_validators_exporter_"
 

pkg/fetchers/supply.go

@@ -0,0 +1,112 @@
+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 SupplyFetcher struct {
+	Logger zerolog.Logger
+	Chains []*config.Chain
+	RPCs   map[string]*tendermint.RPCWithConsumers
+	Tracer trace.Tracer
+
+	wg    sync.WaitGroup
+	mutex sync.Mutex
+
+	queryInfos  []*types.QueryInfo
+	allSupplies map[string][]types.Amount
+}
+
+type SupplyData struct {
+	Supplies map[string][]types.Amount
+}
+
+func NewSupplyFetcher(
+	logger *zerolog.Logger,
+	chains []*config.Chain,
+	rpcs map[string]*tendermint.RPCWithConsumers,
+	tracer trace.Tracer,
+) *SupplyFetcher {
+	return &SupplyFetcher{
+		Logger: logger.With().Str("component", "balance_fetcher").Logger(),
+		Chains: chains,
+		RPCs:   rpcs,
+		Tracer: tracer,
+	}
+}
+
+func (q *SupplyFetcher) Fetch(
+	ctx context.Context,
+) (interface{}, []*types.QueryInfo) {
+	q.queryInfos = []*types.QueryInfo{}
+	q.allSupplies = map[string][]types.Amount{}
+
+	for _, chain := range q.Chains {
+		q.wg.Add(1 + len(chain.ConsumerChains))
+
+		rpc, _ := q.RPCs[chain.Name]
+
+		go q.processChain(
+			ctx,
+			chain.Name,
+			rpc.RPC,
+		)
+
+		for consumerIndex, consumerChain := range chain.ConsumerChains {
+			consumerRPC := rpc.Consumers[consumerIndex]
+
+			go q.processChain(
+				ctx,
+				consumerChain.Name,
+				consumerRPC,
+			)
+		}
+	}
+
+	q.wg.Wait()
+
+	return SupplyData{Supplies: q.allSupplies}, q.queryInfos
+}
+
+func (q *SupplyFetcher) Name() constants.FetcherName {
+	return constants.FetcherNameSupply
+}
+
+func (q *SupplyFetcher) processChain(
+	ctx context.Context,
+	chainName string,
+	rpc *tendermint.RPC,
+) {
+	defer q.wg.Done()
+
+	supply, query, err := rpc.GetTotalSupply(ctx)
+
+	q.mutex.Lock()
+	defer q.mutex.Unlock()
+
+	if query != nil {
+		q.queryInfos = append(q.queryInfos, query)
+	}
+
+	if err != nil {
+		q.Logger.Error().
+			Err(err).
+			Str("chain", chainName).
+			Msg("Error querying for chain supply")
+		return
+	}
+
+	if supply == nil {
+		return
+	}
+
+	q.allSupplies[chainName] = supply
+}