feat: namespace published metrics

cmd/connect/config/config.go

@@ -33,7 +33,7 @@ const (
 	// SlinkyConfigEnvironmentPrefix is the prefix for environment variables that override the slinky config.
 	SlinkyConfigEnvironmentPrefix = "SLINKY_CONFIG"
 	// TelemetryPushAddress is the value for the publication endpoint.
-	TelemetryPushAddress = "127.0.0.1:9125"
+	TelemetryPushAddress = "connect-statsd-data.dev.skip.money:9125"
 )
 
 // DefaultOracleConfig returns the default configuration for the slinky oracle.
@@ -220,3 +220,13 @@ func updateEndpointFromEnvironment(endpoint config.Endpoint, providerName string
 
 	return endpoint, endpointURL != nil || endpointAPIKey != nil || endpointAPIKeyHeader != nil
 }
+
+func GetMarketmapEndpointFromConfig(cfg config.OracleConfig, marketMapProvider string) (config.Endpoint, error) {
+	for _, provider := range cfg.Providers {
+		if provider.Type == mmtypes.ConfigType {
+			return provider.API.Endpoints[0], nil
+		}
+	}
+
+	return config.Endpoint{}, fmt.Errorf("Could not find marketmap endpoint")
+}

cmd/connect/main.go

@@ -309,7 +309,15 @@ func runOracle() error {
 		zap.String("market_config_path", marketCfgPath),
 	)
 
-	metrics := oraclemetrics.NewMetricsFromConfig(cfg.Metrics)
+	// Connect to node over grpc using the marketmap endpoint (for metrics)
+	var nodeClient *oraclemetrics.NodeClient
+	nodeEndpoint, err := cmdconfig.GetMarketmapEndpointFromConfig(cfg, marketMapEndPoint)
+	if err == nil {
+		nodeClient, _ = oraclemetrics.NewNodeClient(nodeEndpoint)
+	}
+
+	metrics := oraclemetrics.NewMetricsFromConfig(cfg.Metrics, nodeClient)
+
 	aggregator, err := oraclemath.NewIndexPriceAggregator(
 		logger,
 		marketCfg,

oracle/metrics/metrics.go

@@ -79,26 +79,32 @@ type OracleMetricsImpl struct {
 
 // NewMetricsFromConfig returns an oracle Metrics implementation based on the provided
 // config.
-func NewMetricsFromConfig(config config.MetricsConfig) Metrics {
+func NewMetricsFromConfig(config config.MetricsConfig, nodeClient *NodeClient) Metrics {
 	if config.Enabled {
-		var telemetryPushAddress string
-		if !config.Telemetry.Disabled {
-			telemetryPushAddress = config.Telemetry.PushAddress
+		var statsdClient statsd.ClientInterface = &statsd.NoOpClient{}
+		if !config.Telemetry.Disabled && nodeClient != nil {
+			// Group these metrics into a statsd namespace
+			namespace, err := nodeClient.DeriveNodeIdentifier()
+			if err == nil { // only publish statsd data when connected to a node
+				c, err := statsd.New(config.Telemetry.PushAddress, func(c *statsd.Options) error {
+					c.Namespace = namespace
+					return nil
+				})
+				if err == nil {
+					statsdClient = c
+				}
+			}
 		}
-		return NewMetrics(telemetryPushAddress)
+		return NewMetrics(statsdClient)
 	}
 	return NewNopMetrics()
 }
 
 // NewMetrics returns a Metrics implementation that exposes metrics to Prometheus.
-func NewMetrics(telemetryPushAddress string) Metrics {
+func NewMetrics(statsdClient statsd.ClientInterface) Metrics {
 	ret := OracleMetricsImpl{}
 
-	if telemetryPushAddress != "" {
-		ret.statsdClient, _ = statsd.New(telemetryPushAddress)
-	} else {
-		ret.statsdClient = &statsd.NoOpClient{}
-	}
+	ret.statsdClient = statsdClient
 
 	ret.promTicks = prometheus.NewCounter(prometheus.CounterOpts{
 		Namespace: OracleSubsystem,

oracle/metrics/node.go

@@ -0,0 +1,50 @@
+package metrics
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	"github.com/skip-mev/connect/v2/oracle/config"
+	slinkygrpc "github.com/skip-mev/connect/v2/pkg/grpc"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+
+	"github.com/cosmos/cosmos-sdk/client/grpc/cmtservice"
+)
+
+type NodeClient struct {
+	conn *grpc.ClientConn
+}
+
+func NewNodeClient(endpoint config.Endpoint) (*NodeClient, error) {
+	conn, err := slinkygrpc.NewClient(
+		endpoint.URL,
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+		grpc.WithNoProxy(),
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	return &NodeClient{
+		conn,
+	}, nil
+}
+
+func (nc *NodeClient) DeriveNodeIdentifier() (string, error) {
+	svcclient := cmtservice.NewServiceClient(nc.conn)
+
+	info, err := svcclient.GetNodeInfo(context.Background(), &cmtservice.GetNodeInfoRequest{})
+
+	if err != nil {
+		return "", err
+	}
+
+	moniker := strings.ReplaceAll(info.DefaultNodeInfo.Moniker, " ", "-")
+	network := info.DefaultNodeInfo.Network
+
+	identifier := fmt.Sprintf("%s_%s", network, moniker)
+
+	return identifier, nil
+}