schedapi | 南风未起｜云原生/Go技术分享博客

# 1.简介

# 1.1.结构

kube-scheduler是kubernetes核心组件之一，根据预选及优先调度算法进行pod的调度工作，将未调度的pod绑定到合适的最优node，组件结构可分为两部分：cmd启动部分及scheduler核心逻辑部分。
注意

scheduler不会干涉pod运行，仅根据声明式语义哲学更新pod与node绑定关系至apiserver，由kubelet管理pod生命周期

# 1.2.原理

kube-scheduler负责为未调度的pod寻找最合适的node节点，基于predicates预选和priority优选将分值最高的node作为调度目标，以驱动kubelet监听及创建真正的pod。
注意

kube-scheduler的调度过程分为调度周期和绑定周期，前者寻找合适节点，后者进行节点绑定

# 2.流程

# 2.1.入口

kube-scheduler基于cobra命令行框架注册NewSchedulerCommand()入口，以初始化及启动scheduler调度器。

// cmd/kube-scheduler
func main() {
	command := app.NewSchedulerCommand()
	code := cli.Run(command)
	os.Exit(code)
}

// NewSchedulerCommand creates a *cobra.Command object with default parameters and registryOptions
func NewSchedulerCommand(registryOptions ...Option) *cobra.Command {
	...

	cmd := &cobra.Command{
		...
		RunE: func(cmd *cobra.Command, args []string) error {
            // 核心
			return runCommand(cmd, opts, registryOptions...)
		},
		...
	}
	...

	return cmd
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

注意

runCommand()是核心入口，会初始化scheduler对象及启动

# 2.2.runCmd

runCommand()作为入口会调用setup()初始化scheduler，执行Run()监听及启动scheduler，此时scheduler真正开始工作。

// runCommand runs the scheduler.
func runCommand(cmd *cobra.Command, opts *options.Options, registryOptions ...Option) error {
	...

    // 初始化scheduler
	cc, sched, err := Setup(ctx, opts, registryOptions...)
	...

    // 执行scheduler
	return Run(ctx, cc, sched)
}

1
2
3
4
5
6
7
8
9
10
11

注意

setup()和Run()是核心方法，setup()会初始化scheduler，Run()会执行scheduler主逻辑

# 2.3.setup

setup()用于填充scheduler config，根据配置初始化scheduler对象及注册监听回调，返回的scheduler实例负责运行调度主逻辑。

// Setup creates a completed config and a scheduler based on the command args and options
func Setup(ctx context.Context, opts *options.Options, outOfTreeRegistryOptions ...Option) (*schedulerserverconfig.CompletedConfig, *scheduler.Scheduler, error) {
	...

    // 配置初始化
	c, err := opts.Config()
	...

	// 配置填充
	cc := c.Complete()
	...
    
	// 构建scheduler对象
	sched, err := scheduler.New(...)
	...

	return &cc, sched, nil
}

// New returns a Scheduler
func New(...) (*Scheduler, error) {
	...
    
    options := defaultSchedulerOptions
    // 渲染调度配置
	for _, opt := range opts {
		opt(&options)
	}

    // 未渲染完成(配置为空)
	if options.applyDefaultProfile {
		...
        // 渲染默认配置
		scheme.Scheme.Default(&versionedCfg)
		...
        // 配置标准化
		scheme.Scheme.Convert(&versionedCfg, &cfg, nil)
        ...
        
        // 回填profiles配置
		options.profiles = cfg.Profiles
	}
    ...

    // 构建registry对象(集成内置插件)
	registry := frameworkplugins.NewInTreeRegistry()
    // 合并扩展插件
	registry.Merge(options.frameworkOutOfTreeRegistry)
    ...

    // 整理调度器extender扩展(http扩展调度器功能)
	extenders, err := buildExtenders(options.extenders, options.profiles)
	...
    
    // 初始化podLister和nodeLister缓存
    podLister := informerFactory.Core().V1().Pods().Lister()
	nodeLister := informerFactory.Core().V1().Nodes().Lister()
	// 实例化快照
	snapshot := internalcache.NewEmptySnapshot()
	...
    
    // profiles记录不同framework扩展插件
	profiles, err := profile.NewMap(options.profiles, registry, recorderFactory,
		frameworkruntime.WithComponentConfigVersion(options.componentConfigVersion),
		frameworkruntime.WithClientSet(client),
		frameworkruntime.WithKubeConfig(options.kubeConfig),
		frameworkruntime.WithInformerFactory(informerFactory),
		frameworkruntime.WithSnapshotSharedLister(snapshot),
		frameworkruntime.WithCaptureProfile(frameworkruntime.CaptureProfile(options.frameworkCapturer)),
		frameworkruntime.WithClusterEventMap(clusterEventMap),
		frameworkruntime.WithParallelism(int(options.parallelism)),
		frameworkruntime.WithExtenders(extenders),
	)
	...

    // 初始化调度队列
	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.WithClusterEventMap(clusterEventMap),
		internalqueue.WithPodMaxInUnschedulablePodsDuration(options.podMaxInUnschedulablePodsDuration),
	)

    // 调度podQueue队列
    for _, fwk := range profiles {
		fwk.SetPodNominator(podQueue)
	}
    
    // 调度器数据缓存,支持TTL
	schedulerCache := internalcache.New(durationToExpireAssumedPod, stopEverything)
	...
    
    // scheduler对象
	sched := newScheduler(...)

    // 事件回调注册
	addAllEventHandlers(sched, informerFactory, dynInformerFactory, unionedGVKs(clusterEventMap))

	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

注意

registry和extender涉及scheduler功能扩展，后续会分析

# 2.4.eventHandler

addAllEventHandlers()用于注册事件回调至scheduler，监听的是pod和node资源对象，资源变更会触发queue和cache的回调处理。

// addAllEventHandlers add event handlers for various informers.
func addAllEventHandlers(...) {
	// scheduled pod cache
	informerFactory.Core().V1().Pods().Informer().AddEventHandler(
		cache.FilteringResourceEventHandler{
            // pod.spec.scheduleName不为空
			FilterFunc: func(obj interface{}) bool {
				...
			},
            // 已调度的pod加入缓存
			Handler: cache.ResourceEventHandlerFuncs{
				AddFunc:    sched.addPodToCache,
				UpdateFunc: sched.updatePodInCache,
				DeleteFunc: sched.deletePodFromCache,
			},
		},
	)
	// unscheduled pod queue
	informerFactory.Core().V1().Pods().Informer().AddEventHandler(
		cache.FilteringResourceEventHandler{
            // pod.spec.scheduleName为空,关联的调度器支持
			FilterFunc: func(obj interface{}) bool {
				...
			},
            // 未调度的pod加入队列
			Handler: cache.ResourceEventHandlerFuncs{
				AddFunc:    sched.addPodToSchedulingQueue,
				UpdateFunc: sched.updatePodInSchedulingQueue,
				DeleteFunc: sched.deletePodFromSchedulingQueue,
			},
		},
	)

    // node cache
	informerFactory.Core().V1().Nodes().Informer().AddEventHandler(
        // node加入缓存
		cache.ResourceEventHandlerFuncs{
			AddFunc:    sched.addNodeToCache,
			UpdateFunc: sched.updateNodeInCache,
			DeleteFunc: sched.deleteNodeFromCache,
		},
	)

    // 注册调度器唤醒回调
	buildEvtResHandler := func(at framework.ActionType, gvk framework.GVK, shortGVK string) cache.ResourceEventHandlerFuncs {
		funcs := cache.ResourceEventHandlerFuncs{}
        // ADD事件
		if at&framework.Add != 0 {
			...
			funcs.AddFunc = func(_ interface{}) {
				sched.SchedulingQueue.MoveAllToActiveOrBackoffQueue(evt, nil)
			}
		}
        // UPDATE事件
		if at&framework.Update != 0 {
			...
			funcs.UpdateFunc = func(_, _ interface{}) {
				sched.SchedulingQueue.MoveAllToActiveOrBackoffQueue(evt, nil)
			}
		}
        // DELETE事件
		if at&framework.Delete != 0 {
			...
			funcs.DeleteFunc = func(_ interface{}) {
				sched.SchedulingQueue.MoveAllToActiveOrBackoffQueue(evt, nil)
			}
		}
		return funcs
	}

    // 某些资源变化,重新触发未调度pod发起调度(资源不足 --> 充足,部分pod可继续调度)
	for gvk, at := range gvkMap {
		switch gvk {
		...
        // CSINode资源变化
		case framework.CSINode:
			informerFactory.Storage().V1().CSINodes().Informer().AddEventHandler(
				buildEvtResHandler(at, framework.CSINode, "CSINode"),
			)
        //CSIDriver资源变化
		case framework.CSIDriver:
			informerFactory.Storage().V1().CSIDrivers().Informer().AddEventHandler(
				buildEvtResHandler(at, framework.CSIDriver, "CSIDriver"),
			)
        // CSIStorageCapacity资源变化
		case framework.CSIStorageCapacity:
			informerFactory.Storage().V1().CSIStorageCapacities().Informer().AddEventHandler(
				buildEvtResHandler(at, framework.CSIStorageCapacity, "CSIStorageCapacity"),
			)
        // PV资源变化
		case framework.PersistentVolume:
			informerFactory.Core().V1().PersistentVolumes().Informer().AddEventHandler(
				buildEvtResHandler(at, framework.PersistentVolume, "Pv"),
			)
        // PVC资源变化
		case framework.PersistentVolumeClaim:
			// MaxPDVolumeCountPredicate: add/update PVC will affect counts of PV when it is bound.
			informerFactory.Core().V1().PersistentVolumeClaims().Informer().AddEventHandler(
				buildEvtResHandler(at, framework.PersistentVolumeClaim, "Pvc"),
			)
        // StorageClass资源变化
		case framework.StorageClass:
			if at&framework.Add != 0 {
				informerFactory.Storage().V1().StorageClasses().Informer().AddEventHandler(
					cache.ResourceEventHandlerFuncs{
						AddFunc: sched.onStorageClassAdd,
					},
				)
			}
			if at&framework.Update != 0 {
				informerFactory.Storage().V1().StorageClasses().Informer().AddEventHandler(
					cache.ResourceEventHandlerFuncs{
						UpdateFunc: func(_, _ interface{}) {
							sched.SchedulingQueue.MoveAllToActiveOrBackoffQueue(queue.StorageClassUpdate, nil)
						},
					},
				)
			}
		default:
			...
			// Fall back to try dynamic informers.
			gvr, _ := schema.ParseResourceArg(string(gvk))
			dynInformer := dynInformerFactory.ForResource(*gvr).Informer()
			dynInformer.AddEventHandler(
				buildEvtResHandler(at, gvk, strings.Title(gvr.Resource)),
			)
		}
	}
}

注意

scheduler监听回调分为两类，pod变化触发回调，存储或其它资源变化触发回调，回调本质上就是驱动未调度pod的调度

# 2.5.election

run()根据参数配置和调度器对象运行调度过程，内部回执行一系列检测工作，scheduler多实例触发选举调度，本质执行的还是sched.Run()。

// Run executes the scheduler based on the given configuration.
func Run(ctx context.Context, cc *schedulerserverconfig.CompletedConfig, sched *scheduler.Scheduler) error {
	...

	// 执行同步
	cc.InformerFactory.Start(ctx.Done())
	...

	// 等待同步完成
	cc.InformerFactory.WaitForCacheSync(ctx.Done())
	...

	// 选举调度
	if cc.LeaderElection != nil {
        // 设置选举回调
		cc.LeaderElection.Callbacks = leaderelection.LeaderCallbacks{
			OnStartedLeading: func(ctx context.Context) {
				close(waitingForLeader)
                // 最终执行的还是普通调度
				sched.Run(ctx)
			}
            ...
		}
        // 初始化选举对象
		leaderElector, err := leaderelection.NewLeaderElector(*cc.LeaderElection)
		...

        // 执行选举及回调
		leaderElector.Run(ctx)

		return fmt.Errorf("lost lease")
	}

	// Leader election is disabled, so runCommand inline until done.
	close(waitingForLeader)
    
    // 普通调度
	sched.Run(ctx)
	return fmt.Errorf("finished without leader elect")
}

// Run starts the leader election loop. Run will not return before leader election loop is stopped by ctx or it // has stopped holding the leader lease
func (le *LeaderElector) Run(ctx context.Context) {
	...

    // 争夺领导权
	if !le.acquire(ctx) {
		return // ctx signalled done
	}
    
    ...
    // 执行调度
	go le.config.Callbacks.OnStartedLeading(ctx)
    
    // 续期
	le.renew(ctx)
}

// acquire loops calling tryAcquireOrRenew and returns true immediately when tryAcquireOrRenew succeeds.
// Returns false if ctx signals done.
func (le *LeaderElector) acquire(ctx context.Context) bool {
	...
    // 尝试抢主
	wait.JitterUntil(func() {
        // 抢占租约
		succeeded = le.tryAcquireOrRenew(ctx)
		...
        // 失败重试
		if !succeeded {
			klog.V(4).Infof("failed to acquire lease %v", desc)
			return
		}
		...
        // 抢主成功,结束wait.util轮询器
        cancel()
	}, le.config.RetryPeriod, JitterFactor, true, ctx.Done())
	return succeeded
}

// renew loops calling tryAcquireOrRenew and returns immediately when tryAcquireOrRenew fails or ctx signals done.
func (le *LeaderElector) renew(ctx context.Context) {
	...
	wait.Until(func() {
		...
        // 尝试申请或续期
		err := wait.PollImmediateUntil(le.config.RetryPeriod, func() (bool, error) {
			return le.tryAcquireOrRenew(timeoutCtx), nil
		}, timeoutCtx.Done())
		...
        // 成功等待下一轮续期
		if err == nil {
			return
		}
		...
        // 续期失败,结束wait.util轮询器
        cancel()
	}, le.config.RetryPeriod, ctx.Done())

	// 结束,释放租约
	if le.config.ReleaseOnCancel {
		le.release()
	}
}

注意

1.kube-scheduler通过选举leaderelection机制保证集群只有一个leader实例运行调度器，其它的follower实例尝试轮询抢锁知道成功

2.kube-scheduler失去leader身份后，会直接退出进程，以避免异步逻辑不好收敛造成的数据不一致问题

# 2.6.run

sched.Run()是scheduler核心入口，先执行schedulingQueue.Run()处理内部延迟队列，再异步启动sched.scheduleOne()执行调度。

// Run begins watching and scheduling. It starts scheduling and blocked until the context is done.
func (sched *Scheduler) Run(ctx context.Context) {
	// 调度队列监听及处理
    sched.SchedulingQueue.Run()
    // 异步执行调度
	wait.UntilWithContext(ctx, sched.scheduleOne, 0)
	sched.SchedulingQueue.Close()
}

// Run starts the goroutine to pump from podBackoffQ to activeQ
func (p *PriorityQueue) Run() {
	// 间隔1s检查backoff队列,退避结束的pod转移到activeQ
    go wait.Until(p.flushBackoffQCompleted, 1.0*time.Second, p.stop)
    // 间隔30s检查unschedulable队列,将可能调度的pod转移到activeQ
	go wait.Until(p.flushUnschedulablePodsLeftover, 30*time.Second, p.stop)
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

注意

scheduler调度流程分为队列维护及消费，前者定期将未调度的回退结束pod推入activeQ，后者消费activeQ的pod调度到不同节点