publish two basic metrics from the RPC client into Beholder: RPC latency, RPC error rate

Author: Larry Li

docs/rpc_observability.md

@@ -0,0 +1,47 @@
+# RPC Observability (Beholder)
+
+RPC client metrics are published to Beholder and surface in Prometheus/Grafana when `RPCClientBase` is constructed with a non-nil `metrics.RPCClientMetrics`.
+
+## Metrics
+
+| Metric | Type | Description |
+|--------|------|-------------|
+| `rpc_request_latency_ms` | Histogram | RPC call latency in milliseconds (per call) |
+| `rpc_request_errors_total` | Counter | Total number of failed RPC requests |
+
+Labels: `env`, `network`, `chain_id`, `rpc_provider`, `call` (e.g. `latest_block`, `latest_finalized_block`).
+
+## Example Prometheus / Grafana Queries
+
+### Latency over time
+
+- **p99 latency by env and chain:**
+  ```promql
+  histogram_quantile(0.99, sum(rate(rpc_request_latency_ms_bucket[5m])) by (le, env, network, chain_id))
+  ```
+- **p50 latency for a given environment:**
+  ```promql
+  histogram_quantile(0.5, sum(rate(rpc_request_latency_ms_bucket{env="staging"}[5m])) by (le, network, chain_id))
+  ```
+
+### Error rate over time
+
+- **Errors per second by env and chain:**
+  ```promql
+  sum(rate(rpc_request_errors_total[5m])) by (env, network, chain_id, rpc_provider)
+  ```
+- **Error rate for a specific RPC provider:**
+  ```promql
+  sum(rate(rpc_request_errors_total{rpc_provider="primary"}[5m])) by (env, network, chain_id)
+  ```
+
+### Request rate
+
+- **Requests per second by call type:**
+  ```promql
+  sum(rate(rpc_request_latency_ms_count[5m])) by (call, env, network)
+  ```
+
+## Enabling metrics
+
+Create `RPCClientMetrics` with `metrics.NewRPCClientMetrics(metrics.RPCClientMetricsConfig{...})` and pass it as the last argument to `multinode.NewRPCClientBase(...)`. The follow-up interface refactor will make it easier for multinode/chain integrations to supply `env`, `network`, `chain_id`, and `rpc_provider`.

metrics/rpc_client.go

@@ -0,0 +1,125 @@
+// RPC client observability using Beholder.
+//
+// This file defines rpc_request_latency_ms and rpc_request_errors_total, emitted
+// from the RPC client when RPCClientBase is constructed with a non-nil RPCClientMetrics.
+// Metrics are queryable in Prometheus/Grafana by env, network, chain_id, and rpc_provider.
+//
+// Example Prometheus/Grafana queries:
+//
+//   - Latency over time (e.g. p99 by env and chain):
+//     histogram_quantile(0.99, sum(rate(rpc_request_latency_ms_bucket[5m])) by (le, env, network, chain_id))
+//
+//   - Error rate over time (errors per second by env and chain):
+//     sum(rate(rpc_request_errors_total[5m])) by (env, network, chain_id, rpc_provider)
+//
+//   - Request rate by call type:
+//     sum(rate(rpc_request_latency_ms_count[5m])) by (call, env, network)
+package metrics
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
+
+	"github.com/smartcontractkit/chainlink-common/pkg/beholder"
+)
+
+const (
+	// RPCRequestLatencyMs is the Beholder/Prometheus metric name for RPC call latency in milliseconds.
+	RPCRequestLatencyMs = "rpc_request_latency_ms"
+	// RPCRequestErrorsTotal is the Beholder/Prometheus metric name for total RPC call errors.
+	RPCRequestErrorsTotal = "rpc_request_errors_total"
+)
+
+var (
+	rpcRequestLatencyBuckets = []float64{
+		5, 10, 25, 50, 100, 250, 500, 1000, 2500, 5000, 10000, 30000,
+	}
+	promRPCRequestLatency = promauto.NewHistogramVec(prometheus.HistogramOpts{
+		Name:    RPCRequestLatencyMs,
+		Help:    "RPC request latency in milliseconds (per call)",
+		Buckets: rpcRequestLatencyBuckets,
+	}, []string{"env", "network", "chain_id", "rpc_provider", "call"})
+	promRPCRequestErrors = promauto.NewCounterVec(prometheus.CounterOpts{
+		Name: RPCRequestErrorsTotal,
+		Help: "Total number of failed RPC requests",
+	}, []string{"env", "network", "chain_id", "rpc_provider", "call"})
+)
+
+// RPCClientMetrics records RPC latency and error metrics for observability via Beholder/Prometheus.
+// Metrics are queryable by environment, network, chain_id, and rpc_provider in Grafana.
+type RPCClientMetrics interface {
+	// RecordRequest records latency for an RPC call. If err is non-nil, also increments the error counter.
+	// callName identifies the operation (e.g. "latest_block", "latest_finalized_block").
+	RecordRequest(ctx context.Context, callName string, latencyMs float64, err error)
+}
+
+var _ RPCClientMetrics = (*rpcClientMetrics)(nil)
+
+type rpcClientMetrics struct {
+	env          string
+	network      string
+	chainID      string
+	rpcProvider  string
+	latency      metric.Float64Histogram
+	errorsTotal  metric.Int64Counter
+}
+
+// RPCClientMetricsConfig holds labels for RPC client metrics.
+// Empty strings are allowed; they will still be emitted as labels for filtering.
+type RPCClientMetricsConfig struct {
+	Env         string // e.g. "staging", "production"
+	Network     string // chain/network name
+	ChainID     string // chain ID
+	RPCProvider string // RPC provider or node name (optional)
+}
+
+// NewRPCClientMetrics creates RPC client metrics that publish to Beholder and Prometheus.
+// Callers (e.g. chain-specific RPC clients or multinode) should pass env, network, chainID, and optionally rpcProvider
+// so metrics can be queried in Grafana by environment, chain/network, and RPC provider.
+func NewRPCClientMetrics(cfg RPCClientMetricsConfig) (RPCClientMetrics, error) {
+	latency, err := beholder.GetMeter().Float64Histogram(RPCRequestLatencyMs)
+	if err != nil {
+		return nil, fmt.Errorf("failed to register RPC request latency metric: %w", err)
+	}
+	errorsTotal, err := beholder.GetMeter().Int64Counter(RPCRequestErrorsTotal)
+	if err != nil {
+		return nil, fmt.Errorf("failed to register RPC request errors metric: %w", err)
+	}
+	return &rpcClientMetrics{
+		env:         cfg.Env,
+		network:     cfg.Network,
+		chainID:     cfg.ChainID,
+		rpcProvider: cfg.RPCProvider,
+		latency:     latency,
+		errorsTotal: errorsTotal,
+	}, nil
+}
+
+func (m *rpcClientMetrics) RecordRequest(ctx context.Context, callName string, latencyMs float64, err error) {
+	attrs := metric.WithAttributes(
+		attribute.String("env", m.env),
+		attribute.String("network", m.network),
+		attribute.String("chain_id", m.chainID),
+		attribute.String("rpc_provider", m.rpcProvider),
+		attribute.String("call", callName),
+	)
+	promRPCRequestLatency.WithLabelValues(m.env, m.network, m.chainID, m.rpcProvider, callName).Observe(latencyMs)
+	m.latency.Record(ctx, latencyMs, attrs)
+	if err != nil {
+		promRPCRequestErrors.WithLabelValues(m.env, m.network, m.chainID, m.rpcProvider, callName).Inc()
+		m.errorsTotal.Add(ctx, 1, attrs)
+	}
+}
+
+// NoopRPCClientMetrics is a no-op implementation for when metrics are disabled.
+type NoopRPCClientMetrics struct{}
+
+func (NoopRPCClientMetrics) RecordRequest(context.Context, string, float64, error) {}
+
+// Ensure NoopRPCClientMetrics implements RPCClientMetrics.
+var _ RPCClientMetrics = NoopRPCClientMetrics{}

metrics/rpc_client_test.go

@@ -0,0 +1,32 @@
+package metrics
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestNewRPCClientMetrics(t *testing.T) {
+	m, err := NewRPCClientMetrics(RPCClientMetricsConfig{
+		Env:         "staging",
+		Network:     "ethereum",
+		ChainID:     "1",
+		RPCProvider: "primary",
+	})
+	require.NoError(t, err)
+	require.NotNil(t, m)
+
+	ctx := context.Background()
+	m.RecordRequest(ctx, "latest_block", 100.0, nil)
+	m.RecordRequest(ctx, "latest_block", 50.0, errors.New("rpc error"))
+}
+
+func TestNoopRPCClientMetrics_RecordRequest(t *testing.T) {
+	var m NoopRPCClientMetrics
+	ctx := context.Background()
+	m.RecordRequest(ctx, "latest_block", 100.0, nil)
+	m.RecordRequest(ctx, "latest_block", 50.0, errors.New("rpc error"))
+	// Noop should not panic
+}

multinode/go.mod

@@ -2,6 +2,8 @@ module github.com/smartcontractkit/chainlink-framework/multinode
 
 go 1.25.3
 
+replace github.com/smartcontractkit/chainlink-framework/metrics => ../metrics
+
 require (
 	github.com/jpillora/backoff v1.0.0
 	github.com/pkg/errors v0.9.1

multinode/rpc_client_base.go

@@ -9,6 +9,8 @@ import (
 
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"
 	"github.com/smartcontractkit/chainlink-common/pkg/services"
+
+	"github.com/smartcontractkit/chainlink-framework/metrics"
 )
 
 type RPCClientBaseConfig interface {
@@ -46,12 +48,18 @@ type RPCClientBase[HEAD Head] struct {
 	highestUserObservations ChainInfo
 	// most recent chain info observed during current lifecycle
 	latestChainInfo ChainInfo
+
+	// rpcMetrics is optional; when set, RPC latency and errors are reported to Beholder/Prometheus.
+	rpcMetrics metrics.RPCClientMetrics
 }
 
+// NewRPCClientBase creates an RPC client base. rpcMetrics is optional; when non-nil,
+// latency and error metrics are emitted to Beholder for observability in Prometheus/Grafana.
 func NewRPCClientBase[HEAD Head](
 	cfg RPCClientBaseConfig, ctxTimeout time.Duration, log logger.Logger,
 	latestBlock func(ctx context.Context) (HEAD, error),
 	latestFinalizedBlock func(ctx context.Context) (HEAD, error),
+	rpcMetrics metrics.RPCClientMetrics,
 ) *RPCClientBase[HEAD] {
 	return &RPCClientBase[HEAD]{
 		cfg:                  cfg,
@@ -61,6 +69,7 @@ func NewRPCClientBase[HEAD Head](
 		latestFinalizedBlock: latestFinalizedBlock,
 		subs:                 make(map[Subscription]struct{}),
 		lifeCycleCh:          make(chan struct{}),
+		rpcMetrics:           rpcMetrics,
 	}
 }
 
@@ -151,20 +160,29 @@ func (m *RPCClientBase[HEAD]) SubscribeToFinalizedHeads(ctx context.Context) (<-
 	return channel, sub, nil
 }
 
+const (
+	callLatestBlock          = "latest_block"
+	callLatestFinalizedBlock = "latest_finalized_block"
+)
+
 func (m *RPCClientBase[HEAD]) LatestBlock(ctx context.Context) (HEAD, error) {
 	// capture lifeCycleCh to ensure we are not updating chainInfo with observations related to previous life cycle
 	ctx, cancel, lifeCycleCh := m.AcquireQueryCtx(ctx, m.ctxTimeout)
 	defer cancel()
 
+	start := time.Now()
 	head, err := m.latestBlock(ctx)
+	latencyMs := float64(time.Since(start).Milliseconds())
+	if err == nil && !head.IsValid() {
+		err = errors.New("invalid head")
+	}
+	if m.rpcMetrics != nil {
+		m.rpcMetrics.RecordRequest(ctx, callLatestBlock, latencyMs, err)
+	}
 	if err != nil {
 		return head, err
 	}
 
-	if !head.IsValid() {
-		return head, errors.New("invalid head")
-	}
-
 	m.OnNewHead(ctx, lifeCycleCh, head)
 	return head, nil
 }
@@ -173,15 +191,19 @@ func (m *RPCClientBase[HEAD]) LatestFinalizedBlock(ctx context.Context) (HEAD, e
 	ctx, cancel, lifeCycleCh := m.AcquireQueryCtx(ctx, m.ctxTimeout)
 	defer cancel()
 
+	start := time.Now()
 	head, err := m.latestFinalizedBlock(ctx)
+	latencyMs := float64(time.Since(start).Milliseconds())
+	if err == nil && !head.IsValid() {
+		err = errors.New("invalid head")
+	}
+	if m.rpcMetrics != nil {
+		m.rpcMetrics.RecordRequest(ctx, callLatestFinalizedBlock, latencyMs, err)
+	}
 	if err != nil {
 		return head, err
 	}
 
-	if !head.IsValid() {
-		return head, errors.New("invalid head")
-	}
-
 	m.OnNewFinalizedHead(ctx, lifeCycleCh, head)
 	return head, nil
 }

multinode/rpc_client_base_test.go

@@ -67,7 +67,7 @@ func newTestRPC(t *testing.T) *testRPC {
 	}
 
 	rpc := &testRPC{}
-	rpc.RPCClientBase = NewRPCClientBase[*testHead](cfg, requestTimeout, lggr, rpc.latestBlock, rpc.latestBlock)
+	rpc.RPCClientBase = NewRPCClientBase[*testHead](cfg, requestTimeout, lggr, rpc.latestBlock, rpc.latestBlock, nil)
 	t.Cleanup(rpc.Close)
 	return rpc
 }