Refactor scheduler service (#958)

Signed-off-by: Gaius <[email protected]>

cdn/supervisor/task/manager.go

@@ -232,7 +232,7 @@ const (
 )
 
 func (tm *manager) GC() error {
-	logger.MetaGCLogger.Info("start the task meta gc job")
+	logger.GCLogger.Info("start the task meta gc job")
 	startTime := time.Now()
 	var gcTasks []string
 	var remainingTasks []string
@@ -247,17 +247,17 @@ func (tm *manager) GC() error {
 		}
 		gcTasks = append(gcTasks, taskID)
 		// gc task memory data
-		logger.MetaGCLogger.With("type", "meta").Infof("gc task: %s", taskID)
+		logger.GCLogger.With("type", "meta").Infof("gc task: %s", taskID)
 		tm.deleteTask(taskID)
 		return true
 	})
 
 	// slow GC detected, report it with a log warning
 	if timeDuring := time.Since(startTime); timeDuring > gcTasksTimeout {
-		logger.MetaGCLogger.With("type", "meta").Warnf("gc tasks: %d cost: %.3f", len(gcTasks), timeDuring.Seconds())
+		logger.GCLogger.With("type", "meta").Warnf("gc tasks: %d cost: %.3f", len(gcTasks), timeDuring.Seconds())
 	}
-	logger.MetaGCLogger.With("type", "meta").Infof("%d tasks were successfully cleared, leaving %d tasks remaining", len(gcTasks),
+	logger.GCLogger.With("type", "meta").Infof("%d tasks were successfully cleared, leaving %d tasks remaining", len(gcTasks),
 		len(remainingTasks))
-	logger.MetaGCLogger.With("type", "meta").Debugf("tasks %s were successfully cleared, leaving tasks %s remaining", gcTasks, remainingTasks)
+	logger.GCLogger.With("type", "meta").Debugf("tasks %s were successfully cleared, leaving tasks %s remaining", gcTasks, remainingTasks)
 	return nil
 }

client/config/dynconfig_test.go

@@ -43,7 +43,7 @@ func TestDynconfigNewDynconfig(t *testing.T) {
 		expect         func(t *testing.T, err error)
 	}{
 		{
-			name:   "new dynconfig succeeded",
+			name:   "new dynconfig",
 			expire: 10 * time.Second,
 			hostOption: HostOption{
 				Hostname: "foo",
@@ -135,7 +135,7 @@ func TestDynconfigGet(t *testing.T) {
 		expect         func(t *testing.T, dynconfig Dynconfig, data *DynconfigData)
 	}{
 		{
-			name:   "get dynconfig cache data succeeded",
+			name:   "get dynconfig cache data",
 			expire: 10 * time.Second,
 			hostOption: HostOption{
 				Hostname: "foo",
@@ -172,7 +172,7 @@ func TestDynconfigGet(t *testing.T) {
 			},
 		},
 		{
-			name:   "get dynconfig data succeeded",
+			name:   "get dynconfig data",
 			expire: 10 * time.Millisecond,
 			hostOption: HostOption{
 				Hostname: "foo",
@@ -317,7 +317,7 @@ func TestDynconfigGetSchedulers(t *testing.T) {
 		expect         func(t *testing.T, dynconfig Dynconfig, data *DynconfigData)
 	}{
 		{
-			name:   "get cache schedulers succeeded",
+			name:   "get cache schedulers",
 			expire: 10 * time.Second,
 			hostOption: HostOption{
 				Hostname: "foo",
@@ -354,7 +354,7 @@ func TestDynconfigGetSchedulers(t *testing.T) {
 			},
 		},
 		{
-			name:   "get schedulers succeeded",
+			name:   "get schedulers",
 			expire: 10 * time.Millisecond,
 			hostOption: HostOption{
 				Hostname: "foo",

cmd/dependency/dependency.go

@@ -162,15 +162,6 @@ func SetupQuitSignalHandler(handler func()) {
 	}()
 }
 
-func GetConfigPath(name string) string {
-	cfgFile := viper.GetString("config")
-	if cfgFile != "" {
-		return cfgFile
-	}
-
-	return filepath.Join(dfpath.DefaultConfigDir, fmt.Sprintf("%s.yaml", name))
-}
-
 // initConfig reads in config file and ENV variables if set.
 func initConfig(useConfigFile bool, name string, config interface{}) {
 	// Use config file and read once.

cmd/scheduler/cmd/root.go

@@ -17,6 +17,7 @@
 package cmd
 
 import (
+	"context"
 	"os"
 
 	"github.com/pkg/errors"
@@ -46,6 +47,9 @@ 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 {
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+
 		// Initialize dfpath
 		d, err := initDfpath(cfg.Server)
 		if err != nil {
@@ -62,12 +66,7 @@ generate and maintain a P2P network during the download process, and push suitab
 			return err
 		}
 
-		// Convert redis host config
-		if err := cfg.Convert(); err != nil {
-			return err
-		}
-
-		return runScheduler(d)
+		return runScheduler(ctx, d)
 	},
 }
 
@@ -100,7 +99,7 @@ func initDfpath(cfg *config.ServerConfig) (dfpath.Dfpath, error) {
 	return dfpath.New(options...)
 }
 
-func runScheduler(d dfpath.Dfpath) error {
+func runScheduler(ctx context.Context, d dfpath.Dfpath) error {
 	logger.Infof("Version:\n%s", version.Version())
 
 	// scheduler config values
@@ -111,7 +110,7 @@ func runScheduler(d dfpath.Dfpath) error {
 	ff := dependency.InitMonitor(cfg.Verbose, cfg.PProfPort, cfg.Telemetry)
 	defer ff()
 
-	svr, err := scheduler.New(cfg, d)
+	svr, err := scheduler.New(ctx, cfg, d)
 	if err != nil {
 		return err
 	}

cmd/scheduler/main.go

@@ -18,7 +18,6 @@ package main
 
 import (
 	"d7y.io/dragonfly/v2/cmd/scheduler/cmd"
-	_ "d7y.io/dragonfly/v2/scheduler/core/scheduler/basic"
 )
 
 func main() {

docs/en/deployment/configuration/cdn.yaml

@@ -1,5 +1,3 @@
-# This file is the template of cdn system configuration file.
-# You can configure your cdn system by change the parameter according your requirement.
 base:
   # listenPort is the port cdn server listens on.
   # default: 8003

docs/en/deployment/configuration/manager.yaml

@@ -1,6 +1,3 @@
-# This file is the template of manager configuration file.
-# You can configure your manager by change the parameter according your requirement.
----
 # current server info used for server
 server:
   # grpc server configure
@@ -34,7 +31,6 @@ database:
     host: dragonfly
     port: 6379
     db: 0
-
 # manager server cache
 # cache:
 #   # redis cache configure

docs/en/deployment/configuration/scheduler.yaml

@@ -1,5 +1,21 @@
-# This file is the template of scheduler configuration file.
-# You can configure your scheduler by change the parameter according your requirement.
+# server scheduler instance configuration
+server:
+  # # ip
+  # ip: 127.0.0.1
+  # # host
+  # host: localhost
+  # port is the ip and port scheduler server listens on.
+  port: 8002
+  # limit the number of requests
+  listenLimit: 1000
+  # cacheDir is dynconfig cache storage directory
+  # in linux, default value is /var/cache/dragonfly
+  # in macos(just for testing), default value is /Users/$USER/.dragonfly/cache
+  cacheDir: ""
+  # logDir is the log storage directory
+  # in linux, default value is /var/log/dragonfly
+  # in macos(just for testing), default value is /Users/$USER/.dragonfly/logs
+  logDir: ""
 
 # scheduler policy configuration
 scheduler:
@@ -10,66 +26,41 @@ scheduler:
   # and the compiled `d7y-scheduler-plugin-evaluator.so` file is added to
   # the dragonfly working directory plugins
   algorithm: default
-  # workerNum is the number of goroutines that perform scheduling tasks
-  # default:
-  workerNum: 12
-  # backSourceCount is the number of back-to-origin clients when the CDN is disable or unavailable
-  # default: 3
+  # backSourceCount is the number of backsource clients when the CDN is unavailable
   backSourceCount: 3
-  # accessWindow is access interval window that can trigger CDN back to the source again
-  # default: 3m
-  accessWindow: 3m
-  # candidateParentCount is number of candidate parent nodes
-  # default: 10
-  candidateParentCount: 10
-  # scheduler is currently effective scheduling policy
-  # default: basic
-  scheduler: basic
-  # openMonitor Whether to enable monitoring, currently only the current peer list status information is monitored
-  # default: false
-  openMonitor: false
+  # retry scheduling limit times
+  retryLimit: 10
+  # retry scheduling interval
+  retryInterval: 1s
   # gc metadata configuration
   gc:
     # peerGCInterval peer's gc interval
     peerGCInterval: 1m
     # peerTTL peer's TTL duration
-    peerTTL: 10m
-    # peerTTI peer's TTI duration
-    peerTTI: 3m
+    peerTTL: 5m
     # taskGCInterval task's gc interval
     taskGCInterval: 1m
     # taskTTL task's TTL duration
     taskTTL: 10m
-    # taskTTI task's TTI duration
-    taskTTI: 3m
-
-# server scheduler instance configuration
-server:
-  # ip
-  # ip: 127.0.0.1
-  # host
-  # host: localhost
-  # ListenPort is the ip and port scheduler server listens on.
-  # default: 8002
-  port: 8002
-
-  # cacheDir is dynconfig cache storage directory
-  # in linux, default value is /var/cache/dragonfly
-  # in macos(just for testing), default value is /Users/$USER/.dragonfly/cache
-  cacheDir: ""
-
-  # logDir is the log storage directory
-  # in linux, default value is /var/log/dragonfly
-  # in macos(just for testing), default value is /Users/$USER/.dragonfly/logs
-  logDir: ""
 
 # dynamic data configuration
 dynConfig:
-  # dynamic data source type
-  type: manager
+  # dynamic config refresh interval
+  refreshInterval: 5 * time.Minute
+
+# scheduler host configuration
+host:
+  # idc is the idc of scheduler instance
+  idc: ""
+  # netTopology is the net topology of scheduler instance
+  netTopology: ""
+  # location is the location of scheduler instance
+  location: ""
 
 # manager configuration
 manager:
+  # scheduler enable contact with manager
+  enable: true
   # addr manager access address
   addr: 127.0.0.1:65003
   # schedulerClusterID cluster id to which scheduler instance belongs
@@ -79,21 +70,16 @@ manager:
     # interval
     interval: 5s
 
-# host scheduler host configuration
-host:
-  # location is the location of scheduler instance
-  location: ""
-  # idc is the idc of scheduler instance
-  idc: ""
-
 # machinery async job configuration, see https://github.com/RichardKnop/machinery
 job:
-  # globalWorkerNum
-  globalWorkerNum: 2
-  # schedulerWorkerNum
-  schedulerWorkerNum: 3
-  # localWorkerNum
-  localWorkerNum: 3
+  # scheduler enable job service
+  enable: true
+  # number of workers in global queue
+  globalWorkerNum: 1
+  # number of workers in scheduler queue
+  schedulerWorkerNum: 1
+  # number of workers in local queue
+  localWorkerNum: 5
   # redis configuration
   redis:
     # host
@@ -103,35 +89,28 @@ job:
     # password
     password: ""
     # brokerDB
-    brokerDB: ""
+    brokerDB: 1
     # backendDB
-    backendDB: ""
+    backendDB: 2
 
-# whether to disable CDN
-# default: false
-disableCDN: false
+# enable prometheus metrics
+metrics:
+  # scheduler enable metrics service
+  enable: false
+  # metrics service address
+  addr: ":8000"
+  # enable peer host metrics
+  enablePeerHost: false
 
 # console shows log on console
-# default: false
 console: false
 
 # Whether to enable debug level logger and enable pprof
-# default: false
 verbose: false
 
 # listen port for pprof, only valid when the verbose option is true
 # default is -1. If it is 0, pprof will use a random port.
 pprofPort: -1
 
 # jaeger endpoint url, like: http://jaeger.dragonfly.svc:14268/api/traces
-# default: ""
 jaeger: ""
-
-# service name used in tracer
-# default: dragonfly-scheduler
-service-name: dragonfly-scheduler
-
-# enable prometheus metrics
-# metrics:
-#  # metrics service address
-#  addr: ":8000"

docs/zh-CN/deployment/configuration/cdn.yaml

@@ -1,5 +1,3 @@
-# CDN 配置模版
-# 你可以通过修改这里的配置项来自定义你的 CDN
 base:
   # CDN 服务监听的端口
   # 默认值：8003

docs/zh-CN/deployment/configuration/manager.yaml

@@ -1,6 +1,4 @@
-# 此文件是 manager 的配置文件模板，你可以通过根据需要改变对应的值来配置 manager 服务。
----
-# 当前的服务配置
+# 服务配置
 server:
   # grpc 服务配置
   grpc:
@@ -32,7 +30,6 @@ database:
     host: dragonfly
     port: 6379
     db: 0
-
 # 缓存配置
 # cache:
 #   # redis 缓存配置