深入kube-scheduler：Kubernetes调度器源码剖析

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63

// Scheduler 调度器主结构体
// 文件路径: pkg/scheduler/scheduler.go
type Scheduler struct {
    // Cache 调度缓存，存储集群状态信息
    Cache internalcache.Cache
    
    // Extenders 调度器扩展器列表
    Extenders []framework.Extender
    
    // NextPod 获取下一个待调度Pod的函数
    NextPod func() *framework.QueuedPodInfo
    
    // Error 错误处理函数
    Error func(*framework.QueuedPodInfo, error)
    
    // StopEverything 停止所有操作的通道
    StopEverything <-chan struct{}
    
    // SchedulingQueue 调度队列
    SchedulingQueue internalqueue.SchedulingQueue
    
    // Profiles 调度配置文件映射
    // key: 调度器名称，value: 对应的Framework实例
    Profiles map[string]framework.Framework
    
    // client Kubernetes客户端
    client clientset.Interface
    
    // nodeInfoSnapshot 节点信息快照
    nodeInfoSnapshot *internalcache.Snapshot
    
    // percentageOfNodesToScore 需要打分的节点百分比
    percentageOfNodesToScore int32
    
    // nextStartNodeIndex 下次开始调度的节点索引
    nextStartNodeIndex int
}

// QueuedPodInfo 队列中的Pod信息
type QueuedPodInfo struct {
    // PodInfo Pod基础信息
    *framework.PodInfo
    
    // Timestamp Pod入队时间
    Timestamp time.Time
    
    // Attempts 调度尝试次数
    Attempts int
    
    // InitialAttemptTimestamp 首次调度尝试时间
    InitialAttemptTimestamp time.Time
}

// PodInfo Pod信息结构体
type PodInfo struct {
    Pod *v1.Pod
    
    // RequiredAffinity 必须满足的亲和性条件
    RequiredAffinity *nodeaffinity.RequiredNodeAffinity
    
    // RequiredAntiAffinity 必须满足的反亲和性条件
    RequiredAntiAffinity *podaffinity.RequiredPodAntiAffinity
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100

// Framework 调度框架接口
// 文件路径: pkg/scheduler/framework/interface.go
type Framework interface {
    // QueueSortFunc 返回队列排序函数
    QueueSortFunc() LessFunc
    
    // RunPreFilterPlugins 运行预过滤插件
    RunPreFilterPlugins(ctx context.Context, state *CycleState, pod *v1.Pod) (status *Status)
    
    // RunFilterPlugins 运行过滤插件
    RunFilterPlugins(ctx context.Context, state *CycleState, pod *v1.Pod, nodeInfo *NodeInfo) PluginToStatus
    
    // RunPostFilterPlugins 运行后过滤插件
    RunPostFilterPlugins(ctx context.Context, state *CycleState, pod *v1.Pod, filteredNodeStatusMap NodeToStatusMap) (*PostFilterResult, *Status)
    
    // RunPreScorePlugins 运行预打分插件
    RunPreScorePlugins(ctx context.Context, state *CycleState, pod *v1.Pod, nodes []*v1.Node) *Status
    
    // RunScorePlugins 运行打分插件
    RunScorePlugins(ctx context.Context, state *CycleState, pod *v1.Pod, nodes []*v1.Node) (PluginToNodeScores, *Status)
    
    // RunNormalizeScorePlugins 运行分数归一化插件
    RunNormalizeScorePlugins(ctx context.Context, state *CycleState, pod *v1.Pod, scores NodeScoreList) *Status
    
    // RunReservePlugins 运行资源预留插件
    RunReservePlugins(ctx context.Context, state *CycleState, pod *v1.Pod, nodeName string) *Status
    
    // RunUnreservePlugins 运行资源释放插件
    RunUnreservePlugins(ctx context.Context, state *CycleState, pod *v1.Pod, nodeName string)
    
    // RunPermitPlugins 运行许可插件
    RunPermitPlugins(ctx context.Context, state *CycleState, pod *v1.Pod, nodeName string) *Status
    
    // RunPreBindPlugins 运行绑定前插件
    RunPreBindPlugins(ctx context.Context, state *CycleState, pod *v1.Pod, nodeName string) *Status
    
    // RunBindPlugins 运行绑定插件
    RunBindPlugins(ctx context.Context, state *CycleState, pod *v1.Pod, nodeName string) *Status
    
    // RunPostBindPlugins 运行绑定后插件
    RunPostBindPlugins(ctx context.Context, state *CycleState, pod *v1.Pod, nodeName string)
    
    // HasFilterPlugins 检查是否有过滤插件
    HasFilterPlugins() bool
    
    // HasPostFilterPlugins 检查是否有后过滤插件
    HasPostFilterPlugins() bool
    
    // HasScorePlugins 检查是否有打分插件
    HasScorePlugins() bool
    
    // ListPlugins 列出所有插件
    ListPlugins() *config.Plugins
    
    // ProfileName 返回配置文件名称
    ProfileName() string
    
    // PercentageOfNodesToScore 返回需要打分的节点百分比
    PercentageOfNodesToScore() *int32
    
    // SetPodNominator 设置Pod提名器
    SetPodNominator(nominator PodNominator)
}

// frameworkImpl 调度框架实现
type frameworkImpl struct {
    registry              Registry
    snapshotSharedLister  SharedLister
    waitingPods          *waitingPodsMap
    scorePluginWeight    map[string]int
    queueSortPlugins     []QueueSortPlugin
    preFilterPlugins     []PreFilterPlugin
    filterPlugins        []FilterPlugin
    postFilterPlugins    []PostFilterPlugin
    preScorePlugins      []PreScorePlugin
    scorePlugins         []ScorePlugin
    reservePlugins       []ReservePlugin
    unreservePlugins     []UnreservePlugin
    permitPlugins        []PermitPlugin
    preBindPlugins       []PreBindPlugin
    bindPlugins          []BindPlugin
    postBindPlugins      []PostBindPlugin
    
    clientSet            clientset.Interface
    kubeConfig           *restclient.Config
    eventRecorder        events.EventRecorder
    informerFactory      informers.SharedInformerFactory
    
    metricsRecorder      *MetricsRecorder
    profileName          string
    percentageOfNodesToScore *int32
    podNominator         PodNominator
    
    extenders            []Extender
    
    parallelizer         parallelize.Parallelizer
    
    // 插件运行时状态
    pluginsNeeded        config.PluginSet
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

// main kube-scheduler主函数入口
// 文件路径: cmd/kube-scheduler/scheduler.go
func main() {
    // 创建kube-scheduler命令对象
    command := app.NewSchedulerCommand()
    
    // 使用component-base的CLI运行器执行命令
    code := cli.Run(command)
    
    // 以返回的退出码结束程序
    os.Exit(code)
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90

// NewSchedulerCommand 创建调度器命令
// 文件路径: cmd/kube-scheduler/app/server.go
func NewSchedulerCommand(registryOptions ...Option) *cobra.Command {
    // 创建默认选项
    opts := options.NewOptions()

    cmd := &cobra.Command{
        Use: "kube-scheduler",
        Long: `The Kubernetes scheduler is a control plane process which assigns
Pods to Nodes. The scheduler determines which Nodes are valid placements for
each Pod in the scheduling queue according to constraints and available
resources. The scheduler then ranks each valid Node and binds the Pod to a
suitable Node. Multiple different schedulers may be used within a cluster;
kube-scheduler is the reference implementation.
See [scheduling](https://kubernetes.io/docs/concepts/scheduling-eviction/)
for more information about scheduling and the kube-scheduler component.`,

        // 禁用错误时的使用说明
        SilenceUsage: true,
        
        // 预运行钩子
        PersistentPreRunE: func(*cobra.Command, []string) error {
            // 设置默认的klog标志
            cliflag.PrintFlags(cmd.Flags())
            return nil
        },
        
        // 主运行函数
        RunE: func(cmd *cobra.Command, args []string) error {
            return runCommand(cmd, opts, registryOptions...)
        },
        
        Args: func(cmd *cobra.Command, args []string) error {
            for _, arg := range args {
                if len(arg) > 0 {
                    return fmt.Errorf("%q does not take any arguments, got %q", cmd.CommandPath(), args)
                }
            }
            return nil
        },
    }

    // 添加标志
    nfs := opts.Flags()
    verflag.AddFlags(nfs.FlagSet("global"))
    globalflag.AddGlobalFlags(nfs.FlagSet("global"), cmd.Name(), logs.SkipLoggingConfigurationFlags())
    
    fs := cmd.Flags()
    for _, f := range nfs.FlagSets {
        fs.AddFlagSet(f)
    }

    cols, _, _ := term.TerminalSize(cmd.OutOrStdout())
    cliflag.SetUsageAndHelpFunc(cmd, nfs, cols)

    return cmd
}

// runCommand 运行调度器命令
func runCommand(cmd *cobra.Command, opts *options.Options, registryOptions ...Option) error {
    verflag.PrintAndExitIfRequested()

    // 验证选项
    if errs := opts.Validate(); len(errs) > 0 {
        return utilerrors.NewAggregate(errs)
    }

    // 获取配置
    c, err := opts.Config()
    if err != nil {
        return err
    }

    // 完成配置
    cc := c.Complete()

    // 应用插件注册选项
    for _, option := range registryOptions {
        if err := option(cc.Registry); err != nil {
            return err
        }
    }

    // 获取调度器上下文
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()
    
    // 运行调度器
    return Run(ctx, cc)
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107

// Run 运行调度器的主函数
// 文件路径: cmd/kube-scheduler/app/server.go
func Run(ctx context.Context, cc *schedulerserverconfig.CompletedConfig) error {
    // 记录调度器启动
    klog.InfoS("Starting Kubernetes Scheduler", "version", version.Get())

    // 创建调度器
    sched, err := scheduler.New(cc.Client,
        cc.InformerFactory,
        cc.DynInformerFactory,
        recorderFactory,
        ctx.Done(),
        scheduler.WithProfiles(cc.ComponentConfig.Profiles...),
        scheduler.WithAlgorithmSource(cc.ComponentConfig.AlgorithmSource),
        scheduler.WithPercentageOfNodesToScore(cc.ComponentConfig.PercentageOfNodesToScore),
        scheduler.WithFrameworkOutOfTreeRegistry(cc.Registry),
        scheduler.WithPodMaxBackoffSeconds(cc.ComponentConfig.PodMaxBackoffSeconds),
        scheduler.WithPodInitialBackoffSeconds(cc.ComponentConfig.PodInitialBackoffSeconds),
        scheduler.WithPodMaxInUnschedulablePodsDuration(cc.PodMaxInUnschedulablePodsDuration),
        scheduler.WithExtenders(cc.ComponentConfig.Extenders...),
        scheduler.WithParallelism(cc.ComponentConfig.Parallelism),
        scheduler.WithBuildFrameworkCapturer(func(profile kubeschedulerconfig.KubeSchedulerProfile) {
            // 捕获框架构建过程，用于调试
        }),
    )
    if err != nil {
        return err
    }

    // 准备事件广播器
    eventBroadcaster := events.NewBroadcaster(&events.EventSinkImpl{Interface: cc.Client.EventsV1()})
    sched.EventRecorder = eventBroadcaster.NewRecorder(scheme.Scheme, "kube-scheduler")
    eventBroadcaster.StartRecordingToSink(ctx.Done())

    // 设置健康检查
    if cc.SecureServing != nil {
        handler := buildHandlerChain(newHealthzAndMetricsHandler(&cc.ComponentConfig, cc.InformerFactory, sched), cc.Authentication.Authenticator, cc.Authorization.Authorizer)
        // 启动安全服务器
        if _, _, err := cc.SecureServing.Serve(handler, 0, ctx.Done()); err != nil {
            return fmt.Errorf("failed to start secure server: %v", err)
        }
    }

    // 启动所有informers
    cc.InformerFactory.Start(ctx.Done())
    cc.DynInformerFactory.Start(ctx.Done())

    // 等待缓存同步
    cc.InformerFactory.WaitForCacheSync(ctx.Done())
    cc.DynInformerFactory.WaitForCacheSync(ctx.Done())

    // 启动调度器
    sched.Run(ctx)
    return nil
}

// New 创建新的调度器实例
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)
    }

    // 创建调度缓存
    cache := internalcache.New(30*time.Second, stopCh)

    // 创建调度队列
    podQueue := internalqueue.NewSchedulingQueue(
        profiles[options.profiles[0].SchedulerName].QueueSortFunc(),
        informerFactory,
        internalqueue.WithPodInitialBackoffDuration(time.Duration(options.podInitialBackoffSeconds)*time.Second),
        internalqueue.WithPodMaxBackoffDuration(time.Duration(options.podMaxBackoffSeconds)*time.Second),
        internalqueue.WithPodLister(podLister),
        internalqueue.WithPodMaxInUnschedulablePodsDuration(options.podMaxInUnschedulablePodsDuration),
        internalqueue.WithPreEnqueuePluginMap(preEnqueuePluginMap),
        internalqueue.WithQueueingHintMapPerProfile(queueingHintsPerProfile),
        internalqueue.WithPluginMetricsSamplePercent(pluginMetricsSamplePercent),
        internalqueue.WithMetricsRecorder(*metricsRecorder),
    )

    // 创建调度器实例
    sched := &Scheduler{
        Cache:                    cache,
        client:                   client,
        nodeInfoSnapshot:         snapshot,
        percentageOfNodesToScore: options.percentageOfNodesToScore,
        Extenders:                extenders,
        StopEverything:           stopCh,
        SchedulingQueue:          podQueue,
        Profiles:                 profiles,
    }

    // 设置调度队列的调度器引用
    sched.SchedulingQueue.SetScheduler(sched)

    // 添加事件处理器
    addAllEventHandlers(sched, informerFactory, dynInformerFactory, unionedGVKs(queueingHintsPerProfile))

    return sched, nil
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135

// Run 运行调度器主循环
// 文件路径: pkg/scheduler/scheduler.go
func (sched *Scheduler) Run(ctx context.Context) {
    klog.InfoS("Starting Kubernetes Scheduler")
    
    // 启动调度队列
    sched.SchedulingQueue.Run()
    
    // 等待缓存同步
    if !cache.WaitForCacheSync(ctx.Done(), sched.Cache.HasSynced) {
        klog.ErrorS(nil, "Cannot sync cache")
        return
    }
    
    // 启动调度循环
    wait.UntilWithContext(ctx, sched.scheduleOne, 0)
    
    // 关闭调度队列
    sched.SchedulingQueue.Close()
    klog.InfoS("Shutting down Kubernetes Scheduler")
}

// scheduleOne 调度单个Pod的核心函数
func (sched *Scheduler) scheduleOne(ctx context.Context) {
    // 从调度队列获取下一个Pod
    podInfo := sched.NextPod()
    if podInfo == nil || podInfo.Pod == nil {
        return
    }
    pod := podInfo.Pod
    
    // 记录调度开始时间
    start := time.Now()
    defer func() {
        metrics.SchedulingAlgorithmLatency.Observe(metrics.SinceInSeconds(start))
    }()
    
    // 获取调度框架
    fwk, ok := sched.Profiles[pod.Spec.SchedulerName]
    if !ok {
        // 如果找不到对应的调度器，使用默认调度器
        klog.V(2).InfoS("Unable to find profile for scheduler name, using default profile", "pod", klog.KObj(pod), "schedulerName", pod.Spec.SchedulerName)
        fwk = sched.Profiles[v1.DefaultSchedulerName]
    }
    
    // 创建调度上下文
    state := framework.NewCycleState()
    state.SetRecordPluginMetrics(rand.Intn(100) < pluginMetricsSamplePercent)
    
    // 执行调度算法
    scheduleResult, err := sched.Algorithm.Schedule(ctx, fwk, state, pod)
    if err != nil {
        // 调度失败处理
        nominatedNode := ""
        if fitError, ok := err.(*framework.FitError); ok {
            if !fwk.HasPostFilterPlugins() {
                klog.V(3).InfoS("No PostFilter plugins are registered, so no preemption will be performed")
            } else {
                // 尝试抢占
                result, status := fwk.RunPostFilterPlugins(ctx, state, pod, fitError.Diagnosis.NodeToStatusMap)
                if status.Code() == framework.Error {
                    klog.ErrorS(nil, "Status after running PostFilter plugins for pod", "pod", klog.KObj(pod), "status", status)
                } else {
                    klog.V(5).InfoS("Status after running PostFilter plugins for pod", "pod", klog.KObj(pod), "status", status)
                    if result != nil {
                        nominatedNode = result.NominatedNodeName
                    }
                }
            }
            
            // 记录调度失败指标
            metrics.PodUnschedulable(fwk.ProfileName(), metrics.SinceInSeconds(start))
        } else if err == ErrNoNodesAvailable {
            nominatedNode = ""
            metrics.PodUnschedulable(fwk.ProfileName(), metrics.SinceInSeconds(start))
        } else {
            klog.ErrorS(err, "Error selecting node for pod", "pod", klog.KObj(pod))
            metrics.PodScheduleError(fwk.ProfileName(), metrics.SinceInSeconds(start))
        }
        
        // 处理调度失败
        sched.handleSchedulingFailure(ctx, fwk, podInfo, err, nominatedNode, start)
        return
    }
    
    // 调度成功，记录指标
    metrics.PodScheduled(fwk.ProfileName(), metrics.SinceInSeconds(start))
    metrics.PodSchedulingAttempts.Observe(float64(podInfo.Attempts))
    
    // 执行绑定操作
    err = sched.bind(ctx, fwk, pod, scheduleResult.SuggestedHost, state)
    if err != nil {
        // 绑定失败处理
        metrics.PodScheduleError(fwk.ProfileName(), metrics.SinceInSeconds(start))
        sched.handleBindingFailure(ctx, fwk, podInfo, scheduleResult, err, start)
        return
    }
    
    // 绑定成功
    klog.V(2).InfoS("Successfully bound pod to node", "pod", klog.KObj(pod), "node", scheduleResult.SuggestedHost, "evaluatedNodes", scheduleResult.EvaluatedNodes, "feasibleNodes", scheduleResult.FeasibleNodes)
}

// handleSchedulingFailure 处理调度失败
func (sched *Scheduler) handleSchedulingFailure(ctx context.Context, fwk framework.Framework, podInfo *framework.QueuedPodInfo, err error, nominatedNodeName string, start time.Time) {
    pod := podInfo.Pod
    
    // 更新Pod的调度失败状态
    if err == ErrNoNodesAvailable {
        klog.V(2).InfoS("Unable to schedule pod; no nodes are registered to the cluster; waiting", "pod", klog.KObj(pod))
    } else if fitError, ok := err.(*framework.FitError); ok {
        // 记录不适合的节点信息
        if sched.Error != nil {
            sched.Error(podInfo, err)
        }
        
        // 记录调度失败事件
        sched.Recorder.Eventf(pod, nil, v1.EventTypeWarning, "FailedScheduling", "Scheduling", err.Error())
        
        // 如果有提名节点，更新Pod状态
        if nominatedNodeName != "" {
            sched.SchedulingQueue.AddNominatedPod(podInfo.PodInfo, nominatedNodeName)
        }
    } else {
        klog.ErrorS(err, "Error selecting node for pod", "pod", klog.KObj(pod))
        if sched.Error != nil {
            sched.Error(podInfo, err)
        }
        sched.Recorder.Eventf(pod, nil, v1.EventTypeWarning, "FailedScheduling", "Scheduling", err.Error())
    }
    
    // 将Pod重新加入队列
    if err := sched.SchedulingQueue.AddUnschedulableIfNotPresent(podInfo, sched.SchedulingQueue.SchedulingCycle()); err != nil {
        klog.ErrorS(err, "Error occurred")
    }
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212

// Schedule 执行调度算法选择节点
// 文件路径: pkg/scheduler/core/generic_scheduler.go
func (g *genericScheduler) Schedule(ctx context.Context, fwk framework.Framework, state *framework.CycleState, pod *v1.Pod) (result ScheduleResult, err error) {
    trace := utiltrace.New("Scheduling", utiltrace.Field{Key: "namespace", Value: pod.Namespace}, utiltrace.Field{Key: "name", Value: pod.Name})
    defer trace.LogIfLong(100 * time.Millisecond)
    
    // 获取节点列表快照
    if err := g.snapshot(); err != nil {
        return result, err
    }
    trace.Step("Snapshotting scheduler cache and node infos done")
    
    if g.nodeInfoSnapshot.NumNodes() == 0 {
        return result, ErrNoNodesAvailable
    }
    
    // 第一阶段：预过滤阶段
    // 运行PreFilter插件进行预处理
    preFilterStatus := fwk.RunPreFilterPlugins(ctx, state, pod)
    if !preFilterStatus.IsSuccess() {
        return result, preFilterStatus.AsError()
    }
    trace.Step("Running PreFilter plugins done")
    
    // 第二阶段：过滤阶段
    // 找出所有可调度的节点
    feasibleNodes, diagnosis, err := g.findNodesThatFitPod(ctx, fwk, state, pod)
    if err != nil {
        return result, err
    }
    trace.Step("Computing predicates done")
    
    // 如果没有可调度的节点，返回错误
    if len(feasibleNodes) == 0 {
        return result, &framework.FitError{
            Pod:         pod,
            NumAllNodes: g.nodeInfoSnapshot.NumNodes(),
            Diagnosis:   diagnosis,
        }
    }
    
    // 如果只有一个节点，直接返回
    if len(feasibleNodes) == 1 {
        return ScheduleResult{
            SuggestedHost:  feasibleNodes[0].Name,
            EvaluatedNodes: 1 + len(diagnosis.NodeToStatusMap),
            FeasibleNodes:  1,
        }, nil
    }
    
    // 第三阶段：打分阶段
    // 对可调度的节点进行打分
    priorityList, err := g.prioritizeNodes(ctx, fwk, state, pod, feasibleNodes)
    if err != nil {
        return result, err
    }
    
    // 选择得分最高的节点
    host, err := g.selectHost(priorityList)
    trace.Step("Prioritizing done")
    
    return ScheduleResult{
        SuggestedHost:  host,
        EvaluatedNodes: len(feasibleNodes) + len(diagnosis.NodeToStatusMap),
        FeasibleNodes:  len(feasibleNodes),
    }, err
}

// findNodesThatFitPod 找出适合Pod的节点
func (g *genericScheduler) findNodesThatFitPod(ctx context.Context, fwk framework.Framework, state *framework.CycleState, pod *v1.Pod) ([]*v1.Node, framework.Diagnosis, error) {
    diagnosis := framework.Diagnosis{
        NodeToStatusMap:      make(framework.NodeToStatusMap),
        UnschedulablePlugins: sets.NewString(),
    }
    
    // 获取所有节点
    allNodes, err := g.nodeInfoSnapshot.NodeInfos().List()
    if err != nil {
        return nil, diagnosis, err
    }
    
    // 计算需要检查的节点数量
    numNodesToFind := g.numFeasibleNodesToFind(int32(len(allNodes)))
    
    // 并发检查节点
    feasibleNodes := make([]*v1.Node, 0, numNodesToFind)
    if !fwk.HasFilterPlugins() {
        // 如果没有过滤插件，所有节点都是可调度的
        for _, nodeInfo := range allNodes {
            feasibleNodes = append(feasibleNodes, nodeInfo.Node())
        }
        return feasibleNodes, diagnosis, nil
    }
    
    // 并发过滤节点
    errCh := parallelize.NewErrorChannel()
    var statusesLock sync.Mutex
    var feasibleNodesLock sync.Mutex
    
    ctx, cancel := context.WithCancel(ctx)
    
    // 检查单个节点的函数
    checkNode := func(i int) {
        nodeInfo := allNodes[i]
        node := nodeInfo.Node()
        
        // 运行Filter插件
        status := fwk.RunFilterPluginsWithNominatedPods(ctx, state, pod, nodeInfo)
        if status.Code() == framework.Error {
            errCh.SendErrorWithCancel(status.AsError(), cancel)
            return
        }
        
        if status.IsSuccess() {
            // 节点通过过滤
            feasibleNodesLock.Lock()
            feasibleNodes = append(feasibleNodes, node)
            feasibleNodesLock.Unlock()
        } else {
            // 节点未通过过滤
            statusesLock.Lock()
            diagnosis.NodeToStatusMap[node.Name] = status
            diagnosis.UnschedulablePlugins.Insert(status.FailedPlugin())
            statusesLock.Unlock()
        }
    }
    
    // 并行执行节点检查
    fwk.Parallelizer().Until(ctx, len(allNodes), checkNode)
    if err := errCh.ReceiveError(); err != nil {
        return nil, diagnosis, err
    }
    
    return feasibleNodes, diagnosis, nil
}

// prioritizeNodes 对节点进行打分
func (g *genericScheduler) prioritizeNodes(ctx context.Context, fwk framework.Framework, state *framework.CycleState, pod *v1.Pod, feasibleNodes []*v1.Node) (framework.NodeScoreList, error) {
    // 如果没有打分插件，给所有节点相同分数
    if !fwk.HasScorePlugins() {
        result := make(framework.NodeScoreList, 0, len(feasibleNodes))
        for _, node := range feasibleNodes {
            result = append(result, framework.NodeScore{
                Name:  node.Name,
                Score: 1,
            })
        }
        return result, nil
    }
    
    // 运行PreScore插件
    preScoreStatus := fwk.RunPreScorePlugins(ctx, state, pod, feasibleNodes)
    if !preScoreStatus.IsSuccess() {
        return nil, preScoreStatus.AsError()
    }
    
    // 运行Score插件
    scoresMap, scoreStatus := fwk.RunScorePlugins(ctx, state, pod, feasibleNodes)
    if !scoreStatus.IsSuccess() {
        return nil, scoreStatus.AsError()
    }
    
    // 应用权重并归一化分数
    result := make(framework.NodeScoreList, 0, len(feasibleNodes))
    for _, node := range feasibleNodes {
        result = append(result, framework.NodeScore{Name: node.Name, Score: 0})
        for plugin, nodeScoreList := range scoresMap {
            weight := fwk.ScorePluginWeight(plugin)
            nodeScore := nodeScoreList[node.Name]
            result[len(result)-1].Score += nodeScore.Score * int64(weight)
        }
    }
    
    // 运行NormalizeScore插件
    if fwk.HasNormalizeScorePlugins() {
        normalizeStatus := fwk.RunNormalizeScorePlugins(ctx, state, pod, result)
        if !normalizeStatus.IsSuccess() {
            return nil, normalizeStatus.AsError()
        }
    }
    
    return result, nil
}

// selectHost 选择得分最高的节点
func (g *genericScheduler) selectHost(nodeScoreList framework.NodeScoreList) (string, error) {
    if len(nodeScoreList) == 0 {
        return "", fmt.Errorf("empty priorityList")
    }
    
    // 找出最高分数
    maxScore := nodeScoreList[0].Score
    selected := nodeScoreList[0].Name
    cntOfMaxScore := 1
    
    for _, ns := range nodeScoreList[1:] {
        if ns.Score > maxScore {
            maxScore = ns.Score
            selected = ns.Name
            cntOfMaxScore = 1
        } else if ns.Score == maxScore {
            cntOfMaxScore++
            // 随机选择一个最高分节点
            if rand.Intn(cntOfMaxScore) == 0 {
                selected = ns.Name
            }
        }
    }
    
    klog.V(2).InfoS("Selected node for pod", "node", selected, "score", maxScore)
    return selected, nil
}