forked from ebhomengo/niki
519 lines
14 KiB
Go
519 lines
14 KiB
Go
package redis
|
|
|
|
import (
|
|
"context"
|
|
"crypto/tls"
|
|
"errors"
|
|
"fmt"
|
|
"net"
|
|
"net/url"
|
|
"runtime"
|
|
"sort"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/redis/go-redis/v9/internal/pool"
|
|
)
|
|
|
|
// Limiter is the interface of a rate limiter or a circuit breaker.
|
|
type Limiter interface {
|
|
// Allow returns nil if operation is allowed or an error otherwise.
|
|
// If operation is allowed client must ReportResult of the operation
|
|
// whether it is a success or a failure.
|
|
Allow() error
|
|
// ReportResult reports the result of the previously allowed operation.
|
|
// nil indicates a success, non-nil error usually indicates a failure.
|
|
ReportResult(result error)
|
|
}
|
|
|
|
// Options keeps the settings to set up redis connection.
|
|
type Options struct {
|
|
// The network type, either tcp or unix.
|
|
// Default is tcp.
|
|
Network string
|
|
// host:port address.
|
|
Addr string
|
|
|
|
// ClientName will execute the `CLIENT SETNAME ClientName` command for each conn.
|
|
ClientName string
|
|
|
|
// Dialer creates new network connection and has priority over
|
|
// Network and Addr options.
|
|
Dialer func(ctx context.Context, network, addr string) (net.Conn, error)
|
|
|
|
// Hook that is called when new connection is established.
|
|
OnConnect func(ctx context.Context, cn *Conn) error
|
|
|
|
// Protocol 2 or 3. Use the version to negotiate RESP version with redis-server.
|
|
// Default is 3.
|
|
Protocol int
|
|
// Use the specified Username to authenticate the current connection
|
|
// with one of the connections defined in the ACL list when connecting
|
|
// to a Redis 6.0 instance, or greater, that is using the Redis ACL system.
|
|
Username string
|
|
// Optional password. Must match the password specified in the
|
|
// requirepass server configuration option (if connecting to a Redis 5.0 instance, or lower),
|
|
// or the User Password when connecting to a Redis 6.0 instance, or greater,
|
|
// that is using the Redis ACL system.
|
|
Password string
|
|
// CredentialsProvider allows the username and password to be updated
|
|
// before reconnecting. It should return the current username and password.
|
|
CredentialsProvider func() (username string, password string)
|
|
|
|
// Database to be selected after connecting to the server.
|
|
DB int
|
|
|
|
// Maximum number of retries before giving up.
|
|
// Default is 3 retries; -1 (not 0) disables retries.
|
|
MaxRetries int
|
|
// Minimum backoff between each retry.
|
|
// Default is 8 milliseconds; -1 disables backoff.
|
|
MinRetryBackoff time.Duration
|
|
// Maximum backoff between each retry.
|
|
// Default is 512 milliseconds; -1 disables backoff.
|
|
MaxRetryBackoff time.Duration
|
|
|
|
// Dial timeout for establishing new connections.
|
|
// Default is 5 seconds.
|
|
DialTimeout time.Duration
|
|
// Timeout for socket reads. If reached, commands will fail
|
|
// with a timeout instead of blocking. Supported values:
|
|
// - `0` - default timeout (3 seconds).
|
|
// - `-1` - no timeout (block indefinitely).
|
|
// - `-2` - disables SetReadDeadline calls completely.
|
|
ReadTimeout time.Duration
|
|
// Timeout for socket writes. If reached, commands will fail
|
|
// with a timeout instead of blocking. Supported values:
|
|
// - `0` - default timeout (3 seconds).
|
|
// - `-1` - no timeout (block indefinitely).
|
|
// - `-2` - disables SetWriteDeadline calls completely.
|
|
WriteTimeout time.Duration
|
|
// ContextTimeoutEnabled controls whether the client respects context timeouts and deadlines.
|
|
// See https://redis.uptrace.dev/guide/go-redis-debugging.html#timeouts
|
|
ContextTimeoutEnabled bool
|
|
|
|
// Type of connection pool.
|
|
// true for FIFO pool, false for LIFO pool.
|
|
// Note that FIFO has slightly higher overhead compared to LIFO,
|
|
// but it helps closing idle connections faster reducing the pool size.
|
|
PoolFIFO bool
|
|
// Base number of socket connections.
|
|
// Default is 10 connections per every available CPU as reported by runtime.GOMAXPROCS.
|
|
// If there is not enough connections in the pool, new connections will be allocated in excess of PoolSize,
|
|
// you can limit it through MaxActiveConns
|
|
PoolSize int
|
|
// Amount of time client waits for connection if all connections
|
|
// are busy before returning an error.
|
|
// Default is ReadTimeout + 1 second.
|
|
PoolTimeout time.Duration
|
|
// Minimum number of idle connections which is useful when establishing
|
|
// new connection is slow.
|
|
// Default is 0. the idle connections are not closed by default.
|
|
MinIdleConns int
|
|
// Maximum number of idle connections.
|
|
// Default is 0. the idle connections are not closed by default.
|
|
MaxIdleConns int
|
|
// Maximum number of connections allocated by the pool at a given time.
|
|
// When zero, there is no limit on the number of connections in the pool.
|
|
MaxActiveConns int
|
|
// ConnMaxIdleTime is the maximum amount of time a connection may be idle.
|
|
// Should be less than server's timeout.
|
|
//
|
|
// Expired connections may be closed lazily before reuse.
|
|
// If d <= 0, connections are not closed due to a connection's idle time.
|
|
//
|
|
// Default is 30 minutes. -1 disables idle timeout check.
|
|
ConnMaxIdleTime time.Duration
|
|
// ConnMaxLifetime is the maximum amount of time a connection may be reused.
|
|
//
|
|
// Expired connections may be closed lazily before reuse.
|
|
// If <= 0, connections are not closed due to a connection's age.
|
|
//
|
|
// Default is to not close idle connections.
|
|
ConnMaxLifetime time.Duration
|
|
|
|
// TLS Config to use. When set, TLS will be negotiated.
|
|
TLSConfig *tls.Config
|
|
|
|
// Limiter interface used to implement circuit breaker or rate limiter.
|
|
Limiter Limiter
|
|
|
|
// Enables read only queries on slave/follower nodes.
|
|
readOnly bool
|
|
|
|
// Disable set-lib on connect. Default is false.
|
|
DisableIndentity bool
|
|
|
|
// Add suffix to client name. Default is empty.
|
|
IdentitySuffix string
|
|
}
|
|
|
|
func (opt *Options) init() {
|
|
if opt.Addr == "" {
|
|
opt.Addr = "localhost:6379"
|
|
}
|
|
if opt.Network == "" {
|
|
if strings.HasPrefix(opt.Addr, "/") {
|
|
opt.Network = "unix"
|
|
} else {
|
|
opt.Network = "tcp"
|
|
}
|
|
}
|
|
if opt.DialTimeout == 0 {
|
|
opt.DialTimeout = 5 * time.Second
|
|
}
|
|
if opt.Dialer == nil {
|
|
opt.Dialer = NewDialer(opt)
|
|
}
|
|
if opt.PoolSize == 0 {
|
|
opt.PoolSize = 10 * runtime.GOMAXPROCS(0)
|
|
}
|
|
switch opt.ReadTimeout {
|
|
case -2:
|
|
opt.ReadTimeout = -1
|
|
case -1:
|
|
opt.ReadTimeout = 0
|
|
case 0:
|
|
opt.ReadTimeout = 3 * time.Second
|
|
}
|
|
switch opt.WriteTimeout {
|
|
case -2:
|
|
opt.WriteTimeout = -1
|
|
case -1:
|
|
opt.WriteTimeout = 0
|
|
case 0:
|
|
opt.WriteTimeout = opt.ReadTimeout
|
|
}
|
|
if opt.PoolTimeout == 0 {
|
|
if opt.ReadTimeout > 0 {
|
|
opt.PoolTimeout = opt.ReadTimeout + time.Second
|
|
} else {
|
|
opt.PoolTimeout = 30 * time.Second
|
|
}
|
|
}
|
|
if opt.ConnMaxIdleTime == 0 {
|
|
opt.ConnMaxIdleTime = 30 * time.Minute
|
|
}
|
|
|
|
if opt.MaxRetries == -1 {
|
|
opt.MaxRetries = 0
|
|
} else if opt.MaxRetries == 0 {
|
|
opt.MaxRetries = 3
|
|
}
|
|
switch opt.MinRetryBackoff {
|
|
case -1:
|
|
opt.MinRetryBackoff = 0
|
|
case 0:
|
|
opt.MinRetryBackoff = 8 * time.Millisecond
|
|
}
|
|
switch opt.MaxRetryBackoff {
|
|
case -1:
|
|
opt.MaxRetryBackoff = 0
|
|
case 0:
|
|
opt.MaxRetryBackoff = 512 * time.Millisecond
|
|
}
|
|
}
|
|
|
|
func (opt *Options) clone() *Options {
|
|
clone := *opt
|
|
return &clone
|
|
}
|
|
|
|
// NewDialer returns a function that will be used as the default dialer
|
|
// when none is specified in Options.Dialer.
|
|
func NewDialer(opt *Options) func(context.Context, string, string) (net.Conn, error) {
|
|
return func(ctx context.Context, network, addr string) (net.Conn, error) {
|
|
netDialer := &net.Dialer{
|
|
Timeout: opt.DialTimeout,
|
|
KeepAlive: 5 * time.Minute,
|
|
}
|
|
if opt.TLSConfig == nil {
|
|
return netDialer.DialContext(ctx, network, addr)
|
|
}
|
|
return tls.DialWithDialer(netDialer, network, addr, opt.TLSConfig)
|
|
}
|
|
}
|
|
|
|
// ParseURL parses an URL into Options that can be used to connect to Redis.
|
|
// Scheme is required.
|
|
// There are two connection types: by tcp socket and by unix socket.
|
|
// Tcp connection:
|
|
//
|
|
// redis://<user>:<password>@<host>:<port>/<db_number>
|
|
//
|
|
// Unix connection:
|
|
//
|
|
// unix://<user>:<password>@</path/to/redis.sock>?db=<db_number>
|
|
//
|
|
// Most Option fields can be set using query parameters, with the following restrictions:
|
|
// - field names are mapped using snake-case conversion: to set MaxRetries, use max_retries
|
|
// - only scalar type fields are supported (bool, int, time.Duration)
|
|
// - for time.Duration fields, values must be a valid input for time.ParseDuration();
|
|
// additionally a plain integer as value (i.e. without unit) is intepreted as seconds
|
|
// - to disable a duration field, use value less than or equal to 0; to use the default
|
|
// value, leave the value blank or remove the parameter
|
|
// - only the last value is interpreted if a parameter is given multiple times
|
|
// - fields "network", "addr", "username" and "password" can only be set using other
|
|
// URL attributes (scheme, host, userinfo, resp.), query paremeters using these
|
|
// names will be treated as unknown parameters
|
|
// - unknown parameter names will result in an error
|
|
//
|
|
// Examples:
|
|
//
|
|
// redis://user:password@localhost:6789/3?dial_timeout=3&db=1&read_timeout=6s&max_retries=2
|
|
// is equivalent to:
|
|
// &Options{
|
|
// Network: "tcp",
|
|
// Addr: "localhost:6789",
|
|
// DB: 1, // path "/3" was overridden by "&db=1"
|
|
// DialTimeout: 3 * time.Second, // no time unit = seconds
|
|
// ReadTimeout: 6 * time.Second,
|
|
// MaxRetries: 2,
|
|
// }
|
|
func ParseURL(redisURL string) (*Options, error) {
|
|
u, err := url.Parse(redisURL)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
switch u.Scheme {
|
|
case "redis", "rediss":
|
|
return setupTCPConn(u)
|
|
case "unix":
|
|
return setupUnixConn(u)
|
|
default:
|
|
return nil, fmt.Errorf("redis: invalid URL scheme: %s", u.Scheme)
|
|
}
|
|
}
|
|
|
|
func setupTCPConn(u *url.URL) (*Options, error) {
|
|
o := &Options{Network: "tcp"}
|
|
|
|
o.Username, o.Password = getUserPassword(u)
|
|
|
|
h, p := getHostPortWithDefaults(u)
|
|
o.Addr = net.JoinHostPort(h, p)
|
|
|
|
f := strings.FieldsFunc(u.Path, func(r rune) bool {
|
|
return r == '/'
|
|
})
|
|
switch len(f) {
|
|
case 0:
|
|
o.DB = 0
|
|
case 1:
|
|
var err error
|
|
if o.DB, err = strconv.Atoi(f[0]); err != nil {
|
|
return nil, fmt.Errorf("redis: invalid database number: %q", f[0])
|
|
}
|
|
default:
|
|
return nil, fmt.Errorf("redis: invalid URL path: %s", u.Path)
|
|
}
|
|
|
|
if u.Scheme == "rediss" {
|
|
o.TLSConfig = &tls.Config{
|
|
ServerName: h,
|
|
MinVersion: tls.VersionTLS12,
|
|
}
|
|
}
|
|
|
|
return setupConnParams(u, o)
|
|
}
|
|
|
|
// getHostPortWithDefaults is a helper function that splits the url into
|
|
// a host and a port. If the host is missing, it defaults to localhost
|
|
// and if the port is missing, it defaults to 6379.
|
|
func getHostPortWithDefaults(u *url.URL) (string, string) {
|
|
host, port, err := net.SplitHostPort(u.Host)
|
|
if err != nil {
|
|
host = u.Host
|
|
}
|
|
if host == "" {
|
|
host = "localhost"
|
|
}
|
|
if port == "" {
|
|
port = "6379"
|
|
}
|
|
return host, port
|
|
}
|
|
|
|
func setupUnixConn(u *url.URL) (*Options, error) {
|
|
o := &Options{
|
|
Network: "unix",
|
|
}
|
|
|
|
if strings.TrimSpace(u.Path) == "" { // path is required with unix connection
|
|
return nil, errors.New("redis: empty unix socket path")
|
|
}
|
|
o.Addr = u.Path
|
|
o.Username, o.Password = getUserPassword(u)
|
|
return setupConnParams(u, o)
|
|
}
|
|
|
|
type queryOptions struct {
|
|
q url.Values
|
|
err error
|
|
}
|
|
|
|
func (o *queryOptions) has(name string) bool {
|
|
return len(o.q[name]) > 0
|
|
}
|
|
|
|
func (o *queryOptions) string(name string) string {
|
|
vs := o.q[name]
|
|
if len(vs) == 0 {
|
|
return ""
|
|
}
|
|
delete(o.q, name) // enable detection of unknown parameters
|
|
return vs[len(vs)-1]
|
|
}
|
|
|
|
func (o *queryOptions) strings(name string) []string {
|
|
vs := o.q[name]
|
|
delete(o.q, name)
|
|
return vs
|
|
}
|
|
|
|
func (o *queryOptions) int(name string) int {
|
|
s := o.string(name)
|
|
if s == "" {
|
|
return 0
|
|
}
|
|
i, err := strconv.Atoi(s)
|
|
if err == nil {
|
|
return i
|
|
}
|
|
if o.err == nil {
|
|
o.err = fmt.Errorf("redis: invalid %s number: %s", name, err)
|
|
}
|
|
return 0
|
|
}
|
|
|
|
func (o *queryOptions) duration(name string) time.Duration {
|
|
s := o.string(name)
|
|
if s == "" {
|
|
return 0
|
|
}
|
|
// try plain number first
|
|
if i, err := strconv.Atoi(s); err == nil {
|
|
if i <= 0 {
|
|
// disable timeouts
|
|
return -1
|
|
}
|
|
return time.Duration(i) * time.Second
|
|
}
|
|
dur, err := time.ParseDuration(s)
|
|
if err == nil {
|
|
return dur
|
|
}
|
|
if o.err == nil {
|
|
o.err = fmt.Errorf("redis: invalid %s duration: %w", name, err)
|
|
}
|
|
return 0
|
|
}
|
|
|
|
func (o *queryOptions) bool(name string) bool {
|
|
switch s := o.string(name); s {
|
|
case "true", "1":
|
|
return true
|
|
case "false", "0", "":
|
|
return false
|
|
default:
|
|
if o.err == nil {
|
|
o.err = fmt.Errorf("redis: invalid %s boolean: expected true/false/1/0 or an empty string, got %q", name, s)
|
|
}
|
|
return false
|
|
}
|
|
}
|
|
|
|
func (o *queryOptions) remaining() []string {
|
|
if len(o.q) == 0 {
|
|
return nil
|
|
}
|
|
keys := make([]string, 0, len(o.q))
|
|
for k := range o.q {
|
|
keys = append(keys, k)
|
|
}
|
|
sort.Strings(keys)
|
|
return keys
|
|
}
|
|
|
|
// setupConnParams converts query parameters in u to option value in o.
|
|
func setupConnParams(u *url.URL, o *Options) (*Options, error) {
|
|
q := queryOptions{q: u.Query()}
|
|
|
|
// compat: a future major release may use q.int("db")
|
|
if tmp := q.string("db"); tmp != "" {
|
|
db, err := strconv.Atoi(tmp)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("redis: invalid database number: %w", err)
|
|
}
|
|
o.DB = db
|
|
}
|
|
|
|
o.Protocol = q.int("protocol")
|
|
o.ClientName = q.string("client_name")
|
|
o.MaxRetries = q.int("max_retries")
|
|
o.MinRetryBackoff = q.duration("min_retry_backoff")
|
|
o.MaxRetryBackoff = q.duration("max_retry_backoff")
|
|
o.DialTimeout = q.duration("dial_timeout")
|
|
o.ReadTimeout = q.duration("read_timeout")
|
|
o.WriteTimeout = q.duration("write_timeout")
|
|
o.PoolFIFO = q.bool("pool_fifo")
|
|
o.PoolSize = q.int("pool_size")
|
|
o.PoolTimeout = q.duration("pool_timeout")
|
|
o.MinIdleConns = q.int("min_idle_conns")
|
|
o.MaxIdleConns = q.int("max_idle_conns")
|
|
o.MaxActiveConns = q.int("max_active_conns")
|
|
if q.has("conn_max_idle_time") {
|
|
o.ConnMaxIdleTime = q.duration("conn_max_idle_time")
|
|
} else {
|
|
o.ConnMaxIdleTime = q.duration("idle_timeout")
|
|
}
|
|
if q.has("conn_max_lifetime") {
|
|
o.ConnMaxLifetime = q.duration("conn_max_lifetime")
|
|
} else {
|
|
o.ConnMaxLifetime = q.duration("max_conn_age")
|
|
}
|
|
if q.err != nil {
|
|
return nil, q.err
|
|
}
|
|
|
|
// any parameters left?
|
|
if r := q.remaining(); len(r) > 0 {
|
|
return nil, fmt.Errorf("redis: unexpected option: %s", strings.Join(r, ", "))
|
|
}
|
|
|
|
return o, nil
|
|
}
|
|
|
|
func getUserPassword(u *url.URL) (string, string) {
|
|
var user, password string
|
|
if u.User != nil {
|
|
user = u.User.Username()
|
|
if p, ok := u.User.Password(); ok {
|
|
password = p
|
|
}
|
|
}
|
|
return user, password
|
|
}
|
|
|
|
func newConnPool(
|
|
opt *Options,
|
|
dialer func(ctx context.Context, network, addr string) (net.Conn, error),
|
|
) *pool.ConnPool {
|
|
return pool.NewConnPool(&pool.Options{
|
|
Dialer: func(ctx context.Context) (net.Conn, error) {
|
|
return dialer(ctx, opt.Network, opt.Addr)
|
|
},
|
|
PoolFIFO: opt.PoolFIFO,
|
|
PoolSize: opt.PoolSize,
|
|
PoolTimeout: opt.PoolTimeout,
|
|
MinIdleConns: opt.MinIdleConns,
|
|
MaxIdleConns: opt.MaxIdleConns,
|
|
MaxActiveConns: opt.MaxActiveConns,
|
|
ConnMaxIdleTime: opt.ConnMaxIdleTime,
|
|
ConnMaxLifetime: opt.ConnMaxLifetime,
|
|
})
|
|
}
|