Merge pull request #414 from k82cn/kb_412

Added --schedule-period

cmd/kube-batch/app/options/options.go

@@ -18,6 +18,7 @@ package options
 
 import (
 	"fmt"
+	"time"
 
 	"github.com/spf13/pflag"
 )
@@ -28,6 +29,7 @@ type ServerOption struct {
 	Kubeconfig           string
 	SchedulerName        string
 	SchedulerConf        string
+	SchedulePeriod       string
 	NamespaceAsQueue     bool
 	EnableLeaderElection bool
 	LockObjectNamespace  string
@@ -46,6 +48,7 @@ func (s *ServerOption) AddFlags(fs *pflag.FlagSet) {
 	// kube-batch will ignore pods with scheduler names other than specified with the option
 	fs.StringVar(&s.SchedulerName, "scheduler-name", "kube-batch", "kube-batch will handle pods with the scheduler-name")
 	fs.StringVar(&s.SchedulerConf, "scheduler-conf", "", "The namespace and name of ConfigMap for scheduler configuration")
+	fs.StringVar(&s.SchedulePeriod, "schedule-period", "1s", "The period between each scheduling cycle")
 	fs.BoolVar(&s.EnableLeaderElection, "leader-elect", s.EnableLeaderElection, "Start a leader election client and gain leadership before "+
 		"executing the main loop. Enable this when running replicated kube-batch for high availability")
 	fs.BoolVar(&s.NamespaceAsQueue, "enable-namespace-as-queue", true, "Make Namespace as Queue with weight one, "+
@@ -57,5 +60,9 @@ func (s *ServerOption) CheckOptionOrDie() error {
 	if s.EnableLeaderElection && s.LockObjectNamespace == "" {
 		return fmt.Errorf("lock-object-namespace must not be nil when LeaderElection is enabled")
 	}
+	if _, err := time.ParseDuration(s.SchedulePeriod); err != nil {
+		return fmt.Errorf("failed to parse --schedule-period: %v", err)
+	}
+
 	return nil
 }

cmd/kube-batch/app/server.go

@@ -60,7 +60,8 @@ func Run(opt *options.ServerOption) error {
 	neverStop := make(chan struct{})
 
 	// Start policy controller to allocate resources.
-	sched, err := scheduler.NewScheduler(config, opt.SchedulerName, opt.SchedulerConf, opt.NamespaceAsQueue)
+	sched, err := scheduler.NewScheduler(config, opt.SchedulerName,
+		opt.SchedulerConf, opt.SchedulePeriod, opt.NamespaceAsQueue)
 	if err != nil {
 		panic(err)
 	}

pkg/scheduler/scheduler.go

@@ -29,23 +29,27 @@ import (
 )
 
 type Scheduler struct {
-	cache         schedcache.Cache
-	config        *rest.Config
-	actions       []framework.Action
-	pluginArgs    []*framework.PluginArgs
-	schedulerConf string
+	cache          schedcache.Cache
+	config         *rest.Config
+	actions        []framework.Action
+	pluginArgs     []*framework.PluginArgs
+	schedulerConf  string
+	schedulePeriod time.Duration
 }
 
 func NewScheduler(
 	config *rest.Config,
 	schedulerName string,
 	conf string,
+	period string,
 	nsAsQueue bool,
 ) (*Scheduler, error) {
+	sp, _ := time.ParseDuration(period)
 	scheduler := &Scheduler{
-		config:        config,
-		schedulerConf: conf,
-		cache:         schedcache.New(config, schedulerName, nsAsQueue),
+		config:         config,
+		schedulerConf:  conf,
+		cache:          schedcache.New(config, schedulerName, nsAsQueue),
+		schedulePeriod: sp,
 	}
 
 	return scheduler, nil
@@ -69,7 +73,7 @@ func (pc *Scheduler) Run(stopCh <-chan struct{}) {
 
 	pc.actions, pc.pluginArgs = loadSchedulerConf(conf)
 
-	go wait.Until(pc.runOnce, 1*time.Second, stopCh)
+	go wait.Until(pc.runOnce, pc.schedulePeriod, stopCh)
 }
 
 func (pc *Scheduler) runOnce() {