Repository: dragonflyoss/Dragonfly2
Pull request #1666 (2022-09-08 02:30:03)

feat: split concurrent back source e2e
Signed-off-by: Jim Ma <majinjing3@gmail.com>

<!--- Provide a general summary of your changes in the Title above -->

## Description

<!--- Describe your changes in detail -->

## Related Issue

<!--- This project only accepts pull requests related to open issues -->
<!--- If suggesting a new feature or change, please discuss it in an issue first -->
<!--- If fixing a bug, there should be an issue describing it with steps to reproduce -->
<!--- Please link to the issue here: -->

## Motivation and Context

<!--- Why is this change required? What problem does it solve? -->

## Screenshots (if appropriate)

## Types of changes

<!--- What types of changes does your code introduce? Put an `x` in all the boxes that apply: -->

- [ ] Bug fix (non-breaking change which fixes an issue)
- [x] New feature (non-breaking change which adds functionality)
- [ ] Breaking change (fix or feature that would cause existing functionality to change)
- [ ] Documentation Update (if none of the other choices apply)

## Checklist

<!--- Go over all the following points, and put an `x` in all the boxes that apply. -->
<!--- If you're unsure about any of these, don't hesitate to ask. We're here to help! -->

- [ ] My change requires a change to the documentation.
- [ ] I have updated the documentation accordingly.
- [ ] I have read the **CONTRIBUTING** document.
- [ ] I have added tests to cover my changes.


codecov-commenter: # [Codecov](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1666?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) Report
> Merging [#1666](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1666?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) (b49fc9d) into [main](https://codecov.io/gh/dragonflyoss/Dragonfly2/commit/39516aa339cf41e30e66d1b6ae5a20e3a9b2a4d3?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) (39516aa) will **increase** coverage by `0.16%`.
> The diff coverage is `n/a`.

> :exclamation: Current head b49fc9d differs from pull request most recent head a7278c0. Consider uploading reports for the commit a7278c0 to get more accurate results

[![Impacted file tree graph](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1666/graphs/tree.svg?width=650&height=150&src=pr&token=PUICJYSAT7&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss)](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1666?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss)

```diff
@@            Coverage Diff             @@
##             main    #1666      +/-   ##
==========================================
+ Coverage   41.82%   41.98%   +0.16%     
==========================================
  Files         125      125              
  Lines       15423    15423              
==========================================
+ Hits         6450     6476      +26     
+ Misses       8457     8428      -29     
- Partials      516      519       +3     
```

| Flag | Coverage Δ | |
|---|---|---|
| unittests | `41.98% <ø> (+0.16%)` | :arrow_up: |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#carryforward-flags-in-the-pull-request-comment) to find out more.

| [Impacted Files](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1666?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) | Coverage Δ | |
|---|---|---|
| [client/daemon/peer/peertask\_piecetask\_poller.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1666/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9wZWVyL3BlZXJ0YXNrX3BpZWNldGFza19wb2xsZXIuZ28=) | `46.99% <0.00%> (+14.20%)` | :arrow_up: |


Patch:
diff --git a/cmd/manager/cmd/root.go b/cmd/manager/cmd/root.go
index d4dc3df9..8a0bd612 100644
--- a/cmd/manager/cmd/root.go
+++ b/cmd/manager/cmd/root.go
@@ -37,7 +37,7 @@ var (
 	cfg *config.Config
 )
 
-// rootCmd represents the commonv1 command when called without any subcommands
+// rootCmd represents the commonv1 command when called without any subcommands.
 var rootCmd = &cobra.Command{
 	Use:   "manager",
 	Short: "The central manager of dragonfly.",
@@ -47,18 +47,23 @@ for managing schedulers and seed peers, offering http apis and portal, etc.`,
 	DisableAutoGenTag: true,
 	SilenceUsage:      true,
 	RunE: func(cmd *cobra.Command, args []string) error {
-		// Validate config
+		// Convert config.
+		if err := cfg.Convert(); err != nil {
+			return err
+		}
+
+		// Validate config.
 		if err := cfg.Validate(); err != nil {
 			return err
 		}
 
-		// Initialize dfpath
+		// Initialize dfpath.
 		d, err := initDfpath(cfg.Server)
 		if err != nil {
 			return err
 		}
 
-		// Initialize logger
+		// Initialize logger.
 		if err := logger.InitManager(cfg.Verbose, cfg.Console, d.LogDir()); err != nil {
 			return fmt.Errorf("init manager logger: %w", err)
 		}
@@ -78,10 +83,10 @@ func Execute() {
 }
 
 func init() {
-	// Initialize default manager config
+	// Initialize default manager config.
 	cfg = config.New()
 
-	// Initialize command and config
+	// Initialize command and config.
 	dependency.InitCommandAndConfig(rootCmd, true, cfg)
 }
 
@@ -100,7 +105,7 @@ func initDfpath(cfg *config.ServerConfig) (dfpath.Dfpath, error) {
 
 func runManager(d dfpath.Dfpath) error {
 	logger.Infof("Version:\n%s", version.Version())
-	// manager config values
+	// manager config values.
 	s, err := yaml.Marshal(cfg)
 
 	if err != nil {
diff --git a/cmd/scheduler/cmd/root.go b/cmd/scheduler/cmd/root.go
index 9eb74a12..b21791ef 100644
--- a/cmd/scheduler/cmd/root.go
+++ b/cmd/scheduler/cmd/root.go
@@ -38,7 +38,7 @@ var (
 	cfg *config.Config
 )
 
-// rootCmd represents the commonv1 command when called without any subcommands
+// rootCmd represents the commonv1 command when called without any subcommands.
 var rootCmd = &cobra.Command{
 	Use:   "scheduler",
 	Short: "the scheduler of dragonfly",
@@ -48,7 +48,12 @@ generate and maintain a P2P network during the download process, and push suitab
 	DisableAutoGenTag: true,
 	SilenceUsage:      true,
 	RunE: func(cmd *cobra.Command, args []string) error {
-		// Validate config
+		// Convert config.
+		if err := cfg.Convert(); err != nil {
+			return err
+		}
+
+		// Validate config.
 		if err := cfg.Validate(); err != nil {
 			return err
 		}
@@ -56,13 +61,13 @@ generate and maintain a P2P network during the download process, and push suitab
 		ctx, cancel := context.WithCancel(context.Background())
 		defer cancel()
 
-		// Initialize dfpath
+		// Initialize dfpath.
 		d, err := initDfpath(cfg.Server)
 		if err != nil {
 			return err
 		}
 
-		// Initialize logger
+		// Initialize logger.
 		if err := logger.InitScheduler(cfg.Verbose, cfg.Console, d.LogDir()); err != nil {
 			return fmt.Errorf("init scheduler logger: %w", err)
 		}
@@ -82,9 +87,9 @@ func Execute() {
 }
 
 func init() {
-	// Initialize default scheduler config
+	// Initialize default scheduler config.
 	cfg = config.New()
-	// Initialize command and config
+	// Initialize command and config.
 	dependency.InitCommandAndConfig(rootCmd, true, cfg)
 }
 
@@ -114,7 +119,7 @@ func initDfpath(cfg *config.ServerConfig) (dfpath.Dfpath, error) {
 func runScheduler(ctx context.Context, d dfpath.Dfpath) error {
 	logger.Infof("Version:\n%s", version.Version())
 
-	// scheduler config values
+	// scheduler config values.
 	s, _ := yaml.Marshal(cfg)
 
 	logger.Infof("scheduler configuration:\n%s", string(s))
diff --git a/deploy/docker-compose/template/manager.template.yaml b/deploy/docker-compose/template/manager.template.yaml
index 560f5013..741f2444 100644
--- a/deploy/docker-compose/template/manager.template.yaml
+++ b/deploy/docker-compose/template/manager.template.yaml
@@ -37,10 +37,13 @@ database:
   #   insecureSkipVerify: true
   # redis configure
   redis:
+    # addresses
+    addrs:
+      - "__IP__:6379"
     password: dragonfly
-    host: __IP__
-    port: 6379
     db: 0
+    brokerDB: 1
+    backendDB: 2
 # manager server cache
 # cache:
 #   # redis cache configure
diff --git a/deploy/docker-compose/template/scheduler.template.yaml b/deploy/docker-compose/template/scheduler.template.yaml
index 308be03f..8f710fe2 100644
--- a/deploy/docker-compose/template/scheduler.template.yaml
+++ b/deploy/docker-compose/template/scheduler.template.yaml
@@ -87,10 +87,9 @@ job:
   localWorkerNum: 5
   # redis configuration
   redis:
-    # host
-    host: "__IP__"
-    # port
-    port: 6379
+    # addresses
+    addrs:
+      - "__IP__:6379"
     # password
     password: dragonfly
     # brokerDB
diff --git a/internal/job/job.go b/internal/job/job.go
index ddcc1c9c..0c577b17 100644
--- a/internal/job/job.go
+++ b/internal/job/job.go
@@ -22,6 +22,7 @@ import (
 	"errors"
 	"fmt"
 	"reflect"
+	"strings"
 	"time"
 
 	"github.com/RichardKnop/machinery/v1"
@@ -32,8 +33,8 @@ import (
 )
 
 type Config struct {
-	Host      string
-	Port      int
+	Addrs     []string
+	Username  string
 	Password  string
 	BrokerDB  int
 	BackendDB int
@@ -49,18 +50,18 @@ func New(cfg *Config, queue Queue) (*Job, error) {
 	// Set logger
 	machineryv1log.Set(&MachineryLogger{})
 
-	broker := fmt.Sprintf("redis://%s@%s:%d/%d", cfg.Password, cfg.Host, cfg.Port, cfg.BrokerDB)
-	if err := ping(&redis.Options{
-		Addr:     fmt.Sprintf("%s:%d", cfg.Host, cfg.Port),
+	if err := ping(&redis.UniversalOptions{
+		Addrs:    cfg.Addrs,
+		Username: cfg.Username,
 		Password: cfg.Password,
 		DB:       cfg.BrokerDB,
 	}); err != nil {
 		return nil, err
 	}
 
-	backend := fmt.Sprintf("redis://%s@%s:%d/%d", cfg.Password, cfg.Host, cfg.Port, cfg.BackendDB)
-	if err := ping(&redis.Options{
-		Addr:     fmt.Sprintf("%s:%d", cfg.Host, cfg.Port),
+	if err := ping(&redis.UniversalOptions{
+		Addrs:    cfg.Addrs,
+		Username: cfg.Username,
 		Password: cfg.Password,
 		DB:       cfg.BackendDB,
 	}); err != nil {
@@ -68,9 +69,9 @@ func New(cfg *Config, queue Queue) (*Job, error) {
 	}
 
 	server, err := machinery.NewServer(&machineryv1config.Config{
-		Broker:          broker,
+		Broker:          fmt.Sprintf("redis://%s:%s@%s/%d", cfg.Username, cfg.Password, strings.Join(cfg.Addrs, ","), cfg.BrokerDB),
 		DefaultQueue:    queue.String(),
-		ResultBackend:   backend,
+		ResultBackend:   fmt.Sprintf("redis://%s:%s@%s/%d", cfg.Username, cfg.Password, strings.Join(cfg.Addrs, ","), cfg.BackendDB),
 		ResultsExpireIn: DefaultResultsExpireIn,
 		Redis: &machineryv1config.RedisConfig{
 			MaxIdle:        DefaultRedisMaxIdle,
@@ -90,8 +91,8 @@ func New(cfg *Config, queue Queue) (*Job, error) {
 	}, nil
 }
 
-func ping(options *redis.Options) error {
-	client := redis.NewClient(options)
+func ping(options *redis.UniversalOptions) error {
+	client := redis.NewUniversalClient(options)
 	return client.Ping(context.Background()).Err()
 }
 
diff --git a/manager/config/config.go b/manager/config/config.go
index 206343f5..5ec56f6e 100644
--- a/manager/config/config.go
+++ b/manager/config/config.go
@@ -19,6 +19,7 @@ package config
 import (
 	"crypto/tls"
 	"errors"
+	"fmt"
 	"time"
 
 	"github.com/docker/go-connections/tlsconfig"
@@ -162,12 +163,18 @@ func (t *TLSConfig) Client() (*tls.Config, error) {
 }
 
 type RedisConfig struct {
-	// Server host.
+	// DEPRECATED: Please use the `addrs` field instead.
 	Host string `yaml:"host" mapstructure:"host"`
 
-	// Server port.
+	// DEPRECATED: Please use the `addrs` field instead.
 	Port int `yaml:"port" mapstructure:"port"`
 
+	// Server addresses.
+	Addrs []string `yaml:"addrs" mapstructure:"addrs"`
+
+	// Server username.
+	Username string `yaml:"username" mapstructure:"username"`
+
 	// Server password.
 	Password string `yaml:"password" mapstructure:"password"`
 
@@ -450,16 +457,22 @@ func (cfg *Config) Validate() error {
 		return errors.New("database requires parameter redis")
 	}
 
-	if cfg.Database.Redis.Host == "" {
-		return errors.New("redis requires parameter host")
+	if len(cfg.Database.Redis.Addrs) == 0 {
+		return errors.New("redis requires parameter addrs")
 	}
 
-	if cfg.Database.Redis.Port <= 0 {
-		return errors.New("redis requires parameter port")
-	}
+	if len(cfg.Database.Redis.Addrs) == 1 {
+		if cfg.Database.Redis.DB < 0 {
+			return errors.New("redis requires parameter db")
+		}
 
-	if cfg.Database.Redis.DB < 0 {
-		return errors.New("redis requires parameter db")
+		if cfg.Database.Redis.BrokerDB < 0 {
+			return errors.New("redis requires parameter brokerDB")
+		}
+
+		if cfg.Database.Redis.BackendDB < 0 {
+			return errors.New("redis requires parameter backendDB")
+		}
 	}
 
 	if cfg.Database.Redis.BrokerDB < 0 {
@@ -550,3 +563,12 @@ func (cfg *Config) Validate() error {
 
 	return nil
 }
+
+func (cfg *Config) Convert() error {
+	// TODO Compatible with deprecated fields host and port.
+	if cfg.Database.Redis.Host != "" && cfg.Database.Redis.Port > 0 {
+		cfg.Database.Redis.Addrs = append(cfg.Database.Redis.Addrs, fmt.Sprintf("%s:%d", cfg.Database.Redis.Host, cfg.Database.Redis.Port))
+	}
+
+	return nil
+}
diff --git a/manager/database/database.go b/manager/database/database.go
index 6015982c..f6110e21 100644
--- a/manager/database/database.go
+++ b/manager/database/database.go
@@ -37,7 +37,7 @@ const (
 
 type Database struct {
 	DB  *gorm.DB
-	RDB *redis.Client
+	RDB redis.UniversalClient
 }
 
 func New(cfg *config.Config) (*Database, error) {
diff --git a/manager/database/redis.go b/manager/database/redis.go
index f3767990..10bb7a36 100644
--- a/manager/database/redis.go
+++ b/manager/database/redis.go
@@ -18,20 +18,19 @@ package database
 
 import (
 	"context"
-	"fmt"
 
 	"github.com/go-redis/redis/v8"
 
 	"d7y.io/dragonfly/v2/manager/config"
 )
 
-func NewRedis(cfg *config.RedisConfig) (*redis.Client, error) {
+func NewRedis(cfg *config.RedisConfig) (redis.UniversalClient, error) {
 	redis.SetLogger(&redisLogger{})
-
-	client := redis.NewClient(&redis.Options{
-		Addr:     fmt.Sprintf("%s:%d", cfg.Host, cfg.Port),
-		Password: cfg.Password,
+	client := redis.NewUniversalClient(&redis.UniversalOptions{
+		Addrs:    cfg.Addrs,
 		DB:       cfg.DB,
+		Username: cfg.Username,
+		Password: cfg.Password,
 	})
 
 	if err := client.Ping(context.Background()).Err(); err != nil {
diff --git a/manager/job/job.go b/manager/job/job.go
index 14a83aa3..6da89bc7 100644
--- a/manager/job/job.go
+++ b/manager/job/job.go
@@ -28,8 +28,8 @@ type Job struct {
 
 func New(cfg *config.Config) (*Job, error) {
 	j, err := internaljob.New(&internaljob.Config{
-		Host:      cfg.Database.Redis.Host,
-		Port:      cfg.Database.Redis.Port,
+		Addrs:     cfg.Database.Redis.Addrs,
+		Username:  cfg.Database.Redis.Username,
 		Password:  cfg.Database.Redis.Password,
 		BrokerDB:  cfg.Database.Redis.BrokerDB,
 		BackendDB: cfg.Database.Redis.BackendDB,
diff --git a/manager/rpcserver/rpcserver.go b/manager/rpcserver/rpcserver.go
index 9b69c0f8..19e668cd 100644
--- a/manager/rpcserver/rpcserver.go
+++ b/manager/rpcserver/rpcserver.go
@@ -78,8 +78,8 @@ type Server struct {
 	// GORM instance.
 	db *gorm.DB
 
-	// Redis client instance.
-	rdb *redis.Client
+	// Redis universal client interface.
+	rdb redis.UniversalClient
 
 	// Cache instance.
 	cache *cache.Cache
diff --git a/manager/service/job.go b/manager/service/job.go
index 74e6a39d..1c52d089 100644
--- a/manager/service/job.go
+++ b/manager/service/job.go
@@ -137,15 +137,15 @@ func (s *service) pollingJob(ctx context.Context, id uint, taskID string) {
 			log.Errorf("polling job failed: %#v", jobStates)
 			return nil, true, nil
 		default:
-			msg := fmt.Sprintf("unknow state %s", job.State)
-			log.Error(msg)
+			msg := fmt.Sprintf("polling job state is %s", job.State)
+			log.Info(msg)
 			return nil, false, errors.New(msg)
 		}
 	}); err != nil {
 		log.Errorf("polling job failed: %s", err.Error())
 	}
 
-	// Polling timeout and failed
+	// Polling timeout and failed.
 	if job.State != machineryv1tasks.StateSuccess && job.State != machineryv1tasks.StateFailure {
 		job := model.Job{}
 		if err := s.db.WithContext(ctx).First(&job, id).Updates(model.Job{
diff --git a/manager/service/service.go b/manager/service/service.go
index cb31bef5..9f612cc1 100644
--- a/manager/service/service.go
+++ b/manager/service/service.go
@@ -153,7 +153,7 @@ type Service interface {
 
 type service struct {
 	db            *gorm.DB
-	rdb           *redis.Client
+	rdb           redis.UniversalClient
 	cache         *cache.Cache
 	job           *job.Job
 	enforcer      *casbin.Enforcer
diff --git a/scheduler/config/config.go b/scheduler/config/config.go
index fe55d9d4..00831d31 100644
--- a/scheduler/config/config.go
+++ b/scheduler/config/config.go
@@ -18,6 +18,7 @@ package config
 
 import (
 	"errors"
+	"fmt"
 	"time"
 
 	"d7y.io/dragonfly/v2/cmd/dependency/base"
@@ -213,12 +214,18 @@ type StorageConfig struct {
 }
 
 type RedisConfig struct {
-	// Server hostname.
+	// DEPRECATED: Please use the `addrs` field instead.
 	Host string `yaml:"host" mapstructure:"host"`
 
-	// Server port.
+	// DEPRECATED: Please use the `addrs` field instead.
 	Port int `yaml:"port" mapstructure:"port"`
 
+	// Server addresses.
+	Addrs []string `yaml:"addrs" mapstructure:"addrs"`
+
+	// Server username.
+	Username string `yaml:"username" mapstructure:"username"`
+
 	// Server password.
 	Password string `yaml:"password" mapstructure:"password"`
 
@@ -315,7 +322,6 @@ func New() *Config {
 			SchedulerWorkerNum: DefaultJobSchedulerWorkerNum,
 			LocalWorkerNum:     DefaultJobLocalWorkerNum,
 			Redis: &RedisConfig{
-				Port:      DefaultJobRedisPort,
 				BrokerDB:  DefaultJobRedisBrokerDB,
 				BackendDB: DefaultJobRedisBackendDB,
 			},
@@ -434,20 +440,18 @@ func (cfg *Config) Validate() error {
 			return errors.New("job requires parameter localWorkerNum")
 		}
 
-		if cfg.Job.Redis.Host == "" {
-			return errors.New("job requires parameter redis host")
-		}
-
-		if cfg.Job.Redis.Port <= 0 {
-			return errors.New("job requires parameter redis port")
+		if len(cfg.Job.Redis.Addrs) == 0 {
+			return errors.New("job requires parameter addrs")
 		}
 
-		if cfg.Job.Redis.BrokerDB <= 0 {
-			return errors.New("job requires parameter redis brokerDB")
-		}
+		if len(cfg.Job.Redis.Addrs) == 1 {
+			if cfg.Job.Redis.BrokerDB <= 0 {
+				return errors.New("job requires parameter redis brokerDB")
+			}
 
-		if cfg.Job.Redis.BackendDB <= 0 {
-			return errors.New("job requires parameter redis backendDB")
+			if cfg.Job.Redis.BackendDB <= 0 {
+				return errors.New("job requires parameter redis backendDB")
+			}
 		}
 	}
 
@@ -489,3 +493,12 @@ func (cfg *Config) Validate() error {
 
 	return nil
 }
+
+func (cfg *Config) Convert() error {
+	// TODO Compatible with deprecated fields host and port.
+	if cfg.Job.Redis.Host != "" && cfg.Job.Redis.Port > 0 {
+		cfg.Job.Redis.Addrs = append(cfg.Job.Redis.Addrs, fmt.Sprintf("%s:%d", cfg.Job.Redis.Host, cfg.Job.Redis.Port))
+	}
+
+	return nil
+}
diff --git a/scheduler/config/constants.go b/scheduler/config/constants.go
index 767ce181..ada8870e 100644
--- a/scheduler/config/constants.go
+++ b/scheduler/config/constants.go
@@ -110,9 +110,6 @@ const (
 	// DefaultJobGlobalWorkerNum is default local worker number for job.
 	DefaultJobLocalWorkerNum = 10
 
-	// DefaultJobRedisPort is default port for redis.
-	DefaultJobRedisPort = 6379
-
 	// DefaultJobRedisBrokerDB is default db for redis broker.
 	DefaultJobRedisBrokerDB = 1
 
diff --git a/scheduler/job/job.go b/scheduler/job/job.go
index a9b17b5a..30d23a18 100644
--- a/scheduler/job/job.go
+++ b/scheduler/job/job.go
@@ -52,8 +52,7 @@ type job struct {
 
 func New(cfg *config.Config, resource resource.Resource) (Job, error) {
 	redisConfig := &internaljob.Config{
-		Host:      cfg.Job.Redis.Host,
-		Port:      cfg.Job.Redis.Port,
+		Addrs:     cfg.Job.Redis.Addrs,
 		Password:  cfg.Job.Redis.Password,
 		BrokerDB:  cfg.Job.Redis.BrokerDB,
 		BackendDB: cfg.Job.Redis.BackendDB,


Test patch:
diff --git a/.github/workflows/e2e.yml b/.github/workflows/e2e.yml
index ac61a362..5fdf1cf9 100644
--- a/.github/workflows/e2e.yml
+++ b/.github/workflows/e2e.yml
@@ -25,13 +25,16 @@ jobs:
     timeout-minutes: 30
     strategy:
       matrix:
-        module: ["normal", "tls", "disable-seed-peer"]
+        module: ["normal", "concurent-back-source", "grpc-tls", "disable-seed-peer"]
         include:
           - module: normal
             charts-config: test/testdata/charts/config.yaml
             skip: ""
-          - module: tls
-            charts-config: test/testdata/charts/config-tls.yaml
+          - module: concurent-back-source
+            charts-config: test/testdata/charts/config-concurent-back-source.yaml
+            skip: ""
+          - module: grpc-tls
+            charts-config: test/testdata/charts/config-grpc-tls.yaml
             skip: ""
           - module: disable-seed-peer
             charts-config: test/testdata/charts/config-disable-seed-peer.yaml
diff --git a/manager/config/config_test.go b/manager/config/config_test.go
index f219ea47..e3818f41 100644
--- a/manager/config/config_test.go
+++ b/manager/config/config_test.go
@@ -76,6 +76,7 @@ func TestManagerConfig_Load(t *testing.T) {
 			Redis: &RedisConfig{
 				Host:      "bar",
 				Password:  "bar",
+				Addrs:     []string{"foo", "bar"},
 				Port:      6379,
 				DB:        0,
 				BrokerDB:  1,
diff --git a/manager/config/testdata/manager.yaml b/manager/config/testdata/manager.yaml
index effc94ea..75e0ca2c 100644
--- a/manager/config/testdata/manager.yaml
+++ b/manager/config/testdata/manager.yaml
@@ -35,6 +35,7 @@ database:
     timezone: UTC
     migrate: true
   redis:
+    addrs: [foo, bar]
     password: bar
     host: bar
     port: 6379
diff --git a/scheduler/config/config_test.go b/scheduler/config/config_test.go
index d4c2031d..f7a68097 100644
--- a/scheduler/config/config_test.go
+++ b/scheduler/config/config_test.go
@@ -85,6 +85,7 @@ func TestConfig_Load(t *testing.T) {
 			SchedulerWorkerNum: 1,
 			LocalWorkerNum:     5,
 			Redis: &RedisConfig{
+				Addrs:     []string{"foo", "bar"},
 				Host:      "127.0.0.1",
 				Port:      6379,
 				Password:  "foo",
diff --git a/scheduler/config/testdata/scheduler.yaml b/scheduler/config/testdata/scheduler.yaml
index d342470d..91f98fd0 100644
--- a/scheduler/config/testdata/scheduler.yaml
+++ b/scheduler/config/testdata/scheduler.yaml
@@ -50,6 +50,7 @@ job:
   schedulerWorkerNum: 1
   localWorkerNum: 5
   redis:
+    addrs: ["foo", "bar"]
     host: 127.0.0.1
     port: 6379
     password: foo
diff --git a/test/testdata/charts/config-concurent-back-source.yaml b/test/testdata/charts/config-concurent-back-source.yaml
new file mode 100644
index 00000000..6be1f0c9
--- /dev/null
+++ b/test/testdata/charts/config-concurent-back-source.yaml
@@ -0,0 +1,130 @@
+scheduler:
+  image: d7yio/scheduler
+  tag: latest
+  replicas: 3
+  resources:
+    requests:
+      cpu: "0"
+      memory: "0"
+    limits:
+      cpu: "2"
+      memory: "4Gi"
+  extraVolumeMounts:
+  - name: logs
+    mountPath: "/var/log/"
+  - name: artifact
+    mountPath: /tmp/artifact
+  extraVolumes:
+  - name: logs
+    emptyDir: { }
+  - name: artifact
+    hostPath:
+      path: /tmp/artifact
+  config:
+    verbose: true
+
+seedPeer:
+  image: d7yio/dfdaemon
+  tag: latest
+  replicas: 3
+  resources:
+    requests:
+      cpu: "0"
+      memory: "0"
+    limits:
+      cpu: "1"
+      memory: "1Gi"
+  extraVolumeMounts:
+  - name: logs
+    mountPath: "/var/log/dragonfly"
+    subPath: dragonfly
+  - name: artifact
+    mountPath: /tmp/artifact
+  extraVolumes:
+  - name: logs
+    emptyDir: { }
+  - name: artifact
+    hostPath:
+      path: /tmp/artifact
+  config:
+    verbose: true
+    download:
+      prefetch: true
+      concurrent:
+        thresholdSize: 10M
+        thresholdSpeed: 2M
+        goroutineCount: 4
+
+dfdaemon:
+  image: d7yio/dfdaemon
+  tag: latest
+  resources:
+    requests:
+      cpu: "0"
+      memory: "0"
+    limits:
+      cpu: "1"
+      memory: "1Gi"
+  extraVolumeMounts:
+  - name: logs
+    mountPath: "/var/log/"
+  - name: artifact
+    mountPath: /tmp/artifact
+  extraVolumes:
+  - name: logs
+    emptyDir: { }
+  - name: artifact
+    hostPath:
+      path: /tmp/artifact
+  config:
+    verbose: true
+    pprofPort: 9999
+    metrics: 127.0.0.1:8888
+    download:
+      prefetch: true
+      concurrent:
+        thresholdSize: 10M
+        thresholdSpeed: 2M
+        goroutineCount: 4
+    scheduler:
+      disableAutoBackSource: true
+    proxy:
+      defaultFilter: "Expires&Signature&ns"
+      security:
+        insecure: true
+      tcpListen:
+        namespace: /run/dragonfly/net
+        listen: 0.0.0.0
+        # if you want to change port, please update hostPort in $.Values.dfdaemon.hostPort
+        # port in configmap is generated from $.Values.dfdaemon.hostPort
+        # port: 65001
+      registryMirror:
+        url: https://index.docker.io
+      proxies:
+        - regx: blobs/sha256.*
+        - regx: file-server
+
+manager:
+  image: d7yio/manager
+  tag: latest
+  replicas: 1
+  resources:
+    requests:
+      cpu: "0"
+      memory: "0"
+    limits:
+      cpu: "1"
+      memory: "2Gi"
+  extraVolumeMounts:
+  - name: logs
+    mountPath: "/var/log/"
+  - name: artifact
+    mountPath: /tmp/artifact
+  extraVolumes:
+  - name: logs
+    emptyDir: { }
+  - name: artifact
+    hostPath:
+      path: /tmp/artifact
+  config:
+    verbose: true
diff --git a/test/testdata/charts/config-tls.yaml b/test/testdata/charts/config-grpc-tls.yaml
similarity index 100%
rename from test/testdata/charts/config-tls.yaml
rename to test/testdata/charts/config-grpc-tls.yaml
diff --git a/test/testdata/charts/config.yaml b/test/testdata/charts/config.yaml
index 6be1f0c9..278c0845 100644
--- a/test/testdata/charts/config.yaml
+++ b/test/testdata/charts/config.yaml
@@ -50,10 +50,6 @@ seedPeer:
     verbose: true
     download:
       prefetch: true
-      concurrent:
-        thresholdSize: 10M
-        thresholdSpeed: 2M
-        goroutineCount: 4
 
 dfdaemon:
   image: d7yio/dfdaemon
@@ -82,10 +78,6 @@ dfdaemon:
     metrics: 127.0.0.1:8888
     download:
       prefetch: true
-      concurrent:
-        thresholdSize: 10M
-        thresholdSpeed: 2M
-        goroutineCount: 4
     scheduler:
       disableAutoBackSource: true
     proxy: