Move churner to new metrics framework.

Signed-off-by: Cody Littley <[email protected]>

common/metrics/metrics_server.go

@@ -145,6 +145,8 @@ func (m *metrics) Start() error {
 	}
 	m.started = true
 
+	m.logger.Infof("Starting metrics server at port %d", m.config.HTTPPort)
+
 	go func() {
 		err := m.server.ListenAndServe()
 		if err != nil && !strings.Contains(err.Error(), "http: Server closed") {

operators/churner/churner.go

@@ -232,7 +232,10 @@ func (c *churner) getOperatorsToChurn(ctx context.Context, quorumIDs []uint8, op
 		// For example, when churnBIPsOfOperatorStake=11000, the operator trying to
 		// register needs to have 1.1 times the stake of the lowest-stake operator.
 		if new(big.Int).Mul(lowestStake, churnBIPsOfOperatorStake).Cmp(new(big.Int).Mul(operatorToRegisterStake, bipMultiplier)) >= 0 {
-			c.metrics.IncrementFailedRequestNum("getOperatorsToChurn", FailReasonInsufficientStakeToRegister)
+			e := c.metrics.IncrementFailedRequestNum("getOperatorsToChurn", FailReasonInsufficientStakeToRegister)
+			if e != nil {
+				c.logger.Error("Failed to increment failed request num", "error", e)
+			}
 			msg := "registering operator must have %f%% more than the stake of the " +
 				"lowest-stake operator. Block number used for this decision: %d, " +
 				"registering operator address: %s, registering operator stake: %d, " +
@@ -249,7 +252,10 @@ func (c *churner) getOperatorsToChurn(ctx context.Context, quorumIDs []uint8, op
 		// (i.e. the lowest-stake operator) needs to have less than 10.01% of the total
 		// stake.
 		if new(big.Int).Mul(lowestStake, bipMultiplier).Cmp(new(big.Int).Mul(totalStake, churnBIPsOfTotalStake)) >= 0 {
-			c.metrics.IncrementFailedRequestNum("getOperatorsToChurn", FailReasonInsufficientStakeToChurn)
+			e := c.metrics.IncrementFailedRequestNum("getOperatorsToChurn", FailReasonInsufficientStakeToChurn)
+			if e != nil {
+				c.logger.Error("Failed to increment failed request num", "error", e)
+			}
 			msg := "operator to churn out must have less than %f%% of the total stake. " +
 				"Block number used for this decision: %d, operatorId of the operator " +
 				"to churn: %x, stake of the operator to churn: %d, total stake in " +

operators/churner/churner_test.go

@@ -28,7 +28,8 @@ func TestProcessChurnRequest(t *testing.T) {
 			NumRetries:       numRetries,
 		},
 	}
-	metrics := churner.NewMetrics("9001", logger)
+	metrics, err := churner.NewMetrics(9001, logger)
+	assert.NoError(t, err)
 	cn, err := churner.NewChurner(config, mockIndexer, transactorMock, logger, metrics)
 	assert.NoError(t, err)
 	assert.NotNil(t, cn)

operators/churner/cmd/main.go

@@ -86,7 +86,10 @@ func run(ctx *cli.Context) error {
 	logger.Info("Connecting to subgraph", "url", config.ChainStateConfig.Endpoint)
 	indexer := thegraph.MakeIndexedChainState(config.ChainStateConfig, cs, logger)
 
-	metrics := churner.NewMetrics(config.MetricsConfig.HTTPPort, logger)
+	metrics, err := churner.NewMetrics(config.MetricsConfig.HTTPPort, logger)
+	if err != nil {
+		log.Fatalf("failed to create metrics: %v", err)
+	}
 
 	cn, err := churner.NewChurner(config, indexer, tx, logger, metrics)
 	if err != nil {

operators/churner/config.go

@@ -37,7 +37,7 @@ func NewConfig(ctx *cli.Context) (*Config, error) {
 		PerPublicKeyRateLimit:         ctx.GlobalDuration(flags.PerPublicKeyRateLimit.Name),
 		ChurnApprovalInterval:         ctx.GlobalDuration(flags.ChurnApprovalInterval.Name),
 		MetricsConfig: MetricsConfig{
-			HTTPPort:      ctx.GlobalString(flags.MetricsHTTPPort.Name),
+			HTTPPort:      ctx.GlobalInt(flags.MetricsHTTPPort.Name),
 			EnableMetrics: ctx.GlobalBool(flags.EnableMetrics.Name),
 		},
 	}, nil

operators/churner/flags/flags.go

@@ -58,11 +58,11 @@ var (
 		EnvVar:   common.PrefixEnvVar(envPrefix, "ENABLE_METRICS"),
 	}
 	/* Optional Flags*/
-	MetricsHTTPPort = cli.StringFlag{
+	MetricsHTTPPort = cli.IntFlag{
 		Name:     common.PrefixFlag(FlagPrefix, "metrics-http-port"),
 		Usage:    "the http port which the metrics prometheus server is listening",
 		Required: false,
-		Value:    "9100",
+		Value:    9100,
 		EnvVar:   common.PrefixEnvVar(envPrefix, "METRICS_HTTP_PORT"),
 	}
 	ChurnApprovalInterval = cli.DurationFlag{

operators/churner/metrics.go

@@ -1,15 +1,12 @@
 package churner
 
 import (
-	"context"
-	"fmt"
-	"net/http"
+	"github.com/Layr-Labs/eigenda/common/metrics"
+	"time"
 
 	"github.com/Layr-Labs/eigensdk-go/logging"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/collectors"
-	"github.com/prometheus/client_golang/prometheus/promauto"
-	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"google.golang.org/grpc/codes"
 )
 
@@ -28,7 +25,7 @@ const (
 )
 
 // Note: statusCodeMap must be maintained in sync with failure reason constants.
-var statusCodeMap map[FailReason]string = map[FailReason]string{
+var statusCodeMap = map[FailReason]string{
 	FailReasonRateLimitExceeded:           codes.ResourceExhausted.String(),
 	FailReasonInsufficientStakeToRegister: codes.InvalidArgument.String(),
 	FailReasonInsufficientStakeToChurn:    codes.InvalidArgument.String(),
@@ -40,93 +37,91 @@ var statusCodeMap map[FailReason]string = map[FailReason]string{
 }
 
 type MetricsConfig struct {
-	HTTPPort      string
+	HTTPPort      int
 	EnableMetrics bool
 }
 
 type Metrics struct {
-	registry *prometheus.Registry
+	metricsServer metrics.Metrics
 
-	NumRequests *prometheus.CounterVec
-	Latency     *prometheus.SummaryVec
+	numRequests metrics.CountMetric
+	latency     metrics.LatencyMetric
 
-	httpPort string
-	logger   logging.Logger
+	logger logging.Logger
 }
 
-func NewMetrics(httpPort string, logger logging.Logger) *Metrics {
-	namespace := "eigenda_churner"
+type latencyLabel struct {
+	method string
+}
+
+type numRequestsLabel struct {
+	status string
+	method string
+	reason string
+}
+
+func NewMetrics(httpPort int, logger logging.Logger) (*Metrics, error) {
 	reg := prometheus.NewRegistry()
 	reg.MustRegister(collectors.NewProcessCollector(collectors.ProcessCollectorOpts{}))
 	reg.MustRegister(collectors.NewGoCollector())
 
-	metrics := &Metrics{
-		NumRequests: promauto.With(reg).NewCounterVec(
-			prometheus.CounterOpts{
-				Namespace: namespace,
-				Name:      "requests",
-				Help:      "the number of requests",
-			},
-			[]string{"status", "reason", "method"},
-		),
-		Latency: promauto.With(reg).NewSummaryVec(
-			prometheus.SummaryOpts{
-				Namespace:  namespace,
-				Name:       "latency_ms",
-				Help:       "latency summary in milliseconds",
-				Objectives: map[float64]float64{0.5: 0.05, 0.9: 0.01, 0.95: 0.01, 0.99: 0.001},
-			},
-			[]string{"method"},
-		),
-		registry: reg,
-		httpPort: httpPort,
-		logger:   logger.With("component", "ChurnerMetrics"),
+	metricsServer := metrics.NewMetrics(logger, &metrics.Config{
+		Namespace: "eigenda_churner",
+		HTTPPort:  httpPort,
+	})
+
+	numRequests, err := metricsServer.NewCountMetric(
+		"requests",
+		"the number of requests",
+		numRequestsLabel{})
+	if err != nil {
+		return nil, err
 	}
-	return metrics
+
+	latency, err := metricsServer.NewLatencyMetric(
+		"latency",
+		"latency summary in milliseconds",
+		latencyLabel{},
+		&metrics.Quantile{Quantile: 0.5, Error: 0.05},
+		&metrics.Quantile{Quantile: 0.9, Error: 0.01},
+		&metrics.Quantile{Quantile: 0.95, Error: 0.01},
+		&metrics.Quantile{Quantile: 0.99, Error: 0.001})
+	if err != nil {
+		return nil, err
+	}
+
+	return &Metrics{
+		metricsServer: metricsServer,
+		numRequests:   numRequests,
+		latency:       latency,
+		logger:        logger.With("component", "ChurnerMetrics"),
+	}, nil
 }
 
-// ObserveLatency observes the latency of a stage in 'stage
-func (g *Metrics) ObserveLatency(method string, latencyMs float64) {
-	g.Latency.WithLabelValues(method).Observe(latencyMs)
+// ObserveLatency observes the latency of a stage
+func (g *Metrics) ObserveLatency(method string, latency time.Duration) error {
+	return g.latency.ReportLatency(latency, latencyLabel{method: method})
 }
 
 // IncrementSuccessfulRequestNum increments the number of successful requests
-func (g *Metrics) IncrementSuccessfulRequestNum(method string) {
-	g.NumRequests.With(prometheus.Labels{
-		"status": "success",
-		"method": method,
-		"reason": "",
-	}).Inc()
+func (g *Metrics) IncrementSuccessfulRequestNum(method string) error {
+	return g.numRequests.Increment(numRequestsLabel{status: "success", method: method})
 }
 
 // IncrementFailedRequestNum increments the number of failed requests
-func (g *Metrics) IncrementFailedRequestNum(method string, reason FailReason) {
+func (g *Metrics) IncrementFailedRequestNum(method string, reason FailReason) error {
 	code, ok := statusCodeMap[reason]
 	if !ok {
 		g.logger.Error("cannot map failure reason to status code", "failure reason", reason)
 		// Treat this as an internal server error. This is a conservative approach to
 		// handle a negligence of mapping from failure reason to status code.
 		code = codes.Internal.String()
 	}
-	g.NumRequests.With(prometheus.Labels{
-		"status": code,
-		"reason": string(reason),
-		"method": method,
-	}).Inc()
+
+	return g.numRequests.Increment(numRequestsLabel{status: code, reason: string(reason), method: method})
 }
 
 // Start starts the metrics server
-func (g *Metrics) Start(ctx context.Context) {
-	g.logger.Info("Starting metrics server at ", "port", g.httpPort)
-	addr := fmt.Sprintf(":%s", g.httpPort)
-	go func() {
-		log := g.logger
-		mux := http.NewServeMux()
-		mux.Handle("/metrics", promhttp.HandlerFor(
-			g.registry,
-			promhttp.HandlerOpts{},
-		))
-		err := http.ListenAndServe(addr, mux)
-		log.Error("Prometheus server failed", "err", err)
-	}()
+func (g *Metrics) Start() error {
+	return g.metricsServer.Start()
 }

operators/churner/server.go

@@ -47,8 +47,12 @@ func NewServer(
 func (s *Server) Start(metricsConfig MetricsConfig) error {
 	// Enable Metrics Block
 	if metricsConfig.EnableMetrics {
-		httpSocket := fmt.Sprintf(":%s", metricsConfig.HTTPPort)
-		s.metrics.Start(context.Background())
+		httpSocket := fmt.Sprintf(":%d", metricsConfig.HTTPPort)
+		err := s.metrics.Start()
+		if err != nil {
+			return fmt.Errorf("failed to start metrics server: %w", err)
+		}
+
 		s.logger.Info("Enabled metrics for Churner", "socket", httpSocket)
 	}
 	return nil
@@ -57,39 +61,57 @@ func (s *Server) Start(metricsConfig MetricsConfig) error {
 func (s *Server) Churn(ctx context.Context, req *pb.ChurnRequest) (*pb.ChurnReply, error) {
 	err := s.validateChurnRequest(ctx, req)
 	if err != nil {
-		s.metrics.IncrementFailedRequestNum("Churn", FailReasonInvalidRequest)
+		e := s.metrics.IncrementFailedRequestNum("Churn", FailReasonInvalidRequest)
+		if e != nil {
+			s.logger.Error("Failed to increment failed request num", "error", e)
+		}
 		return nil, api.NewErrorInvalidArg(fmt.Sprintf("invalid request: %s", err.Error()))
 	}
 
 	timer := prometheus.NewTimer(prometheus.ObserverFunc(func(f float64) {
-		s.metrics.ObserveLatency("Churn", f*1000) // make milliseconds
+		e := s.metrics.ObserveLatency("Churn", time.Duration(f*float64(time.Second)))
+		if e != nil {
+			s.logger.Error("Failed to increment failed request num", "error", e)
+		}
 	}))
 	defer timer.ObserveDuration()
 	s.logger.Info("Received request: ", "QuorumIds", req.GetQuorumIds())
 
 	now := time.Now()
 	// Global rate limiting: check that we are after the previous approval's expiry
 	if now.Unix() < s.latestExpiry {
-		s.metrics.IncrementFailedRequestNum("Churn", FailReasonPrevApprovalNotExpired)
+		e := s.metrics.IncrementFailedRequestNum("Churn", FailReasonPrevApprovalNotExpired)
+		if e != nil {
+			s.logger.Error("Failed to increment failed request num", "error", e)
+		}
 		return nil, api.NewErrorResourceExhausted(fmt.Sprintf("previous approval not expired, retry in %d seconds", s.latestExpiry-now.Unix()))
 	}
 
 	request, err := createChurnRequest(req)
 	if err != nil {
-		s.metrics.IncrementFailedRequestNum("Churn", FailReasonInvalidRequest)
+		e := s.metrics.IncrementFailedRequestNum("Churn", FailReasonInvalidRequest)
+		if e != nil {
+			s.logger.Error("Failed to increment failed request num", "error", e)
+		}
 		return nil, api.NewErrorInvalidArg(err.Error())
 	}
 
 	operatorToRegisterAddress, err := s.churner.VerifyRequestSignature(ctx, request)
 	if err != nil {
-		s.metrics.IncrementFailedRequestNum("Churn", FailReasonInvalidSignature)
+		e := s.metrics.IncrementFailedRequestNum("Churn", FailReasonInvalidSignature)
+		if e != nil {
+			s.logger.Error("Failed to increment failed request num", "error", e)
+		}
 		return nil, api.NewErrorInvalidArg(fmt.Sprintf("failed to verify request signature: %s", err.Error()))
 	}
 
 	// Per-operator rate limiting: check if the request should be rate limited
 	err = s.checkShouldBeRateLimited(now, *request)
 	if err != nil {
-		s.metrics.IncrementFailedRequestNum("Churn", FailReasonRateLimitExceeded)
+		e := s.metrics.IncrementFailedRequestNum("Churn", FailReasonRateLimitExceeded)
+		if e != nil {
+			s.logger.Error("Failed to increment failed request num", "error", e)
+		}
 		return nil, api.NewErrorResourceExhausted(fmt.Sprintf("rate limiter error: %s", err.Error()))
 	}
 
@@ -98,7 +120,10 @@ func (s *Server) Churn(ctx context.Context, req *pb.ChurnRequest) (*pb.ChurnRepl
 		if _, ok := status.FromError(err); ok {
 			return nil, err
 		}
-		s.metrics.IncrementFailedRequestNum("Churn", FailReasonProcessChurnRequestFailed)
+		e := s.metrics.IncrementFailedRequestNum("Churn", FailReasonProcessChurnRequestFailed)
+		if e != nil {
+			s.logger.Error("Failed to increment failed request num", "error", e)
+		}
 		return nil, api.NewErrorInternal(fmt.Sprintf("failed to process churn request: %s", err.Error()))
 	}
 
@@ -107,7 +132,10 @@ func (s *Server) Churn(ctx context.Context, req *pb.ChurnRequest) (*pb.ChurnRepl
 
 	operatorsToChurn := convertToOperatorsToChurnGrpc(response.OperatorsToChurn)
 
-	s.metrics.IncrementSuccessfulRequestNum("Churn")
+	e := s.metrics.IncrementSuccessfulRequestNum("Churn")
+	if e != nil {
+		s.logger.Error("Failed to increment failed request num", "error", e)
+	}
 	return &pb.ChurnReply{
 		SignatureWithSaltAndExpiry: &pb.SignatureWithSaltAndExpiry{
 			Signature: response.SignatureWithSaltAndExpiry.Signature,

operators/churner/server_test.go

@@ -181,7 +181,8 @@ func newTestServer(t *testing.T) *churner.Server {
 
 	setupMockWriter()
 
-	metrics := churner.NewMetrics("9001", logger)
+	metrics, err := churner.NewMetrics(9001, logger)
+	assert.NoError(t, err)
 	cn, err := churner.NewChurner(config, mockIndexer, transactorMock, logger, metrics)
 	if err != nil {
 		log.Fatalln("cannot create churner", err)

operators/churner/tests/churner_test.go

@@ -225,7 +225,8 @@ func newTestServer(t *testing.T) *churner.Server {
 	)
 	assert.NoError(t, err)
 
-	metrics := churner.NewMetrics("9001", logger)
+	metrics, err := churner.NewMetrics(9001, logger)
+	assert.NoError(t, err)
 	cn, err := churner.NewChurner(config, mockIndexer, operatorTransactorChurner, logger, metrics)
 	assert.NoError(t, err)