Merge pull request #227 from onflow/kan/access-node-metrics

Kan/access node metrics

cmd/access/main.go

@@ -67,6 +67,7 @@ func main() {
 		logTxTimeToExecuted          bool
 		logTxTimeToFinalizedExecuted bool
 		retryEnabled                 bool
+		rpcMetricsEnabled            bool
 	)
 
 	cmd.FlowNode(flow.RoleAccess.String()).
@@ -85,6 +86,7 @@ func main() {
 			flags.BoolVar(&logTxTimeToFinalizedExecuted, "log-tx-time-to-finalized-executed", false, "log transaction time to finalized and executed")
 			flags.BoolVar(&pingEnabled, "ping-enabled", false, "whether to enable the ping process that pings all other peers and report the connectivity to metrics")
 			flags.BoolVar(&retryEnabled, "retry-enabled", false, "whether to enable the retry mechanism at the access node level")
+			flags.BoolVar(&rpcMetricsEnabled, "rpc-metrics-enabled", false, "whether to enable the rpc metrics")
 			flags.StringVarP(&nodeInfoFile, "node-info-file", "", "", "full path to a json file which provides more details about nodes when reporting its reachability metrics")
 		}).
 		Module("collection node client", func(node *cmd.FlowNodeBuilder) error {
@@ -188,6 +190,7 @@ func main() {
 				transactionMetrics,
 				collectionGRPCPort,
 				retryEnabled,
+				rpcMetricsEnabled,
 			)
 			return rpcEng, nil
 		}).

engine/access/access_test.go

@@ -447,7 +447,7 @@ func (suite *Suite) TestGetSealedTransaction() {
 		require.NoError(suite.T(), err)
 
 		rpcEng := rpc.New(suite.log, suite.state, rpc.Config{}, nil, nil, nil, blocks, headers, collections, transactions,
-			suite.chainID, metrics, 0, false)
+			suite.chainID, metrics, 0, false, false)
 
 		// create the ingest engine
 		ingestEng, err := ingestion.New(suite.log, suite.net, suite.state, suite.me, suite.request, blocks, headers, collections,

engine/access/ingestion/engine_test.go

@@ -90,7 +90,7 @@ func (suite *Suite) SetupTest() {
 	require.NoError(suite.T(), err)
 
 	rpcEng := rpc.New(log, suite.proto.state, rpc.Config{}, nil, nil, nil, suite.blocks, suite.headers, suite.collections,
-		suite.transactions, flow.Testnet, metrics.NewNoopCollector(), 0, false)
+		suite.transactions, flow.Testnet, metrics.NewNoopCollector(), 0, false, false)
 
 	eng, err := New(log, net, suite.proto.state, suite.me, suite.request, suite.blocks, suite.headers, suite.collections,
 		suite.transactions, metrics.NewNoopCollector(), collectionsToMarkFinalized, collectionsToMarkExecuted,

engine/access/rpc/engine.go

@@ -7,6 +7,7 @@ import (
 	"net"
 	"net/http"
 
+	grpc_prometheus "github.com/grpc-ecosystem/go-grpc-prometheus"
 	"github.com/rs/zerolog"
 	"google.golang.org/grpc"
 
@@ -60,6 +61,7 @@ func New(log zerolog.Logger,
 	transactionMetrics module.TransactionMetrics,
 	collectionGRPCPort uint,
 	retryEnabled bool,
+	rpcMetricsEnabled bool,
 ) *Engine {
 
 	log = log.With().Str("engine", "rpc").Logger()
@@ -69,10 +71,19 @@ func New(log zerolog.Logger,
 	}
 
 	// create a GRPC server to serve GRPC clients
-	grpcServer := grpc.NewServer(
+	grpcOpts := []grpc.ServerOption{
 		grpc.MaxRecvMsgSize(config.MaxMsgSize),
 		grpc.MaxSendMsgSize(config.MaxMsgSize),
-	)
+	}
+	if rpcMetricsEnabled {
+		grpcOpts = append(
+			grpcOpts,
+			grpc.StreamInterceptor(grpc_prometheus.StreamServerInterceptor),
+			grpc.UnaryInterceptor(grpc_prometheus.UnaryServerInterceptor),
+		)
+	}
+
+	grpcServer := grpc.NewServer(grpcOpts...)
 
 	// wrap the GRPC server with an HTTP proxy server to serve HTTP clients
 	httpServer := NewHTTPServer(grpcServer, config.HTTPListenAddr)
@@ -107,6 +118,11 @@ func New(log zerolog.Logger,
 		access.NewHandler(backend, chainID.Chain()),
 	)
 
+	if rpcMetricsEnabled {
+		// Not interested in legacy metrics, so initialize here
+		grpc_prometheus.Register(grpcServer)
+	}
+
 	// Register legacy gRPC handlers for backwards compatibility, to be removed at a later date
 	legacyaccessproto.RegisterAccessAPIServer(
 		eng.grpcServer,

fvm/state/accounts_test.go

@@ -51,7 +51,6 @@ func TestAccounts_GetWithNoKeys(t *testing.T) {
 	})
 }
 
-
 func TestAccounts_GetPublicKey(t *testing.T) {
 
 	t.Run("non-existent key index", func(t *testing.T) {

go.mod

@@ -19,6 +19,7 @@ require (
 	github.com/golang/protobuf v1.4.2
 	github.com/google/go-cmp v0.5.2
 	github.com/google/uuid v1.1.1
+	github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0
 	github.com/hashicorp/go-multierror v1.0.0
 	github.com/hashicorp/golang-lru v0.5.4
 	github.com/improbable-eng/grpc-web v0.12.0

go.sum

@@ -288,6 +288,7 @@ github.com/gorilla/websocket v1.4.2 h1:+/TMaTYc4QFitKJxsQ7Yye35DkWvkdLcvGKqM+x0U
 github.com/gorilla/websocket v1.4.2/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/graph-gophers/graphql-go v0.0.0-20191115155744-f33e81362277/go.mod h1:9CQHMSxwO4MprSdzoIEobiHpoLtHm77vfxsvsIN5Vuc=
 github.com/grpc-ecosystem/go-grpc-middleware v1.0.0/go.mod h1:FiyG127CGDf3tlThmgyCl78X/SZQqEOJBCDaAfeWzPs=
+github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0 h1:Ovs26xHkKqVztRpIrF/92BcuyuQ/YW4NSIpoGtfXNho=
 github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0/go.mod h1:8NvIoxWQoOIhqOTXgfV/d3M/q6VIi02HzZEHgUlZvzk=
 github.com/grpc-ecosystem/grpc-gateway v1.9.0/go.mod h1:vNeuVxBJEsws4ogUvrchl83t/GYV9WGTSLVdBhOQFDY=
 github.com/gxed/hashland/keccakpg v0.0.1/go.mod h1:kRzw3HkwxFU1mpmPP8v1WyQzwdGfmKFJ6tItnhQ67kU=