metrics split by request size

Author: TymKh

go.mod

@@ -8,7 +8,7 @@ require (
 	github.com/VictoriaMetrics/metrics v1.35.1
 	github.com/cenkalti/backoff v2.2.1+incompatible
 	github.com/ethereum/go-ethereum v1.15.5
-	github.com/flashbots/go-utils v0.13.0
+	github.com/flashbots/go-utils v0.13.2
 	github.com/google/uuid v1.6.0
 	github.com/hashicorp/golang-lru/v2 v2.0.7
 	github.com/stretchr/testify v1.10.0

go.sum

@@ -32,8 +32,8 @@ github.com/ethereum/go-ethereum v1.15.5 h1:Fo2TbBWC61lWVkFw9tsMoHCNX1ndpuaQBRJ8H
 github.com/ethereum/go-ethereum v1.15.5/go.mod h1:1LG2LnMOx2yPRHR/S+xuipXH29vPr6BIH6GElD8N/fo=
 github.com/ethereum/go-verkle v0.2.2 h1:I2W0WjnrFUIzzVPwm8ykY+7pL2d4VhlsePn4j7cnFk8=
 github.com/ethereum/go-verkle v0.2.2/go.mod h1:M3b90YRnzqKyyzBEWJGqj8Qff4IDeXnzFw0P9bFw3uk=
-github.com/flashbots/go-utils v0.13.0 h1:rGR/4e8Xh2Y4JUjX+gZ7T82AT3d//m5RawVy+h95PeA=
-github.com/flashbots/go-utils v0.13.0/go.mod h1:i4xxEB6sHDFfNWEIfh+rP6nx3LxynEn8AOZa05EYgwA=
+github.com/flashbots/go-utils v0.13.2 h1:9fdnLKpQP/9i48NDad7u5NOK945DmL3Z61eCA9JGiwQ=
+github.com/flashbots/go-utils v0.13.2/go.mod h1:i4xxEB6sHDFfNWEIfh+rP6nx3LxynEn8AOZa05EYgwA=
 github.com/go-ole/go-ole v1.3.0 h1:Dt6ye7+vXGIKZ7Xtk4s6/xVdGDQynvom7xCFEdWr6uE=
 github.com/go-ole/go-ole v1.3.0/go.mod h1:5LS6F96DhAwUc7C+1HLexzMXY1xGRSryjyPPKW6zv78=
 github.com/gofrs/flock v0.8.1 h1:+gYjHKf32LDeiEEFhQaotPbLuUXjY5ZqxKgXy7n59aw=

proxy/metrics.go

@@ -28,8 +28,9 @@ const (
 
 	shareQueuePeerStallingErrorsLabel = `orderflow_proxy_share_queue_peer_stalling_errors{peer="%s"}`
 	shareQueuePeerRPCErrorsLabel      = `orderflow_proxy_share_queue_peer_rpc_errors{peer="%s"}`
-	shareQueuePeerRPCDurationLabel    = `orderflow_proxy_share_queue_peer_rpc_duration_milliseconds{peer="%s"}`
-	shareQueuePeerE2EDurationLabel    = `orderflow_proxy_share_queue_peer_e2e_duration_milliseconds{peer="%s",method="%s",system_endpoint="%t"}`
+	shareQueuePeerRPCDurationLabel    = `orderflow_proxy_share_queue_peer_rpc_duration_milliseconds{peer="%s",is_big="%t"}`
+	shareQueuePeerE2EDurationLabel    = `orderflow_proxy_share_queue_peer_e2e_duration_milliseconds{peer="%s",method="%s",system_endpoint="%t",is_big="%t"}`
+	shareQueuePeerQueueDurationLabel  = `orderflow_proxy_share_queue_peer_queue_duration_milliseconds{peer="%s",method="%s",system_endpoint="%t",is_big="%t"}`
 
 	requestDurationLabel = `orderflow_proxy_api_request_processing_duration_milliseconds{method="%s",server_name="%s",step="%s"}`
 )
@@ -58,14 +59,20 @@ func incShareQueuePeerRPCErrors(peer string) {
 	metrics.GetOrCreateCounter(l).Inc()
 }
 
-func timeShareQueuePeerRPCDuration(peer string, duration int64) {
-	l := fmt.Sprintf(shareQueuePeerRPCDurationLabel, peer)
+func timeShareQueuePeerRPCDuration(peer string, duration int64, bigRequest bool) {
+	l := fmt.Sprintf(shareQueuePeerRPCDurationLabel, peer, bigRequest)
 	metrics.GetOrCreateSummary(l).Update(float64(duration))
 }
 
-func timeShareQueuePeerE2EDuration(peer string, duration time.Duration, method string, systemEndpoint bool) {
+func timeShareQueuePeerE2EDuration(peer string, duration time.Duration, method string, systemEndpoint, bigRequest bool) {
 	millis := float64(duration.Microseconds()) / 1000.0
-	l := fmt.Sprintf(shareQueuePeerE2EDurationLabel, peer, method, systemEndpoint)
+	l := fmt.Sprintf(shareQueuePeerE2EDurationLabel, peer, method, systemEndpoint, bigRequest)
+	metrics.GetOrCreateSummary(l).Update(float64(millis))
+}
+
+func timeShareQueuePeerQueueDuration(peer string, duration time.Duration, method string, systemEndpoint, bigRequest bool) {
+	millis := float64(duration.Microseconds()) / 1000.0
+	l := fmt.Sprintf(shareQueuePeerQueueDurationLabel, peer, method, systemEndpoint, bigRequest)
 	metrics.GetOrCreateSummary(l).Update(float64(millis))
 }
 

proxy/receiver_api.go

@@ -125,6 +125,7 @@ func (prx *ReceiverProxy) EthSendBundle(ctx context.Context, ethSendBundle rpcty
 		systemEndpoint: systemEndpoint,
 		ethSendBundle:  &ethSendBundle,
 		method:         EthSendBundleMethod,
+		size:           rpcserver.GetRequestSize(ctx),
 	}
 
 	err := prx.ValidateSigner(ctx, &parsedRequest, systemEndpoint)
@@ -192,6 +193,7 @@ func (prx *ReceiverProxy) MevSendBundle(ctx context.Context, mevSendBundle rpcty
 		systemEndpoint: systemEndpoint,
 		mevSendBundle:  &mevSendBundle,
 		method:         MevSendBundleMethod,
+		size:           rpcserver.GetRequestSize(ctx),
 	}
 
 	err := prx.ValidateSigner(ctx, &parsedRequest, systemEndpoint)
@@ -259,6 +261,7 @@ func (prx *ReceiverProxy) EthCancelBundle(ctx context.Context, ethCancelBundle r
 		systemEndpoint:  systemEndpoint,
 		ethCancelBundle: &ethCancelBundle,
 		method:          EthCancelBundleMethod,
+		size:            rpcserver.GetRequestSize(ctx),
 	}
 
 	err := prx.ValidateSigner(ctx, &parsedRequest, systemEndpoint)
@@ -290,6 +293,7 @@ func (prx *ReceiverProxy) EthSendRawTransaction(ctx context.Context, ethSendRawT
 		systemEndpoint:        systemEndpoint,
 		ethSendRawTransaction: &ethSendRawTransaction,
 		method:                EthSendRawTransactionMethod,
+		size:                  rpcserver.GetRequestSize(ctx),
 	}
 	err := prx.ValidateSigner(ctx, &parsedRequest, systemEndpoint)
 	if err != nil {
@@ -319,6 +323,7 @@ func (prx *ReceiverProxy) BidSubsidiseBlock(ctx context.Context, bidSubsidiseBlo
 		systemEndpoint:    systemEndpoint,
 		bidSubsidiseBlock: &bidSubsidiseBlock,
 		method:            BidSubsidiseBlockMethod,
+		size:              rpcserver.GetRequestSize(ctx),
 	}
 
 	err := prx.ValidateSigner(ctx, &parsedRequest, systemEndpoint)
@@ -349,6 +354,7 @@ type ParsedRequest struct {
 	signer                common.Address
 	method                string
 	peerName              string
+	size                  int
 	receivedAt            time.Time
 	requestArgUniqueKey   *uuid.UUID
 	ethSendBundle         *rpctypes.EthSendBundleArgs

proxy/sharing.go

@@ -14,6 +14,10 @@ var (
 	requestTimeout       = time.Second * 10
 )
 
+const (
+	bigRequestSize = 50_000
+)
+
 type ShareQueue struct {
 	name         string
 	log          *slog.Logger
@@ -162,6 +166,7 @@ func (sq *ShareQueue) proxyRequests(peer *shareQueuePeer, worker int) {
 			method string
 			data   any
 		)
+		isBig := req.size > bigRequestSize
 		if req.ethSendBundle != nil {
 			method = EthSendBundleMethod
 			data = req.ethSendBundle
@@ -182,12 +187,13 @@ func (sq *ShareQueue) proxyRequests(peer *shareQueuePeer, worker int) {
 			shareQueueInternalErrors.Inc()
 			continue
 		}
+		timeShareQueuePeerQueueDuration(peer.name, time.Since(req.receivedAt), method, req.systemEndpoint, isBig)
 		start := time.Now()
 		ctx, cancel := context.WithTimeout(context.Background(), requestTimeout)
 		resp, err := peer.client.Call(ctx, method, data)
 		cancel()
-		timeShareQueuePeerRPCDuration(peer.name, time.Since(start).Milliseconds())
-		timeShareQueuePeerE2EDuration(peer.name, time.Since(req.receivedAt), method, req.systemEndpoint)
+		timeShareQueuePeerRPCDuration(peer.name, time.Since(start).Milliseconds(), isBig)
+		timeShareQueuePeerE2EDuration(peer.name, time.Since(req.receivedAt), method, req.systemEndpoint, isBig)
 		logSendErrorLevel := slog.LevelDebug
 		if peer.name == "local-builder" {
 			logSendErrorLevel = slog.LevelWarn