Skip to content

Commit 93c63f9

Browse files
stefannegreastefannegrea
authored
Merge pull request #90 from digitalocean/agandhi/redis-v9
redis: upgrade driver to v9
2 parents 69d6cdd + ea5199a commit 93c63f9

10 files changed

Lines changed: 76 additions & 147 deletions

File tree

‎.github/workflows/ci.yml‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -13,14 +13,14 @@ jobs:
1313
- 6379:6379
1414

1515
steps:
16-
- name: Set up Go 1.13
17-
uses: actions/setup-go@v2
16+
- name: Set up Go
17+
uses: actions/setup-go@v6
1818
with:
19-
go-version: 1.13
19+
go-version: 1.26
2020
id: go
2121

2222
- name: Checkout code
23-
uses: actions/checkout@v2
23+
uses: actions/checkout@v6
2424

2525
- name: Test
2626
run: make test

‎go.mod‎

Lines changed: 15 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -1,23 +1,25 @@
11
module github.com/digitalocean/go-workers2
22

3+
go 1.21
4+
35
require (
46
github.com/bitly/go-simplejson v0.5.0
5-
github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869 // indirect
6-
github.com/go-redis/redis/v8 v8.4.4
7-
github.com/golang/protobuf v1.4.3 // indirect
87
github.com/google/uuid v1.1.4
9-
github.com/kr/text v0.2.0 // indirect
10-
github.com/nxadm/tail v1.4.6 // indirect
8+
github.com/redis/go-redis/v9 v9.18.0
119
github.com/spf13/cobra v1.1.1
1210
github.com/stretchr/testify v1.6.1
13-
golang.org/x/net v0.0.0-20201224014010-6772e930b67b // indirect
14-
golang.org/x/sys v0.0.0-20210105210732-16f7687f5001 // indirect
15-
golang.org/x/text v0.3.4 // indirect
16-
golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect
17-
google.golang.org/protobuf v1.25.0 // indirect
11+
)
12+
13+
require (
14+
github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869 // indirect
15+
github.com/cespare/xxhash/v2 v2.3.0 // indirect
16+
github.com/davecgh/go-spew v1.1.1 // indirect
17+
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
18+
github.com/inconshreveable/mousetrap v1.0.0 // indirect
19+
github.com/kr/text v0.2.0 // indirect
20+
github.com/pmezard/go-difflib v1.0.0 // indirect
21+
github.com/spf13/pflag v1.0.5 // indirect
22+
go.uber.org/atomic v1.11.0 // indirect
1823
gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c // indirect
19-
gopkg.in/yaml.v2 v2.4.0 // indirect
2024
gopkg.in/yaml.v3 v3.0.0-20210105161348-2e78108cf5f8 // indirect
2125
)
22-
23-
go 1.13

‎go.sum‎

Lines changed: 14 additions & 83 deletions
Large diffs are not rendered by default.

‎manager.go‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -9,8 +9,8 @@ import (
99
"time"
1010

1111
"github.com/digitalocean/go-workers2/storage"
12-
"github.com/go-redis/redis/v8"
1312
"github.com/google/uuid"
13+
"github.com/redis/go-redis/v9"
1414
)
1515

1616
// Manager coordinates work, workers, and signaling needed for job processing
@@ -61,6 +61,7 @@ func NewManager(options Options) (*Manager, error) {
6161
}
6262

6363
// GetRedisClient returns the Redis client used by the manager
64+
// Deprecated: the Redis client is an internal implementation and access will be removed
6465
func (m *Manager) GetRedisClient() *redis.Client {
6566
return m.opts.client
6667
}

‎manager_test.go‎

Lines changed: 9 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -9,7 +9,7 @@ import (
99
"time"
1010

1111
"github.com/digitalocean/go-workers2/storage"
12-
"github.com/go-redis/redis/v8"
12+
"github.com/redis/go-redis/v9"
1313
"github.com/stretchr/testify/assert"
1414
)
1515

@@ -38,10 +38,10 @@ func TestNewManagerWithRedisClient(t *testing.T) {
3838
}
3939

4040
client := redis.NewClient(&redis.Options{
41-
IdleTimeout: 1,
42-
Password: "ab",
43-
DB: 2,
44-
TLSConfig: &tls.Config{ServerName: "test_tls2"},
41+
ConnMaxIdleTime: 1,
42+
Password: "ab",
43+
DB: 2,
44+
TLSConfig: &tls.Config{ServerName: "test_tls2"},
4545
})
4646

4747
mgr, err := NewManagerWithRedisClient(opts, client)
@@ -62,10 +62,10 @@ func TestNewManagerWithRedisClientNoProcessID(t *testing.T) {
6262
}
6363

6464
client := redis.NewClient(&redis.Options{
65-
IdleTimeout: 1,
66-
Password: "ab",
67-
DB: 2,
68-
TLSConfig: &tls.Config{ServerName: "test_tls2"},
65+
ConnMaxIdleTime: 1,
66+
Password: "ab",
67+
DB: 2,
68+
TLSConfig: &tls.Config{ServerName: "test_tls2"},
6969
})
7070

7171
mgr, err := NewManagerWithRedisClient(opts, client)
@@ -490,7 +490,6 @@ type testPrioritizedActiveManagerConfig struct {
490490
waitGroup sync.WaitGroup
491491
assertHeartbeat chan bool
492492
assertedHeartbeat bool
493-
assertedActivate bool
494493
}
495494

496495
func TestManager_Run_PrioritizedActiveManager(t *testing.T) {

‎options.go‎

Lines changed: 14 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -9,7 +9,7 @@ import (
99
"time"
1010

1111
"github.com/digitalocean/go-workers2/storage"
12-
"github.com/go-redis/redis/v8"
12+
"github.com/redis/go-redis/v9"
1313
)
1414

1515
const (
@@ -80,26 +80,26 @@ func processOptions(options Options) (Options, error) {
8080

8181
if options.ServerAddr != "" {
8282
options.client = redis.NewClient(&redis.Options{
83-
IdleTimeout: redisIdleTimeout,
84-
Password: options.Password,
85-
DB: options.Database,
86-
PoolSize: options.PoolSize,
87-
Addr: options.ServerAddr,
88-
TLSConfig: options.RedisTLSConfig,
83+
ConnMaxIdleTime: redisIdleTimeout,
84+
Password: options.Password,
85+
DB: options.Database,
86+
PoolSize: options.PoolSize,
87+
Addr: options.ServerAddr,
88+
TLSConfig: options.RedisTLSConfig,
8989
})
9090
} else if options.SentinelAddrs != "" {
9191
if options.RedisMasterName == "" {
9292
return Options{}, errors.New("Sentinel configuration requires a master name")
9393
}
9494

9595
options.client = redis.NewFailoverClient(&redis.FailoverOptions{
96-
IdleTimeout: redisIdleTimeout,
97-
Password: options.Password,
98-
DB: options.Database,
99-
PoolSize: options.PoolSize,
100-
SentinelAddrs: strings.Split(options.SentinelAddrs, ","),
101-
MasterName: options.RedisMasterName,
102-
TLSConfig: options.RedisTLSConfig,
96+
ConnMaxIdleTime: redisIdleTimeout,
97+
Password: options.Password,
98+
DB: options.Database,
99+
PoolSize: options.PoolSize,
100+
SentinelAddrs: strings.Split(options.SentinelAddrs, ","),
101+
MasterName: options.RedisMasterName,
102+
TLSConfig: options.RedisTLSConfig,
103103
})
104104
} else {
105105
return Options{}, errors.New("Options requires either the Server or Sentinels option")

‎producer.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@ import (
88
"io"
99
"time"
1010

11-
"github.com/go-redis/redis/v8"
11+
"github.com/redis/go-redis/v9"
1212
)
1313

1414
const (

‎producer_test.go‎

Lines changed: 9 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,7 @@ import (
77
"testing"
88

99
"github.com/digitalocean/go-workers2/storage"
10-
"github.com/go-redis/redis/v8"
10+
"github.com/redis/go-redis/v9"
1111
"github.com/stretchr/testify/assert"
1212
)
1313

@@ -180,10 +180,10 @@ func TestNewProducerWithRedisClient(t *testing.T) {
180180
}
181181

182182
client := redis.NewClient(&redis.Options{
183-
IdleTimeout: 1,
184-
Password: "ab",
185-
DB: 2,
186-
TLSConfig: &tls.Config{ServerName: "test_tls3"},
183+
ConnMaxIdleTime: 1,
184+
Password: "ab",
185+
DB: 2,
186+
TLSConfig: &tls.Config{ServerName: "test_tls3"},
187187
})
188188

189189
producer, err := NewProducerWithRedisClient(opts, client)
@@ -203,10 +203,10 @@ func TestNewProducerWithRedisClientNoProcessID(t *testing.T) {
203203
}
204204

205205
client := redis.NewClient(&redis.Options{
206-
IdleTimeout: 1,
207-
Password: "ab",
208-
DB: 2,
209-
TLSConfig: &tls.Config{ServerName: "test_tls2"},
206+
ConnMaxIdleTime: 1,
207+
Password: "ab",
208+
DB: 2,
209+
TLSConfig: &tls.Config{ServerName: "test_tls2"},
210210
})
211211

212212
mgr, err := NewProducerWithRedisClient(opts, client)

‎scheduled_test.go‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,7 @@ import (
55
"testing"
66

77
"github.com/digitalocean/go-workers2/storage"
8-
"github.com/go-redis/redis/v8"
8+
"github.com/redis/go-redis/v9"
99
"github.com/stretchr/testify/assert"
1010
)
1111

@@ -25,9 +25,9 @@ func TestScheduled(t *testing.T) {
2525
message2, _ := NewMsg("{\"queue\":\"myqueue\",\"foo\":\"bar2\"}")
2626
message3, _ := NewMsg("{\"queue\":\"default\",\"foo\":\"bar3\"}")
2727

28-
rc.ZAdd(ctx, retryQueue(opts.Namespace), &redis.Z{Score: now - 60.0, Member: message1.ToJson()}).Result()
29-
rc.ZAdd(ctx, retryQueue(opts.Namespace), &redis.Z{Score: now - 10.0, Member: message2.ToJson()}).Result()
30-
rc.ZAdd(ctx, retryQueue(opts.Namespace), &redis.Z{Score: now + 60.0, Member: message3.ToJson()}).Result()
28+
rc.ZAdd(ctx, retryQueue(opts.Namespace), redis.Z{Score: now - 60.0, Member: message1.ToJson()}).Result()
29+
rc.ZAdd(ctx, retryQueue(opts.Namespace), redis.Z{Score: now - 10.0, Member: message2.ToJson()}).Result()
30+
rc.ZAdd(ctx, retryQueue(opts.Namespace), redis.Z{Score: now + 60.0, Member: message3.ToJson()}).Result()
3131

3232
scheduled.poll()
3333

‎storage/redis.go‎

Lines changed: 4 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@ import (
88
"strconv"
99
"time"
1010

11-
"github.com/go-redis/redis/v8"
11+
"github.com/redis/go-redis/v9"
1212
)
1313

1414
type redisStore struct {
@@ -168,10 +168,6 @@ func (r *redisStore) SendHeartbeat(ctx context.Context, heartbeat *Heartbeat) er
168168
return nil
169169
}
170170

171-
func (r *redisStore) getTaskRunnerID(pid int, tid string) string {
172-
return fmt.Sprintf("%d-%s", pid, tid)
173-
}
174-
175171
func (r *redisStore) RequeueMessagesFromInProgressQueue(ctx context.Context, inprogressQueue, queue string) ([]string, error) {
176172
var requeuedMsgs []string
177173
for {
@@ -208,7 +204,7 @@ func (r *redisStore) RemoveHeartbeat(ctx context.Context, heartbeatID string) er
208204
}
209205

210206
func (r *redisStore) EnqueueMessage(ctx context.Context, queue string, priority float64, message string) error {
211-
_, err := r.client.ZAdd(ctx, r.getQueueName(queue), &redis.Z{
207+
_, err := r.client.ZAdd(ctx, r.getQueueName(queue), redis.Z{
212208
Score: priority,
213209
Member: message,
214210
}).Result()
@@ -217,7 +213,7 @@ func (r *redisStore) EnqueueMessage(ctx context.Context, queue string, priority
217213
}
218214

219215
func (r *redisStore) EnqueueScheduledMessage(ctx context.Context, priority float64, message string) error {
220-
_, err := r.client.ZAdd(ctx, r.namespace+ScheduledJobsKey, &redis.Z{
216+
_, err := r.client.ZAdd(ctx, r.namespace+ScheduledJobsKey, redis.Z{
221217
Score: priority,
222218
Member: message,
223219
}).Result()
@@ -256,7 +252,7 @@ func (r *redisStore) DequeueScheduledMessage(ctx context.Context, priority float
256252
}
257253

258254
func (r *redisStore) EnqueueRetriedMessage(ctx context.Context, priority float64, message string) error {
259-
_, err := r.client.ZAdd(ctx, r.namespace+RetryKey, &redis.Z{
255+
_, err := r.client.ZAdd(ctx, r.namespace+RetryKey, redis.Z{
260256
Score: priority,
261257
Member: message,
262258
}).Result()

0 commit comments

Comments
 (0)