Kubernetes源码阅读笔记——Scheduler（之一）

Scheduler是集群中Master节点的重要组件，其功能是根据集群中各Pod的资源需求、亲和性等指标，将Pod合理调度到Kubernetes集群中的各个节点上。

一、入口函数

入口函数与Controller Manager的入口函数结构相同，同样是应用了cobra包，在命令行中注册了kube-scheduler命令。

cmd/kube-scheduler/scheduler.go

func main() {
rand.Seed(time.Now().UnixNano())
command := app.NewSchedulerCommand()
pflag.CommandLine.SetNormalizeFunc(utilflag.WordSepNormalizeFunc)
logs.InitLogs()
defer logs.FlushLogs()
if err := command.Execute(); err != nil {
fmt.Fprintf(os.Stderr, "%vn", err)
os.Exit(1)
}
}

这里核心的方法仍然是NewSchedulerCommand。该方法位于app/server.go中，结构与Controller Manager几乎一样，因此不贴上来了。核心的部分仍然是在cobra.Command结构体的Run字段中调用runCommand方法。

runCommand方法为Scheduler配置Config，最终返回的是Run方法，将Scheduler运行起来。

runCommand方法中间有一行值得注意：

func runCommand(cmd *cobra.Command, args []string, opts *options.Options) error {
...
algorithmprovider.ApplyFeatureGates()
...

return Run(cc, stopCh)

}

这一行的作用是调用ApplyFeatureGates方法，并根据Feature Gate的配置，注册或者删除相应的预选策略。

进入ApplyFeatureGates方法，发现方法就一行，而整个包就这一个方法：

pkg/scheduler/algorithmprovider/plugin.go
package algorithmprovider
import
"k8s.io/kubernetes/pkg/scheduler/algorithmprovider/defaults"
// ApplyFeatureGates applies algorithm by feature gates.
func ApplyFeatureGates() {
defaults.ApplyFeatureGates()
}

事实上，在pkg/scheduler/algorirhmprovider/defaults/defaults.go中，有一个init方法：

pkg/scheduler/algorirhmprovider/defaults/defaults.go

func init() {
registerAlgorithmProvider(defaultPredicates(), defaultPriorities())
}

因此，在导入defaults包时，就已经执行了registerAlgorithmProvider方法，对一些预选与优选方法进行了注册。再配合ApplyFeatureGates方法，根据k8s中一些feature的开启情况，增加或删除一些预选和优选方法。这些feature的位置在pkg/features/kube_features.go中。

详细的预选和优选方法的定义位于pkg/scheduler/algorithm和pkg/scheduler/algorithmprovider中，这里不详细展开。

二、Run

看一下Run方法：

func Run(cc schedulerserverconfig.CompletedConfig, stopCh <-chan struct{}) error {
// Create the scheduler.
sched, err := scheduler.New(cc.Client,
cc.InformerFactory.Core().V1().Nodes(),
cc.PodInformer,
cc.InformerFactory.Core().V1().PersistentVolumes(),
cc.InformerFactory.Core().V1().PersistentVolumeClaims(),
cc.InformerFactory.Core().V1().ReplicationControllers(),
cc.InformerFactory.Apps().V1().ReplicaSets(),
cc.InformerFactory.Apps().V1().StatefulSets(),
cc.InformerFactory.Core().V1().Services(),
cc.InformerFactory.Policy().V1beta1().PodDisruptionBudgets(),
cc.InformerFactory.Storage().V1().StorageClasses(),
cc.Recorder,
cc.ComponentConfig.AlgorithmSource,
stopCh,
scheduler.WithName(cc.ComponentConfig.SchedulerName),
scheduler.WithHardPodAffinitySymmetricWeight(cc.ComponentConfig.HardPodAffinitySymmetricWeight),
scheduler.WithPreemptionDisabled(cc.ComponentConfig.DisablePreemption),
scheduler.WithPercentageOfNodesToScore(cc.ComponentConfig.PercentageOfNodesToScore),
scheduler.WithBindTimeoutSeconds(*cc.ComponentConfig.BindTimeoutSeconds))
if err != nil {
return err
}
// Prepare the event broadcaster.

...
// Setup healthz checks.

...// Start all informers.
go cc.PodInformer.Informer().Run(stopCh)
cc.InformerFactory.Start(stopCh)
// Wait for all caches to sync before scheduling.
cc.InformerFactory.WaitForCacheSync(stopCh)
controller.WaitForCacheSync("scheduler", stopCh, cc.PodInformer.Informer().HasSynced)
// Prepare a reusable runCommand function.
run := func(ctx context.Context) {
sched.Run()
<-ctx.Done()
}
ctx, cancel := context.WithCancel(context.TODO()) // TODO once Run() accepts a context, it should be used here

defer cancel()
go func() {
select {
case <-stopCh:
cancel()
case <-ctx.Done():
}
}()
// If leader election is enabled, runCommand via LeaderElector until done and exit.

...
// Leader election is disabled, so runCommand inline until done.

run(ctx)
return fmt.Errorf("finished without leader elect")
}

Run方法主要包含下面几件事：

（1）创建Scheduler。

Run方法的前几行代码调用了New方法，创建了一个Scheduler对象。这个New方法位于pkg/scheduler/scheduler.go中：

pkg/scheduler/scheduler.go
func New(client clientset.Interface,
nodeInformer coreinformers.NodeInformer,
podInformer coreinformers.PodInformer,
pvInformer coreinformers.PersistentVolumeInformer,
pvcInformer coreinformers.PersistentVolumeClaimInformer,
replicationControllerInformer coreinformers.ReplicationControllerInformer,
replicaSetInformer appsinformers.ReplicaSetInformer,
statefulSetInformer appsinformers.StatefulSetInformer,
serviceInformer coreinformers.ServiceInformer,
pdbInformer policyinformers.PodDisruptionBudgetInformer,
storageClassInformer storageinformers.StorageClassInformer,
recorder record.EventRecorder,
schedulerAlgorithmSource kubeschedulerconfig.SchedulerAlgorithmSource,
stopCh <-chan struct{},
opts ...func(o *schedulerOptions)) (*Scheduler, error) {
options := defaultSchedulerOptions
for _, opt := range opts {
opt(&options)
}
// Set up the configurator which can create schedulers from configs.
configurator := factory.NewConfigFactory(&factory.ConfigFactoryArgs{
SchedulerName:
options.schedulerName,
Client:
client,
NodeInformer:
nodeInformer,
PodInformer:
podInformer,
PvInformer:
pvInformer,
PvcInformer:
pvcInformer,
ReplicationControllerInformer:
replicationControllerInformer,
ReplicaSetInformer:
replicaSetInformer,
StatefulSetInformer:
statefulSetInformer,
ServiceInformer:
serviceInformer,
PdbInformer:
pdbInformer,
StorageClassInformer:
storageClassInformer,
HardPodAffinitySymmetricWeight: options.hardPodAffinitySymmetricWeight,
DisablePreemption:
options.disablePreemption,
PercentageOfNodesToScore:
options.percentageOfNodesToScore,
BindTimeoutSeconds:
options.bindTimeoutSeconds,
})
var config *factory.Config
source := schedulerAlgorithmSource
switch {
case source.Provider != nil:
// Create the config from a named algorithm provider.

...
case source.Policy != nil:
// Create the config from a user specified policy source.

...
default:
return nil, fmt.Errorf("unsupported algorithm source: %v", source)
}
// Additional tweaks to the config produced by the configurator.
config.Recorder = recorder
config.DisablePreemption = options.disablePreemption
config.StopEverything = stopCh
// Create the scheduler.
sched := NewFromConfig(config)
return sched, nil
}

New方法逻辑相对清晰，其本质就是根据传入的Informer、算法等参数，实例化一个Config，然后调用NewFromConfig方法，通过这个Config创建一个scheduler实例并返回。可以看到，scheduler中也用到了包括nodeInformer、podInformer等在内的大量Informer，因为scheduler也需要及时掌握资源的变化，从而调整调度的策略。

中间switch一段代码会判断config的调度算法源是用户自定义的还是给定的provider。如果使用默认的provider，则会将前面注册过的预选、优选方法加载进来。

创建config的NewConfigFactory方法位于pkg/scheduler/factory/factory.go中，进入方法：

pkg/scheduler/factory/factory.go
// NewConfigFactory initializes the default implementation of a Configurator. To encourage eventual privatization of the struct type, we only
// return the interface.
func NewConfigFactory(args *ConfigFactoryArgs) Configurator {
stopEverything := args.StopCh
if stopEverything == nil {
stopEverything = wait.NeverStop
}
schedulerCache := schedulerinternalcache.New(30*time.Second, stopEverything)
// storageClassInformer is only enabled through VolumeScheduling feature gate

var storageClassLister storagelisters.StorageClassLister
if args.StorageClassInformer != nil {
storageClassLister = args.StorageClassInformer.Lister()
}
c := &configFactory{
client:
args.Client,
podLister:
schedulerCache,
podQueue:
internalqueue.NewSchedulingQueue(stopEverything),
nodeLister:
args.NodeInformer.Lister(),
pVLister:
args.PvInformer.Lister(),
pVCLister:
args.PvcInformer.Lister(),
serviceLister:
args.ServiceInformer.Lister(),
controllerLister:
args.ReplicationControllerInformer.Lister(),
replicaSetLister:
args.ReplicaSetInformer.Lister(),
statefulSetLister:
args.StatefulSetInformer.Lister(),
pdbLister:
args.PdbInformer.Lister(),
storageClassLister:
storageClassLister,
schedulerCache:
schedulerCache,
StopEverything:
stopEverything,
schedulerName:
args.SchedulerName,
hardPodAffinitySymmetricWeight: args.HardPodAffinitySymmetricWeight,
disablePreemption:
args.DisablePreemption,
percentageOfNodesToScore:
args.PercentageOfNodesToScore,
}
c.scheduledPodsHasSynced = args.PodInformer.Informer().HasSynced
// scheduled pod cache

args.PodInformer.Informer().AddEventHandler(
cache.FilteringResourceEventHandler{
FilterFunc: func(obj interface{}) bool {
switch t := obj.(type) {
case *v1.Pod:
return assignedPod(t)
case cache.DeletedFinalStateUnknown:
if pod, ok := t.Obj.(*v1.Pod); ok {
return assignedPod(pod)
}
runtime.HandleError(fmt.Errorf("unable to convert object %T to *v1.Pod in %T", obj, c))
return false
default:
runtime.HandleError(fmt.Errorf("unable to handle object in %T: %T", c, obj))
return false
}
},
Handler: cache.ResourceEventHandlerFuncs{
AddFunc:
c.addPodToCache,
UpdateFunc: c.updatePodInCache,
DeleteFunc: c.deletePodFromCache,
},
},
)
// unscheduled pod queue

args.PodInformer.Informer().AddEventHandler(
cache.FilteringResourceEventHandler{
FilterFunc: func(obj interface{}) bool {
switch t := obj.(type) {
case *v1.Pod:
return !assignedPod(t) && responsibleForPod(t, args.SchedulerName)
case cache.DeletedFinalStateUnknown:
if pod, ok := t.Obj.(*v1.Pod); ok {
return !assignedPod(pod) && responsibleForPod(pod, args.SchedulerName)
}
runtime.HandleError(fmt.Errorf("unable to convert object %T to *v1.Pod in %T", obj, c))
return false
default:
runtime.HandleError(fmt.Errorf("unable to handle object in %T: %T", c, obj))
return false
}
},
Handler: cache.ResourceEventHandlerFuncs{
AddFunc:
c.addPodToSchedulingQueue,
UpdateFunc: c.updatePodInSchedulingQueue,
DeleteFunc: c.deletePodFromSchedulingQueue,
},
},
)
// ScheduledPodLister is something we provide to plug-in functions that
// they may need to call.
c.scheduledPodLister = assignedPodLister{args.PodInformer.Lister()}
args.NodeInformer.Informer().AddEventHandler(
cache.ResourceEventHandlerFuncs{
AddFunc:
c.addNodeToCache,
UpdateFunc: c.updateNodeInCache,
DeleteFunc: c.deleteNodeFromCache,
},
)
args.PvInformer.Informer().AddEventHandler(
cache.ResourceEventHandlerFuncs{
// MaxPDVolumeCountPredicate: since it relies on the counts of PV.

AddFunc:
c.onPvAdd,
UpdateFunc: c.onPvUpdate,
},
)
// This is for MaxPDVolumeCountPredicate: add/delete PVC will affect counts of PV when it is bound.

args.PvcInformer.Informer().AddEventHandler(
cache.ResourceEventHandlerFuncs{
AddFunc:
c.onPvcAdd,
UpdateFunc: c.onPvcUpdate,
},
)
// This is for ServiceAffinity: affected by the selector of the service is updated.

args.ServiceInformer.Informer().AddEventHandler(
cache.ResourceEventHandlerFuncs{
AddFunc:
c.onServiceAdd,
UpdateFunc: c.onServiceUpdate,
DeleteFunc: c.onServiceDelete,
},
)
// Setup volume binder
c.volumeBinder = volumebinder.NewVolumeBinder(args.Client, args.NodeInformer, args.PvcInformer, args.PvInformer, args.StorageClassInformer, time.Duration(args.BindTimeoutSeconds)*time.Second)
args.StorageClassInformer.Informer().AddEventHandler(
cache.ResourceEventHandlerFuncs{
AddFunc: c.onStorageClassAdd,
},
)
// Setup cache debugger

...
go func() {
<-c.StopEverything
c.podQueue.Close()
}()
return c
}

该方法为一系列Informer初始化了回调函数。其中最重要的是PodInformer的两个回调函数。

可以看到，方法调用了两次AddEventHandler方法，都经过了过滤。第一次只处理已调度的Pod，第二次只处理未调度的Pod，并定义了对两种Pod的增、改、删方法，分别在缓存和队列中对这两种Pod进行更新。这样，就将已调度和未调度的Pod区分开。

后面为其他informer添加的回调函数，除了NodeInformer的回调函数会在缓存中更新node信息，其他回调函数最终都会调用MoveAllToActiveQueue方法，将待调度的Pod添加进队列。

此外，可以看到，在ConfigFactory中，有一个podQueue字段，维护了一个队列，用于存放待调度的Pod。

（2）运行广播和健康检查。

中间有几行是为Scheduler配置广播和健康检查相关内容，与Controller Manager类似，不提。

（3）Informer启动。

值得注意的是，Scheduler将PodInformer从其他的Informer中独立出来，因为对Pod的调度才是Scheduler的核心。

（4）运行Scheduler。

这是整个方法的核心。通过调用Scheduler的Run方法，将Scheduler运行起来。

进入Run方法，我们发现方法非常简洁，就做了2件事：

pkg/scheduler/scheduler.go
func (sched *Scheduler) Run() {
if !sched.config.WaitForCacheSync() {
return
}
go wait.Until(sched.scheduleOne, 0, sched.config.StopEverything)
}

第一件事是等待各informer的缓存同步，第二件事是调用scheduleOne方法，执行Pod的调度操作。wait.Until的作用是每隔一段时间执行一次sched.scheduleOne方法，除非sched.config.StopEverything被关闭。这里时间段被设置为0，所以scheduleOne方法会一个接一个不停地被调用。

scheduleOne方法的具体逻辑我们下一篇文章再继续分析。https://www.cnblogs.com/00986014w/p/10320844.html

三、总结

总结Scheduler的逻辑，大体上是通过cobra注册一个kube-scheduler命令并运行。命令运行时，首先应用给定的调度算法，然后基于ConfigFactory，创建一个Scheduler的实例，启动相关的Informer，然后开始执行调度。