metrics for signature-aggregator

main/main.go

@@ -27,6 +27,7 @@ import (
 	"github.com/ava-labs/awm-relayer/peers"
 	"github.com/ava-labs/awm-relayer/relayer"
 	"github.com/ava-labs/awm-relayer/signature-aggregator/aggregator"
+	sigAggMetrics "github.com/ava-labs/awm-relayer/signature-aggregator/metrics"
 	"github.com/ava-labs/awm-relayer/utils"
 	"github.com/ava-labs/awm-relayer/vms"
 	"github.com/ava-labs/subnet-evm/ethclient"
@@ -210,7 +211,14 @@ func main() {
 		panic(err)
 	}
 
-	signatureAggregator := aggregator.NewSignatureAggregator(network, logger, messageCreator)
+	signatureAggregator := aggregator.NewSignatureAggregator(
+		network,
+		logger,
+		sigAggMetrics.NewSignatureAggregatorMetrics(
+			prometheus.DefaultRegisterer,
+		),
+		messageCreator,
+	)
 
 	applicationRelayers, minHeights, err := createApplicationRelayers(
 		context.Background(),

signature-aggregator/README.md

@@ -17,6 +17,7 @@ Namely:
 - `PChainAPI`: APIConfig
 - `InfoAPI` : APIConfig
 - `APIPort` : (optional) defaults to 8080
+- `MetricsPort`: (optional) defaults to 8081
 
 Sample config that can be used for local testing is `signature-aggregator/sample-signature-aggregator-config.json`
 

signature-aggregator/aggregator/aggregator.go

@@ -23,6 +23,7 @@ import (
 	"github.com/ava-labs/avalanchego/utils/set"
 	avalancheWarp "github.com/ava-labs/avalanchego/vms/platformvm/warp"
 	"github.com/ava-labs/awm-relayer/peers"
+	"github.com/ava-labs/awm-relayer/signature-aggregator/metrics"
 	"github.com/ava-labs/awm-relayer/utils"
 	coreEthMsg "github.com/ava-labs/coreth/plugin/evm/message"
 	msg "github.com/ava-labs/subnet-evm/plugin/evm/message"
@@ -56,17 +57,20 @@ type SignatureAggregator struct {
 	messageCreator          message.Creator
 	currentRequestID        atomic.Uint32
 	subnetsMapLock          sync.RWMutex
+	metrics                 *metrics.SignatureAggregatorMetrics
 }
 
 func NewSignatureAggregator(
 	network *peers.AppRequestNetwork,
 	logger logging.Logger,
+	metrics *metrics.SignatureAggregatorMetrics,
 	messageCreator message.Creator,
 ) *SignatureAggregator {
 	sa := SignatureAggregator{
 		network:                 network,
 		subnetIDsByBlockchainID: map[ids.ID]ids.ID{},
 		logger:                  logger,
+		metrics:                 metrics,
 		messageCreator:          messageCreator,
 		currentRequestID:        atomic.Uint32{},
 	}
@@ -106,6 +110,7 @@ func (s *SignatureAggregator) AggregateSignaturesAppRequest(
 			zap.String("warpMessageID", unsignedMessage.ID().String()),
 			zap.Error(err),
 		)
+		s.metrics.FailuresToGetValidatorSet.Inc()
 		return nil, fmt.Errorf("%s: %w", msg, err)
 	}
 	if !utils.CheckStakeWeightPercentageExceedsThreshold(
@@ -119,6 +124,7 @@ func (s *SignatureAggregator) AggregateSignaturesAppRequest(
 			zap.Uint64("totalValidatorWeight", connectedValidators.TotalValidatorWeight),
 			zap.Uint64("quorumPercentage", quorumPercentage),
 		)
+		s.metrics.FailuresToConnectToSufficientStake.Inc()
 		return nil, errNotEnoughConnectedStake
 	}
 
@@ -224,6 +230,7 @@ func (s *SignatureAggregator) AggregateSignaturesAppRequest(
 					zap.Error(err),
 				)
 				responsesExpected--
+				s.metrics.FailuresSendingToNode.Inc()
 			}
 		}
 
@@ -247,6 +254,8 @@ func (s *SignatureAggregator) AggregateSignaturesAppRequest(
 					quorumPercentage,
 				)
 				if err != nil {
+					// don't increase node failures metric here, because we did
+					// it in handleResponse
 					return nil, fmt.Errorf(
 						"failed to handle response: %w",
 						err,
@@ -345,6 +354,7 @@ func (s *SignatureAggregator) handleResponse(
 	// This is still a relevant response, since we are no longer expecting a response from that node.
 	if response.Op() == message.AppErrorOp {
 		s.logger.Debug("Request timed out")
+		s.metrics.ValidatorTimeouts.Inc()
 		return nil, true, nil
 	}
 
@@ -368,6 +378,7 @@ func (s *SignatureAggregator) handleResponse(
 			zap.String("warpMessageID", unsignedMessage.ID().String()),
 			zap.String("sourceBlockchainID", unsignedMessage.SourceChainID.String()),
 		)
+		s.metrics.InvalidSignatureResponses.Inc()
 		return nil, true, nil
 	}
 

signature-aggregator/api/api.go

@@ -8,10 +8,12 @@ import (
 	"encoding/json"
 	"net/http"
 	"strings"
+	"time"
 
 	"github.com/ava-labs/avalanchego/ids"
 	"github.com/ava-labs/avalanchego/utils/logging"
 	"github.com/ava-labs/awm-relayer/signature-aggregator/aggregator"
+	"github.com/ava-labs/awm-relayer/signature-aggregator/metrics"
 	"github.com/ava-labs/awm-relayer/types"
 	"github.com/ava-labs/awm-relayer/utils"
 	"go.uber.org/zap"
@@ -40,9 +42,17 @@ type AggregateSignatureResponse struct {
 
 func HandleAggregateSignaturesByRawMsgRequest(
 	logger logging.Logger,
+	metrics *metrics.SignatureAggregatorMetrics,
 	signatureAggregator *aggregator.SignatureAggregator,
 ) {
-	http.Handle(APIPath, signatureAggregationAPIHandler(logger, signatureAggregator))
+	http.Handle(
+		APIPath,
+		signatureAggregationAPIHandler(
+			logger,
+			metrics,
+			signatureAggregator,
+		),
+	)
 }
 
 func writeJSONError(
@@ -65,8 +75,15 @@ func writeJSONError(
 	}
 }
 
-func signatureAggregationAPIHandler(logger logging.Logger, aggregator *aggregator.SignatureAggregator) http.Handler {
+func signatureAggregationAPIHandler(
+	logger logging.Logger,
+	metrics *metrics.SignatureAggregatorMetrics,
+	aggregator *aggregator.SignatureAggregator,
+) http.Handler {
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		metrics.AggregateSignaturesRequestCount.Inc()
+		startTime := time.Now()
+
 		var req AggregateSignatureRequest
 		err := json.NewDecoder(r.Body).Decode(&req)
 		if err != nil {
@@ -150,5 +167,8 @@ func signatureAggregationAPIHandler(logger logging.Logger, aggregator *aggregato
 		if err != nil {
 			logger.Error("Error writing response", zap.Error(err))
 		}
+		metrics.AggregateSignaturesLatencyMS.Set(
+			float64(time.Since(startTime).Milliseconds()),
+		)
 	})
 }

signature-aggregator/config/config.go

@@ -11,7 +11,8 @@ import (
 )
 
 const (
-	defaultAPIPort = uint16(8080)
+	defaultAPIPort     = uint16(8080)
+	defaultMetricsPort = uint16(8081)
 )
 
 var defaultLogLevel = logging.Info.String()
@@ -28,6 +29,8 @@ type Config struct {
 	PChainAPI *baseCfg.APIConfig `mapstructure:"p-chain-api" json:"p-chain-api"`
 	InfoAPI   *baseCfg.APIConfig `mapstructure:"info-api" json:"info-api"`
 	APIPort   uint16             `mapstructure:"api-port" json:"api-port"`
+
+	MetricsPort uint16 `mapstructure:"metrics-port" json:"metrics-port"`
 }
 
 func DisplayUsageText() {

signature-aggregator/config/keys.go

@@ -14,5 +14,6 @@ const (
 	PChainAPIKey         = "p-chain-api"
 	InfoAPIKey           = "info-api"
 	APIPortKey           = "api-port"
+	MetricsPortKey       = "metrics-port"
 	SourceBlockchainsKey = "source-blockchains"
 )

signature-aggregator/config/viper.go

@@ -52,6 +52,7 @@ func BuildViper(fs *pflag.FlagSet) (*viper.Viper, error) {
 func SetDefaultConfigValues(v *viper.Viper) {
 	v.SetDefault(LogLevelKey, defaultLogLevel)
 	v.SetDefault(APIPortKey, defaultAPIPort)
+	v.SetDefault(MetricsPortKey, defaultMetricsPort)
 }
 
 // BuildConfig constructs the signature aggregator config using Viper.

signature-aggregator/main/main.go

@@ -17,6 +17,7 @@ import (
 	"github.com/ava-labs/awm-relayer/signature-aggregator/aggregator"
 	"github.com/ava-labs/awm-relayer/signature-aggregator/api"
 	"github.com/ava-labs/awm-relayer/signature-aggregator/config"
+	"github.com/ava-labs/awm-relayer/signature-aggregator/metrics"
 	"github.com/prometheus/client_golang/prometheus"
 	"go.uber.org/zap"
 )
@@ -105,9 +106,22 @@ func main() {
 		logger.Fatal("Failed to create message creator", zap.Error(err))
 		panic(err)
 	}
-	signatureAggregator := aggregator.NewSignatureAggregator(network, logger, messageCreator)
 
-	api.HandleAggregateSignaturesByRawMsgRequest(logger, signatureAggregator)
+	registry := metrics.Initialize(cfg.MetricsPort)
+	metricsInstance := metrics.NewSignatureAggregatorMetrics(registry)
+
+	signatureAggregator := aggregator.NewSignatureAggregator(
+		network,
+		logger,
+		metricsInstance,
+		messageCreator,
+	)
+
+	api.HandleAggregateSignaturesByRawMsgRequest(
+		logger,
+		metricsInstance,
+		signatureAggregator,
+	)
 
 	err = http.ListenAndServe(fmt.Sprintf(":%d", cfg.APIPort), nil)
 	if errors.Is(err, http.ErrServerClosed) {

signature-aggregator/metrics/metrics.go

@@ -0,0 +1,150 @@
+// Copyright (C) 2023, Ava Labs, Inc. All rights reserved.
+// See the file LICENSE for licensing terms.
+
+package metrics
+
+import (
+	"errors"
+	"fmt"
+	"log"
+	"net/http"
+
+	"github.com/ava-labs/avalanchego/api/metrics"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+)
+
+var (
+	ErrFailedToCreateSignatureAggregatorMetrics = errors.New(
+		"failed to create signature aggregator metrics",
+	)
+)
+
+var Opts = struct {
+	AggregateSignaturesLatencyMS       prometheus.GaugeOpts
+	AggregateSignaturesRequestCount    prometheus.CounterOpts
+	FailuresToGetValidatorSet          prometheus.CounterOpts
+	FailuresToConnectToSufficientStake prometheus.CounterOpts
+	FailuresSendingToNode              prometheus.CounterOpts
+	ValidatorTimeouts                  prometheus.CounterOpts
+	InvalidSignatureResponses          prometheus.CounterOpts
+}{
+	AggregateSignaturesLatencyMS: prometheus.GaugeOpts{
+		Name: "agg_sigs_latency_ms",
+		Help: "Latency of requests for aggregate signatures",
+	},
+	AggregateSignaturesRequestCount: prometheus.CounterOpts{
+		Name: "agg_sigs_req_count",
+		Help: "Number of requests for aggregate signatures",
+	},
+	FailuresToGetValidatorSet: prometheus.CounterOpts{
+		Name: "failures_to_get_validator_set",
+		Help: "Number of failed attempts to retrieve the validator set",
+	},
+	FailuresToConnectToSufficientStake: prometheus.CounterOpts{
+		Name: "failures_to_connect_to_sufficient_stake",
+		Help: "Number of incidents of connecting to some validators but not enough stake weight",
+	},
+	FailuresSendingToNode: prometheus.CounterOpts{
+		Name: "failures_sending_to_node",
+		Help: "Number of failures to send a request to a validator node",
+	},
+	ValidatorTimeouts: prometheus.CounterOpts{
+		Name: "validator_timeouts",
+		Help: "Number of timeouts while waiting for a validator to respond to a request",
+	},
+	InvalidSignatureResponses: prometheus.CounterOpts{
+		Name: "invalid_signature_responses",
+		Help: "Number of responses from validators that were not valid signatures",
+	},
+}
+
+type SignatureAggregatorMetrics struct {
+	AggregateSignaturesLatencyMS       prometheus.Gauge
+	AggregateSignaturesRequestCount    prometheus.Counter
+	FailuresToGetValidatorSet          prometheus.Counter
+	FailuresToConnectToSufficientStake prometheus.Counter
+	FailuresSendingToNode              prometheus.Counter
+	ValidatorTimeouts                  prometheus.Counter
+	InvalidSignatureResponses          prometheus.Counter
+
+	// TODO: consider other failures to monitor. Issue #384 requires
+	// "network failures", but we probably don't handle those directly.
+	// Surely there are some error types specific to this layer that we can
+	// count.
+
+	// TODO: consider how the relayer keeps separate counts of aggregations
+	// by AppRequest vs by Warp API and whether we should have such counts.
+}
+
+func NewSignatureAggregatorMetrics(
+	registerer prometheus.Registerer,
+) *SignatureAggregatorMetrics {
+	m := SignatureAggregatorMetrics{
+		AggregateSignaturesLatencyMS: prometheus.NewGauge(
+			Opts.AggregateSignaturesLatencyMS,
+		),
+		AggregateSignaturesRequestCount: prometheus.NewCounter(
+			Opts.AggregateSignaturesRequestCount,
+		),
+		FailuresToGetValidatorSet: prometheus.NewCounter(
+			Opts.FailuresToGetValidatorSet,
+		),
+		FailuresToConnectToSufficientStake: prometheus.NewCounter(
+			Opts.FailuresToConnectToSufficientStake,
+		),
+		FailuresSendingToNode: prometheus.NewCounter(
+			Opts.FailuresSendingToNode,
+		),
+		ValidatorTimeouts: prometheus.NewCounter(
+			Opts.ValidatorTimeouts,
+		),
+		InvalidSignatureResponses: prometheus.NewCounter(
+			Opts.InvalidSignatureResponses,
+		),
+	}
+
+	registerer.MustRegister(m.AggregateSignaturesLatencyMS)
+	registerer.MustRegister(m.AggregateSignaturesRequestCount)
+	registerer.MustRegister(m.FailuresToGetValidatorSet)
+	registerer.MustRegister(m.FailuresToConnectToSufficientStake)
+	registerer.MustRegister(m.FailuresSendingToNode)
+	registerer.MustRegister(m.ValidatorTimeouts)
+	registerer.MustRegister(m.InvalidSignatureResponses)
+
+	return &m
+}
+
+func (m *SignatureAggregatorMetrics) HandleMetricsRequest(
+	gatherer metrics.MultiGatherer,
+) {
+	http.Handle(
+		"/metrics",
+		promhttp.HandlerFor(gatherer, promhttp.HandlerOpts{}),
+	)
+}
+
+func Initialize(port uint16) *prometheus.Registry {
+	gatherer := metrics.NewPrefixGatherer()
+	registry := prometheus.NewRegistry()
+	err := gatherer.Register("signature-aggregator", registry)
+	if err != nil {
+		panic(
+			fmt.Errorf(
+				"failed to register metrics gatherer: %w",
+				err,
+			),
+		)
+	}
+
+	http.Handle(
+		"/metrics",
+		promhttp.HandlerFor(gatherer, promhttp.HandlerOpts{}),
+	)
+
+	go func() {
+		log.Fatalln(http.ListenAndServe(fmt.Sprintf(":%d", port), nil))
+	}()
+
+	return registry
+}