Add more metrics (#35)

## 📝 Summary

<!--- A general summary of your changes -->

## ⛱ Motivation and Context

<!--- Why is this change required? What problem does it solve? -->

## 📚 References

<!-- Any interesting external links to documentation, articles, tweets
which add value to the PR -->

---

## ✅ I have run these commands

* [ ] `make lint`
* [ ] `make test`
* [ ] `go mod tidy`

Author: dvush

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.10.0
+	github.com/flashbots/go-utils v0.10.1-0.20250416132112-39233b64a8c5
 	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

@@ -34,6 +34,8 @@ github.com/ethereum/go-verkle v0.2.2 h1:I2W0WjnrFUIzzVPwm8ykY+7pL2d4VhlsePn4j7cn
 github.com/ethereum/go-verkle v0.2.2/go.mod h1:M3b90YRnzqKyyzBEWJGqj8Qff4IDeXnzFw0P9bFw3uk=
 github.com/flashbots/go-utils v0.10.0 h1:75XWewRO5GIhdLn8+vqdzzuoqJh+j8wN54A++Id7W0Y=
 github.com/flashbots/go-utils v0.10.0/go.mod h1:i4xxEB6sHDFfNWEIfh+rP6nx3LxynEn8AOZa05EYgwA=
+github.com/flashbots/go-utils v0.10.1-0.20250416132112-39233b64a8c5 h1:oEfjmk2NQ/FtX+1h1Rx4PMpJtbNAa6T/l6otbUmzZp8=
+github.com/flashbots/go-utils v0.10.1-0.20250416132112-39233b64a8c5/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

@@ -2,6 +2,7 @@ package proxy
 
 import (
 	"fmt"
+	"time"
 
 	"github.com/VictoriaMetrics/metrics"
 )
@@ -28,6 +29,8 @@ 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"}`
+
+	requestDurationLabel = `orderflow_proxy_api_request_processing_duration_milliseconds{method="%s",server_name="%s",step="%s"}`
 )
 
 func incAPIIncomingRequestsByPeer(peer string) {
@@ -58,3 +61,9 @@ func timeShareQueuePeerRPCDuration(peer string, duration int64) {
 	l := fmt.Sprintf(shareQueuePeerRPCDurationLabel, peer)
 	metrics.GetOrCreateSummary(l).Update(float64(duration))
 }
+
+func incRequestDurationStep(duration time.Duration, method, serverName, step string) {
+	millis := float64(duration.Microseconds()) / 1000.0
+	l := fmt.Sprintf(requestDurationLabel, method, serverName, step)
+	metrics.GetOrCreateSummary(l).Update(millis)
+}

proxy/receiver_api.go

@@ -120,6 +120,7 @@ func (prx *ReceiverProxy) ValidateSigner(ctx context.Context, req *ParsedRequest
 }
 
 func (prx *ReceiverProxy) EthSendBundle(ctx context.Context, ethSendBundle rpctypes.EthSendBundleArgs, systemEndpoint bool) error {
+	startAt := time.Now()
 	parsedRequest := ParsedRequest{
 		systemEndpoint: systemEndpoint,
 		ethSendBundle:  &ethSendBundle,
@@ -141,6 +142,9 @@ func (prx *ReceiverProxy) EthSendBundle(ctx context.Context, ethSendBundle rpcty
 		return err
 	}
 
+	incRequestDurationStep(time.Since(startAt), parsedRequest.method, "", "validation")
+	startAt = time.Now()
+
 	// For direct orderflow we extract signing address from header
 	// We also default to v2 bundle version if it's not set
 	if !systemEndpoint {
@@ -169,6 +173,8 @@ func (prx *ReceiverProxy) EthSendBundle(ctx context.Context, ethSendBundle rpcty
 	uniqueKey := ethSendBundle.UniqueKey()
 	parsedRequest.requestArgUniqueKey = &uniqueKey
 
+	incRequestDurationStep(time.Since(startAt), parsedRequest.method, "", "add_fields")
+
 	return prx.HandleParsedRequest(ctx, parsedRequest)
 }
 
@@ -181,6 +187,7 @@ func (prx *ReceiverProxy) EthSendBundleUser(ctx context.Context, ethSendBundle r
 }
 
 func (prx *ReceiverProxy) MevSendBundle(ctx context.Context, mevSendBundle rpctypes.MevSendBundleArgs, systemEndpoint bool) error {
+	startAt := time.Now()
 	parsedRequest := ParsedRequest{
 		systemEndpoint: systemEndpoint,
 		mevSendBundle:  &mevSendBundle,
@@ -197,6 +204,9 @@ func (prx *ReceiverProxy) MevSendBundle(ctx context.Context, mevSendBundle rpcty
 		return err
 	}
 
+	incRequestDurationStep(time.Since(startAt), parsedRequest.method, "", "validation")
+	startAt = time.Now()
+
 	if !systemEndpoint {
 		mevSendBundle.Metadata = &rpctypes.MevBundleMetadata{
 			Signer: &parsedRequest.signer,
@@ -231,6 +241,8 @@ func (prx *ReceiverProxy) MevSendBundle(ctx context.Context, mevSendBundle rpcty
 	uniqueKey := mevSendBundle.UniqueKey()
 	parsedRequest.requestArgUniqueKey = &uniqueKey
 
+	incRequestDurationStep(time.Since(startAt), parsedRequest.method, "", "add_fields")
+
 	return prx.HandleParsedRequest(ctx, parsedRequest)
 }
 
@@ -347,6 +359,7 @@ type ParsedRequest struct {
 }
 
 func (prx *ReceiverProxy) HandleParsedRequest(ctx context.Context, parsedRequest ParsedRequest) error {
+	startAt := time.Now()
 	ctx, cancel := context.WithTimeout(ctx, handleParsedRequestTimeout)
 	defer cancel()
 
@@ -362,25 +375,39 @@ func (prx *ReceiverProxy) HandleParsedRequest(ctx context.Context, parsedRequest
 		}
 		prx.requestUniqueKeysRLU.Add(*parsedRequest.requestArgUniqueKey, struct{}{})
 	}
+
+	incRequestDurationStep(time.Since(startAt), parsedRequest.method, "", "validation")
+	startAt = time.Now()
+
 	if !parsedRequest.systemEndpoint {
 		err := prx.userAPIRateLimiter.Wait(ctx)
 		if err != nil {
 			incAPIUserRateLimits()
 			return errors.Join(errRateLimiting, err)
 		}
 	}
+
+	incRequestDurationStep(time.Since(startAt), parsedRequest.method, "", "rate_limiting")
+	startAt = time.Now()
+
 	select {
 	case <-ctx.Done():
 		prx.Log.Error("Shared queue is stalling")
 	case prx.shareQueue <- &parsedRequest:
 	}
+
+	incRequestDurationStep(time.Since(startAt), parsedRequest.method, "", "share_queue")
+	startAt = time.Now()
+
 	if !parsedRequest.systemEndpoint {
 		select {
 		case <-ctx.Done():
 			prx.Log.Error("Archive queue is stalling")
 		case prx.archiveQueue <- &parsedRequest:
 		}
 	}
+
+	incRequestDurationStep(time.Since(startAt), parsedRequest.method, "", "archive_queue")
 	return nil
 }