preemption(Preemption搶占式調度的方法是什么)

ScheduleAlgorithm的變化

在Kubernetes 1.8中,對ScheduleAlgorithm Interface的定義發生了改變,多了一個Preempt(...)。因此,我在博文Kubernetes Scheduler原理解析(當時是基于kubernetes 1.5)中對scheduler調度過程開的一句話概括“將PodSpec.NodeName為空的Pods逐個地,經過預選(Predicates)和優選(Priorities)兩個步驟,挑選最合適的Node作為該Pod的Destination?！睂⒉辉贉蚀_了。

現在應該一句話這樣描述才算準確了:“將PodSpec.NodeName為空的Pods逐個地,經過預選(Predicates)和優選(Priorities)兩個步驟,挑選最合適的Node作為該Pod的Destination。如果經過預選和優選仍然沒有找到合適的節點,并且啟動了Pod Priority,那么該Pod將會進行Preempt搶占式調度找到最合適的節點及需要Evict的Pods?！?/p>

//ScheduleAlgorithmisaninterfaceimplementedbythingsthatknowhowtoschedulepods//ontomachines.typeScheduleAlgorithminterface{Schedule(*v1.Pod,NodeLister)(selectedMachinestring,errerror)//Preemptreceivesschedulingerrorsforapodandtriestocreateroomfor//thepodbypreemptinglowerprioritypodsifpossible.//Itreturnsthenodewherepreemptionhappened,alistofpreemptedpods,anderrorifany.Preempt(*v1.Pod,NodeLister,error)(selectedNode*v1.Node,preemptedPods[]*v1.Pod,errerror)//Predicates()returnsapointertoamapofpredicatefunctions.Thisis//exposedfortesting.Predicates()map[string]FitPredicate//Prioritizersreturnsasliceofpriorityconfig.Thisisexposedfor//testing.Prioritizers()[]PriorityConfig}

Scheduler.scheduleOne開始真正的調度邏輯,每次負責一個Pod的調度,邏輯如下:

• 從PodQueue中獲取一個Pod。

• 執行對應Algorithm的Schedule,進行預選和優選。

• AssumePod

• Bind Pod, 如果Bind Failed,ForgetPod。

在1.8中,但預選和優選調度完整沒有找到合適node時(其實一定會是預選沒有找到nodes,優選只是挑更好的),還會調用sched.preempt進行搶占式調度。

plugin/pkg/scheduler/scheduler.go:293func(sched*Scheduler)scheduleOne(){pod:=sched.config.NextPod()ifpod.DeletionTimestamp!=nil{sched.config.Recorder.Eventf(pod,v1.EventTypeWarning,"FailedScheduling","skipscheduledeletingpod:%v/%v",pod.Namespace,pod.Name)glog.V(3).Infof("Skipscheduledeletingpod:%v/%v",pod.Namespace,pod.Name)return}glog.V(3).Infof("Attemptingtoschedulepod:%v/%v",pod.Namespace,pod.Name)//Synchronouslyattempttofindafitforthepod.start:=time.Now()suggestedHost,err:=sched.schedule(pod)metrics.SchedulingAlgorithmLatency.Observe(metrics.SinceInMicroseconds(start))iferr!=nil{//schedule()mayhavefailedbecausethepodwouldnotfitonanyhost,sowetryto//preempt,withtheexpectationthatthenexttimethepodistriedforschedulingit//willfitduetothepreemption.Itisalsopossiblethatadifferentpodwillschedule//intotheresourcesthatwerepreempted,butthisisharmless.iffitError,ok:=err.(*core.FitError);ok{sched.preempt(pod,fitError)}return}//Tellthecachetoassumethatapodnowisrunningonagivennode,eventhoughithasn'tbeenboundyet.//Thisallowsustokeepschedulingwithoutwaitingonbindingtooccur.assumedPod:=*pod//assumemodifies`assumedPod`bysettingNodeName=suggestedHosterr=sched.assume(&assumedPod,suggestedHost)iferr!=nil{return}//bindthepodtoitshostasynchronously(wecandothisb/coftheassumptionstepabove).gofunc(){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))iferr!=nil{glog.Errorf("Internalerrorbindingpod:(%v)",err)}}()}

Scheduler.preemt

好的,關于預選和優選,我這里不做過多解讀,因為整個源碼邏輯和1.5是一樣,不同的是1.8增加了更多的Predicate和Priority Policys及其實現。下面只看搶占式調度Preempt的代碼。

plugin/pkg/scheduler/scheduler.go:191func(sched*Scheduler)preempt(preemptor*v1.Pod,scheduleErrerror)(string,error){if!utilfeature.DefaultFeatureGate.Enabled(features.PodPriority){glog.V(3).Infof("Podpriorityfeatureisnotenabled.Nopreemptionisperformed.")return"",nil}preemptor,err:=sched.config.PodPreemptor.GetUpdatedPod(preemptor)iferr!=nil{glog.Errorf("Errorgettingtheupdatedpreemptorpodobject:%v",err)return"",err}node,victims,err:=sched.config.Algorithm.Preempt(preemptor,sched.config.NodeLister,scheduleErr)iferr!=nil{glog.Errorf("Errorpreemptingvictimstomakeroomfor%v/%v.",preemptor.Namespace,preemptor.Name)return"",err}ifnode==nil{return"",err}glog.Infof("Preempting%dpod(s)onnode%vtomakeroomfor%v/%v.",len(victims),node.Name,preemptor.Namespace,preemptor.Name)annotations:=map[string]string{core.NominatedNodeAnnotationKey:node.Name}err=sched.config.PodPreemptor.UpdatePodAnnotations(preemptor,annotations)iferr!=nil{glog.Errorf("Errorinpreemptionprocess.Cannotupdatepod%vannotations:%v",preemptor.Name,err)return"",err}for_,victim:=rangevictims{iferr:=sched.config.PodPreemptor.DeletePod(victim);err!=nil{glog.Errorf("Errorpreemptingpod%v/%v:%v",victim.Namespace,victim.Name,err)return"",err}sched.config.Recorder.Eventf(victim,v1.EventTypeNormal,"Preempted","by%v/%vonnode%v",preemptor.Namespace,preemptor.Name,node.Name)}returnnode.Name,err}

• 檢查FeaturesGate中是否開啟了PodPriority,如果沒開啟,則不會進行后續Preemption操作;

• 由于該Pod在Predicate/Priortiy調度過程失敗后,會更新PodCondition,記錄調度失敗狀態及失敗原因。因此需要從apiserver中獲取PodCondition更新后的Pod Object;

• 調用ScheduleAlgorithm.Preempt進行搶占式調度,選出最佳node和待preempt pods(稱為victims);

• 調用apiserver給該pod(稱為Preemptor)打上Annotation:NominatedNodeName=nodeName;

• 遍歷victims,調用apiserver進行逐個刪除這些pods;

注意:在scheduler調用shed.schedule(pod)進行預選和優選調度失敗時,Pod Bind Node失敗,該Pod會requeue unscheduled Cache podqueue中,如果在這個pod調度過程中又有新的pod加入到待調度隊列,那么該pod requeue時它前面就有其他pod,下一次調度就是先調度在它前面的pod,而這些pod的調度有可能會調度到剛剛通過Preempt釋放資源的Node上,導致把剛才Preemptor釋放的resource消耗掉。當再次輪到上次的Preemptor調度時,可能又需要觸發一次某個節點的Preempt。

genericScheduler.Preempt

ScheduleAlgorithm.Preempt是搶占式調度的關鍵實現,其對應的實現在genericScheduler中:

plugin/pkg/scheduler/core/generic_scheduler.go:181//preemptfindsnodeswithpodsthatcanbepreemptedtomakeroomfor"pod"to//schedule.Itchoosesoneofthenodesandpreemptsthepodsonthenodeand//returnsthenodeandthelistofpreemptedpodsifsuchanodeisfound.//TODO(bsalamat):Addpriority-basedscheduling.Moreinfo:todayoneormore//pendingpods(differentfromthepodthattriggeredthepreemption(s))may//scheduleintosomeportionoftheresourcesfreedupbythepreemption(s)//beforethepodthattriggeredthepreemption(s)hasachancetoschedule//there,therebypreventingthepodthattriggeredthepreemption(s)from//scheduling.Solutionisgivenat://https://github.com/kubernetes/community/blob/master/contributors/design-proposals/pod-preemption.md#preemption-mechanicsfunc(g*genericScheduler)Preempt(pod*v1.Pod,nodeListeralgorithm.NodeLister,scheduleErrerror)(*v1.Node,[]*v1.Pod,error){//Schedulermayreturnvarioustypesoferrors.Considerpreemptiononlyif//theerrorisoftypeFitError.fitError,ok:=scheduleErr.(*FitError)if!ok||fitError==nil{returnnil,nil,nil}err:=g.cache.UpdateNodeNameToInfoMap(g.cachedNodeInfoMap)iferr!=nil{returnnil,nil,err}if!podEligibleToPreemptOthers(pod,g.cachedNodeInfoMap){glog.V(5).Infof("Pod%visnoteligibleformorepreemption.",pod.Name)returnnil,nil,nil}allNodes,err:=nodeLister.List()iferr!=nil{returnnil,nil,err}iflen(allNodes)==0{returnnil,nil,ErrNoNodesAvailable}potentialNodes:=nodesWherePreemptionMightHelp(pod,allNodes,fitError.FailedPredicates)iflen(potentialNodes)==0{glog.V(3).Infof("Preemptionwillnothelpschedulepod%vonanynode.",pod.Name)returnnil,nil,nil}nodeToPods,err:=selectNodesForPreemption(pod,g.cachedNodeInfoMap,potentialNodes,g.predicates,g.predicateMetaProducer)iferr!=nil{returnnil,nil,err}forlen(nodeToPods)>0{node:=pickOneNodeForPreemption(nodeToPods)ifnode==nil{returnnil,nil,err}passes,pErr:=nodePassesExtendersForPreemption(pod,node.Name,nodeToPods[node],g.cachedNodeInfoMap,g.extenders)ifpasses&&pErr==nil{returnnode,nodeToPods[node],err}ifpErr!=nil{glog.Errorf("Erroroccurredwhilecheckingextendersforpreemptiononnode%v:%v",node,pErr)}//Removethenodefromthemapandtrytopickadifferentnode.delete(nodeToPods,node)}returnnil,nil,err}

sched.schedule error檢查

• 只有前面sched.schedule()返回的error為FitError類型時,才會觸發后續的Preemption。FitError就是表示pod在Predicate階段進行某些PredicateFunc篩選時不通過。也就是說只有預選失敗的Pod才會進行搶占式調度。

更新scheduler cache中的NodeInfo

• 更新scheduler cache中NodeInfo,主要是更新Node上scheduled 和Assumed Pods,作為后續Preempt Pods時的考慮范圍,確保Preemption是正確的。

podEligibleToPreemptOthers檢查pod是否有資格進行搶占式調度

• invoke podEligibleToPreemptOthers來判斷該pod是否適合進行后續的Preemption,判斷邏輯是:

• 如果該Pod已經包含Annotation:NominatedNodeName=nodeName(說明該pod之前已經Preempted),并且Annotation中的這個Node有比該pod優先級更低的pod正在Terminating,則認為該pod不適合進行后續的Preemption,流程結束。

• 除此之外,繼續后續的流程。

• 對應代碼如下:

  plugin/pkg/scheduler/core/generic_scheduler.go:756funcpodEligibleToPreemptOthers(pod*v1.Pod,nodeNameToInfomap[string]*schedulercache.NodeInfo)bool{ifnodeName,found:=pod.Annotations[NominatedNodeAnnotationKey];found{ifnodeInfo,found:=nodeNameToInfo[nodeName];found{for_,p:=rangenodeInfo.Pods(){ifp.DeletionTimestamp!=nil&&util.GetPodPriority(p)<util.GetPodPriority(pod){//Thereisaterminatingpodonthenominatednode.returnfalse}}}}returntrue}

  
  

nodesWherePreemptionMightHelp篩選出Potential Nodes

• invoke nodesWherePreemptionMightHelp來獲取potential nodes。nodesWherePreemptionMightHelp的邏輯是:

• NodeSelectorNotMatch,

• PodNotMatchHostName,

• TaintsTolerationsNotMatch,

• NodeLabelPresenceViolated,

• NodeNotReady,

• NodeNetworkUnavailable,

• NodeUnschedulable,

• NodeUnknownCondition

• 遍歷所有的nodes,對每個nodes在sched.schedule()在預選階段失敗的Predicate策略(failedPredicates)進行掃描,如果failedPredicates包含以下Policy,則說明該node不適合作為Preempt的備選節點。

• 除此之外的Node均作為Potential Nodes。

• 對應代碼如下:

  funcnodesWherePreemptionMightHelp(pod*v1.Pod,nodes[]*v1.Node,failedPredicatesMapFailedPredicateMap)[]*v1.Node{potentialNodes:=[]*v1.Node{}for_,node:=rangenodes{unresolvableReasonExist:=falsefailedPredicates,found:=failedPredicatesMap[node.Name]//IfweassumethatschedulerlooksatallnodesandpopulatesthefailedPredicateMap//(whichisthecasetoday),the!foundcaseshouldneverhappen,butwe'dprefer//torelylessonsuchassumptionsinthecodewhencheckingdoesnotimpose//significantoverhead.for_,failedPredicate:=rangefailedPredicates{switchfailedPredicate{casepredicates.ErrNodeSelectorNotMatch,predicates.ErrPodNotMatchHostName,predicates.ErrTaintsTolerationsNotMatch,predicates.ErrNodeLabelPresenceViolated,predicates.ErrNodeNotReady,predicates.ErrNodeNetworkUnavailable,predicates.ErrNodeUnschedulable,predicates.ErrNodeUnknownCondition:unresolvableReasonExist=truebreak//TODO(bsalamat):Pleaseaddaffinityfailurecasesoncewehavespecificaffinityfailureerrors.}}if!found||!unresolvableReasonExist{glog.V(3).Infof("Node%visapotentialnodeforpreemption.",node.Name)potentialNodes=append(potentialNodes,node)}}returnpotentialNodes}

  
  

selectNodesForPreemption和selectVictimsOnNode選出可行Nodes及其對應的victims

• invoke selectNodesForPreemption從Potential Nodes中找出所有可行的Nodes及對應的victim Pods,其對應的邏輯如為:啟動max(16, potentialNodesNum)個worker(對應goruntine)通過WaitGroups并發等待所有node的check完成:

• 遍歷該node上所有的scheduled pods(包括assumed pods),將優先級比Preemptor更低的Pods都加入到Potential victims List中,并且將這些victims從NodeInfoCopy中刪除,下次進行Predicate時就意味著Node上有更多資源可用。

• 對Potential victims中元素進行排序,排序規則是按照優先級從高到底排序的,index為0的對應的優先級最高。

• 檢查Preemptor是否能scheduler配置的所有Predicates Policy(基于前面將這些victims從NodeInfoCopy中刪除,將所有更低優先級的pods資源全部釋放了),如果不通過則返回,表示該node不合適。All Predicate通過后,繼續下面流程。

• 遍歷所有的Potential victims list item(已經按照優先級從高到底排序),試著把Potential victims中第一個Pod(優先級最高)加回到NodeInfoCopy中,再檢查Preemptor是否能scheduler配置的所有Predicates Policy,如果不滿足就把該pod再從NodeInfoCopy中刪除,并且正式加入到victims list中。接著對Potential victims中第2,3...個Pod進行同樣處理。這樣做,是為了保證盡量保留優先級更高的Pods,盡量刪除更少的Pods。

• 最終返回每個可行node及其對應victims list。

• selectNodesForPreemption代碼如下,其實核心代碼在selectVictimsOnNode。

  plugin/pkg/scheduler/core/generic_scheduler.go:583funcselectNodesForPreemption(pod*v1.Pod,nodeNameToInfomap[string]*schedulercache.NodeInfo,potentialNodes[]*v1.Node,predicatesmap[string]algorithm.FitPredicate,metadataProduceralgorithm.PredicateMetadataProducer,)(map[*v1.Node][]*v1.Pod,error){nodeNameToPods:=map[*v1.Node][]*v1.Pod{}varresultLocksync.Mutex//Wecanusethesamemetadataproducerforallnodes.meta:=metadataProducer(pod,nodeNameToInfo)checkNode:=func(iint){nodeName:=potentialNodes[i].NamevarmetaCopyalgorithm.PredicateMetadataifmeta!=nil{metaCopy=meta.ShallowCopy()}pods,fits:=selectVictimsOnNode(pod,metaCopy,nodeNameToInfo[nodeName],predicates)iffits{resultLock.Lock()nodeNameToPods[potentialNodes[i]]=podsresultLock.Unlock()}}workqueue.Parallelize(16,len(potentialNodes),checkNode)returnnodeNameToPods,nil}

  
  

  plugin/pkg/scheduler/core/generic_scheduler.go:659funcselectVictimsOnNode(pod*v1.Pod,metaalgorithm.PredicateMetadata,nodeInfo*schedulercache.NodeInfo,fitPredicatesmap[string]algorithm.FitPredicate)([]*v1.Pod,bool){potentialVictims:=util.SortableList{CompFunc:util.HigherPriorityPod}nodeInfoCopy:=nodeInfo.Clone()removePod:=func(rp*v1.Pod){nodeInfoCopy.RemovePod(rp)ifmeta!=nil{meta.RemovePod(rp)}}addPod:=func(ap*v1.Pod){nodeInfoCopy.AddPod(ap)ifmeta!=nil{meta.AddPod(ap,nodeInfoCopy)}}//Asthefirststep,removeallthelowerprioritypodsfromthenodeand//checkifthegivenpodcanbescheduled.podPriority:=util.GetPodPriority(pod)for_,p:=rangenodeInfoCopy.Pods(){ifutil.GetPodPriority(p)<podPriority{potentialVictims.Items=append(potentialVictims.Items,p)removePod(p)}}potentialVictims.Sort()//Ifthenewpoddoesnotfitafterremovingallthelowerprioritypods,//wearealmostdoneandthisnodeisnotsuitableforpreemption.Theonlycondition//thatweshouldcheckisifthe"pod"isfailingtoscheduleduetopodaffinity//failure.//TODO(bsalamat):Considercheckingaffinitytolowerprioritypodsiffeasiblewithreasonableperformance.iffits,_,err:=podFitsOnNode(pod,meta,nodeInfoCopy,fitPredicates,nil);!fits{iferr!=nil{glog.Warningf("Encounterederrorwhileselectingvictimsonnode%v:%v",nodeInfo.Node().Name,err)}returnnil,false}victims:=[]*v1.Pod{}//Trytoreprieveasmanypodsaspossiblestartingfromthehighestpriorityone.for_,p:=rangepotentialVictims.Items{lpp:=p.(*v1.Pod)addPod(lpp)iffits,_,_:=podFitsOnNode(pod,meta,nodeInfoCopy,fitPredicates,nil);!fits{removePod(lpp)victims=append(victims,lpp)glog.V(5).Infof("Pod%visapotentialpreemptionvictimonnode%v.",lpp.Name,nodeInfo.Node().Name)}}returnvictims,true}

  
  

pickOneNodeForPreemption從可行Nodes中找出最合適的一個Node

• 如果上一步至少找到一個可行node,則調用pickOneNodeForPreemption按照以下邏輯選擇一個最合適的node:

• 選擇victims中最高pod優先級最低的那個Node。

• 如果上一步有不止一個Nodes滿足條件,則再對選擇所有victims優先級之和最小的那個Node。

• 如果上一步有不止一個Nodes滿足條件,則再選擇victims pod數最少的Node。

• 如果上一步有不止一個Nodes滿足條件,則再隨機選擇一個Node。

• 以上每一步的Nodes列表,都是基于上一步篩選后的Nodes。

  plugin/pkg/scheduler/core/generic_scheduler.go:501funcpickOneNodeForPreemption(nodesToPodsmap[*v1.Node][]*v1.Pod)*v1.Node{typenodeScorestruct{node*v1.NodehighestPriorityint32sumPrioritiesint64numPodsint}iflen(nodesToPods)==0{returnnil}minHighestPriority:=int32(math.MaxInt32)minPriorityScores:=[]*nodeScore{}fornode,pods:=rangenodesToPods{iflen(pods)==0{//Wefoundanodethatdoesn'tneedanypreemption.Returnit!//Thisshouldhappenrarelywhenoneormorepodsareterminatedbetween//thetimethatschedulertriestoschedulethepodandthetimethat//preemptionlogictriestofindnodesforpreemption.returnnode}//highestPodPriorityisthehighestpriorityamongthevictimsonthisnode.highestPodPriority:=util.GetPodPriority(pods[0])ifhighestPodPriority<minHighestPriority{minHighestPriority=highestPodPriorityminPriorityScores=nil}ifhighestPodPriority==minHighestPriority{minPriorityScores=append(minPriorityScores,&nodeScore{node:node,highestPriority:highestPodPriority,numPods:len(pods)})}}iflen(minPriorityScores)==1{returnminPriorityScores[0].node}//Thereareafewnodeswithminimumhighestpriorityvictim.Findthe//smallestsumofpriorities.minSumPriorities:=int64(math.MaxInt64)minSumPriorityScores:=[]*nodeScore{}for_,nodeScore:=rangeminPriorityScores{varsumPrioritiesint64for_,pod:=rangenodesToPods[nodeScore.node]{//WeaddMaxInt32+1toallprioritiestomakeallofthem>=0.Thisis//neededsothatanodewithafewpodswithnegativepriorityisnot//pickedoveranodewithasmallernumberofpodswiththesamenegative//priority(andsimilarscenarios).sumPriorities+=int64(util.GetPodPriority(pod))+int64(math.MaxInt32+1)}ifsumPriorities<minSumPriorities{minSumPriorities=sumPrioritiesminSumPriorityScores=nil}nodeScore.sumPriorities=sumPrioritiesifsumPriorities==minSumPriorities{minSumPriorityScores=append(minSumPriorityScores,nodeScore)}}iflen(minSumPriorityScores)==1{returnminSumPriorityScores[0].node}//Thereareafewnodeswithminimumhighestpriorityvictimandsumofpriorities.//Findonewiththeminimumnumberofpods.minNumPods:=math.MaxInt32minNumPodScores:=[]*nodeScore{}for_,nodeScore:=rangeminSumPriorityScores{ifnodeScore.numPods<minNumPods{minNumPods=nodeScore.numPodsminNumPodScores=nil}ifnodeScore.numPods==minNumPods{minNumPodScores=append(minNumPodScores,nodeScore)}}//Atthispoint,eveniftherearemorethanonenodewiththesamescore,//returnthefirstone.iflen(minNumPodScores)>0{returnminNumPodScores[0].node}glog.Errorf("Errorinlogicofnodescoringforpreemption.Weshouldneverreachhere!")returnnil}

最合適的Node仍然要交給extender(if configed)檢查

• 如果scheduler配置extender scheduler,則還需要通過invoke nodePassesExtendersForPreemption再次將該pod和(假設)剔除victims的該node交給extender.Filter進行一下檢查,只有檢查通過了才返回該node作為最終選擇的Preempt node。

• 關于extender的理解,請參考如何對kubernetes scheduler進行二次開發和Kubernetes Scheduler源碼分析。其實用的場景不多,現在支持自定義調度器了,就更少需要使用scheduler extender了。

感謝各位的閱讀,以上就是“Preemption搶占式調度的方法是什么”的內容了,經過本文的學習后,相信大家對Preemption搶占式調度的方法是什么這一問題有了更深刻的體會,具體使用情況還需要大家實踐驗證。這里是本站,小編將為大家推送更多相關知識點的文章,歡迎關注!