blob: f2c16c5b8eb1ec3fcafea09046d93ec5c12bdc92 [file] [log] [blame]
serkant.uluderyae5afeff2021-02-23 18:00:23 +03001package redis
2
3import (
4 "context"
5 "crypto/tls"
6 "errors"
7 "fmt"
8 "net"
9 "net/url"
10 "runtime"
11 "strconv"
12 "strings"
13 "time"
14
15 "github.com/go-redis/redis/v8/internal"
16 "github.com/go-redis/redis/v8/internal/pool"
17 "go.opentelemetry.io/otel/api/trace"
18 "go.opentelemetry.io/otel/label"
19)
20
21// Limiter is the interface of a rate limiter or a circuit breaker.
22type Limiter interface {
23 // Allow returns nil if operation is allowed or an error otherwise.
24 // If operation is allowed client must ReportResult of the operation
25 // whether it is a success or a failure.
26 Allow() error
27 // ReportResult reports the result of the previously allowed operation.
28 // nil indicates a success, non-nil error usually indicates a failure.
29 ReportResult(result error)
30}
31
32// Options keeps the settings to setup redis connection.
33type Options struct {
34 // The network type, either tcp or unix.
35 // Default is tcp.
36 Network string
37 // host:port address.
38 Addr string
39
40 // Dialer creates new network connection and has priority over
41 // Network and Addr options.
42 Dialer func(ctx context.Context, network, addr string) (net.Conn, error)
43
44 // Hook that is called when new connection is established.
45 OnConnect func(ctx context.Context, cn *Conn) error
46
47 // Use the specified Username to authenticate the current connection
48 // with one of the connections defined in the ACL list when connecting
49 // to a Redis 6.0 instance, or greater, that is using the Redis ACL system.
50 Username string
51 // Optional password. Must match the password specified in the
52 // requirepass server configuration option (if connecting to a Redis 5.0 instance, or lower),
53 // or the User Password when connecting to a Redis 6.0 instance, or greater,
54 // that is using the Redis ACL system.
55 Password string
56
57 // Database to be selected after connecting to the server.
58 DB int
59
60 // Maximum number of retries before giving up.
61 // Default is 3 retries.
62 MaxRetries int
63 // Minimum backoff between each retry.
64 // Default is 8 milliseconds; -1 disables backoff.
65 MinRetryBackoff time.Duration
66 // Maximum backoff between each retry.
67 // Default is 512 milliseconds; -1 disables backoff.
68 MaxRetryBackoff time.Duration
69
70 // Dial timeout for establishing new connections.
71 // Default is 5 seconds.
72 DialTimeout time.Duration
73 // Timeout for socket reads. If reached, commands will fail
74 // with a timeout instead of blocking. Use value -1 for no timeout and 0 for default.
75 // Default is 3 seconds.
76 ReadTimeout time.Duration
77 // Timeout for socket writes. If reached, commands will fail
78 // with a timeout instead of blocking.
79 // Default is ReadTimeout.
80 WriteTimeout time.Duration
81
82 // Maximum number of socket connections.
83 // Default is 10 connections per every CPU as reported by runtime.NumCPU.
84 PoolSize int
85 // Minimum number of idle connections which is useful when establishing
86 // new connection is slow.
87 MinIdleConns int
88 // Connection age at which client retires (closes) the connection.
89 // Default is to not close aged connections.
90 MaxConnAge time.Duration
91 // Amount of time client waits for connection if all connections
92 // are busy before returning an error.
93 // Default is ReadTimeout + 1 second.
94 PoolTimeout time.Duration
95 // Amount of time after which client closes idle connections.
96 // Should be less than server's timeout.
97 // Default is 5 minutes. -1 disables idle timeout check.
98 IdleTimeout time.Duration
99 // Frequency of idle checks made by idle connections reaper.
100 // Default is 1 minute. -1 disables idle connections reaper,
101 // but idle connections are still discarded by the client
102 // if IdleTimeout is set.
103 IdleCheckFrequency time.Duration
104
105 // Enables read only queries on slave nodes.
106 readOnly bool
107
108 // TLS Config to use. When set TLS will be negotiated.
109 TLSConfig *tls.Config
110
111 // Limiter interface used to implemented circuit breaker or rate limiter.
112 Limiter Limiter
113}
114
115func (opt *Options) init() {
116 if opt.Addr == "" {
117 opt.Addr = "localhost:6379"
118 }
119 if opt.Network == "" {
120 if strings.HasPrefix(opt.Addr, "/") {
121 opt.Network = "unix"
122 } else {
123 opt.Network = "tcp"
124 }
125 }
126 if opt.DialTimeout == 0 {
127 opt.DialTimeout = 5 * time.Second
128 }
129 if opt.Dialer == nil {
130 opt.Dialer = func(ctx context.Context, network, addr string) (net.Conn, error) {
131 netDialer := &net.Dialer{
132 Timeout: opt.DialTimeout,
133 KeepAlive: 5 * time.Minute,
134 }
135 if opt.TLSConfig == nil {
136 return netDialer.DialContext(ctx, network, addr)
137 }
138 return tls.DialWithDialer(netDialer, network, addr, opt.TLSConfig)
139 }
140 }
141 if opt.PoolSize == 0 {
142 opt.PoolSize = 10 * runtime.NumCPU()
143 }
144 switch opt.ReadTimeout {
145 case -1:
146 opt.ReadTimeout = 0
147 case 0:
148 opt.ReadTimeout = 3 * time.Second
149 }
150 switch opt.WriteTimeout {
151 case -1:
152 opt.WriteTimeout = 0
153 case 0:
154 opt.WriteTimeout = opt.ReadTimeout
155 }
156 if opt.PoolTimeout == 0 {
157 opt.PoolTimeout = opt.ReadTimeout + time.Second
158 }
159 if opt.IdleTimeout == 0 {
160 opt.IdleTimeout = 5 * time.Minute
161 }
162 if opt.IdleCheckFrequency == 0 {
163 opt.IdleCheckFrequency = time.Minute
164 }
165
166 if opt.MaxRetries == -1 {
167 opt.MaxRetries = 0
168 } else if opt.MaxRetries == 0 {
169 opt.MaxRetries = 3
170 }
171 switch opt.MinRetryBackoff {
172 case -1:
173 opt.MinRetryBackoff = 0
174 case 0:
175 opt.MinRetryBackoff = 8 * time.Millisecond
176 }
177 switch opt.MaxRetryBackoff {
178 case -1:
179 opt.MaxRetryBackoff = 0
180 case 0:
181 opt.MaxRetryBackoff = 512 * time.Millisecond
182 }
183}
184
185func (opt *Options) clone() *Options {
186 clone := *opt
187 return &clone
188}
189
190// ParseURL parses an URL into Options that can be used to connect to Redis.
191// Scheme is required.
192// There are two connection types: by tcp socket and by unix socket.
193// Tcp connection:
194// redis://<user>:<password>@<host>:<port>/<db_number>
195// Unix connection:
196// unix://<user>:<password>@</path/to/redis.sock>?db=<db_number>
197func ParseURL(redisURL string) (*Options, error) {
198 u, err := url.Parse(redisURL)
199 if err != nil {
200 return nil, err
201 }
202
203 switch u.Scheme {
204 case "redis", "rediss":
205 return setupTCPConn(u)
206 case "unix":
207 return setupUnixConn(u)
208 default:
209 return nil, fmt.Errorf("redis: invalid URL scheme: %s", u.Scheme)
210 }
211}
212
213func setupTCPConn(u *url.URL) (*Options, error) {
214 o := &Options{Network: "tcp"}
215
216 o.Username, o.Password = getUserPassword(u)
217
218 if len(u.Query()) > 0 {
219 return nil, errors.New("redis: no options supported")
220 }
221
222 h, p, err := net.SplitHostPort(u.Host)
223 if err != nil {
224 h = u.Host
225 }
226 if h == "" {
227 h = "localhost"
228 }
229 if p == "" {
230 p = "6379"
231 }
232 o.Addr = net.JoinHostPort(h, p)
233
234 f := strings.FieldsFunc(u.Path, func(r rune) bool {
235 return r == '/'
236 })
237 switch len(f) {
238 case 0:
239 o.DB = 0
240 case 1:
241 if o.DB, err = strconv.Atoi(f[0]); err != nil {
242 return nil, fmt.Errorf("redis: invalid database number: %q", f[0])
243 }
244 default:
245 return nil, fmt.Errorf("redis: invalid URL path: %s", u.Path)
246 }
247
248 if u.Scheme == "rediss" {
249 o.TLSConfig = &tls.Config{ServerName: h}
250 }
251
252 return o, nil
253}
254
255func setupUnixConn(u *url.URL) (*Options, error) {
256 o := &Options{
257 Network: "unix",
258 }
259
260 if strings.TrimSpace(u.Path) == "" { // path is required with unix connection
261 return nil, errors.New("redis: empty unix socket path")
262 }
263 o.Addr = u.Path
264
265 o.Username, o.Password = getUserPassword(u)
266
267 dbStr := u.Query().Get("db")
268 if dbStr == "" {
269 return o, nil // if database is not set, connect to 0 db.
270 }
271
272 db, err := strconv.Atoi(dbStr)
273 if err != nil {
274 return nil, fmt.Errorf("redis: invalid database number: %s", err)
275 }
276 o.DB = db
277
278 return o, nil
279}
280
281func getUserPassword(u *url.URL) (string, string) {
282 var user, password string
283 if u.User != nil {
284 user = u.User.Username()
285 if p, ok := u.User.Password(); ok {
286 password = p
287 }
288 }
289 return user, password
290}
291
292func newConnPool(opt *Options) *pool.ConnPool {
293 return pool.NewConnPool(&pool.Options{
294 Dialer: func(ctx context.Context) (net.Conn, error) {
295 var conn net.Conn
296 err := internal.WithSpan(ctx, "redis.dial", func(ctx context.Context, span trace.Span) error {
297 span.SetAttributes(
298 label.String("db.connection_string", opt.Addr),
299 )
300
301 var err error
302 conn, err = opt.Dialer(ctx, opt.Network, opt.Addr)
303 if err != nil {
304 _ = internal.RecordError(ctx, err)
305 }
306 return err
307 })
308 return conn, err
309 },
310 PoolSize: opt.PoolSize,
311 MinIdleConns: opt.MinIdleConns,
312 MaxConnAge: opt.MaxConnAge,
313 PoolTimeout: opt.PoolTimeout,
314 IdleTimeout: opt.IdleTimeout,
315 IdleCheckFrequency: opt.IdleCheckFrequency,
316 })
317}