rpc: add execution pool metrics (#919)

* rpc: add execution pool metrics

* rpc: stop execution pool and report metrics using ticker

* fix lint

* update go.mod, update metric report interval

* handle empty workerpool case to fix tests

* fix lint

* refactor ep metrics collection based on each service

* remove log

* rpc: convert processed metric to histogram

cmd/clef/main.go

@@ -656,7 +656,7 @@ func signer(c *cli.Context) error {
 		vhosts := utils.SplitAndTrim(c.GlobalString(utils.HTTPVirtualHostsFlag.Name))
 		cors := utils.SplitAndTrim(c.GlobalString(utils.HTTPCORSDomainFlag.Name))
 
-		srv := rpc.NewServer(0, 0)
+		srv := rpc.NewServer("", 0, 0)
 		err := node.RegisterApis(rpcAPI, []string{"account"}, srv, false)
 		if err != nil {
 			utils.Fatalf("Could not register API: %w", err)

go.mod

@@ -7,7 +7,7 @@ require (
 	github.com/BurntSushi/toml v1.1.0
 	github.com/JekaMas/crand v1.0.1
 	github.com/JekaMas/go-grpc-net-conn v0.0.0-20220708155319-6aff21f2d13d
-	github.com/JekaMas/workerpool v1.1.5
+	github.com/JekaMas/workerpool v1.1.8
 	github.com/VictoriaMetrics/fastcache v1.6.0
 	github.com/aws/aws-sdk-go-v2 v1.2.0
 	github.com/aws/aws-sdk-go-v2/config v1.1.1

go.sum

@@ -54,8 +54,8 @@ github.com/JekaMas/crand v1.0.1 h1:FMPxkUQqH/hExl0aUXsr0UCGYZ4lJH9IJ5H/KbM6Y9A=
 github.com/JekaMas/crand v1.0.1/go.mod h1:GGzGpMCht/tbaNQ5A4kSiKSqEoNAhhyTfSDQyIENBQU=
 github.com/JekaMas/go-grpc-net-conn v0.0.0-20220708155319-6aff21f2d13d h1:RO27lgfZF8s9lZ3pWyzc0gCE0RZC+6/PXbRjAa0CNp8=
 github.com/JekaMas/go-grpc-net-conn v0.0.0-20220708155319-6aff21f2d13d/go.mod h1:romz7UPgSYhfJkKOalzEEyV6sWtt/eAEm0nX2aOrod0=
-github.com/JekaMas/workerpool v1.1.5 h1:xmrx2Zyft95CEGiEqzDxiawptCIRZQ0zZDhTGDFOCaw=
-github.com/JekaMas/workerpool v1.1.5/go.mod h1:IoDWPpwMcA27qbuugZKeBslDrgX09lVmksuh9sjzbhc=
+github.com/JekaMas/workerpool v1.1.8 h1:IbDbTITrDgt1xAzdzTS9aLk4Q/4dCsjUOopiyFkDFZ4=
+github.com/JekaMas/workerpool v1.1.8/go.mod h1:IoDWPpwMcA27qbuugZKeBslDrgX09lVmksuh9sjzbhc=
 github.com/Knetic/govaluate v3.0.1-0.20171022003610-9aa49832a739+incompatible/go.mod h1:r7JcOSlj0wfOMncg0iLm8Leh48TZaKVeNIfJntJ2wa0=
 github.com/Masterminds/goutils v1.1.1 h1:5nUrii3FMTL5diU80unEVvNevw1nH4+ZV4DSLVJLSYI=
 github.com/Masterminds/goutils v1.1.1/go.mod h1:8cTjp+g8YejhMuvIA5y2vz3BpJxksy863GQaJW2MFNU=

node/node.go

@@ -105,7 +105,7 @@ func New(conf *Config) (*Node, error) {
 
 	node := &Node{
 		config:        conf,
-		inprocHandler: rpc.NewServer(0, 0),
+		inprocHandler: rpc.NewServer("inproc", 0, 0),
 		eventmux:      new(event.TypeMux),
 		log:           conf.Logger,
 		stop:          make(chan struct{}),

node/rpcstack.go

@@ -293,7 +293,7 @@ func (h *httpServer) enableRPC(apis []rpc.API, config httpConfig) error {
 	}
 
 	// Create RPC server and handler.
-	srv := rpc.NewServer(config.executionPoolSize, config.executionPoolRequestTimeout)
+	srv := rpc.NewServer("http", config.executionPoolSize, config.executionPoolRequestTimeout)
 	srv.SetRPCBatchLimit(h.RPCBatchLimit)
 	if err := RegisterApis(apis, config.Modules, srv, false); err != nil {
 		return err
@@ -325,7 +325,7 @@ func (h *httpServer) enableWS(apis []rpc.API, config wsConfig) error {
 		return fmt.Errorf("JSON-RPC over WebSocket is already enabled")
 	}
 	// Create RPC server and handler.
-	srv := rpc.NewServer(config.executionPoolSize, config.executionPoolRequestTimeout)
+	srv := rpc.NewServer("ws", config.executionPoolSize, config.executionPoolRequestTimeout)
 	srv.SetRPCBatchLimit(h.RPCBatchLimit)
 	if err := RegisterApis(apis, config.Modules, srv, false); err != nil {
 		return err

rpc/client.go

@@ -112,7 +112,7 @@ func (c *Client) newClientConn(conn ServerCodec) *clientConn {
 	ctx := context.Background()
 	ctx = context.WithValue(ctx, clientContextKey{}, c)
 	ctx = context.WithValue(ctx, peerInfoContextKey{}, conn.peerInfo())
-	handler := newHandler(ctx, conn, c.idgen, c.services, NewExecutionPool(100, 0))
+	handler := newHandler(ctx, conn, c.idgen, c.services, NewExecutionPool(100, 0, "rpcclient", true))
 	return &clientConn{conn, handler}
 }
 

rpc/client_test.go

@@ -410,7 +410,7 @@ func TestClientSubscriptionUnsubscribeServer(t *testing.T) {
 	t.Parallel()
 
 	// Create the server.
-	srv := NewServer(0, 0)
+	srv := NewServer("test", 0, 0)
 	srv.RegisterName("nftest", new(notificationTestService))
 	p1, p2 := net.Pipe()
 	recorder := &unsubscribeRecorder{ServerCodec: NewCodec(p1)}
@@ -446,7 +446,7 @@ func TestClientSubscriptionChannelClose(t *testing.T) {
 	t.Parallel()
 
 	var (
-		srv     = NewServer(0, 0)
+		srv     = NewServer("test", 0, 0)
 		httpsrv = httptest.NewServer(srv.WebsocketHandler(nil))
 		wsURL   = "ws:" + strings.TrimPrefix(httpsrv.URL, "http:")
 	)

rpc/endpoints.go

@@ -27,7 +27,7 @@ import (
 func StartIPCEndpoint(ipcEndpoint string, apis []API) (net.Listener, *Server, error) {
 	// Register all the APIs exposed by the services.
 	var (
-		handler    = NewServer(0, 0)
+		handler    = NewServer("", 0, 0) // we can skip reporting ipc metrics, hence empty service name
 		regMap     = make(map[string]struct{})
 		registered []string
 	)

rpc/execution_pool.go

@@ -7,24 +7,32 @@ import (
 	"time"
 
 	"github.com/JekaMas/workerpool"
+	"github.com/ethereum/go-ethereum/metrics"
 )
 
 type SafePool struct {
-	executionPool *atomic.Pointer[workerpool.WorkerPool]
+	executionPool atomic.Pointer[workerpool.WorkerPool]
 
 	sync.RWMutex
 
-	timeout time.Duration
 	size    int
+	timeout time.Duration
+
+	service   string       // the service using ep
+	processed atomic.Int64 // keeps count of total processed requests
+
+	close chan struct{}
 
 	// Skip sending task to execution pool
 	fastPath bool
 }
 
-func NewExecutionPool(initialSize int, timeout time.Duration) *SafePool {
+func NewExecutionPool(initialSize int, timeout time.Duration, service string, report bool) *SafePool {
 	sp := &SafePool{
 		size:    initialSize,
 		timeout: timeout,
+		service: service,
+		close:   make(chan struct{}),
 	}
 
 	if initialSize == 0 {
@@ -33,11 +41,11 @@ func NewExecutionPool(initialSize int, timeout time.Duration) *SafePool {
 		return sp
 	}
 
-	var ptr atomic.Pointer[workerpool.WorkerPool]
+	sp.executionPool.Store(workerpool.New(initialSize))
 
-	p := workerpool.New(initialSize)
-	ptr.Store(p)
-	sp.executionPool = &ptr
+	if metrics.Enabled && report {
+		go sp.reportMetrics(3 * time.Second)
+	}
 
 	return sp
 }
@@ -51,10 +59,6 @@ func (s *SafePool) Submit(ctx context.Context, fn func() error) (<-chan error, b
 		return nil, true
 	}
 
-	if s.executionPool == nil {
-		return nil, false
-	}
-
 	pool := s.executionPool.Load()
 	if pool == nil {
 		return nil, false
@@ -97,3 +101,42 @@ func (s *SafePool) Size() int {
 
 	return s.size
 }
+
+func (s *SafePool) Stop() {
+	close(s.close)
+
+	if s.executionPool.Load() != nil {
+		s.executionPool.Load().Stop()
+	}
+}
+
+// reportMetrics reports the metrics after every `refresh` time interval
+// regarding the execution pool.
+func (s *SafePool) reportMetrics(refresh time.Duration) {
+	var (
+		epWorkerCountGuage           metrics.Gauge
+		epWaitingQueueGuage          metrics.Gauge
+		epProcessedRequestsHistogram metrics.Histogram
+	)
+
+	ticker := time.NewTicker(refresh)
+
+	for {
+		select {
+		case <-ticker.C:
+			ep := s.executionPool.Load()
+
+			epWorkerCountGuage, epWaitingQueueGuage, epProcessedRequestsHistogram = newEpMetrics(s.service)
+
+			epWorkerCountGuage.Update(ep.GetWorkerCount())
+			epWaitingQueueGuage.Update(int64(ep.WaitingQueueSize()))
+			epProcessedRequestsHistogram.Update(s.processed.Load())
+
+			s.processed.Store(0)
+		case <-s.close:
+			ticker.Stop()
+
+			return
+		}
+	}
+}

rpc/handler.go

@@ -231,6 +231,8 @@ func (h *handler) startCallProc(fn func(*callProc)) {
 		defer cancel()
 		fn(&callProc{ctx: ctx})
 
+		h.executionPool.processed.Add(1)
+
 		return nil
 	})
 }
@@ -269,7 +271,6 @@ func (h *handler) handleSubscriptionResult(msg *jsonrpcMessage) {
 
 // handleResponse processes method call responses.
 func (h *handler) handleResponse(msg *jsonrpcMessage) {
-
 	op := h.respWait[string(msg.ID)]
 	if op == nil {
 		h.log.Debug("Unsolicited RPC response", "reqid", idForLog{msg.ID})
@@ -292,6 +293,7 @@ func (h *handler) handleResponse(msg *jsonrpcMessage) {
 	if op.err = json.Unmarshal(msg.Result, &op.sub.subid); op.err == nil {
 		h.executionPool.Submit(context.Background(), func() error {
 			op.sub.run()
+			h.executionPool.processed.Add(1)
 			return nil
 		})
 

rpc/http_test.go

@@ -103,7 +103,7 @@ func TestHTTPResponseWithEmptyGet(t *testing.T) {
 func TestHTTPRespBodyUnlimited(t *testing.T) {
 	const respLength = maxRequestContentLength * 3
 
-	s := NewServer(0, 0)
+	s := NewServer("test", 0, 0)
 	defer s.Stop()
 	s.RegisterName("test", largeRespService{respLength})
 	ts := httptest.NewServer(s)

rpc/inproc.go

@@ -30,6 +30,7 @@ func DialInProc(handler *Server) *Client {
 		//nolint:contextcheck
 		handler.executionPool.Submit(initctx, func() error {
 			handler.ServeCodec(NewCodec(p1), 0)
+			handler.executionPool.processed.Add(1)
 			return nil
 		})
 

rpc/ipc.go

@@ -38,6 +38,7 @@ func (s *Server) ServeListener(l net.Listener) error {
 
 		s.executionPool.Submit(context.Background(), func() error {
 			s.ServeCodec(NewCodec(conn), 0)
+			s.executionPool.processed.Add(1)
 			return nil
 		})
 	}

rpc/metrics.go

@@ -37,3 +37,11 @@ func newRPCServingTimer(method string, valid bool) metrics.Timer {
 	m := fmt.Sprintf("rpc/duration/%s/%s", method, flag)
 	return metrics.GetOrRegisterTimer(m, nil)
 }
+
+func newEpMetrics(service string) (metrics.Gauge, metrics.Gauge, metrics.Histogram) {
+	epWorkerCount := fmt.Sprintf("rpc/ep/workers/%s", service)
+	epWaitingQueue := fmt.Sprintf("rpc/ep/queue/%s", service)
+	epProcessedRequests := fmt.Sprintf("rpc/ep/processed/%s", service)
+
+	return metrics.GetOrRegisterGauge(epWorkerCount, nil), metrics.GetOrRegisterGauge(epWaitingQueue, nil), metrics.GetOrRegisterHistogram(epProcessedRequests, nil, metrics.NewExpDecaySample(1028, 0.015))
+}

rpc/server.go

@@ -56,18 +56,24 @@ type Server struct {
 }
 
 // NewServer creates a new server instance with no registered handlers.
-func NewServer(executionPoolSize uint64, executionPoolRequesttimeout time.Duration) *Server {
+func NewServer(service string, executionPoolSize uint64, executionPoolRequesttimeout time.Duration) *Server {
+	reportEpStats := true
+	if service == "" || service == "test" {
+		reportEpStats = false
+	}
+
 	server := &Server{
 		idgen:         randomIDGenerator(),
 		codecs:        mapset.NewSet(),
 		run:           1,
-		executionPool: NewExecutionPool(int(executionPoolSize), executionPoolRequesttimeout),
+		executionPool: NewExecutionPool(int(executionPoolSize), executionPoolRequesttimeout, service, reportEpStats),
 	}
 
 	// Register the default service providing meta information about the RPC service such
 	// as the services and methods it offers.
 	rpcService := &RPCService{server}
 	server.RegisterName(MetadataApi, rpcService)
+
 	return server
 }
 
@@ -166,6 +172,9 @@ func (s *Server) serveSingleRequest(ctx context.Context, codec ServerCodec) {
 // requests to finish, then closes all codecs which will cancel pending requests and
 // subscriptions.
 func (s *Server) Stop() {
+	// Stop the execution pool
+	s.executionPool.Stop()
+
 	if atomic.CompareAndSwapInt32(&s.run, 1, 0) {
 		log.Debug("RPC server shutting down")
 		s.codecs.Each(func(c interface{}) bool {

rpc/server_test.go

@@ -29,7 +29,7 @@ import (
 )
 
 func TestServerRegisterName(t *testing.T) {
-	server := NewServer(0, 0)
+	server := NewServer("test", 0, 0)
 	service := new(testService)
 
 	if err := server.RegisterName("test", service); err != nil {

rpc/subscription_test.go

@@ -53,7 +53,7 @@ func TestSubscriptions(t *testing.T) {
 		subCount          = len(namespaces)
 		notificationCount = 3
 
-		server                 = NewServer(0, 0)
+		server                 = NewServer("test", 0, 0)
 		clientConn, serverConn = net.Pipe()
 		out                    = json.NewEncoder(clientConn)
 		in                     = json.NewDecoder(clientConn)

rpc/testservice_test.go

@@ -26,7 +26,7 @@ import (
 )
 
 func newTestServer() *Server {
-	server := NewServer(0, 0)
+	server := NewServer("test", 0, 0)
 	server.idgen = sequentialIDGenerator()
 	if err := server.RegisterName("test", new(testService)); err != nil {
 		panic(err)

rpc/websocket_test.go

@@ -203,7 +203,7 @@ func TestClientWebsocketPing(t *testing.T) {
 // This checks that the websocket transport can deal with large messages.
 func TestClientWebsocketLargeMessage(t *testing.T) {
 	var (
-		srv     = NewServer(0, 0)
+		srv     = NewServer("test", 0, 0)
 		httpsrv = httptest.NewServer(srv.WebsocketHandler(nil))
 		wsURL   = "ws:" + strings.TrimPrefix(httpsrv.URL, "http:")
 	)