聚焦目标
理解kube-scheduler启动的流程
目录
run
// kube-scheduler 类似于kube-apiserver,是个常驻进程,查看其对应的Run函数
func runCommand(cmd *cobra.Command, opts *options.Options, registryOptions ...Option) error {
// 根据入参,返回配置cc与调度sched
cc, sched, err := Setup(ctx, opts, registryOptions...)
// 运行
return Run(ctx, cc, sched)
}
// 运行调度策略
func Run(ctx context.Context, cc *schedulerserverconfig.CompletedConfig, sched *scheduler.Scheduler) error {
// 将配置注册到configz中,会保存在一个全局map里
if cz, err := configz.New("componentconfig"); err == nil {
cz.Set(cc.ComponentConfig)
} else {
return fmt.Errorf("unable to register configz: %s", err)
}
// 事件广播管理器,涉及到k8s里的一个核心资源 - Event事件,暂时不细讲
cc.EventBroadcaster.StartRecordingToSink(ctx.Done())
// 健康监测的服务
var checks []healthz.HealthChecker
// 异步各个Informer。Informer是kube-scheduler的一个重点
go cc.PodInformer.Informer().Run(ctx.Done())
cc.InformerFactory.Start(ctx.Done())
cc.InformerFactory.WaitForCacheSync(ctx.Done())
// 选举Leader的工作,因为Master节点可以存在多个,选举一个作为Leader
if cc.LeaderElection != nil {
cc.LeaderElection.Callbacks = leaderelection.LeaderCallbacks{
// 两个钩子函数,开启Leading时运行调度,结束时打印报错
OnStartedLeading: sched.Run,
OnStoppedLeading: func() {
klog.Fatalf("leaderelection lost")
},
}
leaderElector, err := leaderelection.NewLeaderElector(*cc.LeaderElection)
if err != nil {
return fmt.Errorf("couldn't create leader elector: %v", err)
}
// 参与选举的会持续通信
leaderElector.Run(ctx)
return fmt.Errorf("lost lease")
}
// 不参与选举的,也就是单节点的情况时,在这里运行
sched.Run(ctx)
return fmt.Errorf("finished without leader elect")
}
/*
到这里,我们已经接触了kube-scheduler的2个核心概念:
1. scheduler:正如程序名kube-scheduler,这个进程的核心作用是进行调度,会涉及到多种调度策略
2. Informer:k8s中有各种类型的资源,包括自定义的。而Informer的实现就将调度和资源结合了起来
*/
Scheduler
// 在创建scheduler的函数
func Setup() {
// 创建scheduler,包括多个选项
sched, err := scheduler.New(cc.Client,
cc.InformerFactory,
cc.PodInformer,
recorderFactory,
ctx.Done(),
scheduler.WithProfiles(cc.ComponentConfig.Profiles...),
scheduler.WithAlgorithmSource(cc.ComponentConfig.AlgorithmSource),
scheduler.WithPercentageOfNodesToScore(cc.ComponentConfig.PercentageOfNodesToScore),
scheduler.WithFrameworkOutOfTreeRegistry(outOfTreeRegistry),
scheduler.WithPodMaxBackoffSeconds(cc.ComponentConfig.PodMaxBackoffSeconds),
scheduler.WithPodInitialBackoffSeconds(cc.ComponentConfig.PodInitialBackoffSeconds),
scheduler.WithExtenders(cc.ComponentConfig.Extenders...),
)
return &cc, sched, nil
}
// 我们再看一下New这个函数
func New() (*Scheduler, error) {
// 先注册了所有的算法,保存到一个 map[string]PluginFactory 中
registry := frameworkplugins.NewInTreeRegistry()
// 重点看一下Scheduler的创建过程
var sched *Scheduler
source := options.schedulerAlgorithmSource
switch {
// 根据Provider创建,重点看这里
case source.Provider != nil:
sc, err := configurator.createFromProvider(*source.Provider)
if err != nil {
return nil, fmt.Errorf("couldn't create scheduler using provider %q: %v", *source.Provider, err)
}
sched = sc
// 根据用户设置创建,来自文件或者ConfigMap
case source.Policy != nil:
policy := &schedulerapi.Policy{
}
switch {
case source.Policy.File != nil:
if err := initPolicyFromFile(source.Policy.File.Path, policy); err != nil {
return nil, err
}
case source.Policy.ConfigMap != nil:
if err := initPolicyFromConfigMap(client, source.Policy.ConfigMap, policy); err != nil {
return nil, err
}
}
configurator.extenders = policy.Extenders
sc, err := configurator.createFromConfig(*policy)
if err != nil {
return nil, fmt.Errorf("couldn't create scheduler from policy: %v", err)
}
sched = sc
default:
return nil, fmt.Errorf("unsupported algorithm source: %v", source)
}
}
// 创建
func (c *Configurator) createFromProvider(providerName string) (*Scheduler, error) {
klog.V(2).Infof("Creating scheduler from algorithm provider '%v'", providerName)
// 实例化算法的Registry
r := algorithmprovider.NewRegistry()
defaultPlugins, exist := r[providerName]
if !exist {
return nil, fmt.Errorf("algorithm provider %q is not registered", providerName)
}
// 将各种算法作为plugin进行设置
for i := range c.profiles {
prof := &c.profiles[i]
plugins := &schedulerapi.Plugins{
}
plugins.Append(defaultPlugins)
plugins.Apply(prof.Plugins)
prof.Plugins = plugins
}
return c.create()
}
// 从这个初始化中可以看到,主要分为2类:默认与ClusterAutoscaler两种算法
func NewRegistry() Registry {
// 默认算法包括过滤、打分、绑定等,有兴趣的去源码中逐个阅读
defaultConfig := getDefaultConfig()
applyFeatureGates(defaultConfig)
// ClusterAutoscaler 是集群自动扩展的算法,被单独拎出来,
caConfig := getClusterAutoscalerConfig()
applyFeatureGates(caConfig)
return Registry{
schedulerapi.SchedulerDefaultProviderName: defaultConfig,
ClusterAutoscalerProvider: caConfig,
}
}
/*
在这里,熟悉k8s的朋友会有个疑问:以前听说kubernets的调度有个Predicate和Priority两个算法,这里怎么没有分类?
这个疑问,我们在后面具体场景时再进行分析。
*/
NodeName
// 为了加深大家对Plugin的印象,我选择一个最简单的示例:根据Pod的spec字段中的NodeName,分配到指定名称的节点
package nodename
import (
"context"
v1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/runtime"
framework "k8s.io/kubernetes/pkg/scheduler/framework/v1alpha1"
)
type NodeName struct{
}
var _ framework.FilterPlugin = &NodeName{
}
// 这个调度算法的名称和错误信息
const (
Name = "NodeName"
ErrReason = "node(s) didn't match the requested hostname"
)
// 调度算法的明明
func (pl *NodeName) Name() string {
return Name
}
// 过滤功能,这个就是NodeName算法的实现
func (pl *NodeName) Filter(ctx context.Context, _ *framework.CycleState, pod *v1.Pod, nodeInfo *framework.NodeInfo) *framework.Status {
// 找不到Node
if nodeInfo.Node() == nil {
return framework.NewStatus(framework.Error, "node not found")
}
// 匹配不到,返回错误
if !Fits(pod, nodeInfo) {
return framework.NewStatus(framework.UnschedulableAndUnresolvable, ErrReason)
}
return nil
}
/*
匹配的算法,两种条件满足一个就认为成功
1. spec没有填NodeName
2.spec的NodeName和节点匹配
*/
func Fits(pod *v1.Pod, nodeInfo *framework.NodeInfo) bool {
return len(pod.Spec.NodeName) == 0 || pod.Spec.NodeName == nodeInfo.Node().Name
}
// 初始化
func New(_ runtime.Object, _ framework.FrameworkHandle) (framework.Plugin, error) {
return &NodeName{
}, nil
}