diff --git a/cmd/kube-batch/app/options/options.go b/cmd/kube-batch/app/options/options.go index a61532f95..104fa7b6e 100644 --- a/cmd/kube-batch/app/options/options.go +++ b/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 } diff --git a/cmd/kube-batch/app/server.go b/cmd/kube-batch/app/server.go index 36e87e503..32b58bd69 100644 --- a/cmd/kube-batch/app/server.go +++ b/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) } diff --git a/pkg/scheduler/scheduler.go b/pkg/scheduler/scheduler.go index 5c7d0bd6e..7c3e11252 100644 --- a/pkg/scheduler/scheduler.go +++ b/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() {