[Vendor] Update go-redis to v8.5.0 (#13749)

* Update go-redis to v8.4.0

* github.com/go-redis/redis/v8  v8.4.0 -> v8.5.0

* Apply suggestions from code review

Co-authored-by: zeripath <[email protected]>

* TODO

* Use the Queue termination channel as the default context for pushes

Signed-off-by: Andrew Thornton <[email protected]>

* missed one

Signed-off-by: Andrew Thornton <[email protected]>

Co-authored-by: zeripath <[email protected]>

Author: 6543

go.mod

@@ -35,7 +35,7 @@ require (
 	github.com/go-git/go-billy/v5 v5.0.0
 	github.com/go-git/go-git/v5 v5.2.0
 	github.com/go-ldap/ldap/v3 v3.2.4
-	github.com/go-redis/redis/v7 v7.4.0
+	github.com/go-redis/redis/v8 v8.5.0
 	github.com/go-sql-driver/mysql v1.5.0
 	github.com/go-swagger/go-swagger v0.26.0
 	github.com/go-testfixtures/testfixtures/v3 v3.4.1

go.sum

@@ -8,10 +8,11 @@ import (
 	"fmt"
 	"time"
 
+	"code.gitea.io/gitea/modules/graceful"
 	"code.gitea.io/gitea/modules/nosql"
 
 	"gitea.com/go-chi/cache"
-	"github.com/go-redis/redis/v7"
+	"github.com/go-redis/redis/v8"
 	"github.com/unknwon/com"
 )
 
@@ -28,28 +29,28 @@ type RedisCacher struct {
 func (c *RedisCacher) Put(key string, val interface{}, expire int64) error {
 	key = c.prefix + key
 	if expire == 0 {
-		if err := c.c.Set(key, com.ToStr(val), 0).Err(); err != nil {
+		if err := c.c.Set(graceful.GetManager().HammerContext(), key, com.ToStr(val), 0).Err(); err != nil {
 			return err
 		}
 	} else {
 		dur, err := time.ParseDuration(com.ToStr(expire) + "s")
 		if err != nil {
 			return err
 		}
-		if err = c.c.Set(key, com.ToStr(val), dur).Err(); err != nil {
+		if err = c.c.Set(graceful.GetManager().HammerContext(), key, com.ToStr(val), dur).Err(); err != nil {
 			return err
 		}
 	}
 
 	if c.occupyMode {
 		return nil
 	}
-	return c.c.HSet(c.hsetName, key, "0").Err()
+	return c.c.HSet(graceful.GetManager().HammerContext(), c.hsetName, key, "0").Err()
 }
 
 // Get gets cached value by given key.
 func (c *RedisCacher) Get(key string) interface{} {
-	val, err := c.c.Get(c.prefix + key).Result()
+	val, err := c.c.Get(graceful.GetManager().HammerContext(), c.prefix+key).Result()
 	if err != nil {
 		return nil
 	}
@@ -59,58 +60,58 @@ func (c *RedisCacher) Get(key string) interface{} {
 // Delete deletes cached value by given key.
 func (c *RedisCacher) Delete(key string) error {
 	key = c.prefix + key
-	if err := c.c.Del(key).Err(); err != nil {
+	if err := c.c.Del(graceful.GetManager().HammerContext(), key).Err(); err != nil {
 		return err
 	}
 
 	if c.occupyMode {
 		return nil
 	}
-	return c.c.HDel(c.hsetName, key).Err()
+	return c.c.HDel(graceful.GetManager().HammerContext(), c.hsetName, key).Err()
 }
 
 // Incr increases cached int-type value by given key as a counter.
 func (c *RedisCacher) Incr(key string) error {
 	if !c.IsExist(key) {
 		return fmt.Errorf("key '%s' not exist", key)
 	}
-	return c.c.Incr(c.prefix + key).Err()
+	return c.c.Incr(graceful.GetManager().HammerContext(), c.prefix+key).Err()
 }
 
 // Decr decreases cached int-type value by given key as a counter.
 func (c *RedisCacher) Decr(key string) error {
 	if !c.IsExist(key) {
 		return fmt.Errorf("key '%s' not exist", key)
 	}
-	return c.c.Decr(c.prefix + key).Err()
+	return c.c.Decr(graceful.GetManager().HammerContext(), c.prefix+key).Err()
 }
 
 // IsExist returns true if cached value exists.
 func (c *RedisCacher) IsExist(key string) bool {
-	if c.c.Exists(c.prefix+key).Val() == 1 {
+	if c.c.Exists(graceful.GetManager().HammerContext(), c.prefix+key).Val() == 1 {
 		return true
 	}
 
 	if !c.occupyMode {
-		c.c.HDel(c.hsetName, c.prefix+key)
+		c.c.HDel(graceful.GetManager().HammerContext(), c.hsetName, c.prefix+key)
 	}
 	return false
 }
 
 // Flush deletes all cached data.
 func (c *RedisCacher) Flush() error {
 	if c.occupyMode {
-		return c.c.FlushDB().Err()
+		return c.c.FlushDB(graceful.GetManager().HammerContext()).Err()
 	}
 
-	keys, err := c.c.HKeys(c.hsetName).Result()
+	keys, err := c.c.HKeys(graceful.GetManager().HammerContext(), c.hsetName).Result()
 	if err != nil {
 		return err
 	}
-	if err = c.c.Del(keys...).Err(); err != nil {
+	if err = c.c.Del(graceful.GetManager().HammerContext(), keys...).Err(); err != nil {
 		return err
 	}
-	return c.c.Del(c.hsetName).Err()
+	return c.c.Del(graceful.GetManager().HammerContext(), c.hsetName).Err()
 }
 
 // StartAndGC starts GC routine based on config string settings.
@@ -132,7 +133,7 @@ func (c *RedisCacher) StartAndGC(opts cache.Options) error {
 		}
 	}
 
-	return c.c.Ping().Err()
+	return c.c.Ping(graceful.GetManager().HammerContext()).Err()
 }
 
 func init() {

modules/cache/cache_redis.go

@@ -9,7 +9,7 @@ import (
 	"sync"
 	"time"
 
-	"github.com/go-redis/redis/v7"
+	"github.com/go-redis/redis/v8"
 	"github.com/syndtr/goleveldb/leveldb"
 )
 

modules/nosql/manager.go

@@ -10,7 +10,7 @@ import (
 	"strconv"
 	"strings"
 
-	"github.com/go-redis/redis/v7"
+	"github.com/go-redis/redis/v8"
 )
 
 var replacer = strings.NewReplacer("_", "", "-", "")

modules/nosql/manager_redis.go

@@ -163,6 +163,11 @@ func (q *ByteFIFOQueue) Shutdown() {
 	log.Debug("%s: %s Shutdown", q.typ, q.name)
 }
 
+// IsShutdown returns a channel which is closed when this Queue is shutdown
+func (q *ByteFIFOQueue) IsShutdown() <-chan struct{} {
+	return q.closed
+}
+
 // Terminate this queue and close the queue
 func (q *ByteFIFOQueue) Terminate() {
 	log.Trace("%s: %s Terminating", q.typ, q.name)
@@ -185,6 +190,11 @@ func (q *ByteFIFOQueue) Terminate() {
 	log.Debug("%s: %s Terminated", q.typ, q.name)
 }
 
+// IsTerminated returns a channel which is closed when this Queue is terminated
+func (q *ByteFIFOQueue) IsTerminated() <-chan struct{} {
+	return q.terminated
+}
+
 var _ (UniqueQueue) = &ByteFIFOUniqueQueue{}
 
 // ByteFIFOUniqueQueue represents a UniqueQueue formed from a UniqueByteFifo

modules/queue/queue_bytefifo.go

@@ -5,10 +5,14 @@
 package queue
 
 import (
+	"context"
+	"fmt"
+
+	"code.gitea.io/gitea/modules/graceful"
 	"code.gitea.io/gitea/modules/log"
 	"code.gitea.io/gitea/modules/nosql"
 
-	"github.com/go-redis/redis/v7"
+	"github.com/go-redis/redis/v8"
 )
 
 // RedisQueueType is the type for redis queue
@@ -43,6 +47,8 @@ func NewRedisQueue(handle HandlerFunc, cfg, exemplar interface{}) (Queue, error)
 		return nil, err
 	}
 
+	byteFIFO.ctx = graceful.NewChannelContext(byteFIFOQueue.IsTerminated(), fmt.Errorf("queue has been terminated"))
+
 	queue := &RedisQueue{
 		ByteFIFOQueue: byteFIFOQueue,
 	}
@@ -53,20 +59,21 @@ func NewRedisQueue(handle HandlerFunc, cfg, exemplar interface{}) (Queue, error)
 }
 
 type redisClient interface {
-	RPush(key string, args ...interface{}) *redis.IntCmd
-	LPop(key string) *redis.StringCmd
-	LLen(key string) *redis.IntCmd
-	SAdd(key string, members ...interface{}) *redis.IntCmd
-	SRem(key string, members ...interface{}) *redis.IntCmd
-	SIsMember(key string, member interface{}) *redis.BoolCmd
-	Ping() *redis.StatusCmd
+	RPush(ctx context.Context, key string, args ...interface{}) *redis.IntCmd
+	LPop(ctx context.Context, key string) *redis.StringCmd
+	LLen(ctx context.Context, key string) *redis.IntCmd
+	SAdd(ctx context.Context, key string, members ...interface{}) *redis.IntCmd
+	SRem(ctx context.Context, key string, members ...interface{}) *redis.IntCmd
+	SIsMember(ctx context.Context, key string, member interface{}) *redis.BoolCmd
+	Ping(ctx context.Context) *redis.StatusCmd
 	Close() error
 }
 
 var _ (ByteFIFO) = &RedisByteFIFO{}
 
 // RedisByteFIFO represents a ByteFIFO formed from a redisClient
 type RedisByteFIFO struct {
+	ctx       context.Context
 	client    redisClient
 	queueName string
 }
@@ -82,8 +89,9 @@ func NewRedisByteFIFO(config RedisByteFIFOConfiguration) (*RedisByteFIFO, error)
 	fifo := &RedisByteFIFO{
 		queueName: config.QueueName,
 	}
+	fifo.ctx = graceful.GetManager().TerminateContext()
 	fifo.client = nosql.GetManager().GetRedisClient(config.ConnectionString)
-	if err := fifo.client.Ping().Err(); err != nil {
+	if err := fifo.client.Ping(graceful.GetManager().ShutdownContext()).Err(); err != nil {
 		return nil, err
 	}
 	return fifo, nil
@@ -96,12 +104,12 @@ func (fifo *RedisByteFIFO) PushFunc(data []byte, fn func() error) error {
 			return err
 		}
 	}
-	return fifo.client.RPush(fifo.queueName, data).Err()
+	return fifo.client.RPush(fifo.ctx, fifo.queueName, data).Err()
 }
 
 // Pop pops data from the start of the fifo
 func (fifo *RedisByteFIFO) Pop() ([]byte, error) {
-	data, err := fifo.client.LPop(fifo.queueName).Bytes()
+	data, err := fifo.client.LPop(fifo.ctx, fifo.queueName).Bytes()
 	if err == nil || err == redis.Nil {
 		return data, nil
 	}
@@ -115,7 +123,7 @@ func (fifo *RedisByteFIFO) Close() error {
 
 // Len returns the length of the fifo
 func (fifo *RedisByteFIFO) Len() int64 {
-	val, err := fifo.client.LLen(fifo.queueName).Result()
+	val, err := fifo.client.LLen(fifo.ctx, fifo.queueName).Result()
 	if err != nil {
 		log.Error("Error whilst getting length of redis queue %s: Error: %v", fifo.queueName, err)
 		return -1

modules/queue/queue_redis.go

@@ -4,7 +4,12 @@
 
 package queue
 
-import "github.com/go-redis/redis/v7"
+import (
+	"fmt"
+
+	"code.gitea.io/gitea/modules/graceful"
+	"github.com/go-redis/redis/v8"
+)
 
 // RedisUniqueQueueType is the type for redis queue
 const RedisUniqueQueueType Type = "unique-redis"
@@ -46,6 +51,8 @@ func NewRedisUniqueQueue(handle HandlerFunc, cfg, exemplar interface{}) (Queue,
 		return nil, err
 	}
 
+	byteFIFO.ctx = graceful.NewChannelContext(byteFIFOQueue.IsTerminated(), fmt.Errorf("queue has been terminated"))
+
 	queue := &RedisUniqueQueue{
 		ByteFIFOUniqueQueue: byteFIFOQueue,
 	}
@@ -86,7 +93,7 @@ func NewRedisUniqueByteFIFO(config RedisUniqueByteFIFOConfiguration) (*RedisUniq
 
 // PushFunc pushes data to the end of the fifo and calls the callback if it is added
 func (fifo *RedisUniqueByteFIFO) PushFunc(data []byte, fn func() error) error {
-	added, err := fifo.client.SAdd(fifo.setName, data).Result()
+	added, err := fifo.client.SAdd(fifo.ctx, fifo.setName, data).Result()
 	if err != nil {
 		return err
 	}
@@ -98,12 +105,12 @@ func (fifo *RedisUniqueByteFIFO) PushFunc(data []byte, fn func() error) error {
 			return err
 		}
 	}
-	return fifo.client.RPush(fifo.queueName, data).Err()
+	return fifo.client.RPush(fifo.ctx, fifo.queueName, data).Err()
 }
 
 // Pop pops data from the start of the fifo
 func (fifo *RedisUniqueByteFIFO) Pop() ([]byte, error) {
-	data, err := fifo.client.LPop(fifo.queueName).Bytes()
+	data, err := fifo.client.LPop(fifo.ctx, fifo.queueName).Bytes()
 	if err != nil && err != redis.Nil {
 		return data, err
 	}
@@ -112,13 +119,13 @@ func (fifo *RedisUniqueByteFIFO) Pop() ([]byte, error) {
 		return data, nil
 	}
 
-	err = fifo.client.SRem(fifo.setName, data).Err()
+	err = fifo.client.SRem(fifo.ctx, fifo.setName, data).Err()
 	return data, err
 }
 
 // Has returns whether the fifo contains this data
 func (fifo *RedisUniqueByteFIFO) Has(data []byte) (bool, error) {
-	return fifo.client.SIsMember(fifo.setName, data).Result()
+	return fifo.client.SIsMember(fifo.ctx, fifo.setName, data).Result()
 }
 
 func init() {