feat(observability): add prometheus metrics

cmd/feeder/main.go

@@ -2,10 +2,12 @@ package main
 
 import (
 	"flag"
+	"net/http"
 	"os"
 	"os/signal"
 
 	"github.com/NibiruChain/nibiru/app"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"github.com/rs/zerolog"
 
 	"github.com/NibiruChain/pricefeeder/config"
@@ -49,6 +51,13 @@ func main() {
 
 	handleInterrupt(logger, f)
 
+	http.Handle("/metrics", promhttp.Handler())
+	go func() {
+		if err := http.ListenAndServe(":3000", nil); err != nil {
+			logger.Error().Err(err).Msg("Metrics HTTP server failed")
+		}
+	}()
+
 	select {}
 }
 

feeder/priceposter/client.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/NibiruChain/nibiru/app"
 	oracletypes "github.com/NibiruChain/nibiru/x/oracle/types"
+	"github.com/NibiruChain/pricefeeder/metrics"
 	"github.com/NibiruChain/pricefeeder/types"
 	"github.com/cosmos/cosmos-sdk/client"
 	codectypes "github.com/cosmos/cosmos-sdk/codec/types"
@@ -117,6 +118,8 @@ func (c *Client) SendPrices(vp types.VotingPeriod, prices []types.Price) {
 
 	c.previousPrevote = newPrevote
 	logger.Info().Str("tx-hash", resp.TxHash).Msg("successfully forwarded prices")
+
+	metrics.NumVotingPeriods.Inc()
 }
 
 func (c *Client) Close() {

feeder/priceprovider/priceprovider.go

@@ -39,13 +39,13 @@ func NewPriceProvider(
 	var source types.Source
 	switch sourceName {
 	case sources.Bitfinex:
-		source = sources.NewTickSource(symbolsFromPairToSymbolMapping(pairToSymbolMap), sources.BitfinexPriceUpdate, logger)
+		source = sources.NewTickSource(symbolsFromPairToSymbolMapping(pairToSymbolMap), sources.BitfinexPriceUpdate, logger, sources.Bitfinex)
 	case sources.Binance:
-		source = sources.NewTickSource(symbolsFromPairToSymbolMapping(pairToSymbolMap), sources.BinancePriceUpdate, logger)
+		source = sources.NewTickSource(symbolsFromPairToSymbolMapping(pairToSymbolMap), sources.BinancePriceUpdate, logger, sources.Binance)
 	case sources.Coingecko:
-		source = sources.NewTickSource(symbolsFromPairToSymbolMapping(pairToSymbolMap), sources.CoingeckoPriceUpdate(config), logger)
+		source = sources.NewTickSource(symbolsFromPairToSymbolMapping(pairToSymbolMap), sources.CoingeckoPriceUpdate(config), logger, sources.Coingecko)
 	case sources.CoinMarketCap:
-		source = sources.NewTickSource(symbolsFromPairToSymbolMapping(pairToSymbolMap), sources.CoinmarketcapPriceUpdate(config), logger)
+		source = sources.NewTickSource(symbolsFromPairToSymbolMapping(pairToSymbolMap), sources.CoinmarketcapPriceUpdate(config), logger, sources.CoinMarketCap)
 	default:
 		panic("unknown price provider: " + sourceName)
 	}

feeder/priceprovider/sources/tick_source.go

@@ -4,20 +4,19 @@ import (
 	"time"
 
 	"github.com/NibiruChain/nibiru/x/common/set"
+	"github.com/NibiruChain/pricefeeder/metrics"
 	"github.com/NibiruChain/pricefeeder/types"
 	"github.com/rs/zerolog"
 )
 
-var (
-	// UpdateTick defines the wait time between price updates.
-	UpdateTick = 8 * time.Second
-)
+// UpdateTick defines the wait time between price updates.
+var UpdateTick = 8 * time.Second
 
 var _ types.Source = (*TickSource)(nil)
 
 // NewTickSource instantiates a new TickSource instance, given the symbols and a price updater function
 // which returns the latest prices for the provided symbols.
-func NewTickSource(symbols set.Set[types.Symbol], fetchPricesFunc types.FetchPricesFunc, logger zerolog.Logger) *TickSource {
+func NewTickSource(symbols set.Set[types.Symbol], fetchPricesFunc types.FetchPricesFunc, logger zerolog.Logger, source string) *TickSource {
 	ts := &TickSource{
 		logger:             logger,
 		stopSignal:         make(chan struct{}),
@@ -26,6 +25,7 @@ func NewTickSource(symbols set.Set[types.Symbol], fetchPricesFunc types.FetchPri
 		symbols:            symbols,
 		fetchPrices:        fetchPricesFunc,
 		priceUpdateChannel: make(chan map[types.Symbol]types.RawPrice),
+		source:             source,
 	}
 
 	go ts.loop()
@@ -43,6 +43,7 @@ type TickSource struct {
 	symbols            set.Set[types.Symbol] // symbols as named on the third party data source
 	fetchPrices        func(symbols set.Set[types.Symbol]) (map[types.Symbol]float64, error)
 	priceUpdateChannel chan map[types.Symbol]types.RawPrice
+	source             string
 }
 
 func (s *TickSource) loop() {
@@ -56,11 +57,14 @@ func (s *TickSource) loop() {
 		case <-s.tick.C:
 			s.logger.Debug().Msg("received tick, updating prices")
 
+			start := time.Now()
 			rawPrices, err := s.fetchPrices(s.symbols)
 			if err != nil {
 				s.logger.Err(err).Msg("failed to update prices")
 				break // breaks the current select case, not the for cycle
 			}
+			duration := time.Since(start)
+			metrics.PriceSourceLatency.WithLabelValues(s.source).Observe(duration.Seconds())
 
 			priceUpdate := make(map[types.Symbol]types.RawPrice, len(rawPrices))
 			for symbol, price := range rawPrices {

feeder/priceprovider/sources/tick_source_test.go

@@ -29,7 +29,7 @@ func TestTickSource(t *testing.T) {
 		ts := NewTickSource(expectedSymbols, func(symbols set.Set[types.Symbol]) (map[types.Symbol]float64, error) {
 			require.Equal(t, expectedSymbols, symbols)
 			return expectedPrices, nil
-		}, zerolog.New(io.Discard))
+		}, zerolog.New(io.Discard), "test_source")
 
 		defer ts.Close()
 
@@ -63,7 +63,7 @@ func TestTickSource(t *testing.T) {
 
 		ts := NewTickSource(expectedSymbols, func(symbols set.Set[types.Symbol]) (map[types.Symbol]float64, error) {
 			return expectedPrices, nil
-		}, zerolog.New(mw))
+		}, zerolog.New(mw), "test_source")
 
 		<-time.After(UpdateTick + 1*time.Second) // wait for a tick update
 		ts.Close()                               // make the update be dropped because of close
@@ -82,7 +82,7 @@ func TestTickSource(t *testing.T) {
 
 		ts := NewTickSource(set.New[types.Symbol]("tBTCUSDT"), func(symbols set.Set[types.Symbol]) (map[types.Symbol]float64, error) {
 			return nil, fmt.Errorf("sentinel error")
-		}, zerolog.New(mw))
+		}, zerolog.New(mw), "test_source")
 		defer ts.Close()
 
 		<-time.After(UpdateTick + 1*time.Second) // wait for a tick update

metrics/README.md

@@ -0,0 +1,28 @@
+# Metrics in metrics.go
+
+This file defines two metrics that are used to monitor the performance and behavior of the pricefeeder application.
+
+## pricefeeder_voting_periods_total
+
+This is a counter metric that keeps track of the total number of voting periods the pricefeeder has participated in. A voting period is a specific duration during which votes are collected. This metric is incremented each time the pricefeeder participates in a voting period.
+
+```go
+var NumVotingPeriods = promauto.NewCounter(prometheus.CounterOpts{
+    Name: "pricefeeder_voting_periods_total",
+    Help: "The total number of voting periods this pricefeeder has participated in",
+})
+```
+
+## pricefeeder_price_source_latency_seconds
+
+This is a histogram metric that measures the latency of querying a price source in seconds. The latency is the amount of time it takes for the pricefeeder to receive a response after it has sent a request to the price source. This metric is useful for monitoring the performance of the price sources and identifying any potential issues or delays.
+
+```go
+var PriceSourceLatency = promauto.NewHistogramVec(prometheus.HistogramOpts{
+    Name:    "pricefeeder_price_source_latency_seconds",
+    Help:    "The latency of querying a price source in seconds",
+    Buckets: prometheus.DefBuckets,
+}, []string{"source"})
+```
+
+The source label is used to differentiate the latencies of different price sources.

metrics/metrics.go

@@ -0,0 +1,17 @@
+package metrics
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+)
+
+var NumVotingPeriods = promauto.NewCounter(prometheus.CounterOpts{
+	Name: "pricefeeder_voting_periods_total",
+	Help: "The total number of voting periods this pricefeeder has participated in",
+})
+
+var PriceSourceLatency = promauto.NewHistogramVec(prometheus.HistogramOpts{
+	Name:    "pricefeeder_price_source_latency_seconds",
+	Help:    "The latency of querying a price source in seconds",
+	Buckets: prometheus.DefBuckets,
+}, []string{"source"})