kubernetes 调度组件 kube-scheduler 1.16.3 源代码阅读指引

// Flags returns flags for a specific scheduler by section name
func (o *Options) Flags() (nfs cliflag.NamedFlagSets) {
    fs := nfs.FlagSet("misc")
    fs.StringVar(&o.ConfigFile, "config", o.ConfigFile, "The path to the configuration file. Flags override values in this file.")
    fs.StringVar(&o.WriteConfigTo, "write-config-to", o.WriteConfigTo, "If set, write the configuration values to this file and exit.")
    fs.StringVar(&o.Master, "master", o.Master, "The address of the Kubernetes API server (overrides any value in kubeconfig)")

    o.SecureServing.AddFlags(nfs.FlagSet("secure serving"))
    o.CombinedInsecureServing.AddFlags(nfs.FlagSet("insecure serving"))
    o.Authentication.AddFlags(nfs.FlagSet("authentication"))
    o.Authorization.AddFlags(nfs.FlagSet("authorization"))
    o.Deprecated.AddFlags(nfs.FlagSet("deprecated"), &o.ComponentConfig)

    leaderelectionconfig.BindFlags(&o.ComponentConfig.LeaderElection.LeaderElectionConfiguration, nfs.FlagSet("leader election"))
    utilfeature.DefaultMutableFeatureGate.AddFlag(nfs.FlagSet("feature gate"))

    return nfs
}

// NewOptions returns default scheduler app options.
func NewOptions() (*Options, error) {
    cfg, err := newDefaultComponentConfig()    //运行时配置的主要组成
    if err != nil {
        return nil, err
    }
    ...
    o := &Options{
    ComponentConfig: *cfg,
    ...

func newDefaultComponentConfig() (*kubeschedulerconfig.KubeSchedulerConfiguration, error) {
    cfgv1alpha1 := kubeschedulerconfigv1alpha1.KubeSchedulerConfiguration{}
    kubeschedulerscheme.Scheme.Default(&cfgv1alpha1)     //这里是关键
    cfg := kubeschedulerconfig.KubeSchedulerConfiguration{}
    if err := kubeschedulerscheme.Scheme.Convert(&cfgv1alpha1, &cfg, nil); err != nil {
        return nil, err
    }
    return &cfg, nil
}

kubeschedulerconfigv1alpha1 "k8s.io/kube-scheduler/config/v1alpha1"
kubeschedulerscheme "k8s.io/kubernetes/pkg/scheduler/apis/config/scheme"

// Default sets defaults on the provided Object.
func (s *Scheme) Default(src Object) {
    if fn, ok := s.defaulterFuncs[reflect.TypeOf(src)]; ok {
        fn(src)
    }
}

func SetDefaults_KubeSchedulerConfiguration(obj *kubescedulerconfigv1alpha1.KubeSchedulerConfiguration) {
    if len(obj.SchedulerName) == 0 {
        obj.SchedulerName = api.DefaultSchedulerName
    }

    if obj.HardPodAffinitySymmetricWeight == 0 {
        obj.HardPodAffinitySymmetricWeight = api.DefaultHardPodAffinitySymmetricWeight
    }

    if obj.AlgorithmSource.Policy == nil &&
        (obj.AlgorithmSource.Provider == nil || len(*obj.AlgorithmSource.Provider) == 0) {
        val := kubescedulerconfigv1alpha1.SchedulerDefaultProviderName
        obj.AlgorithmSource.Provider = &val
    }
    ...
}

// New returns a Scheduler
func New(client clientset.Interface,
    ...
    config = sc
    ...
    sched := NewFromConfig(config)
    ...
    AddAllEventHandlers(sched, options.schedulerName, nodeInformer, podInformer, pvInformer, pvcInformer, serviceInformer, storageClassInformer)
    ...
}

// Run begins watching and scheduling. It waits for cache to be synced, then starts a goroutine and returns immediately.
func (sched *Scheduler) Run() {
    if !sched.config.WaitForCacheSync() {
        return
    }

    go wait.Until(sched.scheduleOne, 0, sched.config.StopEverything)
}

// pkg/scheduler/core/generic_scheduler.go
// NewGenericScheduler creates a genericScheduler object.
func NewGenericScheduler(
    ...
) ScheduleAlgorithm {
    ...
}

// pkg/scheduler/scheduler.go
func (sched *Scheduler) scheduleOne() {
    ...
    scheduleResult, err := sched.schedule(pod, pluginContext)
    ...
}

func (sched *Scheduler) schedule(pod *v1.Pod, pluginContext *framework.PluginContext) (core.ScheduleResult, error) {
    result, err := sched.config.Algorithm.Schedule(pod, sched.config.NodeLister, pluginContext)
    if err != nil {
        pod = pod.DeepCopy()
        sched.recordSchedulingFailure(pod, err, v1.PodReasonUnschedulable, err.Error())
        return core.ScheduleResult{}, err
    }
    return result, err
}

// pkg/scheduler/core/generic_scheduler.go
func (g *genericScheduler) Schedule(pod *v1.Pod, nodeLister algorithm.NodeLister, pluginContext *framework.PluginContext) (result ScheduleResult, err error) {
    ...
}

// pkg/scheduler/scheduler.go
func (sched *Scheduler) Run() {
    ...
    //每次只调度一个 pod 的 sched.scheduleOne，无限重复执行
    go wait.Until(sched.scheduleOne, 0, sched.config.StopEverything)
}

// pkg/scheduler/scheduler.go
func (sched *Scheduler) scheduleOne() {
    fwk := sched.config.Framework
    ...
    // Run "reserve" plugins.
    if sts := fwk.RunReservePlugins(pluginContext, assumedPod, scheduleResult.SuggestedHost); !sts.IsSuccess() {
        sched.recordSchedulingFailure(assumedPod, sts.AsError(), SchedulerError, sts.Message())
        metrics.PodScheduleErrors.Inc()
        return
    }
    ...
        // Run "permit" plugins.
        permitStatus := fwk.RunPermitPlugins(pluginContext, assumedPod, scheduleResult.SuggestedHost)
        ...
        // Run "prebind" plugins.
        prebindStatus := fwk.RunPrebindPlugins(pluginContext, assumedPod, scheduleResult.SuggestedHost)

            // Run "postbind" plugins.
            fwk.RunPostbindPlugins(pluginContext, assumedPod, scheduleResult.SuggestedHost)
    ...
}

type Framework interface {
    FrameworkHandle
    QueueSortFunc() LessFunc
    RunPrefilterPlugins(pc *PluginContext, pod *v1.Pod) *Status
    RunPrebindPlugins(pc *PluginContext, pod *v1.Pod, nodeName string) *Status
    RunPostbindPlugins(pc *PluginContext, pod *v1.Pod, nodeName string)
    RunReservePlugins(pc *PluginContext, pod *v1.Pod, nodeName string) *Status
    RunUnreservePlugins(pc *PluginContext, pod *v1.Pod, nodeName string)
    RunPermitPlugins(pc *PluginContext, pod *v1.Pod, nodeName string) *Status
}

func NewRegistry() Registry {
    return Registry{
        // FactoryMap:
        // New plugins are registered here.
        // example:
        // {
        //  stateful_plugin.Name: stateful.NewStatefulMultipointExample,
        //  fooplugin.Name: fooplugin.New,
        // }
    }
}