kube-scheduler源码分析（一）之 NewSchedulerCommand

scheduler的cmd代码目录结构如下：

1. Main函数

kube-scheduler的入口函数Main函数，仍然是采用统一的代码风格，使用命令行框架。

  1. func main() {
  2.     rand.Seed(time.Now().UTC().UnixNano())
  4.     command := app.NewSchedulerCommand()
  6.     // TODO: once we switch everything over to Cobra commands, we can go back to calling
  7.     // utilflag.InitFlags() (by removing its pflag.Parse() call). For now, we have to set the
  8.     // normalize func and add the go flag set by hand.
  9.     pflag.CommandLine.SetNormalizeFunc(utilflag.WordSepNormalizeFunc)
  10.     pflag.CommandLine.AddGoFlagSet(goflag.CommandLine)
  11.     // utilflag.InitFlags()
  12.     logs.InitLogs()
  13.     defer logs.FlushLogs()
  15.     if err := command.Execute(); err != nil {
  16.         fmt.Fprintf(os.Stderr, "%v\n", err)
  17.         os.Exit(1)
  18.     }
  19. }

核心代码：

  1. // 初始化scheduler命令结构体
  2. command := app.NewSchedulerCommand()
  3. // 执行Execute
  4. err := command.Execute()

2.

NewSchedulerCommand主要用来构造和初始化SchedulerCommand结构体，

  1. // NewSchedulerCommand creates a *cobra.Command object with default parameters
  2. func NewSchedulerCommand() *cobra.Command {
  3.     opts, err := options.NewOptions()
  4.     if err != nil {
  5.         glog.Fatalf("unable to initialize command options: %v", err)
  6.     }
  8.     cmd := &cobra.Command{
  9.         Use: "kube-scheduler",
  10.         Long: `The Kubernetes scheduler is a policy-rich, topology-aware,
  11. workload-specific function that significantly impacts availability, performance,
  12. and capacity. The scheduler needs to take into account individual and collective
  13. resource requirements, quality of service requirements, hardware/software/policy
  14. constraints, affinity and anti-affinity specifications, data locality, inter-workload
  15. interference, deadlines, and so on. Workload-specific requirements will be exposed
  16. through the API as necessary.`,
  17.         Run: func(cmd *cobra.Command, args []string) {
  18.             verflag.PrintAndExitIfRequested()
  19.             utilflag.PrintFlags(cmd.Flags())
  21.             if len(args) != 0 {
  22.                 fmt.Fprint(os.Stderr, "arguments are not supported\n")
  23.             }
  25.             if errs := opts.Validate(); len(errs) > 0 {
  26.                 fmt.Fprintf(os.Stderr, "%v\n", utilerrors.NewAggregate(errs))
  27.                 os.Exit(1)
  28.             }
  30.             if len(opts.WriteConfigTo) > 0 {
  31.                 if err := options.WriteConfigFile(opts.WriteConfigTo, &opts.ComponentConfig); err != nil {
  32.                     fmt.Fprintf(os.Stderr, "%v\n", err)
  33.                     os.Exit(1)
  34.                 }
  35.                 glog.Infof("Wrote configuration to: %s\n", opts.WriteConfigTo)
  36.                 return
  37.             }
  39.             c, err := opts.Config()
  40.             if err != nil {
  41.                 fmt.Fprintf(os.Stderr, "%v\n", err)
  42.                 os.Exit(1)
  43.             }
  45.             stopCh := make(chan struct{})
  46.             if err := Run(c.Complete(), stopCh); err != nil {
  47.                 fmt.Fprintf(os.Stderr, "%v\n", err)
  48.                 os.Exit(1)
  49.             }
  50.         },
  51.     }
  53.     opts.AddFlags(cmd.Flags())
  54.     cmd.MarkFlagFilename("config", "yaml", "yml", "json")
  56.     return cmd
  57. }

核心代码：

  1. // 构造option
  2. opts, err := options.NewOptions()
  3. // 初始化config对象
  4. c, err := opts.Config()
  5. // 执行run函数
  6. err := Run(c.Complete(), stopCh)
  7. // 添加参数
  8. opts.AddFlags(cmd.Flags())

NewOptions主要用来构造SchedulerServer使用的参数和上下文，其中核心参数是KubeSchedulerConfiguration。

    NewOptions:

    1. // NewOptions returns default scheduler app options.
    2. func NewOptions() (*Options, error) {
    3.     cfg, err := newDefaultComponentConfig()
    4.     if err != nil {
    5.         return nil, err
    8.     hhost, hport, err := splitHostIntPort(cfg.HealthzBindAddress)
    9.     if err != nil {
    10.         return nil, err
    11.     }
    13.     o := &Options{
    14.         ComponentConfig: *cfg,
    15.         SecureServing:   nil, // TODO: enable with apiserveroptions.NewSecureServingOptions()
    16.         CombinedInsecureServing: &CombinedInsecureServingOptions{
    17.             Healthz: &apiserveroptions.DeprecatedInsecureServingOptions{
    18.                 BindNetwork: "tcp",
    19.             },
    20.             Metrics: &apiserveroptions.DeprecatedInsecureServingOptions{
    21.                 BindNetwork: "tcp",
    22.             },
    23.             BindPort:    hport,
    24.             BindAddress: hhost,
    25.         },
    26.         Authentication: nil, // TODO: enable with apiserveroptions.NewDelegatingAuthenticationOptions()
    27.         Authorization:  nil, // TODO: enable with apiserveroptions.NewDelegatingAuthorizationOptions()
    28.         Deprecated: &DeprecatedOptions{
    29.             UseLegacyPolicyConfig:    false,
    30.             PolicyConfigMapNamespace: metav1.NamespaceSystem,
    31.         },
    32.     }
    34.     return o, nil
    35. }

    2.2. Options.Config

    Config函数主要执行以下操作：

    • 构建scheduler client、leaderElectionClient、eventClient。
    • 创建event recorder
    • 设置leader选举
    • 创建informer对象，主要函数有NewSharedInformerFactory和NewPodInformer。

    Config具体代码如下：

    1. // Config return a scheduler config object
    2. func (o *Options) Config() (*schedulerappconfig.Config, error) {
    3.     c := &schedulerappconfig.Config{}
    4.     if err := o.ApplyTo(c); err != nil {
    5.         return nil, err
    6.     }
    8.     // prepare kube clients.
    9.     client, leaderElectionClient, eventClient, err := createClients(c.ComponentConfig.ClientConnection, o.Master, c.ComponentConfig.LeaderElection.RenewDeadline.Duration)
    10.     if err != nil {
    11.         return nil, err
    12.     }
    14.     // Prepare event clients.
    15.     eventBroadcaster := record.NewBroadcaster()
    16.     recorder := eventBroadcaster.NewRecorder(legacyscheme.Scheme, corev1.EventSource{Component: c.ComponentConfig.SchedulerName})
    18.     // Set up leader election if enabled.
    19.     var leaderElectionConfig *leaderelection.LeaderElectionConfig
    20.     if c.ComponentConfig.LeaderElection.LeaderElect {
    21.         leaderElectionConfig, err = makeLeaderElectionConfig(c.ComponentConfig.LeaderElection, leaderElectionClient, recorder)
    22.         if err != nil {
    23.             return nil, err
    24.         }
    25.     }
    27.     c.Client = client
    28.     c.InformerFactory = informers.NewSharedInformerFactory(client, 0)
    29.     c.PodInformer = factory.NewPodInformer(client, 0)
    30.     c.EventClient = eventClient
    31.     c.Recorder = recorder
    32.     c.Broadcaster = eventBroadcaster
    33.     c.LeaderElection = leaderElectionConfig
    35.     return c, nil
    36. }

    AddFlags为SchedulerServer添加指定的参数。

    1. opts.AddFlags(cmd.Flags())

    AddFlags函数的具体代码如下：

    1. // AddFlags adds flags for the scheduler options.
    2. func (o *Options) AddFlags(fs *pflag.FlagSet) {
    3.     fs.StringVar(&o.ConfigFile, "config", o.ConfigFile, "The path to the configuration file. Flags override values in this file.")
    4.     fs.StringVar(&o.WriteConfigTo, "write-config-to", o.WriteConfigTo, "If set, write the configuration values to this file and exit.")
    5.     fs.StringVar(&o.Master, "master", o.Master, "The address of the Kubernetes API server (overrides any value in kubeconfig)")
    7.     o.SecureServing.AddFlags(fs)
    8.     o.CombinedInsecureServing.AddFlags(fs)
    9.     o.Authentication.AddFlags(fs)
    10.     o.Authorization.AddFlags(fs)
    11.     o.Deprecated.AddFlags(fs, &o.ComponentConfig)
    13.     leaderelectionconfig.BindFlags(&o.ComponentConfig.LeaderElection.LeaderElectionConfiguration, fs)
    14.     utilfeature.DefaultFeatureGate.AddFlag(fs)
    15. }

    3. Run

    1. err := Run(c.Complete(), stopCh)

    Run运行一个不退出的常驻进程，来执行scheduler的相关操作。

    Run函数的主要内容如下：

    • 通过scheduler config来创建scheduler的结构体。
    • 运行event broadcaster、healthz server、metrics server。
    • 运行所有的informer并在调度前等待cache的同步（重点）。
    • 执行sched.Run()来运行scheduler的调度逻辑。
    • 如果多个scheduler并开启了LeaderElect，则执行leader选举。

    以下对重点代码分开分析：

    3.1. NewSchedulerConfig

    1. // Build a scheduler config from the provided algorithm source.
    2. schedulerConfig, err := NewSchedulerConfig(c)
    3. if err != nil {
    4.     return err
    6. // Create the scheduler.
    7. sched := scheduler.NewFromConfig(schedulerConfig)
    1. // Start all informers.
    2. go c.PodInformer.Informer().Run(stopCh)
    3. c.InformerFactory.Start(stopCh)

    3.3. WaitForCacheSync

    在调度前等待cache同步。

    等待所有启动的informer的cache进行同步，保持本地的store信息与etcd的信息是最新一致的。

    1. // WaitForCacheSync waits for all started informers' cache were synced.
    2. func (f *sharedInformerFactory) WaitForCacheSync(stopCh <-chan struct{}) map[reflect.Type]bool {
    3.     informers := func() map[reflect.Type]cache.SharedIndexInformer {
    4.         f.lock.Lock()
    5.         defer f.lock.Unlock()
    7.         informers := map[reflect.Type]cache.SharedIndexInformer{}
    8.         for informerType, informer := range f.informers {
    9.             if f.startedInformers[informerType] {
    10.                 informers[informerType] = informer
    11.             }
    12.         }
    13.         return informers
    14.     }()
    16.     res := map[reflect.Type]bool{}
    17.     for informType, informer := range informers {
    18.         res[informType] = cache.WaitForCacheSync(stopCh, informer.HasSynced)
    19.     }
    20.     return res
    21. }

    接着调用cache.WaitForCacheSync。

    1. // WaitForCacheSync waits for caches to populate.  It returns true if it was successful, false
    2. // if the controller should shutdown
    3. func WaitForCacheSync(stopCh <-chan struct{}, cacheSyncs ...InformerSynced) bool {
    4.     err := wait.PollUntil(syncedPollPeriod,
    5.         func() (bool, error) {
    6.             for _, syncFunc := range cacheSyncs {
    7.                 if !syncFunc() {
    8.                     return false, nil
    9.                 }
    10.             }
    11.             return true, nil
    12.         },
    13.         stopCh)
    14.     if err != nil {
    15.         glog.V(2).Infof("stop requested")
    16.         return false
    17.     }
    19.     glog.V(4).Infof("caches populated")
    20.     return true
    21. }

    3.3.2. controller.WaitForCacheSync

    controller.WaitForCacheSync是对cache.WaitForCacheSync的一层封装，通过不同的controller的名字来记录不同controller等待cache同步。

    1. controller.WaitForCacheSync("scheduler", stop, s.PodInformer.Informer().HasSynced)

    controller.WaitForCacheSync具体代码如下：

    1. // WaitForCacheSync is a wrapper around cache.WaitForCacheSync that generates log messages
    2. // indicating that the controller identified by controllerName is waiting for syncs, followed by
    3. // either a successful or failed sync.
    4. func WaitForCacheSync(controllerName string, stopCh <-chan struct{}, cacheSyncs ...cache.InformerSynced) bool {
    5.     glog.Infof("Waiting for caches to sync for %s controller", controllerName)
    7.     if !cache.WaitForCacheSync(stopCh, cacheSyncs...) {
    8.         utilruntime.HandleError(fmt.Errorf("Unable to sync caches for %s controller", controllerName))
    9.         return false
    10.     }
    12.     glog.Infof("Caches are synced for %s controller", controllerName)
    13.     return true
    14. }

    如果有多个scheduler，并开启leader选举，则运行LeaderElector直到选举结束或退出。

    1. // If leader election is enabled, run via LeaderElector until done and exit.
    2. if c.LeaderElection != nil {
    3.     c.LeaderElection.Callbacks = leaderelection.LeaderCallbacks{
    4.         OnStartedLeading: run,
    5.         OnStoppedLeading: func() {
    6.             utilruntime.HandleError(fmt.Errorf("lost master"))
    7.         },
    8.     }
    9.     leaderElector, err := leaderelection.NewLeaderElector(*c.LeaderElection)
    10.     if err != nil {
    11.         return fmt.Errorf("couldn't create leader elector: %v", err)
    12.     }
    14.     leaderElector.Run(ctx)
    16.     return fmt.Errorf("lost lease")
    17. }

    3.5. Scheduler.Run

    1. // Prepare a reusable run function.
    2. run := func(ctx context.Context) {
    3.     sched.Run()
    4.     <-ctx.Done()
    5. }
    7. ctx, cancel := context.WithCancel(context.TODO()) // TODO once Run() accepts a context, it should be used here
    8. defer cancel()
    10. go func() {
    11.     select {
    12.     case <-stopCh:
    13.         cancel()
    14.     case <-ctx.Done():
    15.     }
    16. }()
    17. ...
    18. run(ctx)

    Scheduler.Run先等待cache同步，然后开启调度逻辑的goroutine。

    Scheduler.Run的具体代码如下：

    以上是对/cmd/kube-scheduler/scheduler.go部分代码的分析，后续的具体代码位于pkg/scheduler/scheduler.go待后续文章分析。