【kubernetes/k8s源码分析】kube-scheduler 源码分析

左手的ㄟ右手 2022-06-10 09:18 267阅读 0赞

*kubernetes 1.12.1版本*

# 前言 #

在 kubernetes 体系中，scheduler 是以 plugin 形式存在的模块，这种可插拔的设计方便用户自定义所需要的调度算法，所以源码路径为 plugin 目录下的 cmd 以及 pkg/scheduler

###  ###

Scheduler 负责安排 Pod 到具体的Node，通过监听API server提供的watch等接口，获取到未调度的pods和node的相关信息，根据一系列的预选策略和优选策略给各个 Node 节点打分排序，选择出最合适的也就是优先级最高的node节点，将其与pods进行绑定，并将绑定的结果固化到etcd中去，然后由 kubelet 负责创建 Pod。

Kubernetes 调度分为 Predicate（预选）和 Priority（优选），分为两个过程：

*  预选：遍历所有 Node，按照预选筛选出符合要求的 Node，如果没有 Node 符合 Predicate 策略，那该 Pod 就会被挂起，直到有 Node 能够满足所有策略；
 *  优选，在第一步基础上，按照优选为待选 Node 打分，获取最高分；

*    也不知道哪位大神画的这么好，借来一用*

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob25nbGluemhhbmc_size_16_color_FFFFFF_t_70][]

## **接口ScheduleAlgorithm** ##

如果未自定义调度器，则启用默认的调度器genericScheduler

// ScheduleAlgorithm is an interface implemented by things that know how to schedule pods
    // onto machines.
    type ScheduleAlgorithm interface {
    	Schedule(*v1.Pod, NodeLister) (selectedMachine string, err error)
    	// Preempt receives scheduling errors for a pod and tries to create room for
    	// the pod by preempting lower priority pods if possible.
    	// It returns the node where preemption happened, a list of preempted pods, a
    	// list of pods whose nominated node name should be removed, and error if any.
    	Preempt(*v1.Pod, NodeLister, error) (selectedNode *v1.Node, preemptedPods []*v1.Pod, cleanupNominatedPods []*v1.Pod, err error)
    	// Predicates() returns a pointer to a map of predicate functions. This is
    	// exposed for testing.
    	Predicates() map[string]FitPredicate
    	// Prioritizers returns a slice of priority config. This is exposed for
    	// testing.
    	Prioritizers() []PriorityConfig
    }

# ***kubernetes scheduler 源码分析-启动流程*** #

###      ###

# 一. 入口 main 函数 #

路径 cmd/kube-scheduler/scheduler.go

func main() {
    	rand.Seed(time.Now().UTC().UnixNano())
    
    	command := app.NewSchedulerCommand()
    
    	// TODO: once we switch everything over to Cobra commands, we can go back to calling
    	// utilflag.InitFlags() (by removing its pflag.Parse() call). For now, we have to set the
    	// normalize func and add the go flag set by hand.
    	pflag.CommandLine.SetNormalizeFunc(utilflag.WordSepNormalizeFunc)
    	pflag.CommandLine.AddGoFlagSet(goflag.CommandLine)
    	// utilflag.InitFlags()
    	logs.InitLogs()
    	defer logs.FlushLogs()
    
    	if err := command.Execute(); err != nil {
    		fmt.Fprintf(os.Stderr, "%v\n", err)
    		os.Exit(1)
    	}
    }

###   1.1 NewSchedulerCommand函数 ###

*  NewSchedulerCommand初始化一个cobra comand，定义了命令行参数，并进行验证
 *  \--write-config-to如果命令行有该参数，将配置写入文件就退出
 *  主要逻辑是执行 Run 第二章节分析

// NewSchedulerCommand creates a *cobra.Command object with default parameters
    func NewSchedulerCommand() *cobra.Command {
    	opts, err := options.NewOpteions()
    	if err != nil {
    		glog.Fatalf("unable to initialize command options: %v", err)
    	}
    
    	cmd := &cobra.Command{
    		Use: "kube-scheduler",
    		Long: `The Kubernetes scheduler is a policy-rich, topology-aware,
    workload-specific function that significantly impacts availability, performance,
    and capacity. The scheduler needs to take into account individual and collective
    resource requirements, quality of service requirements, hardware/software/policy
    constraints, affinity and anti-affinity specifications, data locality, inter-workload
    interference, deadlines, and so on. Workload-specific requirements will be exposed
    through the API as necessary.`,
    		Run: func(cmd *cobra.Command, args []string) {
    			verflag.PrintAndExitIfRequested()
    			utilflag.PrintFlags(cmd.Flags())
    
    			if len(args) != 0 {
    				fmt.Fprint(os.Stderr, "arguments are not supported\n")
    			}
    
    			if errs := opts.Validate(); len(errs) > 0 {
    				fmt.Fprintf(os.Stderr, "%v\n", utilerrors.NewAggregate(errs))
    				os.Exit(1)
    			}
    
    			if len(opts.WriteConfigTo) > 0 {
    				if err := options.WriteConfigFile(opts.WriteConfigTo, &opts.ComponentConfig); err != nil {
    					fmt.Fprintf(os.Stderr, "%v\n", err)
    					os.Exit(1)
    				}
    				glog.Infof("Wrote configuration to: %s\n", opts.WriteConfigTo)
    				return
    			}
    
    			c, err := opts.Config()
    			if err != nil {
    				fmt.Fprintf(os.Stderr, "%v\n", err)
    				os.Exit(1)
    			}
    
    			stopCh := make(chan struct{})
    			if err := Run(c.Complete(), stopCh); err != nil {
    				fmt.Fprintf(os.Stderr, "%v\n", err)
    				os.Exit(1)
    			}
    		},
    	}
    
    	opts.AddFlags(cmd.Flags())
    	cmd.MarkFlagFilename("config", "yaml", "yml", "json")
    
    	return cmd
    }

###   1.2 Config函数 ###

生成Config配置，包括kube client，eventBroadcaster，如果--leader-elect为true则设置选举配置

// Config return a scheduler config object
    func (o *Options) Config() (*schedulerappconfig.Config, error) {
    	c := &schedulerappconfig.Config{}
    	if err := o.ApplyTo(c); err != nil {
    		return nil, err
    	}
    
    	// prepare kube clients.
    	client, leaderElectionClient, eventClient, err := createClients(c.ComponentConfig.ClientConnection, o.Master, c.ComponentConfig.LeaderElection.RenewDeadline.Duration)
    	if err != nil {
    		return nil, err
    	}
    
    	// Prepare event clients.
    	eventBroadcaster := record.NewBroadcaster()
    	recorder := eventBroadcaster.NewRecorder(legacyscheme.Scheme, corev1.EventSource{Component: c.ComponentConfig.SchedulerName})
    
    	// Set up leader election if enabled.
    	var leaderElectionConfig *leaderelection.LeaderElectionConfig
    	if c.ComponentConfig.LeaderElection.LeaderElect {
    		leaderElectionConfig, err = makeLeaderElectionConfig(c.ComponentConfig.LeaderElection, leaderElectionClient, recorder)
    		if err != nil {
    			return nil, err
    		}
    	}
    
    	c.Client = client
    	c.InformerFactory = informers.NewSharedInformerFactory(client, 0)
    	c.PodInformer = factory.NewPodInformer(client, 0)
    	c.EventClient = eventClient
    	c.Recorder = recorder
    	c.Broadcaster = eventBroadcaster
    	c.LeaderElection = leaderElectionConfig
    
    	return c, nil
    }

## ##

###   1.3 Execute函数 ###

命令执行第三章节讲解

// Execute uses the args (os.Args[1:] by default)
    // and run through the command tree finding appropriate matches
    // for commands and then corresponding flags.
    func (c *Command) Execute() error {
    	_, err := c.ExecuteC()
    	return err
    }

###     ###

# 二. Run 函数 #

路径 plugin/cmd/kube-scheduler/app/server.go

// Run runs the Scheduler.
    func Run(c schedulerserverconfig.CompletedConfig, stopCh <-chan struct{}) error

###   2.1 ApplyFeatureGates函数 ###

特性开关（Feature Gates）是 Kubernetes 中用来开启实验性功能的配置，可以通过选项 `--feature-gates` 来给不同的组件（如 kube-apiserver、kube-controller-manager、kube-scheduler、kubelet、kube-proxy等）开启功能特性

*         如果设置了TaintNodesByCondition=true|false (BETA - default=true)，删除预选策略CheckNodeCondition，CheckNodeMemoryPressure，CheckNodeDiskPressure，CheckNodePIDPressure，`注册预选策略PodToleratesNodeTaints，CheckNodeUnschedulable`
 *  `   如果设置了ResourceLimitsPriorityFunction=true|false (ALPHA - default=false),则注册`优选策略ResourceLimitsPriorityFunction

// ApplyFeatureGates applies algorithm by feature gates.
    func ApplyFeatureGates() {
    	defaults.ApplyFeatureGates()
    }
    
    // ApplyFeatureGates applies algorithm by feature gates.
    func ApplyFeatureGates() {
    	if utilfeature.DefaultFeatureGate.Enabled(features.TaintNodesByCondition) {
    		// Remove "CheckNodeCondition", "CheckNodeMemoryPressure", "CheckNodePIDPressurePred"
    		// and "CheckNodeDiskPressure" predicates
    		factory.RemoveFitPredicate(predicates.CheckNodeConditionPred)
    		factory.RemoveFitPredicate(predicates.CheckNodeMemoryPressurePred)
    		factory.RemoveFitPredicate(predicates.CheckNodeDiskPressurePred)
    		factory.RemoveFitPredicate(predicates.CheckNodePIDPressurePred)
    		// Remove key "CheckNodeCondition", "CheckNodeMemoryPressure" and "CheckNodeDiskPressure"
    		// from ALL algorithm provider
    		// The key will be removed from all providers which in algorithmProviderMap[]
    		// if you just want remove specific provider, call func RemovePredicateKeyFromAlgoProvider()
    		factory.RemovePredicateKeyFromAlgorithmProviderMap(predicates.CheckNodeConditionPred)
    		factory.RemovePredicateKeyFromAlgorithmProviderMap(predicates.CheckNodeMemoryPressurePred)
    		factory.RemovePredicateKeyFromAlgorithmProviderMap(predicates.CheckNodeDiskPressurePred)
    		factory.RemovePredicateKeyFromAlgorithmProviderMap(predicates.CheckNodePIDPressurePred)
    
    		// Fit is determined based on whether a pod can tolerate all of the node's taints
    		factory.RegisterMandatoryFitPredicate(predicates.PodToleratesNodeTaintsPred, predicates.PodToleratesNodeTaints)
    		// Fit is determined based on whether a pod can tolerate unschedulable of node
    		factory.RegisterMandatoryFitPredicate(predicates.CheckNodeUnschedulablePred, predicates.CheckNodeUnschedulablePredicate)
    		// Insert Key "PodToleratesNodeTaints" and "CheckNodeUnschedulable" To All Algorithm Provider
    		// The key will insert to all providers which in algorithmProviderMap[]
    		// if you just want insert to specific provider, call func InsertPredicateKeyToAlgoProvider()
    		factory.InsertPredicateKeyToAlgorithmProviderMap(predicates.PodToleratesNodeTaintsPred)
    		factory.InsertPredicateKeyToAlgorithmProviderMap(predicates.CheckNodeUnschedulablePred)
    
    		glog.Warningf("TaintNodesByCondition is enabled, PodToleratesNodeTaints predicate is mandatory")
    	}
    
    	// Prioritizes nodes that satisfy pod's resource limits
    	if utilfeature.DefaultFeatureGate.Enabled(features.ResourceLimitsPriorityFunction) {
    		factory.RegisterPriorityFunction2("ResourceLimitsPriority", priorities.ResourceLimitsPriorityMap, nil, 1)
    	}
    
    }

###   2.2 使用Config记录，防止重复启动 ###

// Configz registration.
    	if cz, err := configz.New("componentconfig"); err == nil {
    		cz.Set(c.ComponentConfig)
    	} else {
    		return fmt.Errorf("unable to register configz: %s", err)
    	}

###   2.3 NewSchedulerConfig函数 ###

// Set up the configurator which can create schedulers from configs.
    	configurator := factory.NewConfigFactory(&factory.ConfigFactoryArgs{
    		SchedulerName:                  s.ComponentConfig.SchedulerName,
    		Client:                         s.Client,
    		NodeInformer:                   s.InformerFactory.Core().V1().Nodes(),
    		PodInformer:                    s.PodInformer,
    		PvInformer:                     s.InformerFactory.Core().V1().PersistentVolumes(),
    		PvcInformer:                    s.InformerFactory.Core().V1().PersistentVolumeClaims(),
    		ReplicationControllerInformer:  s.InformerFactory.Core().V1().ReplicationControllers(),
    		ReplicaSetInformer:             s.InformerFactory.Apps().V1().ReplicaSets(),
    		StatefulSetInformer:            s.InformerFactory.Apps().V1().StatefulSets(),
    		ServiceInformer:                s.InformerFactory.Core().V1().Services(),
    		PdbInformer:                    s.InformerFactory.Policy().V1beta1().PodDisruptionBudgets(),
    		StorageClassInformer:           storageClassInformer,
    		HardPodAffinitySymmetricWeight: s.ComponentConfig.HardPodAffinitySymmetricWeight,
    		EnableEquivalenceClassCache:    utilfeature.DefaultFeatureGate.Enabled(features.EnableEquivalenceClassCache),
    		DisablePreemption:              s.ComponentConfig.DisablePreemption,
    		PercentageOfNodesToScore:       s.ComponentConfig.PercentageOfNodesToScore,
    		BindTimeoutSeconds:             *s.ComponentConfig.BindTimeoutSeconds,
    	})

###   2.4 根据调度算法 ###

*  预选策略：NoVolumeZoneConflict，MaxAzureDiskVolumeCount，MaxCSIVolumeCountPred，GeneralPredicates，MaxGCEPDVolumeCount，CheckNodeUnschedulable，PodToleratesNodeTaints
 *  优选策略：NodePreferAvoidPodsPriority，NodeAffinityPriority，TaintTolerationPriority，ImageLocalityPriority，SelectorSpreadPriority，InterPodAffinityPriority，LeastRequestedPriority，BalancedResourceAllocation

source := s.ComponentConfig.AlgorithmSource
    	var config *scheduler.Config
    	switch {
    	case source.Provider != nil:
    		// Create the config from a named algorithm provider.
    		sc, err := configurator.CreateFromProvider(*source.Provider)
    		if err != nil {
    			return nil, fmt.Errorf("couldn't create scheduler using provider %q: %v", *source.Provider, err)
    		}
    		config = sc
    	case source.Policy != nil:
    		// Create the config from a user specified policy source.
    		policy := &schedulerapi.Policy{}
    		switch {
    		case source.Policy.File != nil:
    			// Use a policy serialized in a file.
    			policyFile := source.Policy.File.Path
    			_, err := os.Stat(policyFile)
    			if err != nil {
    				return nil, fmt.Errorf("missing policy config file %s", policyFile)
    			}
    			data, err := ioutil.ReadFile(policyFile)
    			if err != nil {
    				return nil, fmt.Errorf("couldn't read policy config: %v", err)
    			}
    			err = runtime.DecodeInto(latestschedulerapi.Codec, []byte(data), policy)
    			if err != nil {
    				return nil, fmt.Errorf("invalid policy: %v", err)
    			}
    		case source.Policy.ConfigMap != nil:
    			// Use a policy serialized in a config map value.
    			policyRef := source.Policy.ConfigMap
    			policyConfigMap, err := s.Client.CoreV1().ConfigMaps(policyRef.Namespace).Get(policyRef.Name, metav1.GetOptions{})
    			if err != nil {
    				return nil, fmt.Errorf("couldn't get policy config map %s/%s: %v", policyRef.Namespace, policyRef.Name, err)
    			}
    			data, found := policyConfigMap.Data[kubeschedulerconfig.SchedulerPolicyConfigMapKey]
    			if !found {
    				return nil, fmt.Errorf("missing policy config map value at key %q", kubeschedulerconfig.SchedulerPolicyConfigMapKey)
    			}
    			err = runtime.DecodeInto(latestschedulerapi.Codec, []byte(data), policy)
    			if err != nil {
    				return nil, fmt.Errorf("invalid policy: %v", err)
    			}
    		}
    		sc, err := configurator.CreateFromConfig(*policy)
    		if err != nil {
    			return nil, fmt.Errorf("couldn't create scheduler from policy: %v", err)
    		}
    		config = sc
    	default:
    		return nil, fmt.Errorf("unsupported algorithm source: %v", source)
    	}

**  2.4.1 調用CreateFromProvider函數生成scheduler.Config配置，包含预选优选函數**

algo := core.NewGenericScheduler(
    		c.schedulerCache,
    		c.equivalencePodCache,
    		c.podQueue,
    		predicateFuncs,
    		predicateMetaProducer,
    		priorityConfigs,
    		priorityMetaProducer,
    		extenders,
    		c.volumeBinder,
    		c.pVCLister,
    		c.alwaysCheckAllPredicates,
    		c.disablePreemption,
    		c.percentageOfNodesToScore,
    	)
    
    	podBackoff := util.CreateDefaultPodBackoff()
    	return &scheduler.Config{
    		SchedulerCache: c.schedulerCache,
    		Ecache:         c.equivalencePodCache,
    		// The scheduler only needs to consider schedulable nodes.
    		NodeLister:          &nodeLister{c.nodeLister},
    		Algorithm:           algo,
    		GetBinder:           c.getBinderFunc(extenders),
    		PodConditionUpdater: &podConditionUpdater{c.client},
    		PodPreemptor:        &podPreemptor{c.client},
    		WaitForCacheSync: func() bool {
    			return cache.WaitForCacheSync(c.StopEverything, c.scheduledPodsHasSynced)
    		},
    		NextPod: func() *v1.Pod {
    			return c.getNextPod()
    		},
    		Error:          c.MakeDefaultErrorFunc(podBackoff, c.podQueue),
    		StopEverything: c.StopEverything,
    		VolumeBinder:   c.volumeBinder,
    	}, nil

###   2.5 調用NewFronConfig生成Scheduler配置，將schedulerConfig封裝了一下，註冊metrics ###

// Create the scheduler.
    	sched := scheduler.NewFromConfig(schedulerConfig)

###   2.6 Start up the healthz server ###

启动healthz和metrics http服务

// Start up the healthz server.
    	if c.InsecureServing != nil {
    		separateMetrics := c.InsecureMetricsServing != nil
    		handler := buildHandlerChain(newHealthzHandler(&c.ComponentConfig, separateMetrics), nil, nil)
    		if err := c.InsecureServing.Serve(handler, 0, stopCh); err != nil {
    			return fmt.Errorf("failed to start healthz server: %v", err)
    		}
    	}
    	if c.InsecureMetricsServing != nil {
    		handler := buildHandlerChain(newMetricsHandler(&c.ComponentConfig), nil, nil)
    		if err := c.InsecureMetricsServing.Serve(handler, 0, stopCh); err != nil {
    			return fmt.Errorf("failed to start metrics server: %v", err)
    		}
    	}
    	if c.SecureServing != nil {
    		handler := buildHandlerChain(newHealthzHandler(&c.ComponentConfig, false), c.Authentication.Authenticator, c.Authorization.Authorizer)
    		if err := c.SecureServing.Serve(handler, 0, stopCh); err != nil {
    			// fail early for secure handlers, removing the old error loop from above
    			return fmt.Errorf("failed to start healthz server: %v", err)
    		}
    	}

###   2.7 Start all informers ###

Informer().Run函数创建队列，设置lister watcher pod变化

等待同步到cache中，调用sched.Run函数

// Start all informers.
    	go c.PodInformer.Informer().Run(stopCh)
    	c.InformerFactory.Start(stopCh)
    
    	// Wait for all caches to sync before scheduling.
    	c.InformerFactory.WaitForCacheSync(stopCh)
    	controller.WaitForCacheSync("scheduler", stopCh, c.PodInformer.Informer().HasSynced)
    
    	// Prepare a reusable run function.
    	run := func(ctx context.Context) {
    		sched.Run()
    		<-ctx.Done()
    	}
    
    	ctx, cancel := context.WithCancel(context.TODO()) // TODO once Run() accepts a context, it should be used here
    	defer cancel()
    
    	go func() {
    		select {
    		case <-stopCh:
    			cancel()
    		case <-ctx.Done():
    		}
    	}()

**  2.7.1 Run函数**

主要逻辑函数为 scheduleOne 路径 plugin/pkg/scheduler/scheduler.go， 启动 goroutine，循环执行 scheduleOne 方法，直到收到 shut down 的信号

scheduleOne 每次选择一个 pod 进行处理，使用 scheduler 函数进行预选（predicate）和优选 (priority)，选择一个合适的 host，将 pod 和 host 进行绑定关联，第四章节讲解

// 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)
    }

###   2.8 如果设置--leader-elect为true ###

// If leader election is enabled, run via LeaderElector until done and exit.
    	if c.LeaderElection != nil {
    		c.LeaderElection.Callbacks = leaderelection.LeaderCallbacks{
    			OnStartedLeading: run,
    			OnStoppedLeading: func() {
    				utilruntime.HandleError(fmt.Errorf("lost master"))
    			},
    		}
    		leaderElector, err := leaderelection.NewLeaderElector(*c.LeaderElection)
    		if err != nil {
    			return fmt.Errorf("couldn't create leader elector: %v", err)
    		}
    
    		leaderElector.Run(ctx)
    
    		return fmt.Errorf("lost lease")
    	}

# 四. scheduleOne函数 #

scheduleOne 每次选择一个 pod 进行处理，使用 scheduler 函数进行预选（predicate）和优选 (priority)，选择一个合适的 host，将 pod 和 host 进行绑定关联

###   4.1 从队列中取出一个pod ###

如果没有pod则阻塞，如果已经标记为删除则记录事件

pod := sched.config.NextPod()
    	if pod.DeletionTimestamp != nil {
    		sched.config.Recorder.Eventf(pod, v1.EventTypeWarning, "FailedScheduling", "skip schedule deleting pod: %v/%v", pod.Namespace, pod.Name)
    		glog.V(3).Infof("Skip schedule deleting pod: %v/%v", pod.Namespace, pod.Name)
    		return
    	}

###   4.2 sched.schedule(pod)函数 ###

*  `  根据调度算法执行预选和优选策略，选出最合适pod部署node`
 *  `  如果未自定义调度器，默认调度器genericScheduler实现路径为pkg/scheduler/core/generic_scheduler.go（第五章节讲解）`

// schedule implements the scheduling algorithm and returns the suggested host.
    func (sched *Scheduler) schedule(pod *v1.Pod) (string, error) {
    	host, err := sched.config.Algorithm.Schedule(pod, sched.config.NodeLister)
    	if err != nil {
    		pod = pod.DeepCopy()
    		sched.config.Error(pod, err)
    		sched.config.Recorder.Eventf(pod, v1.EventTypeWarning, "FailedScheduling", "%v", err)
    		sched.config.PodConditionUpdater.Update(pod, &v1.PodCondition{
    			Type:    v1.PodScheduled,
    			Status:  v1.ConditionFalse,
    			Reason:  v1.PodReasonUnschedulable,
    			Message: err.Error(),
    		})
    		return "", err
    	}
    	return host, err
    }

如果调度失败则调用sched.preempt进行处理

if err != nil {
    		// schedule() may have failed because the pod would not fit on any host, so we try to
    		// preempt, with the expectation that the next time the pod is tried for scheduling it
    		// will fit due to the preemption. It is also possible that a different pod will schedule
    		// into the resources that were preempted, but this is harmless.
    		if fitError, ok := err.(*core.FitError); ok {
    			preemptionStartTime := time.Now()
    			sched.preempt(pod, fitError)
    			metrics.PreemptionAttempts.Inc()
    			metrics.SchedulingAlgorithmPremptionEvaluationDuration.Observe(metrics.SinceInMicroseconds(preemptionStartTime))
    			metrics.SchedulingLatency.WithLabelValues(metrics.PreemptionEvaluation).Observe(metrics.SinceInSeconds(preemptionStartTime))
    		}
    		return
    	}

**  4.2.1 preempt函数**

*  检查PodPriority是否开启，如果未开启，直接返回（PodPriority=true|false (BETA - default=true)）
 *  GetUpdatedPod会更新PodCondition，记录调度失败状态及失败原因
 *  sched.config.Algorithm.Preempt选出要执行优先调度的node以及node上要删除的pod

// preempt tries to create room for a pod that has failed to schedule, by preempting lower priority pods if possible.
    // If it succeeds, it adds the name of the node where preemption has happened to the pod annotations.
    // It returns the node name and an error if any.
    func (sched *Scheduler) preempt(preemptor *v1.Pod, scheduleErr error) (string, error) {
    	if !util.PodPriorityEnabled() || sched.config.DisablePreemption {
    		glog.V(3).Infof("Pod priority feature is not enabled or preemption is disabled by scheduler configuration." +
    			" No preemption is performed.")
    		return "", nil
    	}
    	preemptor, err := sched.config.PodPreemptor.GetUpdatedPod(preemptor)
    	if err != nil {
    		glog.Errorf("Error getting the updated preemptor pod object: %v", err)
    		return "", err
    	}
    
    	node, victims, nominatedPodsToClear, err := sched.config.Algorithm.Preempt(preemptor, sched.config.NodeLister, scheduleErr)
    	metrics.PreemptionVictims.Set(float64(len(victims)))
    	if err != nil {
    		glog.Errorf("Error preempting victims to make room for %v/%v.", preemptor.Namespace, preemptor.Name)
    		return "", err
    	}
    	var nodeName = ""
    	if node != nil {
    		nodeName = node.Name
    		err = sched.config.PodPreemptor.SetNominatedNodeName(preemptor, nodeName)
    		if err != nil {
    			glog.Errorf("Error in preemption process. Cannot update pod %v/%v annotations: %v", preemptor.Namespace, preemptor.Name, err)
    			return "", err
    		}
    		for _, victim := range victims {
    			if err := sched.config.PodPreemptor.DeletePod(victim); err != nil {
    				glog.Errorf("Error preempting pod %v/%v: %v", victim.Namespace, victim.Name, err)
    				return "", err
    			}
    			sched.config.Recorder.Eventf(victim, v1.EventTypeNormal, "Preempted", "by %v/%v on node %v", preemptor.Namespace, preemptor.Name, nodeName)
    		}
    	}
    	// Clearing nominated pods should happen outside of "if node != nil". Node could
    	// be nil when a pod with nominated node name is eligible to preempt again,
    	// but preemption logic does not find any node for it. In that case Preempt()
    	// function of generic_scheduler.go returns the pod itself for removal of the annotation.
    	for _, p := range nominatedPodsToClear {
    		rErr := sched.config.PodPreemptor.RemoveNominatedNodeName(p)
    		if rErr != nil {
    			glog.Errorf("Cannot remove nominated node annotation of pod: %v", rErr)
    			// We do not return as this error is not critical.
    		}
    	}
    	return nodeName, err
    }

###   4.3 sched.bind函数 ###

调用sched.bind绑定host

// bind the pod to its host asynchronously (we can do this b/c of the assumption step above).
    	go func() {
    		// Bind volumes first before Pod
    		if !allBound {
    			err = sched.bindVolumes(assumedPod)
    			if err != nil {
    				return
    			}
    		}
    
    		err := sched.bind(assumedPod, &v1.Binding{
    			ObjectMeta: metav1.ObjectMeta{Namespace: assumedPod.Namespace, Name: assumedPod.Name, UID: assumedPod.UID},
    			Target: v1.ObjectReference{
    				Kind: "Node",
    				Name: suggestedHost,
    			},
    		})
    		metrics.E2eSchedulingLatency.Observe(metrics.SinceInMicroseconds(start))
    		if err != nil {
    			glog.Errorf("Internal error binding pod: (%v)", err)
    		}
    	}()

# **5 Scheduler函数** #

如果未自定义调度器，默认调度器genericScheduler实现路径为pkg/scheduler/core/generic\_scheduler.go

*  findNodesThatFit：根据所有预选算法过滤符合的 node 列表
 *  PrioritizeNodes: 对符合的节点进行优选评分，一个排序的列表
 *  selectHost: 对优选的 node 列表选择一个最优的节点

###   5.1 podPassesBasicChecks函数主要关注pod的pvc信息 ###

// podPassesBasicChecks makes sanity checks on the pod if it can be scheduled.
    func podPassesBasicChecks(pod *v1.Pod, pvcLister corelisters.PersistentVolumeClaimLister) error {
    	// Check PVCs used by the pod
    	namespace := pod.Namespace
    	manifest := &(pod.Spec)
    	for i := range manifest.Volumes {
    		volume := &manifest.Volumes[i]
    		if volume.PersistentVolumeClaim == nil {
    			// Volume is not a PVC, ignore
    			continue
    		}
    		pvcName := volume.PersistentVolumeClaim.ClaimName
    		pvc, err := pvcLister.PersistentVolumeClaims(namespace).Get(pvcName)
    		if err != nil {
    			// The error has already enough context ("persistentvolumeclaim "myclaim" not found")
    			return err
    		}
    
    		if pvc.DeletionTimestamp != nil {
    			return fmt.Errorf("persistentvolumeclaim %q is being deleted", pvc.Name)
    		}
    	}
    
    	return nil
    }

###   5.2 findNodesThatFit函数根据所有预选算法过滤不符合的 node ###

**    5.2.1 调用podFitsOnNode函数是否node可以满足pod调度**

最多开启16个goroutine调用checkNode，会调用podFitsOnNode对该Node进行所有预选策略检查，包括的策略如下：

>     checkNodeConditionPred, CheckNodeUnschedulablePred,
>     GeneralPred, HostNamePred, PodFitsHostPortsPred,
>     MatchNodeSelectorPred, PodFitsResourcesPred, NoDiskConflictPred,
>     PodToleratesNodeTaintsPred, PodToleratesNodeNoExecuteTaintsPred, 
>     CheckNodeLabelPresencePred,CheckServiceAffinityPred, MaxEBSVolumeCountPred, MaxGCEPDVolumeCountPred, MaxCSIVolumeCountPred,MaxAzureDiskVolumeCountPred, CheckVolumeBindingPred, NoVolumeZoneConflictPred,CheckNodeMemoryPressurePred, CheckNodePIDPressurePred, CheckNodeDiskPressurePred, MatchInterPodAffinityPred

if len(g.predicates) == 0 {
    		filtered = nodes
    	} else {
    		allNodes := int32(g.cache.NodeTree().NumNodes)
    		numNodesToFind := g.numFeasibleNodesToFind(allNodes)
    
    		// Create filtered list with enough space to avoid growing it
    		// and allow assigning.
    		filtered = make([]*v1.Node, numNodesToFind)
    		errs := errors.MessageCountMap{}
    		var (
    			predicateResultLock sync.Mutex
    			filteredLen         int32
    			equivClass          *equivalence.Class
    		)
    
    		ctx, cancel := context.WithCancel(context.Background())
    
    		// We can use the same metadata producer for all nodes.
    		meta := g.predicateMetaProducer(pod, g.cachedNodeInfoMap)
    
    		if g.equivalenceCache != nil {
    			// getEquivalenceClassInfo will return immediately if no equivalence pod found
    			equivClass = equivalence.NewClass(pod)
    		}
    
    		checkNode := func(i int) {
    			var nodeCache *equivalence.NodeCache
    			nodeName := g.cache.NodeTree().Next()
    			if g.equivalenceCache != nil {
    				nodeCache, _ = g.equivalenceCache.GetNodeCache(nodeName)
    			}
    			fits, failedPredicates, err := podFitsOnNode(
    				pod,
    				meta,
    				g.cachedNodeInfoMap[nodeName],
    				g.predicates,
    				g.cache,
    				nodeCache,
    				g.schedulingQueue,
    				g.alwaysCheckAllPredicates,
    				equivClass,
    			)

###   5.3 PrioritizeNodes函数 ###

**   5.3.1 异步执行所有优选策略，default策略有：**

<table> 
 <tbody> 
  <tr> 
   <td>优选算法</td> 
   <td>Weight</td> 
   <td>map</td> 
   <td>reduce</td> 
   <td>Function</td> 
  </tr> 
  <tr> 
   <td>BalancedResourceAllocation</td> 
   <td>1</td> 
   <td> <pre><code>balancedResourcePriority.PriorityMap</code></pre> </td> 
   <td>&nbsp;</td> 
   <td>nil</td> 
  </tr> 
  <tr> 
   <td>ImageLocalityPriority</td> 
   <td>1</td> 
   <td> <pre><code>ImageLocalityPriorityMap</code></pre> </td> 
   <td>&nbsp;</td> 
   <td>nil</td> 
  </tr> 
  <tr> 
   <td>InterPodAffinityPriority</td> 
   <td>1</td> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
   <td>nil</td> 
  </tr> 
  <tr> 
   <td>NodeAffinityPriority</td> 
   <td>1</td> 
   <td> <pre><code>CalculateNodeAffinityPriorityMap</code></pre> </td> 
   <td>&nbsp;</td> 
   <td>nil</td> 
  </tr> 
  <tr> 
   <td>NodePreferAvoidPodsPriority</td> 
   <td>10000</td> 
   <td> <pre><code>CalculateNodePreferAvoidPodsPriorityMap</code></pre> </td> 
   <td>&nbsp;</td> 
   <td>nil</td> 
  </tr> 
  <tr> 
   <td>SelectorSpreadPriority</td> 
   <td>1</td> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
   <td>nil</td> 
  </tr> 
  <tr> 
   <td>TaintTolerationPriority</td> 
   <td>1</td> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
   <td>nil</td> 
  </tr> 
 </tbody> 
</table>

priorityConfig.Function为nil，没有执行这个函数

results := make([]schedulerapi.HostPriorityList, len(priorityConfigs), len(priorityConfigs))
    
    	for i, priorityConfig := range priorityConfigs {
    		if priorityConfig.Function != nil {
    			// DEPRECATED
    			wg.Add(1)
    			go func(index int, config algorithm.PriorityConfig) {
    				defer wg.Done()
    				var err error
    				results[index], err = config.Function(pod, nodeNameToInfo, nodes)
    				if err != nil {
    					appendError(err)
    				}
    			}(i, priorityConfig)
    		} else {
    			results[i] = make(schedulerapi.HostPriorityList, len(nodes))
    		}
    	}

**   5.3.2 使用二维数组记录分数**

调用Map函数，可以参照第八章节结算的优选算法记录分数

processNode := func(index int) {
    		nodeInfo := nodeNameToInfo[nodes[index].Name]
    		var err error
    		for i := range priorityConfigs {
    			if priorityConfigs[i].Function != nil {
    				continue
    			}
    			results[i][index], err = priorityConfigs[i].Map(pod, meta, nodeInfo)
    			if err != nil {
    				appendError(err)
    				results[i][index].Host = nodes[index].Name
    			}
    		}
    	}
    	workqueue.Parallelize(16, len(nodes), processNode)
    	for i, priorityConfig := range priorityConfigs {
    		if priorityConfig.Reduce == nil {
    			continue
    		}
    		wg.Add(1)
    		go func(index int, config algorithm.PriorityConfig) {
    			defer wg.Done()
    			if err := config.Reduce(pod, meta, nodeNameToInfo, results[index]); err != nil {
    				appendError(err)
    			}
    			if glog.V(10) {
    				for _, hostPriority := range results[index] {
    					glog.Infof("zzlin %v -> %v: %v, Score: (%d)", pod.Name, hostPriority.Host, config.Name, hostPriority.Score)
    				}
    			}
    		}(i, priorityConfig)
    	}
    	// Wait for all computations to be finished.
    	wg.Wait()
    	if len(errs) != 0 {
    		return schedulerapi.HostPriorityList{}, errors.NewAggregate(errs)
    	}

**   5.3.3 统计分数，按照每一项的分数乘上权重，所有相加的结果**

for i := range nodes {
    		result = append(result, schedulerapi.HostPriority{Host: nodes[i].Name, Score: 0})
    		for j := range priorityConfigs {
    			result[i].Score += results[j][i].Score * priorityConfigs[j].Weight
    		}
    	}

###   5.4 selectHost函数 ###

对优选的 node 列表选择一个最优的节点

// selectHost takes a prioritized list of nodes and then picks one
    // in a round-robin manner from the nodes that had the highest score.
    func (g *genericScheduler) selectHost(priorityList schedulerapi.HostPriorityList) (string, error) {
    	if len(priorityList) == 0 {
    		return "", fmt.Errorf("empty priorityList")
    	}
    
    	maxScores := findMaxScores(priorityList)
    	ix := int(g.lastNodeIndex % uint64(len(maxScores)))
    	g.lastNodeIndex++
    
    	return priorityList[maxScores[ix]].Host, nil
    }

# **六. 算法初始化** #

###   全局变量如下所示： ###

路徑pkg/scheduler/factory/plugins.go

var (
           schedulerFactoryMutex sync.Mutex
    
           // maps that hold registered algorithm types
           fitPredicateMap          = make(map[string]FitPredicateFactory)
           mandatoryFitPredicateMap = make(map[string]FitPredicateFactory)
           priorityFunctionMap      = make(map[string]PriorityConfigFactory)
           algorithmProviderMap     = make(map[string]AlgorithmProviderConfig)
    
           // Registered metadata producers
           priorityMetadataProducer  MetadataProducerFactory
           predicateMetadataProducer MetadataProducerFactory
    
           // get equivalence pod function
           getEquivalencePodFunc algorithm.GetEquivalencePodFunc
    )

###   6.1 init函数 ###

路径 plugin/pkg/scheduler/algorighmprovider/defaults.go，启动会注册一大堆 defalutProvidor，主要是初始化 plugin/pkg/scheduler/factory/plugins.go 全局变量

func init() {
    	// Register functions that extract metadata used by predicates and priorities computations.
    	factory.RegisterPredicateMetadataProducerFactory(
    		func(args factory.PluginFactoryArgs) algorithm.PredicateMetadataProducer {
    			return predicates.NewPredicateMetadataFactory(args.PodLister)
    		})
    	factory.RegisterPriorityMetadataProducerFactory(
    		func(args factory.PluginFactoryArgs) algorithm.PriorityMetadataProducer {
    			return priorities.NewPriorityMetadataFactory(args.ServiceLister, args.ControllerLister, args.ReplicaSetLister, args.StatefulSetLister)
    		})
    
    	registerAlgorithmProvider(defaultPredicates(), defaultPriorities())
    
    	// IMPORTANT NOTES for predicate developers:
    	// We are using cached predicate result for pods belonging to the same equivalence class.
    	// So when implementing a new predicate, you are expected to check whether the result
    	// of your predicate function can be affected by related API object change (ADD/DELETE/UPDATE).
    	// If yes, you are expected to invalidate the cached predicate result for related API object change.
    	// For example:
    	// https://github.com/kubernetes/kubernetes/blob/36a218e/plugin/pkg/scheduler/factory/factory.go#L422
    
    	// Registers predicates and priorities that are not enabled by default, but user can pick when creating their
    	// own set of priorities/predicates.
    
    	// PodFitsPorts has been replaced by PodFitsHostPorts for better user understanding.
    	// For backwards compatibility with 1.0, PodFitsPorts is registered as well.
    	factory.RegisterFitPredicate("PodFitsPorts", predicates.PodFitsHostPorts)
    	// Fit is defined based on the absence of port conflicts.
    	// This predicate is actually a default predicate, because it is invoked from
    	// predicates.GeneralPredicates()
    	factory.RegisterFitPredicate(predicates.PodFitsHostPortsPred, predicates.PodFitsHostPorts)
    	// Fit is determined by resource availability.
    	// This predicate is actually a default predicate, because it is invoked from
    	// predicates.GeneralPredicates()
    	factory.RegisterFitPredicate(predicates.PodFitsResourcesPred, predicates.PodFitsResources)
    	// Fit is determined by the presence of the Host parameter and a string match
    	// This predicate is actually a default predicate, because it is invoked from
    	// predicates.GeneralPredicates()
    	factory.RegisterFitPredicate(predicates.HostNamePred, predicates.PodFitsHost)
    	// Fit is determined by node selector query.
    	factory.RegisterFitPredicate(predicates.MatchNodeSelectorPred, predicates.PodMatchNodeSelector)
    
    	// ServiceSpreadingPriority is a priority config factory that spreads pods by minimizing
    	// the number of pods (belonging to the same service) on the same node.
    	// Register the factory so that it's available, but do not include it as part of the default priorities
    	// Largely replaced by "SelectorSpreadPriority", but registered for backward compatibility with 1.0
    	factory.RegisterPriorityConfigFactory(
    		"ServiceSpreadingPriority",
    		factory.PriorityConfigFactory{
    			MapReduceFunction: func(args factory.PluginFactoryArgs) (algorithm.PriorityMapFunction, algorithm.PriorityReduceFunction) {
    				return priorities.NewSelectorSpreadPriority(args.ServiceLister, algorithm.EmptyControllerLister{}, algorithm.EmptyReplicaSetLister{}, algorithm.EmptyStatefulSetLister{})
    			},
    			Weight: 1,
    		},
    	)
    	// EqualPriority is a prioritizer function that gives an equal weight of one to all nodes
    	// Register the priority function so that its available
    	// but do not include it as part of the default priorities
    	factory.RegisterPriorityFunction2("EqualPriority", core.EqualPriorityMap, nil, 1)
    	// Optional, cluster-autoscaler friendly priority function - give used nodes higher priority.
    	factory.RegisterPriorityFunction2("MostRequestedPriority", priorities.MostRequestedPriorityMap, nil, 1)
    	factory.RegisterPriorityFunction2(
    		"RequestedToCapacityRatioPriority",
    		priorities.RequestedToCapacityRatioResourceAllocationPriorityDefault().PriorityMap,
    		nil,
    		1)
    }

###   6.2 defaultPredicates函数 ###

调用RegisterFitPredicateFactory注册预选策略，存在全局变量fitPredicateMap

包括预选策略有：

>  *  NoVolumeZoneConflict：defines the name of predicate NoVolumeZoneConflict
>  *  MaxEBSVolumeCount：defines the name of predicate MaxEBSVolumeCount
>  *  MaxGCEPDVolumeCount: defines the name of predicate MaxGCEPDVolumeCount
>  *  MaxAzureDiskVolumeCount: defines the name of predicate MaxAzureDiskVolumeCount
>  *  MaxCSIVolumeCountPred: defines the predicate that decides how many CSI volumes should be attached
>  *  MatchInterPodAffinity: defines the name of predicate MatchInterPodAffinity
>  *  NoDiskConflict: defines the name of predicate NoDiskConflict
>  *  GeneralPredicates: defines the name of predicate GeneralPredicates
>  *  CheckNodeMemoryPressure: defines the name of predicate CheckNodeMemoryPressure
>  *  CheckNodeDiskPressure: defines the name of predicate CheckNodeDiskPressure
>  *  CheckNodePIDPressure: defines the name of predicate CheckNodePIDPressure
>  *  CheckNodeCondition: defines the name of predicate CheckNodeCondition
>  *  PodToleratesNodeTaints: defines the name of predicate PodToleratesNodeTaints
>  *  CheckVolumeBinding: defines the name of predicate CheckVolumeBinding

func defaultPredicates() sets.String {
    	return sets.NewString(
    		// Fit is determined by volume zone requirements.
    		factory.RegisterFitPredicateFactory(
    			predicates.NoVolumeZoneConflictPred,
    			func(args factory.PluginFactoryArgs) algorithm.FitPredicate {
    				return predicates.NewVolumeZonePredicate(args.PVInfo, args.PVCInfo, args.StorageClassInfo)
    			},
    		),
    		// Fit is determined by whether or not there would be too many AWS EBS volumes attached to the node
    		factory.RegisterFitPredicateFactory(
    			predicates.MaxEBSVolumeCountPred,
    			func(args factory.PluginFactoryArgs) algorithm.FitPredicate {
    				return predicates.NewMaxPDVolumeCountPredicate(predicates.EBSVolumeFilterType, args.PVInfo, args.PVCInfo)
    			},
    		),
    		// Fit is determined by whether or not there would be too many GCE PD volumes attached to the node
    		factory.RegisterFitPredicateFactory(
    			predicates.MaxGCEPDVolumeCountPred,
    			func(args factory.PluginFactoryArgs) algorithm.FitPredicate {
    				return predicates.NewMaxPDVolumeCountPredicate(predicates.GCEPDVolumeFilterType, args.PVInfo, args.PVCInfo)
    			},
    		),
    		// Fit is determined by whether or not there would be too many Azure Disk volumes attached to the node
    		factory.RegisterFitPredicateFactory(
    			predicates.MaxAzureDiskVolumeCountPred,
    			func(args factory.PluginFactoryArgs) algorithm.FitPredicate {
    				return predicates.NewMaxPDVolumeCountPredicate(predicates.AzureDiskVolumeFilterType, args.PVInfo, args.PVCInfo)
    			},
    		),
    		factory.RegisterFitPredicateFactory(
    			predicates.MaxCSIVolumeCountPred,
    			func(args factory.PluginFactoryArgs) algorithm.FitPredicate {
    				return predicates.NewCSIMaxVolumeLimitPredicate(args.PVInfo, args.PVCInfo)
    			},
    		),
    		// Fit is determined by inter-pod affinity.
    		factory.RegisterFitPredicateFactory(
    			predicates.MatchInterPodAffinityPred,
    			func(args factory.PluginFactoryArgs) algorithm.FitPredicate {
    				return predicates.NewPodAffinityPredicate(args.NodeInfo, args.PodLister)
    			},
    		),
    
    		// Fit is determined by non-conflicting disk volumes.
    		factory.RegisterFitPredicate(predicates.NoDiskConflictPred, predicates.NoDiskConflict),
    
    		// GeneralPredicates are the predicates that are enforced by all Kubernetes components
    		// (e.g. kubelet and all schedulers)
    		factory.RegisterFitPredicate(predicates.GeneralPred, predicates.GeneralPredicates),
    
    		// Fit is determined by node memory pressure condition.
    		factory.RegisterFitPredicate(predicates.CheckNodeMemoryPressurePred, predicates.CheckNodeMemoryPressurePredicate),
    
    		// Fit is determined by node disk pressure condition.
    		factory.RegisterFitPredicate(predicates.CheckNodeDiskPressurePred, predicates.CheckNodeDiskPressurePredicate),
    
    		// Fit is determined by node pid pressure condition.
    		factory.RegisterFitPredicate(predicates.CheckNodePIDPressurePred, predicates.CheckNodePIDPressurePredicate),
    
    		// Fit is determined by node conditions: not ready, network unavailable or out of disk.
    		factory.RegisterMandatoryFitPredicate(predicates.CheckNodeConditionPred, predicates.CheckNodeConditionPredicate),
    
    		// Fit is determined based on whether a pod can tolerate all of the node's taints
    		factory.RegisterFitPredicate(predicates.PodToleratesNodeTaintsPred, predicates.PodToleratesNodeTaints),
    
    		// Fit is determined by volume topology requirements.
    		factory.RegisterFitPredicateFactory(
    			predicates.CheckVolumeBindingPred,
    			func(args factory.PluginFactoryArgs) algorithm.FitPredicate {
    				return predicates.NewVolumeBindingPredicate(args.VolumeBinder)
    			},
    		),
    	)
    }

###   6.3 defaultPriorities函数 ###

调用RegisterPriorityConfigFactory注册优选策略，存入全局变量priorityFunctionMap

优选策略包括如下：

>  *      SelectorSpreadPriority: spreads pods by minimizing the number of pods (belonging to the same service or replication controller) on the same node
>  *      InterPodAffinityPriority: pods should be placed in the same topological domain (e.g. same node, same rack, same zone, same power domain, etc.)  as some other pods, or, conversely, should not be placed in the same topological domain as some other pods.
>  *      LeastRequestedPriority: Prioritize nodes by least requested utilization
>  *      BalancedResourceAllocation: Prioritizes nodes to help achieve balanced resource usage
>  *      NodePreferAvoidPodsPriority: Set this weight large enough to override all other priority functions
>  *      NodeAffinityPriority: Prioritizes nodes that have labels matching NodeAffinity
>  *      TaintTolerationPriority: Prioritizes nodes that marked with taint which pod can tolerate
>  *      ImageLocalityPriority: ImageLocalityPriority prioritizes nodes that have images requested by the pod present

func defaultPriorities() sets.String {
    	return sets.NewString(
    		// spreads pods by minimizing the number of pods (belonging to the same service or replication controller) on the same node.
    		factory.RegisterPriorityConfigFactory(
    			"SelectorSpreadPriority",
    			factory.PriorityConfigFactory{
    				MapReduceFunction: func(args factory.PluginFactoryArgs) (algorithm.PriorityMapFunction, algorithm.PriorityReduceFunction) {
    					return priorities.NewSelectorSpreadPriority(args.ServiceLister, args.ControllerLister, args.ReplicaSetLister, args.StatefulSetLister)
    				},
    				Weight: 1,
    			},
    		),
    		// pods should be placed in the same topological domain (e.g. same node, same rack, same zone, same power domain, etc.)
    		// as some other pods, or, conversely, should not be placed in the same topological domain as some other pods.
    		factory.RegisterPriorityConfigFactory(
    			"InterPodAffinityPriority",
    			factory.PriorityConfigFactory{
    				Function: func(args factory.PluginFactoryArgs) algorithm.PriorityFunction {
    					return priorities.NewInterPodAffinityPriority(args.NodeInfo, args.NodeLister, args.PodLister, args.HardPodAffinitySymmetricWeight)
    				},
    				Weight: 1,
    			},
    		),
    
    		// Prioritize nodes by least requested utilization.
    		factory.RegisterPriorityFunction2("LeastRequestedPriority", priorities.LeastRequestedPriorityMap, nil, 1),
    
    		// Prioritizes nodes to help achieve balanced resource usage
    		factory.RegisterPriorityFunction2("BalancedResourceAllocation", priorities.BalancedResourceAllocationMap, nil, 1),
    
    		// Set this weight large enough to override all other priority functions.
    		// TODO: Figure out a better way to do this, maybe at same time as fixing #24720.
    		factory.RegisterPriorityFunction2("NodePreferAvoidPodsPriority", priorities.CalculateNodePreferAvoidPodsPriorityMap, nil, 10000),
    
    		// Prioritizes nodes that have labels matching NodeAffinity
    		factory.RegisterPriorityFunction2("NodeAffinityPriority", priorities.CalculateNodeAffinityPriorityMap, priorities.CalculateNodeAffinityPriorityReduce, 1),
    
    		// Prioritizes nodes that marked with taint which pod can tolerate.
    		factory.RegisterPriorityFunction2("TaintTolerationPriority", priorities.ComputeTaintTolerationPriorityMap, priorities.ComputeTaintTolerationPriorityReduce, 1),
    
    		// ImageLocalityPriority prioritizes nodes that have images requested by the pod present.
    		factory.RegisterPriorityFunction2("ImageLocalityPriority", priorities.ImageLocalityPriorityMap, nil, 1),
    	)
    }

# 七. 预选算法－predicate #

路径 plugin/pkg/scheduler/algorithm/predicates/predicates.go，选出符合预选策略的 node 列表

*  **PodFitsHostPorts:** 检查节点的端口可以分配给 pod 使用
 *  **PodFitsResources:** 检查节点是否有足够的资源，CPU，内存，GPU，storage
 *  **NoDiskConflict:** 检查节点卷是否冲突，是否已经被挂载，如果已经被其他使用则不能调度到该节点，可以看具体解释，包括 GCE，AWS，Ceph，ISCSI 说明
 *  **PodMatchNodeSelector:**  检查 pod 的选择器是否匹配到节点 label
 *  **CheckNodeLabelPresence:**
 *  **InterPodAffinityMatches:**
 *  **CheckNodeMemoryPressurePredicate**
 *  **CheckNodeConditionPredicate**

###   7.1 **CheckNodeConditionPredicate** ###

**    根据node状态进行过略**

// CheckNodeConditionPredicate checks if a pod can be scheduled on a node reporting out of disk,
    // network unavailable and not ready condition. Only node conditions are accounted in this predicate.
    func CheckNodeConditionPredicate(pod *v1.Pod, meta algorithm.PredicateMetadata, nodeInfo *schedulercache.NodeInfo) (bool, []algorithm.PredicateFailureReason, error) {
    	reasons := []algorithm.PredicateFailureReason{}
    
    	if nodeInfo == nil || nodeInfo.Node() == nil {
    		return false, []algorithm.PredicateFailureReason{ErrNodeUnknownCondition}, nil
    	}
    
    	node := nodeInfo.Node()
    	for _, cond := range node.Status.Conditions {
    		// We consider the node for scheduling only when its:
    		// - NodeReady condition status is ConditionTrue,
    		// - NodeOutOfDisk condition status is ConditionFalse,
    		// - NodeNetworkUnavailable condition status is ConditionFalse.
    		if cond.Type == v1.NodeReady && cond.Status != v1.ConditionTrue {
    			reasons = append(reasons, ErrNodeNotReady)
    		} else if cond.Type == v1.NodeOutOfDisk && cond.Status != v1.ConditionFalse {
    			reasons = append(reasons, ErrNodeOutOfDisk)
    		} else if cond.Type == v1.NodeNetworkUnavailable && cond.Status != v1.ConditionFalse {
    			reasons = append(reasons, ErrNodeNetworkUnavailable)
    		}
    	}
    
    	if node.Spec.Unschedulable {
    		reasons = append(reasons, ErrNodeUnschedulable)
    	}
    
    	return len(reasons) == 0, reasons, nil
    }

# **八. 优选算法－predicate** #

路径 plugin/pkg/scheduler/algorithm/priorities/priorities.go，结果为一个二维数组，含有节点与各项 优选算法的分数（0-10 之间）

###   8.1 BalancedResourceAllocation ###

注册函数scorer为balancedResourceScorer，资源使用均衡分数最高，计算公式 score = 10 - abs(cpuFraction-memoryFraction)\*10

var (
    	balancedResourcePriority = &ResourceAllocationPriority{"BalancedResourceAllocation", balancedResourceScorer}
    
    	// BalancedResourceAllocationMap favors nodes with balanced resource usage rate.
    	// BalancedResourceAllocationMap should **NOT** be used alone, and **MUST** be used together
    	// with LeastRequestedPriority. It calculates the difference between the cpu and memory fraction
    	// of capacity, and prioritizes the host based on how close the two metrics are to each other.
    	// Detail: score = 10 - variance(cpuFraction,memoryFraction,volumeFraction)*10. The algorithm is partly inspired by:
    	// "Wei Huang et al. An Energy Efficient Virtual Machine Placement Algorithm with Balanced
    	// Resource Utilization"
    	BalancedResourceAllocationMap = balancedResourcePriority.PriorityMap
    )
    
    // PriorityMap priorities nodes according to the resource allocations on the node.
    // It will use `scorer` function to calculate the score.
    func (r *ResourceAllocationPriority) PriorityMap(
    	pod *v1.Pod,
    	meta interface{},
    	nodeInfo *schedulercache.NodeInfo) (schedulerapi.HostPriority, error) {
    	node := nodeInfo.Node()
    	if node == nil {
    		return schedulerapi.HostPriority{}, fmt.Errorf("node not found")
    	}
    	allocatable := nodeInfo.AllocatableResource()
    
    	glog.Infof("zzlin ResourceAllocationPriority PriorityMap allocatable: %#v", allocatable)
    
    	var requested schedulercache.Resource
    	if priorityMeta, ok := meta.(*priorityMetadata); ok {
    		requested = *priorityMeta.nonZeroRequest
    	} else {
    		// We couldn't parse metadata - fallback to computing it.
    		requested = *getNonZeroRequests(pod)
    	}
    
    	requested.MilliCPU += nodeInfo.NonZeroRequest().MilliCPU
    	requested.Memory += nodeInfo.NonZeroRequest().Memory
    	var score int64
    	// Check if the pod has volumes and this could be added to scorer function for balanced resource allocation.
    	if len(pod.Spec.Volumes) >= 0 && utilfeature.DefaultFeatureGate.Enabled(features.BalanceAttachedNodeVolumes) && nodeInfo.TransientInfo != nil {
    		score = r.scorer(&requested, &allocatable, true, nodeInfo.TransientInfo.TransNodeInfo.RequestedVolumes, nodeInfo.TransientInfo.TransNodeInfo.AllocatableVolumesCount)
    	} else {
    		score = r.scorer(&requested, &allocatable, false, 0, 0)
    	}
    
    	if glog.V(10) {
    		if len(pod.Spec.Volumes) >= 0 && utilfeature.DefaultFeatureGate.Enabled(features.BalanceAttachedNodeVolumes) && nodeInfo.TransientInfo != nil {
    			glog.Infof(
    				"%v -> %v: %v, capacity %d millicores %d memory bytes, %d volumes, total request %d millicores %d memory bytes %d volumes, score %d",
    				pod.Name, node.Name, r.Name,
    				allocatable.MilliCPU, allocatable.Memory, nodeInfo.TransientInfo.TransNodeInfo.AllocatableVolumesCount,
    				requested.MilliCPU, requested.Memory,
    				nodeInfo.TransientInfo.TransNodeInfo.RequestedVolumes,
    				score,
    			)
    		} else {
    			glog.Infof(
    				"%v -> %v: %v, capacity %d millicores %d memory bytes, total request %d millicores %d memory bytes, score %d",
    				pod.Name, node.Name, r.Name,
    				allocatable.MilliCPU, allocatable.Memory,
    				requested.MilliCPU, requested.Memory,
    				score,
    			)
    		}
    	}
    
    	return schedulerapi.HostPriority{
    		Host:  node.Name,
    		Score: int(score),
    	}, nil
    }

**  8.1.1 balancedResourceScorer函数**

**   cpu memory或者volume(可能没有)分数计算公式为：**

*  **cpuFraction = **float64(requested) / float64(capacity)
 *  memoryFraction = float64(requested) / float64(capacity)
 *  cpuFraction相减的绝对值相乘10为分数

func balancedResourceScorer(requested, allocable *schedulercache.Resource, includeVolumes bool, requestedVolumes int, allocatableVolumes int) int64 {
    	cpuFraction := fractionOfCapacity(requested.MilliCPU, allocable.MilliCPU)
    	memoryFraction := fractionOfCapacity(requested.Memory, allocable.Memory)
    	// This to find a node which has most balanced CPU, memory and volume usage.
    	if includeVolumes && utilfeature.DefaultFeatureGate.Enabled(features.BalanceAttachedNodeVolumes) && allocatableVolumes > 0 {
    		volumeFraction := float64(requestedVolumes) / float64(allocatableVolumes)
    		if cpuFraction >= 1 || memoryFraction >= 1 || volumeFraction >= 1 {
    			// if requested >= capacity, the corresponding host should never be preferred.
    			return 0
    		}
    		// Compute variance for all the three fractions.
    		mean := (cpuFraction + memoryFraction + volumeFraction) / float64(3)
    		variance := float64((((cpuFraction - mean) * (cpuFraction - mean)) + ((memoryFraction - mean) * (memoryFraction - mean)) + ((volumeFraction - mean) * (volumeFraction - mean))) / float64(3))
    		// Since the variance is between positive fractions, it will be positive fraction. 1-variance lets the
    		// score to be higher for node which has least variance and multiplying it with 10 provides the scaling
    		// factor needed.
    		return int64((1 - variance) * float64(schedulerapi.MaxPriority))
    	}
    
    	if cpuFraction >= 1 || memoryFraction >= 1 {
    		// if requested >= capacity, the corresponding host should never be preferred.
    		return 0
    	}
    	// Upper and lower boundary of difference between cpuFraction and memoryFraction are -1 and 1
    	// respectively. Multiplying the absolute value of the difference by 10 scales the value to
    	// 0-10 with 0 representing well balanced allocation and 10 poorly balanced. Subtracting it from
    	// 10 leads to the score which also scales from 0 to 10 while 10 representing well balanced.
    	diff := math.Abs(cpuFraction - memoryFraction)
    	return int64((1 - diff) * float64(schedulerapi.MaxPriority))
    }

###   8.2 NodePreferAvoidPodsPriority ###

根据注解中的scheduler.alpha.kubernetes.io/preferAvoidPods，要么0分要么10分

/ CalculateNodePreferAvoidPodsPriorityMap priorities nodes according to the node annotation
    // "scheduler.alpha.kubernetes.io/preferAvoidPods".
    func CalculateNodePreferAvoidPodsPriorityMap(pod *v1.Pod, meta interface{}, nodeInfo *schedulercache.NodeInfo) (schedulerapi.HostPriority, error) {
    	node := nodeInfo.Node()
    	if node == nil {
    		return schedulerapi.HostPriority{}, fmt.Errorf("node not found")
    	}
    	var controllerRef *metav1.OwnerReference
    	if priorityMeta, ok := meta.(*priorityMetadata); ok {
    		controllerRef = priorityMeta.controllerRef
    	} else {
    		// We couldn't parse metadata - fallback to the podspec.
    		controllerRef = metav1.GetControllerOf(pod)
    	}
    
    	if controllerRef != nil {
    		// Ignore pods that are owned by other controller than ReplicationController
    		// or ReplicaSet.
    		if controllerRef.Kind != "ReplicationController" && controllerRef.Kind != "ReplicaSet" {
    			controllerRef = nil
    		}
    	}
    	if controllerRef == nil {
    		return schedulerapi.HostPriority{Host: node.Name, Score: schedulerapi.MaxPriority}, nil
    	}
    
    	avoids, err := v1helper.GetAvoidPodsFromNodeAnnotations(node.Annotations)
    	if err != nil {
    		// If we cannot get annotation, assume it's schedulable there.
    		return schedulerapi.HostPriority{Host: node.Name, Score: schedulerapi.MaxPriority}, nil
    	}
    	for i := range avoids.PreferAvoidPods {
    		avoid := &avoids.PreferAvoidPods[i]
    		if avoid.PodSignature.PodController.Kind == controllerRef.Kind && avoid.PodSignature.PodController.UID == controllerRef.UID {
    			return schedulerapi.HostPriority{Host: node.Name, Score: 0}, nil
    		}
    	}
    	return schedulerapi.HostPriority{Host: node.Name, Score: schedulerapi.MaxPriority}, nil
    }

###   8.2 SelectorSpreadPriority ###

根据注册SelectorSpreadPriority，对于属于同一个 service、replication controller 的 Pod，尽量分散在不同的主机上。调度一个 Pod 的时候，先查找 Pod 对于的 service或者 replication controller，然后查找 service 或 replication controller 中已存在的 Pod，主机上运行的已存在的 Pod 越少，主机的打分越高。

// spreads pods by minimizing the number of pods (belonging to the same service or replication controller) on the same node.
    		factory.RegisterPriorityConfigFactory(
    			"SelectorSpreadPriority",
    			factory.PriorityConfigFactory{
    				MapReduceFunction: func(args factory.PluginFactoryArgs) (algorithm.PriorityMapFunction, algorithm.PriorityReduceFunction) {
    					return priorities.NewSelectorSpreadPriority(args.ServiceLister, args.ControllerLister, args.ReplicaSetLister, args.StatefulSetLister)
    				},
    				Weight: 1,
    			},
    		),

**  8.3.1  CalculateSpreadPriorityReduce函数**

尽量将同一rc/replica的多个pod分配到不同的node上

// CalculateSpreadPriorityReduce calculates the source of each node
    // based on the number of existing matching pods on the node
    // where zone information is included on the nodes, it favors nodes
    // in zones with fewer existing matching pods.
    func (s *SelectorSpread) CalculateSpreadPriorityReduce(pod *v1.Pod, meta interface{}, nodeNameToInfo map[string]*schedulercache.NodeInfo, result schedulerapi.HostPriorityList) error {
    	countsByZone := make(map[string]int, 10)
    	maxCountByZone := int(0)
    	maxCountByNodeName := int(0)
    
    	for i := range result {
    		if result[i].Score > maxCountByNodeName {
    			maxCountByNodeName = result[i].Score
    		}
    		zoneID := utilnode.GetZoneKey(nodeNameToInfo[result[i].Host].Node())
    		if zoneID == "" {
    			continue
    		}
    		countsByZone[zoneID] += result[i].Score
    	}
    
    	for zoneID := range countsByZone {
    		if countsByZone[zoneID] > maxCountByZone {
    			maxCountByZone = countsByZone[zoneID]
    		}
    	}
    
    	haveZones := len(countsByZone) != 0
    
    	maxCountByNodeNameFloat64 := float64(maxCountByNodeName)
    	maxCountByZoneFloat64 := float64(maxCountByZone)
    	MaxPriorityFloat64 := float64(schedulerapi.MaxPriority)
    
    	for i := range result {
    		// initializing to the default/max node score of maxPriority
    		fScore := MaxPriorityFloat64
    		if maxCountByNodeName > 0 {
    			fScore = MaxPriorityFloat64 * (float64(maxCountByNodeName-result[i].Score) / maxCountByNodeNameFloat64)
    		}
    		// If there is zone information present, incorporate it
    		if haveZones {
    			zoneID := utilnode.GetZoneKey(nodeNameToInfo[result[i].Host].Node())
    			if zoneID != "" {
    				zoneScore := MaxPriorityFloat64
    				if maxCountByZone > 0 {
    					zoneScore = MaxPriorityFloat64 * (float64(maxCountByZone-countsByZone[zoneID]) / maxCountByZoneFloat64)
    				}
    				fScore = (fScore * (1.0 - zoneWeighting)) + (zoneWeighting * zoneScore)
    			}
    		}
    		result[i].Score = int(fScore)
    		if glog.V(10) {
    			glog.Infof(
    				"%v -> %v: SelectorSpreadPriority, Score: (%d)", pod.Name, result[i].Host, int(fScore),
    			)
    		}
    	}
    	return nil
    }

### Predicate 算法： ###

> PodFitsResources：节点上剩余的资源是否大于 pod 请求的资源
> 
> PodFitsHost：如果 Pod 制定了 NodeName，检查节点是否和 NodeName 匹配
> 
> PodFitsHostsPorts：节点上已经使用 port 是否和 Pod 申请的冲突
> 
> PodSelectorMatches：过略掉和 Pod 制定的 Label 不匹配的节点
> 
> NoDiskConfict：已经 mount 的 volume 和 Pod 制定的 volume 不冲突

### Priorities 算法 ###

> LeastRequestedPriority：通过计算 CPU 和 Memory 的使用率来决定权重，使用率越低权重越高（倾向于资源使用率比例更低的节点）
> 
> BalancedResourceAllocation：节点上 CPU 和 Memory 使用率接近，权重越高
> 
> ImageLocalityPriority：倾向于已经有要使用镜像的节点，镜像值总大小越大，权重越高

### 节点亲和性 ###

pod.spec.nodeAffnity

*   preferredDuringSchedulingIgnoredDuringExecution：软策略
 *   requiredDuringSchedulingIgnoredDuringExecution：硬策略

> spec:
> 
>   containers:
> 
>   - name: xxxx
> 
>     image: xxxx-image
> 
>   affinity:
> 
>     nodeAffinity:
> 
>       requiredDuringSchedulingIgnoredExecution:
> 
>         nodeSelectorItems:
> 
>         - matchExpressions:
> 
>           - key: xxxxxx-key
> 
>             operator: In
> 
>             values:
> 
>             - xxxxxxx-value

### Pod 亲和性 ###

pod.spec.podAffnity

*   preferredDuringSchedulingIgnoredDuringExecution：软策略
 *   requiredDuringSchedulingIgnoredDuringExecution：硬策略

键值运算关系

> In：label 的值在某个列表中
> 
> NotIn：
> 
> Gt：label 的值大于某个值
> 
> Lt：小于
> 
> Exists：某个 label 存在
> 
> DoesNotExist：某个 label 不存在

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob25nbGluemhhbmc_size_16_color_FFFFFF_t_70]: /images/20220610/774e76f7cdf241e7ad7db44c7063936a.png