// Scheduler watches for new unscheduled pods. It attempts to find
// nodes that they fit on and writes bindings back to the api server.
type Scheduler struct {
  // 通过 NodeLister 和 Algorithm 将观察到通过 Cache 所做的任何更改
  // It is expected that changes made via Cache will be observed
  // by NodeLister and Algorithm.
  Cache internalcache.Cache
​
  // 扩展器是外部进程影响Kubernetes做出的调度决策的接口,通常是不由Kubernetes直接管理的资源所需要的。这种方式用的不多
  Extenders []framework.Extender
​
  // 以阻塞的形式获取下一个有效的待调度的Pod。这里不使用channel，主要是因为对一个pod的调度可能需要一些时间
  // NextPod should be a function that blocks until the next pod
  // is available. We don't use a channel for this, because scheduling
  // a pod may take some amount of time and we don't want pods to get
  // stale while they sit in a channel.
  NextPod func() *framework.QueuedPodInfo
​
  // 调度失败处理 Handler
  // FailureHandler is called upon a scheduling failure.
  FailureHandler FailureHandlerFn
​
  // 试图将一个Pod调度到其中一个节点上，成功则返回 ScheduleResult, 否则返回带有失败原来的 FitError
  // SchedulePod tries to schedule the given pod to one of the nodes in the node list.
  // Return a struct of ScheduleResult with the name of suggested host on success,
  // otherwise will return a FitError with reasons.
  SchedulePod func(ctx context.Context, fwk framework.Framework, state *framework.CycleState, pod *v1.Pod) (ScheduleResult, error)
​
  // Close this to shut down the scheduler.
  StopEverything <-chan struct{}
​
  // 等待被调度的Pod队列，也就是说只有从这个队列里获取的Pod才能被调度
  // SchedulingQueue holds pods to be scheduled
  SchedulingQueue internalqueue.SchedulingQueue
​
  // 调度配置profiles
  // Profiles are the scheduling profiles.
  Profiles profile.Map
​
  // api server 专用 client
  client clientset.Interface
​
  // 节点信息快照
  nodeInfoSnapshot *internalcache.Snapshot
​
  // 节点打分相关
  percentageOfNodesToScore int32
​
  nextStartNodeIndex int
}

func New(client clientset.Interface,
  informerFactory informers.SharedInformerFactory,
  dynInformerFactory dynamicinformer.DynamicSharedInformerFactory,
  recorderFactory profile.RecorderFactory,
  stopCh <-chan struct{},
  opts ...Option) (*Scheduler, error) {}

// 函数选项式模块初始化调度器结构体
  options := defaultSchedulerOptions
  for _, opt := range opts {
    opt(&options)
  }

type schedulerOptions struct {
  componentConfigVersion string
  kubeConfig             *restclient.Config
  // Overridden by profile level percentageOfNodesToScore if set in v1.
  percentageOfNodesToScore          int32
  podInitialBackoffSeconds          int64
  podMaxBackoffSeconds              int64
  podMaxInUnschedulablePodsDuration time.Duration
  // Contains out-of-tree plugins to be merged with the in-tree registry.
  frameworkOutOfTreeRegistry frameworkruntime.Registry
  profiles                   []schedulerapi.KubeSchedulerProfile
  extenders                  []schedulerapi.Extender
  frameworkCapturer          FrameworkCapturer
  parallelism                int32
  applyDefaultProfile        bool
}

// KubeSchedulerProfile is a scheduling profile.
type KubeSchedulerProfile struct {
    SchedulerName            string  // 调度器名称
    PercentageOfNodesToScore *int32  // 集群中可用节点的百分比阈值
    Plugins                  *Plugins  // 所有插件，包含启用和禁用
    PluginConfig             []PluginConfig // 每个插件的一组可选的自定义插件参数,如果省略则使用默认配置
}

type Plugins struct {
  // PreEnqueue is a list of plugins that should be invoked before adding pods to the scheduling queue.
  PreEnqueue PluginSet
​
  // QueueSort is a list of plugins that should be invoked when sorting pods in the scheduling queue.
  QueueSort PluginSet
​
  // PreFilter is a list of plugins that should be invoked at "PreFilter" extension point of the scheduling framework.
  PreFilter PluginSet
​
  // Filter is a list of plugins that should be invoked when filtering out nodes that cannot run the Pod.
  Filter PluginSet
​
  // PostFilter is a list of plugins that are invoked after filtering phase, but only when no feasible nodes were found for the pod.
  PostFilter PluginSet
​
  // PreScore is a list of plugins that are invoked before scoring.
  PreScore PluginSet
​
  // Score is a list of plugins that should be invoked when ranking nodes that have passed the filtering phase.
  Score PluginSet
​
  // Reserve is a list of plugins invoked when reserving/unreserving resources
  // after a node is assigned to run the pod.
  Reserve PluginSet
​
  // Permit is a list of plugins that control binding of a Pod. These plugins can prevent or delay binding of a Pod.
  Permit PluginSet
​
  // PreBind is a list of plugins that should be invoked before a pod is bound.
  PreBind PluginSet
​
  // Bind is a list of plugins that should be invoked at "Bind" extension point of the scheduling framework.
  // The scheduler call these plugins in order. Scheduler skips the rest of these plugins as soon as one returns success.
  Bind PluginSet
​
  // PostBind is a list of plugins that should be invoked after a pod is successfully bound.
  PostBind PluginSet
​
  // MultiPoint is a simplified config field for enabling plugins for all valid extension points
  MultiPoint PluginSet
}
​
type PluginSet struct {
  // Enabled指定除了默认插件之外还应启用的插件。这些是在默认插件之后调用的，并且按照此处指定的相同顺序
  Enabled []Plugin
  // Disabled指定应禁用的默认插件。当需要禁用所有默认插件时，只包含一个“*”的数组应该be provided.
  Disabled []Plugin
}
​
type Plugin struct {
  Name string  // 插件名
  Weight int32  // 权重，仅在评分时使用
}

if options.applyDefaultProfile {
    // 调度器配置，很重要，特别是  Profiles  和 Extenders 字段

    // 对象转换，将 versiondCfg 转为 cfg
    var versionedCfg configv1.KubeSchedulerConfiguration
    scheme.Scheme.Default(&versionedCfg)
    cfg := schedulerapi.KubeSchedulerConfiguration{}
    if err := scheme.Scheme.Convert(&versionedCfg, &cfg, nil); err != nil {
      return nil, err
    }

    // 通过指定 "scheduler name" profile，可以用来控制Pod的调度行为，如何为空则默认使用 "default-scheduler" profile
    options.profiles = cfg.Profiles
  }

// KubeSchedulerConfiguration configures a scheduler
type KubeSchedulerConfiguration struct {
  ...
​
  // Profiles are scheduling profiles that kube-scheduler supports. Pods can
  // choose to be scheduled under a particular profile by setting its associated
  // scheduler name. Pods that don't specify any scheduler name are scheduled
  // with the "default-scheduler" profile, if present here.
  // +listType=map
  // +listMapKey=schedulerName
  Profiles []KubeSchedulerProfile `json:"profiles,omitempty"`
​
  // Extenders are the list of scheduler extenders, each holding the values of how to communicate
  // with the extender. These extenders are shared by all scheduler profiles.
  // +listType=set
  Extenders []Extender `json:"extenders,omitempty"`
}

// 插件构建函数注册表，
  // 注册表是所有可用插件的集合，framework 使用注册表来启用和初始化配置的插件。在初始化框架之前，所有插件都必须在注册表中。
  registry := frameworkplugins.NewInTreeRegistry()
  if err := registry.Merge(options.frameworkOutOfTreeRegistry); err != nil {
    return nil, err
  }

func NewInTreeRegistry() runtime.Registry {
  fts := plfeature.Features{
    EnableDynamicResourceAllocation:              feature.DefaultFeatureGate.Enabled(features.DynamicResourceAllocation),
    EnableReadWriteOncePod:                       feature.DefaultFeatureGate.Enabled(features.ReadWriteOncePod),
    EnableVolumeCapacityPriority:                 feature.DefaultFeatureGate.Enabled(features.VolumeCapacityPriority),
    EnableMinDomainsInPodTopologySpread:          feature.DefaultFeatureGate.Enabled(features.MinDomainsInPodTopologySpread),
    EnableNodeInclusionPolicyInPodTopologySpread: feature.DefaultFeatureGate.Enabled(features.NodeInclusionPolicyInPodTopologySpread),
    EnableMatchLabelKeysInPodTopologySpread:      feature.DefaultFeatureGate.Enabled(features.MatchLabelKeysInPodTopologySpread),
    EnablePodSchedulingReadiness:                 feature.DefaultFeatureGate.Enabled(features.PodSchedulingReadiness),
    EnablePodDisruptionConditions:                feature.DefaultFeatureGate.Enabled(features.PodDisruptionConditions),
    EnableInPlacePodVerticalScaling:              feature.DefaultFeatureGate.Enabled(features.InPlacePodVerticalScaling),
  }
​
  // 所有内置插件
  registry := runtime.Registry{
    dynamicresources.Name:                runtime.FactoryAdapter(fts, dynamicresources.New),
    selectorspread.Name:                  selectorspread.New,
    imagelocality.Name:                   imagelocality.New,
    tainttoleration.Name:                 tainttoleration.New,
    nodename.Name:                        nodename.New,
    nodeports.Name:                       nodeports.New,
    nodeaffinity.Name:                    nodeaffinity.New,
    podtopologyspread.Name:               runtime.FactoryAdapter(fts, podtopologyspread.New),
    nodeunschedulable.Name:               nodeunschedulable.New,
    noderesources.Name:                   runtime.FactoryAdapter(fts, noderesources.NewFit),
    noderesources.BalancedAllocationName: runtime.FactoryAdapter(fts, noderesources.NewBalancedAllocation),
    volumebinding.Name:                   runtime.FactoryAdapter(fts, volumebinding.New),
    volumerestrictions.Name:              runtime.FactoryAdapter(fts, volumerestrictions.New),
    volumezone.Name:                      volumezone.New,
    nodevolumelimits.CSIName:             runtime.FactoryAdapter(fts, nodevolumelimits.NewCSI),
    nodevolumelimits.EBSName:             runtime.FactoryAdapter(fts, nodevolumelimits.NewEBS),
    nodevolumelimits.GCEPDName:           runtime.FactoryAdapter(fts, nodevolumelimits.NewGCEPD),
    nodevolumelimits.AzureDiskName:       runtime.FactoryAdapter(fts, nodevolumelimits.NewAzureDisk),
    nodevolumelimits.CinderName:          runtime.FactoryAdapter(fts, nodevolumelimits.NewCinder),
    interpodaffinity.Name:                interpodaffinity.New,
    queuesort.Name:                       queuesort.New,
    defaultbinder.Name:                   defaultbinder.New,
    defaultpreemption.Name:               runtime.FactoryAdapter(fts, defaultpreemption.New),
    schedulinggates.Name:                 runtime.FactoryAdapter(fts, schedulinggates.New),
  }
​
  return registry
}

type Registry map[string]PluginFactory
func (r Registry) Register(name string, factory PluginFactory) error {}
func (r Registry) Unregister(name string) error {}
func (r Registry) Merge(in Registry) error {}

// pkg/scheduler/apis/config/types.go#L261-L301
type Extender struct {
    URLPrefix        string
    FilterVerb       string
    PreemptVerb      string
    PrioritizeVerb   string
    Weight           int64
    BindVerb         string
    EnableHTTPS      bool
    TLSConfig        *ExtenderTLSConfig
    HTTPTimeout      v1.Duration
    NodeCacheCapable bool
    ManagedResources []ExtenderManagedResource
    Ignorable        bool
}

extenders, err := buildExtenders(options.extenders, options.profiles)

func buildExtenders(extenders []schedulerapi.Extender, profiles []schedulerapi.KubeSchedulerProfile) ([]framework.Extender, error) {
  var fExtenders []framework.Extender
  if len(extenders) == 0 {
    return nil, nil
  }
​
  var ignoredExtendedResources []string
  var ignorableExtenders []framework.Extender
  for i := range extenders {
    // 根据每个扩展器参数配置创建一个 httpClient 对象的wrapper
    extender, err := NewHTTPExtender(&extenders[i])
    if err != nil {
      return nil, err
    }
    // 在调度时，当扩展程序返回错误或无法访问时，扩展器是否被忽略
    if !extender.IsIgnorable() {
      fExtenders = append(fExtenders, extender)
    } else {
      ignorableExtenders = append(ignorableExtenders, extender)
    }
    for _, r := range extenders[i].ManagedResources {
      if r.IgnoredByScheduler {
        ignoredExtendedResources = append(ignoredExtendedResources, r.Name)
      }
    }
  }
  // 将允许忽略的扩展器放在最后面
  fExtenders = append(fExtenders, ignorableExtenders...)
​
  // If there are any extended resources found from the Extenders, append them to the pluginConfig for each profile.
  // This should only have an effect on ComponentConfig, where it is possible to configure Extenders and
  // plugin args (and in which case the extender ignored resources take precedence).
  if len(ignoredExtendedResources) == 0 {
    return fExtenders, nil
  }
​
  // 更新每个scheduler name 中插件 noderesources.Name 的参数(prof.PluginConfig[k].Args.IgnoredResources)
  for i := range profiles {
    prof := &profiles[i]
    var found = false
    for k := range prof.PluginConfig {
      if prof.PluginConfig[k].Name == noderesources.Name {
        // Update the existing args
        pc := &prof.PluginConfig[k]
        args, ok := pc.Args.(*schedulerapi.NodeResourcesFitArgs)
        if !ok {
          return nil, fmt.Errorf("want args to be of type NodeResourcesFitArgs, got %T", pc.Args)
        }
        // 允许出错被忽略的扩展器
        args.IgnoredResources = ignoredExtendedResources
        found = true
        break
      }
    }
    if !found {
      return nil, fmt.Errorf("can't find NodeResourcesFitArgs in plugin config")
    }
  }
  return fExtenders, nil
}

type NodeResourcesFitArgs struct {
    v1.TypeMeta
    IgnoredResources      []string
    IgnoredResourceGroups []string
    ScoringStrategy       *ScoringStrategy
}

podLister := informerFactory.Core().V1().Pods().Lister()
nodeLister := informerFactory.Core().V1().Nodes().Lister()

// 节点快照
  snapshot := internalcache.NewEmptySnapshot()

  // clusterEvent 集合
  clusterEventMap := make(map[framework.ClusterEvent]sets.String)
  metricsRecorder := metrics.NewMetricsAsyncRecorder(1000, time.Second, stopCh)
​
  // 根据给出的profiles配置文件 生成对应的 frameworks，(Map类型，格式为 SchedulerName:Framework)
  // 而每一个 frameworks 都会对自己的 plugins 集进行管理，然后在调度上下文中的各种埋点执行插件
  // registry 其实是 frameworkImpl.registry 字段
  profiles, err := profile.NewMap(
    // 调度器配置
    options.profiles,

    // 插件构建器注册表
    registry,

    recorderFactory, stopCh,

    // 下面的全是 Options functions 模式，可任意数量
    // componentConfig 版本号
    frameworkruntime.WithComponentConfigVersion(options.componentConfigVersion),
​
    // api server 的客户端，一般通过读取 $HOME/.kube/config 文件创建
    frameworkruntime.WithClientSet(client),

    // kubeConfig,一般是 $HOME/kube/config 文件
    frameworkruntime.WithKubeConfig(options.kubeConfig),

    // SharedInformerFactory
    frameworkruntime.WithInformerFactory(informerFactory),

    // 设置 SharedLister，用途？
    frameworkruntime.WithSnapshotSharedLister(snapshot),

// 回调函数         frameworkruntime.WithCaptureProfile(frameworkruntime.CaptureProfile(options.frameworkCapturer)),
​
    // 设置一个 clusterEvent 的 Set 类型
    frameworkruntime.WithClusterEventMap(clusterEventMap),

    // 设置并行调度数量，默认为16
    frameworkruntime.WithParallelism(int(options.parallelism)),

    // 设置外部扩展器
    frameworkruntime.WithExtenders(extenders),

    // 指标记录器
    frameworkruntime.WithMetricsRecorder(metricsRecorder),
  )

// NewMap builds the frameworks given by the configuration, indexed by name.
func NewMap(cfgs []config.KubeSchedulerProfile, r frameworkruntime.Registry, recorderFact RecorderFactory,
  stopCh <-chan struct{}, opts ...frameworkruntime.Option) (Map, error) {
  m := make(Map)
  v := cfgValidator{m: m}
​
  for _, cfg := range cfgs {
   // 根据给的配置构建 Framework
    p, err := newProfile(cfg, r, recorderFact, stopCh, opts...)
    if err != nil {
      return nil, fmt.Errorf("creating profile for scheduler name %s: %v", cfg.SchedulerName, err)
    }
    if err := v.validate(cfg, p); err != nil {
      return nil, err
    }

    // scheduler name 为key, 以 Framework 为value
    m[cfg.SchedulerName] = p
  }
  return m, nil
}

type Framework interface {
    // 插件需要的数据和一些工具，在插件初始化的时候通过 pluginFactory 传入,见 NewInTreeRegistry() 函数
    Handle

    // 返回注册过的 preEnqueue 插件列表
    PreEnqueuePlugins() []PreEnqueuePlugin

    // 获取 pod 在队列中的排序函数
    QueueSortFunc() LessFunc

    // PreFilter 插件
    RunPreFilterPlugins(ctx context.Context, state *CycleState, pod *v1.Pod) (*PreFilterResult, *Status)

    // PostFilter
    RunPostFilterPlugins(ctx context.Context, state *CycleState, pod *v1.Pod, filteredNodeStatusMap NodeToStatusMap) (*PostFilterResult, *Status)
​
    // PreBind
    RunPreBindPlugins(ctx context.Context, state *CycleState, pod *v1.Pod, nodeName string) *Status

    // PostBind
    RunPostBindPlugins(ctx context.Context, state *CycleState, pod *v1.Pod, nodeName string)

    // Reserve
    RunReservePluginsReserve(ctx context.Context, state *CycleState, pod *v1.Pod, nodeName string) *Status
    // Unreserve
    RunReservePluginsUnreserve(ctx context.Context, state *CycleState, pod *v1.Pod, nodeName string)

    // Permit
    RunPermitPlugins(ctx context.Context, state *CycleState, pod *v1.Pod, nodeName string) *Status

    // 如果pod处于waiting状态，则将处于block状态，直到pod被 rejected 或 allowed
    WaitOnPermit(ctx context.Context, pod *v1.Pod) *Status

    // Bind
    RunBindPlugins(ctx context.Context, state *CycleState, pod *v1.Pod, nodeName string) *Status

    // 是否定义了 Filter 插件
    HasFilterPlugins() bool
    // 是否定义了 PostFilter 插件
    HasPostFilterPlugins() bool
    // 是否定义了 Score
    HasScorePlugins() bool

    // 返回定义的plugins 对象
    ListPlugins() *config.Plugins
    ProfileName() string
    PercentageOfNodesToScore() *int32

    // 设置 SetPodNominator
    SetPodNominator(nominator PodNominator)
}

preEnqueuePluginMap := make(map[string][]framework.PreEnqueuePlugin)
  for profileName, profile := range profiles {
    preEnqueuePluginMap[profileName] = profile.PreEnqueuePlugins()
  }

// 初始化一个优先队列作为调度队列 sched.SchedulingQueue
  podQueue := internalqueue.NewSchedulingQueue(
    // 获取profile 设置的调度队列Pod里的Pod排序函数，这里指定获取第 options.profiles[0]个profile
    profiles[options.profiles[0].SchedulerName].QueueSortFunc(),

    // sharedInformerFactory
    informerFactory,

    // 设置 pod 的 Initial阶段的 Backoff 的持续时间
internalqueue.WithPodInitialBackoffDuration(time.Duration(options.podInitialBackoffSeconds)*time.Second),

    // 最大backoff持续时间     internalqueue.WithPodMaxBackoffDuration(time.Duration(options.podMaxBackoffSeconds)*time.Second),

    // 设置podLister
    internalqueue.WithPodLister(podLister),

    // 设置 clusterEvent
    internalqueue.WithClusterEventMap(clusterEventMap),

    // 一个pod在 unschedulablePods 停留的最长时间
    internalqueue.WithPodMaxInUnschedulablePodsDuration(options.podMaxInUnschedulablePodsDuration),

    // preEnqueuePluginMap 插件注册
    internalqueue.WithPreEnqueuePluginMap(preEnqueuePluginMap),

    // 指标相关
    internalqueue.WithPluginMetricsSamplePercent(pluginMetricsSamplePercent),
    internalqueue.WithMetricsRecorder(*metricsRecorder),
  )

// pkg/scheduler/internal/queue/scheduling_queue.go#L291-L330
// NewPriorityQueue creates a PriorityQueue object.
func NewPriorityQueue(
  lessFn framework.LessFunc,
  informerFactory informers.SharedInformerFactory,
  opts ...Option,
) *PriorityQueue {
  options := defaultPriorityQueueOptions
  if options.podLister == nil {
    options.podLister = informerFactory.Core().V1().Pods().Lister()
  }
  for _, opt := range opts {
    opt(&options)
  }
​
  // pod 排序
  comp := func(podInfo1, podInfo2 interface{}) bool {
    pInfo1 := podInfo1.(*framework.QueuedPodInfo)
    pInfo2 := podInfo2.(*framework.QueuedPodInfo)
    return lessFn(pInfo1, pInfo2)
  }
​
  // 初始化优先队列
  pq := &PriorityQueue{
    // 存储维护pod与其运行在哪个节点关系
    nominator:                         newPodNominator(options.podLister),
    clock:                             options.clock,
    stop:                              make(chan struct{}),
    podInitialBackoffDuration:         options.podInitialBackoffDuration,
    podMaxBackoffDuration:             options.podMaxBackoffDuration,

    // 一个pod在 unschedulablePods 停留的最长时间
    podMaxInUnschedulablePodsDuration: options.podMaxInUnschedulablePodsDuration,
    activeQ:                           heap.NewWithRecorder(podInfoKeyFunc, comp, metrics.NewActivePodsRecorder()),

    // 包含已尝试并确定为不可调度的pod, 其实是一个map类型，对应 podInfoMap map[string]*framework.QueuedPodInfo
    unschedulablePods:                 newUnschedulablePods(metrics.NewUnschedulablePodsRecorder(), metrics.NewGatedPodsRecorder()),
    moveRequestCycle:                  -1,

    // clusterEvent
    clusterEventMap:                   options.clusterEventMap,
​
    // preEnqueuePlugin
    preEnqueuePluginMap:               options.preEnqueuePluginMap,
    metricsRecorder:                   options.metricsRecorder,
    pluginMetricsSamplePercent:        options.pluginMetricsSamplePercent,
  }
  pq.cond.L = &pq.lock
  pq.podBackoffQ = heap.NewWithRecorder(podInfoKeyFunc, pq.podsCompareBackoffCompleted, metrics.NewBackoffPodsRecorder())
  pq.nsLister = informerFactory.Core().V1().Namespaces().Lister()
​
  return pq
}

// 从 activeQ 队列中以阻塞的方式获取一个Pod，会将增加 scheduling cycle 的值
func (p *PriorityQueue) Pop() (*framework.QueuedPodInfo, error){}
​
// 更新一个队列，分两种情况：
// 不存在：如果 pod 是全新的，则直接添加到 activeQ 队列
// 已存在：如果 pod 在 activeQ 或 backoffQ 队列中已存在，则直接更新；否则，如果pod以可调度的方式更新，它将从不可调度队列 unschedulablePods 中删除，并将更新后的pod添加到 activeQ 中（如修正了原来pod的错误后恢复正常）。
func (p *PriorityQueue) Update(oldPod, newPod *v1.Pod) error {}
​
// 从三个队列中删除 pod
func (p *PriorityQueue) Delete(pod *v1.Pod) error {}
​
// 将pod添加到 activeQ 队列，如果添加的pod已经在activeQ、backoffQ或unschedulablePods 存在，则不会有任何效果
func (p *PriorityQueue) Add(pod *v1.Pod) error {)
​
// 将给定的一组 Pod 从 unschedulablePods 或 backoffQ 队列中移动到 activeQ 队列中，前提是这些 Pod 正在 unschedulablePods 或 backoffQ 队列中
func (p *PriorityQueue) Activate(pods map[string]*v1.Pod) {}

// 根据 profiles 设置PodNominator
  for _, fwk := range profiles {
    fwk.SetPodNominator(podQueue)
  }

// 设置缓存，此时缓存服务自动处于运行状态
  schedulerCache := internalcache.New(durationToExpireAssumedPod, stopEverything)
​
  // Setup cache debugger.
  debugger := cachedebugger.New(nodeLister, podLister, schedulerCache, podQueue)
  debugger.ListenForSignal(stopEverything)

sched := &Scheduler{
    Cache:                    schedulerCache, // 缓存
    client:                   client, // api server 客户端
    nodeInfoSnapshot:         snapshot, // node 快照信息
    percentageOfNodesToScore: options.percentageOfNodesToScore,
    Extenders:                extenders, // 扩展器

    // 获取Pod函数，对应的是优先队列里的 Pop() 方法
    NextPod:                  internalqueue.MakeNextPodFunc(podQueue),
    StopEverything:           stopEverything,
    SchedulingQueue:          podQueue, // 队列
    Profiles:                 profiles, // 配置文件
  }
  sched.applyDefaultHandlers()
​
  addAllEventHandlers(sched, informerFactory, dynInformerFactory, unionedGVKs(clusterEventMap))
​

func (s *Scheduler) applyDefaultHandlers() {
  s.SchedulePod = s.schedulePod
  s.FailureHandler = s.handleSchedulingFailure
}

// cmd/kube-scheduler/app/server.go#L243
// Run executes the scheduler based on the given configuration. It only returns on error or when context is done.
func Run(ctx context.Context, cc *schedulerserverconfig.CompletedConfig, sched *scheduler.Scheduler) error {
  ...
  // 调度器执行
  sched.Run(ctx)
}

// pkg/scheduler/scheduler.go#L337-L391
// Run begins watching and scheduling. It starts scheduling and blocked until the context is done.
func (sched *Scheduler) Run(ctx context.Context) {
  // 1. 优先队列运行
  sched.SchedulingQueue.Run()
​
  // We need to start scheduleOne loop in a dedicated goroutine,
  // because scheduleOne function hangs on getting the next item
  // from the SchedulingQueue.
  // If there are no new pods to schedule, it will be hanging there
  // and if done in this goroutine it will be blocking closing
  // SchedulingQueue, in effect causing a deadlock on shutdown.
  // 2. 在单独的goroutine里以 loop 方式调度Pod
  go wait.UntilWithContext(ctx, sched.scheduleOne, 0)
​
  // 调度服务结束
  <-ctx.Done()
  sched.SchedulingQueue.Close()
}

func (p *PriorityQueue) Run() {
  go wait.Until(p.flushBackoffQCompleted, 1.0*time.Second, p.stop)
  go wait.Until(p.flushUnschedulablePodsLeftover, 30*time.Second, p.stop)
}

func (p *PriorityQueue) flushBackoffQCompleted() {
  p.lock.Lock()
  defer p.lock.Unlock()
  activated := false
  for {
    // 获取队列头部元素，但不从队列里删除
    rawPodInfo := p.podBackoffQ.Peek()
    if rawPodInfo == nil {
      break
    }
    pInfo := rawPodInfo.(*framework.QueuedPodInfo)
    pod := pInfo.Pod
    if p.isPodBackingoff(pInfo) {
      break
    }

    // 删除获取队列头部元素
    _, err := p.podBackoffQ.Pop()
    if err != nil {
      klog.ErrorS(err, "Unable to pop pod from backoff queue despite backoff completion", "pod", klog.KObj(pod))
      break
    }

    // 移入 activeQ 队列
    if added, _ := p.addToActiveQ(pInfo); added {
      activated = true
    }
  }
​
  if activated {
    p.cond.Broadcast()
  }
}

func (p *PriorityQueue) addToActiveQ(pInfo *framework.QueuedPodInfo) (bool, error) {
  // 执行插件 runPreEnqueuePlugins
  pInfo.Gated = !p.runPreEnqueuePlugins(context.Background(), pInfo)
  if pInfo.Gated {
    // Add the Pod to unschedulablePods if it's not passing PreEnqueuePlugins.
    p.unschedulablePods.addOrUpdate(pInfo)
    return false, nil
  }

  // 加入队列，如果已在队列则直接更新（什么情况下会在activeQ队列存在？）
  if err := p.activeQ.Add(pInfo); err != nil {
    klog.ErrorS(err, "Error adding pod to the active queue", "pod", klog.KObj(pInfo.Pod))
    return false, err
  }
  return true, nil
}

func (p *PriorityQueue) flushUnschedulablePodsLeftover() {
  var podsToMove []*framework.QueuedPodInfo
  currentTime := p.clock.Now()
  for _, pInfo := range p.unschedulablePods.podInfoMap {
    lastScheduleTime := pInfo.Timestamp
    if currentTime.Sub(lastScheduleTime) > p.podMaxInUnschedulablePodsDuration {
      podsToMove = append(podsToMove, pInfo)
    }
  }
​
  // 移入activeQ 或 BackoffQueue
  if len(podsToMove) > 0 {
    p.movePodsToActiveOrBackoffQueue(podsToMove, UnschedulableTimeout)
  }
}
​
// 将 Pod 移动到 activeQ 或 backoffQ
func (p *PriorityQueue) movePodsToActiveOrBackoffQueue(podInfoList []*framework.QueuedPodInfo, event framework.ClusterEvent) {
  activated := false
  for _, pInfo := range podInfoList {
    if len(pInfo.UnschedulablePlugins) != 0 && !p.podMatchesEvent(pInfo, event) {
      continue
    }
    pod := pInfo.Pod
    if p.isPodBackingoff(pInfo) {
      if err := p.podBackoffQ.Add(pInfo); err != nil {
        klog.ErrorS(err, "Error adding pod to the backoff queue", "pod", klog.KObj(pod))
      } else {
        klog.V(5).InfoS("Pod moved to an internal scheduling queue", "pod", klog.KObj(pInfo.Pod), "event", event, "queue", backoffQName)
        metrics.SchedulerQueueIncomingPods.WithLabelValues("backoff", event.Label).Inc()

        // 从 unschedulablePods 里删除Pod
        p.unschedulablePods.delete(pod, pInfo.Gated)
      }
    } else {
      gated := pInfo.Gated

      // 添加到 activeQ 队列，并从 unschedulablePods 中将其删除
      if added, _ := p.addToActiveQ(pInfo); added {
        klog.V(5).InfoS("Pod moved to an internal scheduling queue", "pod", klog.KObj(pInfo.Pod), "event", event, "queue", activeQName)
        activated = true
        metrics.SchedulerQueueIncomingPods.WithLabelValues("active", event.Label).Inc()
        p.unschedulablePods.delete(pod, gated)
      }
    }
  }
  p.moveRequestCycle = p.schedulingCycle
  if activated {
    p.cond.Broadcast()
  }
}

func (sched *Scheduler) scheduleOne(ctx context.Context) {
  // 1. 获取一个 Pod, 如果为 nil 则直接返回取消
  podInfo := sched.NextPod()
}

// MakeNextPodFunc returns a function to retrieve the next pod from a given
// scheduling queue
func MakeNextPodFunc(queue SchedulingQueue) func() *framework.QueuedPodInfo {
  return func() *framework.QueuedPodInfo {
    podInfo, err := queue.Pop()
    if err == nil {
      klog.V(4).InfoS("About to try and schedule pod", "pod", klog.KObj(podInfo.Pod))
      for plugin := range podInfo.UnschedulablePlugins {
        metrics.UnschedulableReason(plugin, podInfo.Pod.Spec.SchedulerName).Dec()
      }
      return podInfo
    }
    klog.ErrorS(err, "Error while retrieving next pod from scheduling queue")
    return nil
  }
}

// Pop removes the head of the active queue and returns it. It blocks if the
// activeQ is empty and waits until a new item is added to the queue. It
// increments scheduling cycle when a pod is popped.
func (p *PriorityQueue) Pop() (*framework.QueuedPodInfo, error) {
  p.lock.Lock()
  defer p.lock.Unlock()

  // 阻塞
  for p.activeQ.Len() == 0 {
    // When the queue is empty, invocation of Pop() is blocked until new item is enqueued.
    // When Close() is called, the p.closed is set and the condition is broadcast,
    // which causes this loop to continue and return from the Pop().
    if p.closed {
      return nil, fmt.Errorf(queueClosed)
    }
    p.cond.Wait()
  }

  // 弹出Pod
  obj, err := p.activeQ.Pop()
  if err != nil {
    return nil, err
  }

  // 递增
  pInfo := obj.(*framework.QueuedPodInfo)
  pInfo.Attempts++
  p.schedulingCycle++
  return pInfo, nil
}

func (sched *Scheduler) scheduleOne(ctx context.Context) {
  // 2. 根据当前Pod调度器 pod.schedulerName 获取其对应的 Framework
  fwk, err := sched.frameworkForPod(pod)

  // 检查是否需要跳过调度
  if sched.skipPodSchedule(fwk, pod) {
    return
  }
}

// 3. 获取一个最合适的节点
scheduleResult, assumedPodInfo, status := sched.schedulingCycle(schedulingCycleCtx, state, fwk, podInfo, start, podsToActivate)

// schedulingCycle tries to schedule a single Pod.
func (sched *Scheduler) schedulingCycle(
  ctx context.Context,
  state *framework.CycleState,
  fwk framework.Framework,
  podInfo *framework.QueuedPodInfo,
  start time.Time,
  podsToActivate *framework.PodsToActivate,
) (ScheduleResult, *framework.QueuedPodInfo, *framework.Status) {
  pod := podInfo.Pod

  // 一、 获取一个合适的节点，这块下面有详细介绍
  scheduleResult, err := sched.SchedulePod(ctx, fwk, state, pod)
  if err != nil {
    ...
    return err
  }

// schedulePod tries to schedule the given pod to one of the nodes in the node list.
func (sched *Scheduler) schedulePod(ctx context.Context, fwk framework.Framework, state *framework.CycleState, pod *v1.Pod) (result ScheduleResult, err error) {
​
  // 1. 更新节点快照，为什么更新？为什么用？每调度一个Pod就会更新Node快照，性能是否存在问题
  if err := sched.Cache.UpdateSnapshot(sched.nodeInfoSnapshot); err != nil {
    return result, err
  }
}

// schedulePod tries to schedule the given pod to one of the nodes in the node list.
func (sched *Scheduler) schedulePod(ctx context.Context, fwk framework.Framework, state *framework.CycleState, pod *v1.Pod) (result ScheduleResult, err error) {
  // 2. 检查节点快照有无节点
  if sched.nodeInfoSnapshot.NumNodes() == 0 {
    return result, ErrNoNodesAvailable
  }
}

// 3. 查找合适的节点
feasibleNodes, diagnosis, err := sched.findNodesThatFitPod(ctx, fwk, state, pod)

// 4. 正好只有一个节点，直接使用
  // When only one node after predicate, just use it.
  if len(feasibleNodes) == 1 {
    return ScheduleResult{
      SuggestedHost:  feasibleNodes[0].Name,
      EvaluatedNodes: 1 + len(diagnosis.NodeToStatusMap),
      FeasibleNodes:  1,
    }, nil
  }

// 5. 从多个节点中选择最合适的节点
priorityList, err := prioritizeNodes(ctx, sched.Extenders, fwk, state, pod, feasibleNodes)

// 6. 从丛多节点中遍历出来评分最高的那个，每个节点都会调用所有plugins对其进行评分
host, err := selectHost(priorityList)

return ScheduleResult{
    SuggestedHost:  host,  // 选择的节点名称（这里是指主机名还是系统内部分配的一个标识？）
    EvaluatedNodes: len(feasibleNodes) + len(diagnosis.NodeToStatusMap),
    FeasibleNodes:  len(feasibleNodes),
  }, err

// ScheduleResult represents the result of scheduling a pod.
type ScheduleResult struct {
  // 节点名称
  SuggestedHost string

  //调度程序在筛选阶段及以后评估pod的节点数
  EvaluatedNodes int

  // 评估后，适合pod的节点数量
  FeasibleNodes int

  // 调度周期相关
  nominatingInfo *framework.NominatingInfo
}

func (sched *Scheduler) schedulingCycle(...) {
  assumedPodInfo := podInfo.DeepCopy()
  assumedPod := assumedPodInfo.Pod

  // 二、修改 Pod.Spec.NodeName = NodeName，建立 Pod 与 Node 的映射关系
  err = sched.assume(assumedPod, scheduleResult.SuggestedHost)
  ...
}

func (sched *Scheduler) assume(assumed *v1.Pod, host string) error {
  // pod 所在Node
  assumed.Spec.NodeName = host

  // 缓存更新
  if err := sched.Cache.AssumePod(assumed); err != nil {
    klog.ErrorS(err, "Scheduler cache AssumePod failed")
    return err
  }
  // 从内部缓存中删除
  if sched.SchedulingQueue != nil {
    sched.SchedulingQueue.DeleteNominatedPodIfExists(assumed)
  }

  return nil
}

func (sched *Scheduler) schedulingCycle(...) {
  // 三、执行一些插件
  // Run the Reserve method of reserve plugins.
  if sts := fwk.RunReservePluginsReserve(ctx, state, assumedPod, scheduleResult.SuggestedHost); !sts.IsSuccess() {}


  // Run "permit" plugins.
  runPermitStatus := fwk.RunPermitPlugins(ctx, state, assumedPod, scheduleResult.SuggestedHost)
  if !runPermitStatus.IsWait() && !runPermitStatus.IsSuccess() {
  }

}

func (sched *Scheduler) schedulingCycle(...) {
  // 四、podsToActivate.Map值被插件修改，什么情况下会插件会修改？

  // At the end of a successful scheduling cycle, pop and move up Pods if needed.
  if len(podsToActivate.Map) != 0 {
    // 将pod 从 backoffQ 或unschedulablePods 中移到 activeQ
    sched.SchedulingQueue.Activate(podsToActivate.Map)
    // Clear the entries after activation.
    podsToActivate.Map = make(map[string]*v1.Pod)
  }
​
  return scheduleResult, assumedPodInfo, nil
}

// 异步绑定 Pod 与 node
  go func() {
    bindingCycleCtx, cancel := context.WithCancel(ctx)
    defer cancel()

    // 绑定
    status := sched.bindingCycle(bindingCycleCtx, state, fwk, scheduleResult, assumedPodInfo, start, podsToActivate)
    if !status.IsSuccess() {
      sched.handleBindingCycleError(bindingCycleCtx, state, fwk, assumedPodInfo, start, scheduleResult, status)
    }
  }()

func (sched *Scheduler) bindingCycle( ctx context.Context,
  state *framework.CycleState,
  fwk framework.Framework,
  scheduleResult ScheduleResult,
  assumedPodInfo *framework.QueuedPodInfo,
  start time.Time,
  podsToActivate *framework.PodsToActivate) *framework.Status {

  assumedPod := assumedPodInfo.Pod

  // 执行 "permit" 插件.
  if status := fwk.WaitOnPermit(ctx, assumedPod); !status.IsSuccess() {
    return status
  }

  // Run "prebind" plugins.
  if status := fwk.RunPreBindPlugins(ctx, state, assumedPod, scheduleResult.SuggestedHost); !status.IsSuccess() {
    return status
  }

  // Run "bind" plugins. 真正绑定的地方
  if status := sched.bind(ctx, fwk, assumedPod, scheduleResult.SuggestedHost, state); !status.IsSuccess() {
    return status
  }

  // Run "postbind" plugins.
  fwk.RunPostBindPlugins(ctx, state, assumedPod, scheduleResult.SuggestedHost)

  // At the end of a successful binding cycle, move up Pods if needed.
  if len(podsToActivate.Map) != 0 {
    sched.SchedulingQueue.Activate(podsToActivate.Map)
    // Unlike the logic in schedulingCycle(), we don't bother deleting the entries
    // as `podsToActivate.Map` is no longer consumed.
  }

  return nil
}

// RunBindPlugins runs the set of configured bind plugins until one returns a non `Skip` status.
func (f *frameworkImpl) RunBindPlugins(ctx context.Context, state *framework.CycleState, pod *v1.Pod, nodeName string) (status *framework.Status) {
	startTime := time.Now()
	defer func() {
		metrics.FrameworkExtensionPointDuration.WithLabelValues(metrics.Bind, status.Code().String(), f.profileName).Observe(metrics.SinceInSeconds(startTime))
	}()
  
  // bind插件不存在
	if len(f.bindPlugins) == 0 {
		return framework.NewStatus(framework.Skip, "")
	}
  
	logger := klog.FromContext(ctx)
	logger = klog.LoggerWithName(logger, "Bind")
	// TODO(knelasevero): Remove duplicated keys from log entry calls
	// When contextualized logging hits GA
	// https://github.com/kubernetes/kubernetes/issues/111672
	logger = klog.LoggerWithValues(logger, "pod", klog.KObj(pod), "node", klog.ObjectRef{Name: nodeName})
  
  // 执行注册的所有 bind 插件 
	for _, pl := range f.bindPlugins {
		logger := klog.LoggerWithName(logger, pl.Name())
		ctx := klog.NewContext(ctx, logger)
    
    // 执行绑定插件 
		status = f.runBindPlugin(ctx, pl, state, pod, nodeName)
		if status.IsSkip() {
			continue
		}
		if !status.IsSuccess() {
			if status.IsUnschedulable() {
				logger.V(4).Info("Pod rejected by Bind plugin", "pod", klog.KObj(pod), "node", nodeName, "plugin", pl.Name(), "status", status.Message())
				status.SetFailedPlugin(pl.Name())
				return status
			}
			err := status.AsError()
			logger.Error(err, "Plugin Failed", "plugin", pl.Name(), "pod", klog.KObj(pod), "node", nodeName)
			return framework.AsStatus(fmt.Errorf("running Bind plugin %q: %w", pl.Name(), err))
		}
		return status
	}
	return status
}

// 执行绑定插件
func (f *frameworkImpl) runBindPlugin(ctx context.Context, bp framework.BindPlugin, state *framework.CycleState, pod *v1.Pod, nodeName string) *framework.Status {
	if !state.ShouldRecordPluginMetrics() {
    // 执行插件的 Bind() 方法
		return bp.Bind(ctx, state, pod, nodeName)
	}
	startTime := time.Now()
    // 执行插件的 Bind() 方法
	status := bp.Bind(ctx, state, pod, nodeName)
  
	f.metricsRecorder.ObservePluginDurationAsync(metrics.Bind, bp.Name(), status.Code().String(), metrics.SinceInSeconds(startTime))
	return status
}