package redis import ( "context" "errors" "fmt" "sync/atomic" "time" "github.com/go-redis/redis/v8/internal" "github.com/go-redis/redis/v8/internal/pool" "github.com/go-redis/redis/v8/internal/proto" "go.opentelemetry.io/otel/attribute" ) // Nil reply returned by Redis when key does not exist. const Nil = proto.Nil func SetLogger(logger internal.Logging) { internal.Logger = logger } //------------------------------------------------------------------------------ type Hook interface { BeforeProcess(ctx context.Context, cmd Cmder) (context.Context, error) AfterProcess(ctx context.Context, cmd Cmder) error BeforeProcessPipeline(ctx context.Context, cmds []Cmder) (context.Context, error) AfterProcessPipeline(ctx context.Context, cmds []Cmder) error } type hooks struct { hooks []Hook } func (hs *hooks) lock() { hs.hooks = hs.hooks[:len(hs.hooks):len(hs.hooks)] } func (hs hooks) clone() hooks { clone := hs clone.lock() return clone } func (hs *hooks) AddHook(hook Hook) { hs.hooks = append(hs.hooks, hook) } func (hs hooks) process( ctx context.Context, cmd Cmder, fn func(context.Context, Cmder) error, ) error { if len(hs.hooks) == 0 { err := hs.withContext(ctx, func() error { return fn(ctx, cmd) }) cmd.SetErr(err) return err } var hookIndex int var retErr error for ; hookIndex < len(hs.hooks) && retErr == nil; hookIndex++ { ctx, retErr = hs.hooks[hookIndex].BeforeProcess(ctx, cmd) if retErr != nil { cmd.SetErr(retErr) } } if retErr == nil { retErr = hs.withContext(ctx, func() error { return fn(ctx, cmd) }) cmd.SetErr(retErr) } for hookIndex--; hookIndex >= 0; hookIndex-- { if err := hs.hooks[hookIndex].AfterProcess(ctx, cmd); err != nil { retErr = err cmd.SetErr(retErr) } } return retErr } func (hs hooks) processPipeline( ctx context.Context, cmds []Cmder, fn func(context.Context, []Cmder) error, ) error { if len(hs.hooks) == 0 { err := hs.withContext(ctx, func() error { return fn(ctx, cmds) }) return err } var hookIndex int var retErr error for ; hookIndex < len(hs.hooks) && retErr == nil; hookIndex++ { ctx, retErr = hs.hooks[hookIndex].BeforeProcessPipeline(ctx, cmds) if retErr != nil { setCmdsErr(cmds, retErr) } } if retErr == nil { retErr = hs.withContext(ctx, func() error { return fn(ctx, cmds) }) } for hookIndex--; hookIndex >= 0; hookIndex-- { if err := hs.hooks[hookIndex].AfterProcessPipeline(ctx, cmds); err != nil { retErr = err setCmdsErr(cmds, retErr) } } return retErr } func (hs hooks) processTxPipeline( ctx context.Context, cmds []Cmder, fn func(context.Context, []Cmder) error, ) error { cmds = wrapMultiExec(ctx, cmds) return hs.processPipeline(ctx, cmds, fn) } func (hs hooks) withContext(ctx context.Context, fn func() error) error { return fn() } //------------------------------------------------------------------------------ type baseClient struct { opt *Options connPool pool.Pooler onClose func() error // hook called when client is closed } func newBaseClient(opt *Options, connPool pool.Pooler) *baseClient { return &baseClient{ opt: opt, connPool: connPool, } } func (c *baseClient) clone() *baseClient { clone := *c return &clone } func (c *baseClient) withTimeout(timeout time.Duration) *baseClient { opt := c.opt.clone() opt.ReadTimeout = timeout opt.WriteTimeout = timeout clone := c.clone() clone.opt = opt return clone } func (c *baseClient) String() string { return fmt.Sprintf("Redis<%s db:%d>", c.getAddr(), c.opt.DB) } func (c *baseClient) newConn(ctx context.Context) (*pool.Conn, error) { cn, err := c.connPool.NewConn(ctx) if err != nil { return nil, err } err = c.initConn(ctx, cn) if err != nil { _ = c.connPool.CloseConn(cn) return nil, err } return cn, nil } func (c *baseClient) getConn(ctx context.Context) (*pool.Conn, error) { if c.opt.Limiter != nil { err := c.opt.Limiter.Allow() if err != nil { return nil, err } } cn, err := c._getConn(ctx) if err != nil { if c.opt.Limiter != nil { c.opt.Limiter.ReportResult(err) } return nil, err } return cn, nil } func (c *baseClient) _getConn(ctx context.Context) (*pool.Conn, error) { cn, err := c.connPool.Get(ctx) if err != nil { return nil, err } if cn.Inited { return cn, nil } if err := c.initConn(ctx, cn); err != nil { c.connPool.Remove(ctx, cn, err) if err := errors.Unwrap(err); err != nil { return nil, err } return nil, err } return cn, nil } func (c *baseClient) initConn(ctx context.Context, cn *pool.Conn) error { if cn.Inited { return nil } cn.Inited = true if c.opt.Password == "" && c.opt.DB == 0 && !c.opt.readOnly && c.opt.OnConnect == nil { return nil } ctx, span := internal.StartSpan(ctx, "redis.init_conn") defer span.End() connPool := pool.NewSingleConnPool(c.connPool, cn) conn := newConn(ctx, c.opt, connPool) _, err := conn.Pipelined(ctx, func(pipe Pipeliner) error { if c.opt.Password != "" { if c.opt.Username != "" { pipe.AuthACL(ctx, c.opt.Username, c.opt.Password) } else { pipe.Auth(ctx, c.opt.Password) } } if c.opt.DB > 0 { pipe.Select(ctx, c.opt.DB) } if c.opt.readOnly { pipe.ReadOnly(ctx) } return nil }) if err != nil { return err } if c.opt.OnConnect != nil { return c.opt.OnConnect(ctx, conn) } return nil } func (c *baseClient) releaseConn(ctx context.Context, cn *pool.Conn, err error) { if c.opt.Limiter != nil { c.opt.Limiter.ReportResult(err) } if isBadConn(err, false) { c.connPool.Remove(ctx, cn, err) } else { c.connPool.Put(ctx, cn) } } func (c *baseClient) withConn( ctx context.Context, fn func(context.Context, *pool.Conn) error, ) error { ctx, span := internal.StartSpan(ctx, "redis.with_conn") defer span.End() cn, err := c.getConn(ctx) if err != nil { return err } if span.IsRecording() { if remoteAddr := cn.RemoteAddr(); remoteAddr != nil { span.SetAttributes(attribute.String("net.peer.ip", remoteAddr.String())) } } defer func() { c.releaseConn(ctx, cn, err) }() done := ctx.Done() //nolint:ifshort if done == nil { err = fn(ctx, cn) return err } errc := make(chan error, 1) go func() { errc <- fn(ctx, cn) }() select { case <-done: _ = cn.Close() // Wait for the goroutine to finish and send something. <-errc err = ctx.Err() return err case err = <-errc: return err } } func (c *baseClient) process(ctx context.Context, cmd Cmder) error { var lastErr error for attempt := 0; attempt <= c.opt.MaxRetries; attempt++ { attempt := attempt retry, err := c._process(ctx, cmd, attempt) if err == nil || !retry { return err } lastErr = err } return lastErr } func (c *baseClient) _process(ctx context.Context, cmd Cmder, attempt int) (bool, error) { if attempt > 0 { if err := internal.Sleep(ctx, c.retryBackoff(attempt)); err != nil { return false, err } } retryTimeout := uint32(1) err := c.withConn(ctx, func(ctx context.Context, cn *pool.Conn) error { err := cn.WithWriter(ctx, c.opt.WriteTimeout, func(wr *proto.Writer) error { return writeCmd(wr, cmd) }) if err != nil { return err } err = cn.WithReader(ctx, c.cmdTimeout(cmd), cmd.readReply) if err != nil { if cmd.readTimeout() == nil { atomic.StoreUint32(&retryTimeout, 1) } return err } return nil }) if err == nil { return false, nil } retry := shouldRetry(err, atomic.LoadUint32(&retryTimeout) == 1) return retry, err } func (c *baseClient) retryBackoff(attempt int) time.Duration { return internal.RetryBackoff(attempt, c.opt.MinRetryBackoff, c.opt.MaxRetryBackoff) } func (c *baseClient) cmdTimeout(cmd Cmder) time.Duration { if timeout := cmd.readTimeout(); timeout != nil { t := *timeout if t == 0 { return 0 } return t + 10*time.Second } return c.opt.ReadTimeout } // Close closes the client, releasing any open resources. // // It is rare to Close a Client, as the Client is meant to be // long-lived and shared between many goroutines. func (c *baseClient) Close() error { var firstErr error if c.onClose != nil { if err := c.onClose(); err != nil { firstErr = err } } if err := c.connPool.Close(); err != nil && firstErr == nil { firstErr = err } return firstErr } func (c *baseClient) getAddr() string { return c.opt.Addr } func (c *baseClient) processPipeline(ctx context.Context, cmds []Cmder) error { return c.generalProcessPipeline(ctx, cmds, c.pipelineProcessCmds) } func (c *baseClient) processTxPipeline(ctx context.Context, cmds []Cmder) error { return c.generalProcessPipeline(ctx, cmds, c.txPipelineProcessCmds) } type pipelineProcessor func(context.Context, *pool.Conn, []Cmder) (bool, error) func (c *baseClient) generalProcessPipeline( ctx context.Context, cmds []Cmder, p pipelineProcessor, ) error { err := c._generalProcessPipeline(ctx, cmds, p) if err != nil { setCmdsErr(cmds, err) return err } return cmdsFirstErr(cmds) } func (c *baseClient) _generalProcessPipeline( ctx context.Context, cmds []Cmder, p pipelineProcessor, ) error { var lastErr error for attempt := 0; attempt <= c.opt.MaxRetries; attempt++ { if attempt > 0 { if err := internal.Sleep(ctx, c.retryBackoff(attempt)); err != nil { return err } } var canRetry bool lastErr = c.withConn(ctx, func(ctx context.Context, cn *pool.Conn) error { var err error canRetry, err = p(ctx, cn, cmds) return err }) if lastErr == nil || !canRetry || !shouldRetry(lastErr, true) { return lastErr } } return lastErr } func (c *baseClient) pipelineProcessCmds( ctx context.Context, cn *pool.Conn, cmds []Cmder, ) (bool, error) { err := cn.WithWriter(ctx, c.opt.WriteTimeout, func(wr *proto.Writer) error { return writeCmds(wr, cmds) }) if err != nil { return true, err } err = cn.WithReader(ctx, c.opt.ReadTimeout, func(rd *proto.Reader) error { return pipelineReadCmds(rd, cmds) }) return true, err } func pipelineReadCmds(rd *proto.Reader, cmds []Cmder) error { for _, cmd := range cmds { err := cmd.readReply(rd) cmd.SetErr(err) if err != nil && !isRedisError(err) { return err } } return nil } func (c *baseClient) txPipelineProcessCmds( ctx context.Context, cn *pool.Conn, cmds []Cmder, ) (bool, error) { err := cn.WithWriter(ctx, c.opt.WriteTimeout, func(wr *proto.Writer) error { return writeCmds(wr, cmds) }) if err != nil { return true, err } err = cn.WithReader(ctx, c.opt.ReadTimeout, func(rd *proto.Reader) error { statusCmd := cmds[0].(*StatusCmd) // Trim multi and exec. cmds = cmds[1 : len(cmds)-1] err := txPipelineReadQueued(rd, statusCmd, cmds) if err != nil { return err } return pipelineReadCmds(rd, cmds) }) return false, err } func wrapMultiExec(ctx context.Context, cmds []Cmder) []Cmder { if len(cmds) == 0 { panic("not reached") } cmdCopy := make([]Cmder, len(cmds)+2) cmdCopy[0] = NewStatusCmd(ctx, "multi") copy(cmdCopy[1:], cmds) cmdCopy[len(cmdCopy)-1] = NewSliceCmd(ctx, "exec") return cmdCopy } func txPipelineReadQueued(rd *proto.Reader, statusCmd *StatusCmd, cmds []Cmder) error { // Parse queued replies. if err := statusCmd.readReply(rd); err != nil { return err } for range cmds { if err := statusCmd.readReply(rd); err != nil && !isRedisError(err) { return err } } // Parse number of replies. line, err := rd.ReadLine() if err != nil { if err == Nil { err = TxFailedErr } return err } switch line[0] { case proto.ErrorReply: return proto.ParseErrorReply(line) case proto.ArrayReply: // ok default: err := fmt.Errorf("redis: expected '*', but got line %q", line) return err } return nil } //------------------------------------------------------------------------------ // Client is a Redis client representing a pool of zero or more // underlying connections. It's safe for concurrent use by multiple // goroutines. type Client struct { *baseClient cmdable hooks ctx context.Context } // NewClient returns a client to the Redis Server specified by Options. func NewClient(opt *Options) *Client { opt.init() c := Client{ baseClient: newBaseClient(opt, newConnPool(opt)), ctx: context.Background(), } c.cmdable = c.Process return &c } func (c *Client) clone() *Client { clone := *c clone.cmdable = clone.Process clone.hooks.lock() return &clone } func (c *Client) WithTimeout(timeout time.Duration) *Client { clone := c.clone() clone.baseClient = c.baseClient.withTimeout(timeout) return clone } func (c *Client) Context() context.Context { return c.ctx } func (c *Client) WithContext(ctx context.Context) *Client { if ctx == nil { panic("nil context") } clone := c.clone() clone.ctx = ctx return clone } func (c *Client) Conn(ctx context.Context) *Conn { return newConn(ctx, c.opt, pool.NewStickyConnPool(c.connPool)) } // Do creates a Cmd from the args and processes the cmd. func (c *Client) Do(ctx context.Context, args ...interface{}) *Cmd { cmd := NewCmd(ctx, args...) _ = c.Process(ctx, cmd) return cmd } func (c *Client) Process(ctx context.Context, cmd Cmder) error { return c.hooks.process(ctx, cmd, c.baseClient.process) } func (c *Client) processPipeline(ctx context.Context, cmds []Cmder) error { return c.hooks.processPipeline(ctx, cmds, c.baseClient.processPipeline) } func (c *Client) processTxPipeline(ctx context.Context, cmds []Cmder) error { return c.hooks.processTxPipeline(ctx, cmds, c.baseClient.processTxPipeline) } // Options returns read-only Options that were used to create the client. func (c *Client) Options() *Options { return c.opt } type PoolStats pool.Stats // PoolStats returns connection pool stats. func (c *Client) PoolStats() *PoolStats { stats := c.connPool.Stats() return (*PoolStats)(stats) } func (c *Client) Pipelined(ctx context.Context, fn func(Pipeliner) error) ([]Cmder, error) { return c.Pipeline().Pipelined(ctx, fn) } func (c *Client) Pipeline() Pipeliner { pipe := Pipeline{ ctx: c.ctx, exec: c.processPipeline, } pipe.init() return &pipe } func (c *Client) TxPipelined(ctx context.Context, fn func(Pipeliner) error) ([]Cmder, error) { return c.TxPipeline().Pipelined(ctx, fn) } // TxPipeline acts like Pipeline, but wraps queued commands with MULTI/EXEC. func (c *Client) TxPipeline() Pipeliner { pipe := Pipeline{ ctx: c.ctx, exec: c.processTxPipeline, } pipe.init() return &pipe } func (c *Client) pubSub() *PubSub { pubsub := &PubSub{ opt: c.opt, newConn: func(ctx context.Context, channels []string) (*pool.Conn, error) { return c.newConn(ctx) }, closeConn: c.connPool.CloseConn, } pubsub.init() return pubsub } // Subscribe subscribes the client to the specified channels. // Channels can be omitted to create empty subscription. // Note that this method does not wait on a response from Redis, so the // subscription may not be active immediately. To force the connection to wait, // you may call the Receive() method on the returned *PubSub like so: // // sub := client.Subscribe(queryResp) // iface, err := sub.Receive() // if err != nil { // // handle error // } // // // Should be *Subscription, but others are possible if other actions have been // // taken on sub since it was created. // switch iface.(type) { // case *Subscription: // // subscribe succeeded // case *Message: // // received first message // case *Pong: // // pong received // default: // // handle error // } // // ch := sub.Channel() func (c *Client) Subscribe(ctx context.Context, channels ...string) *PubSub { pubsub := c.pubSub() if len(channels) > 0 { _ = pubsub.Subscribe(ctx, channels...) } return pubsub } // PSubscribe subscribes the client to the given patterns. // Patterns can be omitted to create empty subscription. func (c *Client) PSubscribe(ctx context.Context, channels ...string) *PubSub { pubsub := c.pubSub() if len(channels) > 0 { _ = pubsub.PSubscribe(ctx, channels...) } return pubsub } //------------------------------------------------------------------------------ type conn struct { baseClient cmdable statefulCmdable hooks // TODO: inherit hooks } // Conn is like Client, but its pool contains single connection. type Conn struct { *conn ctx context.Context } func newConn(ctx context.Context, opt *Options, connPool pool.Pooler) *Conn { c := Conn{ conn: &conn{ baseClient: baseClient{ opt: opt, connPool: connPool, }, }, ctx: ctx, } c.cmdable = c.Process c.statefulCmdable = c.Process return &c } func (c *Conn) Process(ctx context.Context, cmd Cmder) error { return c.hooks.process(ctx, cmd, c.baseClient.process) } func (c *Conn) processPipeline(ctx context.Context, cmds []Cmder) error { return c.hooks.processPipeline(ctx, cmds, c.baseClient.processPipeline) } func (c *Conn) processTxPipeline(ctx context.Context, cmds []Cmder) error { return c.hooks.processTxPipeline(ctx, cmds, c.baseClient.processTxPipeline) } func (c *Conn) Pipelined(ctx context.Context, fn func(Pipeliner) error) ([]Cmder, error) { return c.Pipeline().Pipelined(ctx, fn) } func (c *Conn) Pipeline() Pipeliner { pipe := Pipeline{ ctx: c.ctx, exec: c.processPipeline, } pipe.init() return &pipe } func (c *Conn) TxPipelined(ctx context.Context, fn func(Pipeliner) error) ([]Cmder, error) { return c.TxPipeline().Pipelined(ctx, fn) } // TxPipeline acts like Pipeline, but wraps queued commands with MULTI/EXEC. func (c *Conn) TxPipeline() Pipeliner { pipe := Pipeline{ ctx: c.ctx, exec: c.processTxPipeline, } pipe.init() return &pipe }