feat: validation service

cmd/connect/main.go

@@ -4,18 +4,21 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"github.com/skip-mev/connect/v2/service/validation"
 	"net/http"
+	_ "net/http/pprof"
 	"os"
 	"os/signal"
 	"path/filepath"
 	"syscall"
-
-	_ "net/http/pprof" //nolint: gosec
+	"time"
 
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
 	"go.uber.org/zap"
 
+	//nolint: gosec
+
 	"github.com/skip-mev/connect/v2/cmd/build"
 	cmdconfig "github.com/skip-mev/connect/v2/cmd/connect/config"
 	"github.com/skip-mev/connect/v2/oracle"
@@ -58,6 +61,8 @@ var (
 	flagPort                     = "port"
 	flagUpdateInterval           = "update-interval"
 	flagMaxPriceAge              = "max-price-age"
+	flagMode                     = "mode"
+	flagValidationPeriod         = "validation-period"
 
 	// flag-bound values.
 	oracleCfgPath       string
@@ -75,12 +80,21 @@ var (
 	maxAge              int
 	disableCompressLogs bool
 	disableRotatingLogs bool
+	mode                string
+	validationPeriod    int
 )
 
 const (
 	DefaultLegacyConfigPath = "./oracle.json"
 )
 
+type runMode string
+
+var (
+	modeExec     runMode = "exec"
+	modeValidate runMode = "validate"
+)
+
 func init() {
 	rootCmd.Flags().StringVarP(
 		&marketMapProvider,
@@ -187,6 +201,20 @@ func init() {
 		"",
 		"Use a custom listen-to endpoint for market-map (overwrites what is provided in oracle-config).",
 	)
+	rootCmd.Flags().StringVarP(
+		&mode,
+		flagMode,
+		"m",
+		string(modeExec),
+		"Select the mode to run the oracle in.  Default is \"exec\" which will fetch prices as configured.  \"validate\" mode will run the oracle for a set period of time to validate the configuration.",
+	)
+	rootCmd.Flags().IntVarP(
+		&validationPeriod,
+		flagValidationPeriod,
+		"",
+		int(validation.DefaultValidationPeriod),
+		"Duration to run in validation mode.  Note: this flag is only used if mode == \"validate\"",
+	)
 
 	// these flags are connected to the OracleConfig.
 	rootCmd.Flags().Bool(
@@ -316,6 +344,8 @@ func runOracle() error {
 		nodeClient, _ = oraclemetrics.NewNodeClient(nodeEndpoint)
 	}
 
+	isValidateMode := runMode(mode) == modeValidate
+
 	metrics := oraclemetrics.NewMetricsFromConfig(cfg.Metrics, nodeClient)
 
 	aggregator, err := oraclemath.NewIndexPriceAggregator(
@@ -395,6 +425,28 @@ func runOracle() error {
 		}()
 	}
 
+	// run validation service if enabled and tear down if completed successfully
+	if isValidateMode {
+		valCfg := validation.DefaultConfig()
+		valCfg.ValidationPeriod = time.Duration(validationPeriod)
+		validatorService := validation.NewValidator(logger, metrics, valCfg)
+
+		go func(c context.CancelFunc) {
+			defer c()
+
+			_, err := validatorService.Run()
+			if err != nil {
+				logger.Error("failed to validate metrics", zap.Error(err))
+
+				// kill the process
+				os.Exit(1)
+			}
+
+			logger.Info("shutting down gracefully after validation")
+
+		}(cancel)
+	}
+
 	// start server (blocks).
 	if err := srv.StartServer(ctx, cfg.Host, cfg.Port); err != nil {
 		logger.Error("stopping server", zap.Error(err))

contrib/compose/docker-compose-dev.yml

@@ -44,6 +44,7 @@ services:
       "--market-config-path", "/data/markets.json",
       "--pprof-port", "6060",
       "--run-pprof",
+      # "-m", "validate" uncomment to run in validation mode with default config
     ]
     ports:
       - "8080:8080" # main oracle port

go.sum

@@ -947,6 +947,7 @@ github.com/jmespath/go-jmespath/internal/testify v1.5.1/go.mod h1:L3OGu8Wl2/fWfC
 github.com/jmhodges/levigo v1.0.0 h1:q5EC36kV79HWeTBWsod3mG11EgStG3qArTKcvlksN1U=
 github.com/jmhodges/levigo v1.0.0/go.mod h1:Q6Qx+uH3RAqyK4rFQroq9RL7mdkABMcfhEI+nNuzMJQ=
 github.com/jonboulle/clockwork v0.1.0/go.mod h1:Ii8DK3G1RaLaWxj9trq07+26W01tbo22gdxWY5EU2bo=
+github.com/jpillora/backoff v1.0.0 h1:uvFg412JmmHBHw7iwprIxkPMI+sGQ4kzOWsMeHnm2EA=
 github.com/jpillora/backoff v1.0.0/go.mod h1:J/6gKK9jxlEcS3zixgDgUAsiuZ7yrSoa/FX5e0EB2j4=
 github.com/json-iterator/go v1.1.6/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCVDaaPEHmU=
 github.com/json-iterator/go v1.1.7/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=

oracle/metrics/metrics.go

@@ -3,6 +3,7 @@ package metrics
 import (
 	"fmt"
 	"strings"
+	"sync"
 
 	"github.com/DataDog/datadog-go/statsd"
 	"github.com/prometheus/client_golang/prometheus"
@@ -33,6 +34,7 @@ const (
 	ProviderCountMetricName    = "health_check_market_providers"
 	SlinkyBuildInfoMetricName  = "slinky_build_info"
 	ConnectBuildInfoMetricName = "connect_build_info"
+	MissingPricesName          = "missing_prices"
 )
 
 // Metrics is an interface that defines the API for oracle metrics.
@@ -45,7 +47,7 @@ type Metrics interface {
 
 	// AddTickerTick increments the number of ticks for a given ticker. Specifically, this
 	// is used to track the number of times a ticker was updated.
-	AddTickerTick(ticker string)
+	AddTickerTick(pairID string)
 
 	// UpdatePrice price updates the price for the given pairID for the provider.
 	UpdatePrice(name, pairID string, decimals uint64, price float64)
@@ -60,24 +62,33 @@ type Metrics interface {
 
 	// AddProviderCountForMarket increments the number of providers that were utilized
 	// to calculate the final price for a given market.
-	AddProviderCountForMarket(market string, count int)
+	AddProviderCountForMarket(pairID string, count int)
 
 	// SetConnectBuildInfo sets the build information for the Slinky binary.
 	SetConnectBuildInfo()
+
+	// MissingPrices sets a list of missing prices for the given aggregation tick.
+	MissingPrices(pairIDs []string)
+
+	// GetMissingPrices gets the current list of missing prices.
+	GetMissingPrices() []string
 }
 
 // OracleMetricsImpl is a Metrics implementation that does nothing.
 type OracleMetricsImpl struct {
-	promTicks            prometheus.Counter
-	promTickerTicks      *prometheus.CounterVec
-	promPrices           *prometheus.GaugeVec
-	promAggregatePrices  *prometheus.GaugeVec
-	promProviderTick     *prometheus.CounterVec
-	promProviderCount    *prometheus.GaugeVec
-	promSlinkyBuildInfo  *prometheus.GaugeVec
-	promConnectBuildInfo *prometheus.GaugeVec
-	statsdClient         statsd.ClientInterface
-	nodeIdentifier       string
+	promTicks             prometheus.Counter
+	promTickerTicks       *prometheus.CounterVec
+	promPrices            *prometheus.GaugeVec
+	promAggregatePrices   *prometheus.GaugeVec
+	promProviderTick      *prometheus.CounterVec
+	promProviderCount     *prometheus.GaugeVec
+	promSlinkyBuildInfo   *prometheus.GaugeVec
+	promConnectBuildInfo  *prometheus.GaugeVec
+	promMissingPrices     *prometheus.GaugeVec
+	statsdClient          statsd.ClientInterface
+	nodeIdentifier        string
+	missingPricesInternal []string
+	missingPricesMtx      sync.Mutex
 }
 
 // NewMetricsFromConfig returns an oracle Metrics implementation based on the provided
@@ -110,7 +121,9 @@ func NewMetricsFromConfig(config config.MetricsConfig, nodeClient NodeClient) Me
 
 // NewMetrics returns a Metrics implementation that exposes metrics to Prometheus.
 func NewMetrics(statsdClient statsd.ClientInterface, nodeIdentifier string) Metrics {
-	ret := OracleMetricsImpl{}
+	ret := OracleMetricsImpl{
+		missingPricesInternal: make([]string, 0),
+	}
 
 	ret.statsdClient = statsdClient
 	ret.nodeIdentifier = nodeIdentifier
@@ -156,6 +169,11 @@ func NewMetrics(statsdClient statsd.ClientInterface, nodeIdentifier string) Metr
 		Name:      ConnectBuildInfoMetricName,
 		Help:      "Information about the connect build",
 	}, []string{Version})
+	ret.promMissingPrices = prometheus.NewGaugeVec(prometheus.GaugeOpts{
+		Namespace: OracleSubsystem,
+		Name:      MissingPricesName,
+		Help:      "Missing Prices",
+	}, []string{Version})
 
 	prometheus.MustRegister(ret.promTicks)
 	prometheus.MustRegister(ret.promTickerTicks)
@@ -165,6 +183,7 @@ func NewMetrics(statsdClient statsd.ClientInterface, nodeIdentifier string) Metr
 	prometheus.MustRegister(ret.promProviderCount)
 	prometheus.MustRegister(ret.promSlinkyBuildInfo)
 	prometheus.MustRegister(ret.promConnectBuildInfo)
+	prometheus.MustRegister(ret.promMissingPrices)
 
 	return &ret
 }
@@ -176,6 +195,10 @@ func NewNopMetrics() Metrics {
 	return &noOpOracleMetrics{}
 }
 
+func (m *noOpOracleMetrics) MissingPrices(_ []string) {}
+
+func (m *noOpOracleMetrics) GetMissingPrices() []string { return []string{} }
+
 // AddTick increments the total number of ticks that have been processed by the oracle.
 func (m *noOpOracleMetrics) AddTick() {}
 
@@ -279,6 +302,26 @@ func (m *OracleMetricsImpl) AddProviderCountForMarket(market string, count int)
 	m.statsdClient.Gauge(metricName, float64(count), []string{}, 1)
 }
 
+// MissingPrices updates the list of missing prices for the given tick.
+func (m *OracleMetricsImpl) MissingPrices(pairIDs []string) {
+	if len(pairIDs) == 0 {
+		return
+	}
+
+	m.missingPricesMtx.Lock()
+	defer m.missingPricesMtx.Unlock()
+	m.missingPricesInternal = pairIDs
+
+}
+
+// GetMissingPrices gets the internal missing prices array.
+func (m *OracleMetricsImpl) GetMissingPrices() []string {
+	m.missingPricesMtx.Lock()
+	defer m.missingPricesMtx.Unlock()
+
+	return m.missingPricesInternal
+}
+
 // SetConnectBuildInfo sets the build information for the Connect binary. The version exported
 // is determined by the build time version in accordance with the build pkg.
 func (m *OracleMetricsImpl) SetConnectBuildInfo() {

pkg/http/address.go

@@ -0,0 +1,16 @@
+package http
+
+import "net"
+
+func IsValidAddress(address string) bool {
+	host, port, err := net.SplitHostPort(address)
+	if err != nil {
+		return false
+	}
+
+	if host == "" || port == "" {
+		return false
+	}
+
+	return true
+}

pkg/math/oracle/aggregator.go

@@ -132,6 +132,7 @@ func (m *IndexPriceAggregator) AggregatePrices() {
 	m.logger.Debug("calculated median prices for price feeds", zap.Int("num_prices", len(indexPrices)))
 	if len(missingPrices) > 0 {
 		m.logger.Info("failed to calculate prices for price feeds", zap.Strings("missing_prices", missingPrices))
+		m.metrics.MissingPrices(missingPrices)
 	}
 	m.indexPrices = indexPrices
 	m.scaledPrices = scaledPrices

pkg/math/oracle/utils.go

@@ -53,7 +53,7 @@ func (m *IndexPriceAggregator) GetIndexPrice(
 	return price, nil
 }
 
-// SetIndexPrice sets the index price for the given currency pair.
+// SetIndexPrices sets the index price for the given currency pair.
 func (m *IndexPriceAggregator) SetIndexPrices(
 	prices types.Prices,
 ) {

service/clients/prometheus/client.go

@@ -0,0 +1,50 @@
+package prometheus
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/prometheus/client_golang/api"
+	v1 "github.com/prometheus/client_golang/api/prometheus/v1"
+	"go.uber.org/zap"
+
+	libhttp "github.com/skip-mev/connect/v2/pkg/http"
+)
+
+type Client struct {
+	v1.API
+	logger *zap.Logger
+}
+
+func NewClient(address string, logger *zap.Logger) (Client, error) {
+	if logger == nil {
+		logger = zap.NewNop()
+	}
+
+	logger = logger.With(zap.String("service", "prometheus_client"))
+	logger.Info("creating prometheus client", zap.String("address", address))
+
+	// get the prometheus server address
+	if address == "" || !libhttp.IsValidAddress(address) {
+		return Client{}, fmt.Errorf("invalid prometheus server address: %s", address)
+	}
+
+	const httpPrefix = "http://"
+	if !strings.HasPrefix(address, httpPrefix) {
+		address = httpPrefix + address
+	}
+
+	// Create a Prometheus API client
+	client, err := api.NewClient(api.Config{
+		Address: address, // Address of your Prometheus server
+	})
+	if err != nil {
+		return Client{}, fmt.Errorf("failed to create prometheus client: %w", err)
+	}
+
+	// Create a new Prometheus API v1 client
+	return Client{
+		API:    v1.NewAPI(client),
+		logger: logger,
+	}, nil
+}

service/servers/prometheus/server.go

@@ -3,14 +3,14 @@
 import (
 	"errors"
 	"fmt"
-	"net"
 	"net/http"
 	"time"
 
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"go.uber.org/zap"
 
+	libhttp "github.com/skip-mev/connect/v2/pkg/http"
 	"github.com/skip-mev/connect/v2/pkg/sync"
 )
 
@@ -34,7 +34,7 @@
 // address is set, and valid. Notice, this method does not start the server.
 func NewPrometheusServer(prometheusAddress string, logger *zap.Logger) (*PrometheusServer, error) {
 	// get the prometheus server address
-	if prometheusAddress == "" || !isValidAddress(prometheusAddress) {
+	if prometheusAddress == "" || !libhttp.IsValidAddress(prometheusAddress) {
 		return nil, fmt.Errorf("invalid prometheus server address: %s", prometheusAddress)
 	}
 	srv := &http.Server{
@@ -79,16 +79,3 @@
 	// close the done channel
 	close(ps.done)
 }
-
-func isValidAddress(address string) bool {
-	host, port, err := net.SplitHostPort(address)
-	if err != nil {
-		return false
-	}
-
-	if host == "" || port == "" {
-		return false
-	}
-
-	return true
-}