1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
package client
import (
"context"
"fmt"
"regexp"
"time"
"github.com/ethereum-optimism/optimism/op-service/backoff"
"github.com/ethereum/go-ethereum"
"github.com/ethereum/go-ethereum/log"
"github.com/prometheus/client_golang/prometheus"
"github.com/ethereum-optimism/optimism/op-node/metrics"
"github.com/ethereum/go-ethereum/rpc"
)
var httpRegex = regexp.MustCompile("^http(s)?://")
type RPC interface {
Close()
CallContext(ctx context.Context, result any, method string, args ...any) error
BatchCallContext(ctx context.Context, b []rpc.BatchElem) error
EthSubscribe(ctx context.Context, channel any, args ...any) (ethereum.Subscription, error)
}
// NewRPC returns the correct client.RPC instance for a given RPC url.
func NewRPC(ctx context.Context, lgr log.Logger, addr string, opts ...rpc.ClientOption) (RPC, error) {
underlying, err := DialRPCClientWithBackoff(ctx, lgr, addr, opts...)
if err != nil {
return nil, err
}
wrapped := &BaseRPCClient{
c: underlying,
}
if httpRegex.MatchString(addr) {
return NewPollingClient(ctx, lgr, wrapped), nil
}
return wrapped, nil
}
// Dials a JSON-RPC endpoint repeatedly, with a backoff, until a client connection is established. Auth is optional.
func DialRPCClientWithBackoff(ctx context.Context, log log.Logger, addr string, opts ...rpc.ClientOption) (*rpc.Client, error) {
bOff := backoff.Exponential()
var ret *rpc.Client
err := backoff.DoCtx(ctx, 10, bOff, func() error {
client, err := rpc.DialOptions(ctx, addr, opts...)
if err != nil {
if client == nil {
return fmt.Errorf("failed to dial address (%s): %w", addr, err)
}
log.Warn("failed to dial address, but may connect later", "addr", addr, "err", err)
}
ret = client
return nil
})
if err != nil {
return nil, err
}
return ret, nil
}
// BaseRPCClient is a wrapper around a concrete *rpc.Client instance to make it compliant
// with the client.RPC interface.
// It sets a timeout of 10s on CallContext & 20s on BatchCallContext made through it.
type BaseRPCClient struct {
c *rpc.Client
}
func NewBaseRPCClient(c *rpc.Client) *BaseRPCClient {
return &BaseRPCClient{c: c}
}
func (b *BaseRPCClient) Close() {
b.c.Close()
}
func (b *BaseRPCClient) CallContext(ctx context.Context, result any, method string, args ...any) error {
cCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
defer cancel()
return b.c.CallContext(cCtx, result, method, args...)
}
func (b *BaseRPCClient) BatchCallContext(ctx context.Context, batch []rpc.BatchElem) error {
cCtx, cancel := context.WithTimeout(ctx, 20*time.Second)
defer cancel()
return b.c.BatchCallContext(cCtx, batch)
}
func (b *BaseRPCClient) EthSubscribe(ctx context.Context, channel any, args ...any) (ethereum.Subscription, error) {
return b.c.EthSubscribe(ctx, channel, args...)
}
// InstrumentedRPCClient is an RPC client that tracks
// Prometheus metrics for each call.
type InstrumentedRPCClient struct {
c RPC
m *metrics.Metrics
}
// NewInstrumentedRPC creates a new instrumented RPC client.
func NewInstrumentedRPC(c RPC, m *metrics.Metrics) *InstrumentedRPCClient {
return &InstrumentedRPCClient{
c: c,
m: m,
}
}
func (ic *InstrumentedRPCClient) Close() {
ic.c.Close()
}
func (ic *InstrumentedRPCClient) CallContext(ctx context.Context, result any, method string, args ...any) error {
return instrument1(ic.m, method, func() error {
return ic.c.CallContext(ctx, result, method, args...)
})
}
func (ic *InstrumentedRPCClient) BatchCallContext(ctx context.Context, b []rpc.BatchElem) error {
return instrumentBatch(ic.m, func() error {
return ic.c.BatchCallContext(ctx, b)
}, b)
}
func (ic *InstrumentedRPCClient) EthSubscribe(ctx context.Context, channel any, args ...any) (ethereum.Subscription, error) {
return ic.c.EthSubscribe(ctx, channel, args...)
}
// instrumentBatch handles metrics for batch calls. Request metrics are
// increased for each batch element. Request durations are tracked for
// the batch as a whole using a special <batch> method. Errors are tracked
// for each individual batch response, unless the overall request fails in
// which case the <batch> method is used.
func instrumentBatch(m *metrics.Metrics, cb func() error, b []rpc.BatchElem) error {
m.RPCClientRequestsTotal.WithLabelValues(metrics.BatchMethod).Inc()
for _, elem := range b {
m.RPCClientRequestsTotal.WithLabelValues(elem.Method).Inc()
}
timer := prometheus.NewTimer(m.RPCClientRequestDurationSeconds.WithLabelValues(metrics.BatchMethod))
defer timer.ObserveDuration()
// Track response times for batch requests separately.
if err := cb(); err != nil {
m.RecordRPCClientResponse(metrics.BatchMethod, err)
return err
}
for _, elem := range b {
m.RecordRPCClientResponse(elem.Method, elem.Error)
}
return nil
}