feat: support redis sentinal (#1910)

Signed-off-by: Gaius <gaius.qi@gmail.com>

internal/job/job.go

@@ -36,6 +36,7 @@ import (
 
 type Config struct {
 	Addrs      []string
+	MasterName string
 	Username   string
 	Password   string
 	BrokerDB   int
@@ -54,6 +55,7 @@ func New(cfg *Config, queue Queue) (*Job, error) {
 
 	if err := ping(&redis.UniversalOptions{
 		Addrs:      cfg.Addrs,
+		MasterName: cfg.MasterName,
 		Username:   cfg.Username,
 		Password:   cfg.Password,
 		DB:         cfg.BrokerDB,
@@ -63,6 +65,7 @@ func New(cfg *Config, queue Queue) (*Job, error) {
 
 	if err := ping(&redis.UniversalOptions{
 		Addrs:      cfg.Addrs,
+		MasterName: cfg.MasterName,
 		Username:   cfg.Username,
 		Password:   cfg.Password,
 		DB:         cfg.BackendDB,
@@ -76,6 +79,7 @@ func New(cfg *Config, queue Queue) (*Job, error) {
 		ResultBackend:   fmt.Sprintf("redis://%s:%s@%s/%d", cfg.Username, cfg.Password, strings.Join(cfg.Addrs, ","), cfg.BackendDB),
 		ResultsExpireIn: DefaultResultsExpireIn,
 		Redis: &machineryv1config.RedisConfig{
+			MasterName:     cfg.MasterName,
 			MaxIdle:        DefaultRedisMaxIdle,
 			IdleTimeout:    DefaultRedisIdleTimeout,
 			ReadTimeout:    DefaultRedisReadTimeout,

manager/config/config.go

@@ -167,22 +167,25 @@ type RedisConfig struct {
 	// DEPRECATED: Please use the `addrs` field instead.
 	Port int `yaml:"port" mapstructure:"port"`
 
-	// Server addresses.
+	// Addrs is server addresses.
 	Addrs []string `yaml:"addrs" mapstructure:"addrs"`
 
-	// Server username.
+	// MasterName is the sentinel master name.
+	MasterName string `yaml:"masterName" mapstructure:"masterName"`
+
+	// Username is server username.
 	Username string `yaml:"username" mapstructure:"username"`
 
-	// Server password.
+	// Password is server password.
 	Password string `yaml:"password" mapstructure:"password"`
 
-	// Server cache DB name.
+	// DB is server cache DB name.
 	DB int `yaml:"db" mapstructure:"db"`
 
-	// Server broker DB name.
+	// BrokerDB is server broker DB name.
 	BrokerDB int `yaml:"brokerDB" mapstructure:"brokerDB"`
 
-	// Server backend DB name.
+	// BackendDB is server backend DB name.
 	BackendDB int `yaml:"backendDB" mapstructure:"backendDB"`
 }
 

manager/config/config_test.go

@@ -80,6 +80,7 @@ func TestManagerConfig_Load(t *testing.T) {
 				Host:       "bar",
 				Password:   "bar",
 				Addrs:      []string{"foo", "bar"},
+				MasterName: "baz",
 				Port:       6379,
 				DB:         0,
 				BrokerDB:   1,

manager/config/testdata/manager.yaml

@@ -39,6 +39,7 @@ database:
     migrate: true
   redis:
     addrs: [foo, bar]
+    masterName: "baz"
     password: bar
     host: bar
     port: 6379

manager/database/redis.go

@@ -28,6 +28,7 @@ func NewRedis(cfg *config.RedisConfig) (redis.UniversalClient, error) {
 	redis.SetLogger(&redisLogger{})
 	client := redis.NewUniversalClient(&redis.UniversalOptions{
 		Addrs:      cfg.Addrs,
+		MasterName: cfg.MasterName,
 		DB:         cfg.DB,
 		Username:   cfg.Username,
 		Password:   cfg.Password,

manager/job/job.go

@@ -29,6 +29,7 @@ type Job struct {
 func New(cfg *config.Config) (*Job, error) {
 	j, err := internaljob.New(&internaljob.Config{
 		Addrs:      cfg.Database.Redis.Addrs,
+		MasterName: cfg.Database.Redis.MasterName,
 		Username:   cfg.Database.Redis.Username,
 		Password:   cfg.Database.Redis.Password,
 		BrokerDB:   cfg.Database.Redis.BrokerDB,

scheduler/config/config.go

@@ -235,19 +235,22 @@ type RedisConfig struct {
 	// DEPRECATED: Please use the `addrs` field instead.
 	Port int `yaml:"port" mapstructure:"port"`
 
-	// Server addresses.
+	// Addrs is server addresses.
 	Addrs []string `yaml:"addrs" mapstructure:"addrs"`
 
-	// Server username.
+	// MasterName is the sentinel master name.
+	MasterName string `yaml:"masterName" mapstructure:"masterName"`
+
+	// Username is server username.
 	Username string `yaml:"username" mapstructure:"username"`
 
-	// Server password.
+	// Password is server password.
 	Password string `yaml:"password" mapstructure:"password"`
 
-	// Broker database name.
+	// BrokerDB is broker database name.
 	BrokerDB int `yaml:"brokerDB" mapstructure:"brokerDB"`
 
-	// Backend database name.
+	// BackendDB is backend database name.
 	BackendDB int `yaml:"backendDB" mapstructure:"backendDB"`
 }
 

scheduler/config/config_test.go

@@ -86,6 +86,7 @@ func TestConfig_Load(t *testing.T) {
 			LocalWorkerNum:     5,
 			Redis: RedisConfig{
 				Addrs:      []string{"foo", "bar"},
+				MasterName: "baz",
 				Host:       "127.0.0.1",
 				Port:       6379,
 				Password:   "foo",

scheduler/config/testdata/scheduler.yaml

@@ -51,6 +51,7 @@ job:
   localWorkerNum: 5
   redis:
     addrs: ["foo", "bar"]
+    masterName: "baz"
     host: 127.0.0.1
     port: 6379
     password: foo

scheduler/job/job.go

@@ -58,6 +58,7 @@ type job struct {
 func New(cfg *config.Config, resource resource.Resource) (Job, error) {
 	redisConfig := &internaljob.Config{
 		Addrs:      cfg.Job.Redis.Addrs,
+		MasterName: cfg.Job.Redis.MasterName,
 		Username:   cfg.Job.Redis.Username,
 		Password:   cfg.Job.Redis.Password,
 		BrokerDB:   cfg.Job.Redis.BrokerDB,