scheduler.go 25 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626
  1. /*
  2. Copyright 2014 The Kubernetes Authors.
  3. Licensed under the Apache License, Version 2.0 (the "License");
  4. you may not use this file except in compliance with the License.
  5. You may obtain a copy of the License at
  6. http://www.apache.org/licenses/LICENSE-2.0
  7. Unless required by applicable law or agreed to in writing, software
  8. distributed under the License is distributed on an "AS IS" BASIS,
  9. WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  10. See the License for the specific language governing permissions and
  11. limitations under the License.
  12. */
  13. package scheduler
  14. import (
  15. "fmt"
  16. "io/ioutil"
  17. "os"
  18. "time"
  19. "k8s.io/klog"
  20. v1 "k8s.io/api/core/v1"
  21. metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
  22. "k8s.io/apimachinery/pkg/runtime"
  23. "k8s.io/apimachinery/pkg/util/wait"
  24. appsinformers "k8s.io/client-go/informers/apps/v1"
  25. coreinformers "k8s.io/client-go/informers/core/v1"
  26. policyinformers "k8s.io/client-go/informers/policy/v1beta1"
  27. storageinformers "k8s.io/client-go/informers/storage/v1"
  28. clientset "k8s.io/client-go/kubernetes"
  29. "k8s.io/client-go/tools/record"
  30. schedulerapi "k8s.io/kubernetes/pkg/scheduler/api"
  31. latestschedulerapi "k8s.io/kubernetes/pkg/scheduler/api/latest"
  32. kubeschedulerconfig "k8s.io/kubernetes/pkg/scheduler/apis/config"
  33. "k8s.io/kubernetes/pkg/scheduler/core"
  34. "k8s.io/kubernetes/pkg/scheduler/customcache"
  35. "k8s.io/kubernetes/pkg/scheduler/factory"
  36. framework "k8s.io/kubernetes/pkg/scheduler/framework/v1alpha1"
  37. internalcache "k8s.io/kubernetes/pkg/scheduler/internal/cache"
  38. "k8s.io/kubernetes/pkg/scheduler/metrics"
  39. "k8s.io/kubernetes/pkg/scheduler/util"
  40. )
  41. const (
  42. // BindTimeoutSeconds defines the default bind timeout
  43. BindTimeoutSeconds = 100
  44. // SchedulerError is the reason recorded for events when an error occurs during scheduling a pod.
  45. SchedulerError = "SchedulerError"
  46. )
  47. // Scheduler watches for new unscheduled pods. It attempts to find
  48. // nodes that they fit on and writes bindings back to the api server.
  49. type Scheduler struct {
  50. config *factory.Config
  51. }
  52. // Cache returns the cache in scheduler for test to check the data in scheduler.
  53. func (sched *Scheduler) Cache() internalcache.Cache {
  54. return sched.config.SchedulerCache
  55. }
  56. type schedulerOptions struct {
  57. schedulerName string
  58. hardPodAffinitySymmetricWeight int32
  59. disablePreemption bool
  60. percentageOfNodesToScore int32
  61. bindTimeoutSeconds int64
  62. }
  63. // Option configures a Scheduler
  64. type Option func(*schedulerOptions)
  65. // WithName sets schedulerName for Scheduler, the default schedulerName is default-scheduler
  66. func WithName(schedulerName string) Option {
  67. return func(o *schedulerOptions) {
  68. o.schedulerName = schedulerName
  69. }
  70. }
  71. // WithHardPodAffinitySymmetricWeight sets hardPodAffinitySymmetricWeight for Scheduler, the default value is 1
  72. func WithHardPodAffinitySymmetricWeight(hardPodAffinitySymmetricWeight int32) Option {
  73. return func(o *schedulerOptions) {
  74. o.hardPodAffinitySymmetricWeight = hardPodAffinitySymmetricWeight
  75. }
  76. }
  77. // WithPreemptionDisabled sets disablePreemption for Scheduler, the default value is false
  78. func WithPreemptionDisabled(disablePreemption bool) Option {
  79. return func(o *schedulerOptions) {
  80. o.disablePreemption = disablePreemption
  81. }
  82. }
  83. // WithPercentageOfNodesToScore sets percentageOfNodesToScore for Scheduler, the default value is 50
  84. func WithPercentageOfNodesToScore(percentageOfNodesToScore int32) Option {
  85. return func(o *schedulerOptions) {
  86. o.percentageOfNodesToScore = percentageOfNodesToScore
  87. }
  88. }
  89. // WithBindTimeoutSeconds sets bindTimeoutSeconds for Scheduler, the default value is 100
  90. func WithBindTimeoutSeconds(bindTimeoutSeconds int64) Option {
  91. return func(o *schedulerOptions) {
  92. o.bindTimeoutSeconds = bindTimeoutSeconds
  93. }
  94. }
  95. var defaultSchedulerOptions = schedulerOptions{
  96. schedulerName: v1.DefaultSchedulerName,
  97. hardPodAffinitySymmetricWeight: v1.DefaultHardPodAffinitySymmetricWeight,
  98. disablePreemption: false,
  99. percentageOfNodesToScore: schedulerapi.DefaultPercentageOfNodesToScore,
  100. bindTimeoutSeconds: BindTimeoutSeconds,
  101. }
  102. // New returns a Scheduler
  103. func New(client clientset.Interface,
  104. nodeInformer coreinformers.NodeInformer,
  105. podInformer coreinformers.PodInformer,
  106. pvInformer coreinformers.PersistentVolumeInformer,
  107. pvcInformer coreinformers.PersistentVolumeClaimInformer,
  108. replicationControllerInformer coreinformers.ReplicationControllerInformer,
  109. replicaSetInformer appsinformers.ReplicaSetInformer,
  110. statefulSetInformer appsinformers.StatefulSetInformer,
  111. serviceInformer coreinformers.ServiceInformer,
  112. pdbInformer policyinformers.PodDisruptionBudgetInformer,
  113. storageClassInformer storageinformers.StorageClassInformer,
  114. recorder record.EventRecorder,
  115. schedulerAlgorithmSource kubeschedulerconfig.SchedulerAlgorithmSource,
  116. stopCh <-chan struct{},
  117. registry framework.Registry,
  118. plugins *kubeschedulerconfig.Plugins,
  119. pluginConfig []kubeschedulerconfig.PluginConfig,
  120. opts ...func(o *schedulerOptions)) (*Scheduler, error) {
  121. options := defaultSchedulerOptions
  122. for _, opt := range opts {
  123. opt(&options)
  124. }
  125. // Set up the configurator which can create schedulers from configs.
  126. configurator := factory.NewConfigFactory(&factory.ConfigFactoryArgs{
  127. SchedulerName: options.schedulerName,
  128. Client: client,
  129. NodeInformer: nodeInformer,
  130. PodInformer: podInformer,
  131. PvInformer: pvInformer,
  132. PvcInformer: pvcInformer,
  133. ReplicationControllerInformer: replicationControllerInformer,
  134. ReplicaSetInformer: replicaSetInformer,
  135. StatefulSetInformer: statefulSetInformer,
  136. ServiceInformer: serviceInformer,
  137. PdbInformer: pdbInformer,
  138. StorageClassInformer: storageClassInformer,
  139. HardPodAffinitySymmetricWeight: options.hardPodAffinitySymmetricWeight,
  140. DisablePreemption: options.disablePreemption,
  141. PercentageOfNodesToScore: options.percentageOfNodesToScore,
  142. BindTimeoutSeconds: options.bindTimeoutSeconds,
  143. Registry: registry,
  144. Plugins: plugins,
  145. PluginConfig: pluginConfig,
  146. })
  147. var config *factory.Config
  148. source := schedulerAlgorithmSource
  149. switch {
  150. case source.Provider != nil:
  151. // Create the config from a named algorithm provider.
  152. sc, err := configurator.CreateFromProvider(*source.Provider)
  153. if err != nil {
  154. return nil, fmt.Errorf("couldn't create scheduler using provider %q: %v", *source.Provider, err)
  155. }
  156. config = sc
  157. case source.Policy != nil:
  158. // Create the config from a user specified policy source.
  159. policy := &schedulerapi.Policy{}
  160. switch {
  161. case source.Policy.File != nil:
  162. if err := initPolicyFromFile(source.Policy.File.Path, policy); err != nil {
  163. return nil, err
  164. }
  165. case source.Policy.ConfigMap != nil:
  166. if err := initPolicyFromConfigMap(client, source.Policy.ConfigMap, policy); err != nil {
  167. return nil, err
  168. }
  169. }
  170. sc, err := configurator.CreateFromConfig(*policy)
  171. if err != nil {
  172. return nil, fmt.Errorf("couldn't create scheduler from policy: %v", err)
  173. }
  174. config = sc
  175. default:
  176. return nil, fmt.Errorf("unsupported algorithm source: %v", source)
  177. }
  178. // Additional tweaks to the config produced by the configurator.
  179. config.Recorder = recorder
  180. config.DisablePreemption = options.disablePreemption
  181. config.StopEverything = stopCh
  182. // Create the scheduler.
  183. sched := NewFromConfig(config)
  184. AddAllEventHandlers(sched, options.schedulerName, nodeInformer, podInformer, pvInformer, pvcInformer, serviceInformer, storageClassInformer)
  185. return sched, nil
  186. }
  187. // initPolicyFromFile initialize policy from file
  188. func initPolicyFromFile(policyFile string, policy *schedulerapi.Policy) error {
  189. // Use a policy serialized in a file.
  190. _, err := os.Stat(policyFile)
  191. if err != nil {
  192. return fmt.Errorf("missing policy config file %s", policyFile)
  193. }
  194. data, err := ioutil.ReadFile(policyFile)
  195. if err != nil {
  196. return fmt.Errorf("couldn't read policy config: %v", err)
  197. }
  198. err = runtime.DecodeInto(latestschedulerapi.Codec, []byte(data), policy)
  199. if err != nil {
  200. return fmt.Errorf("invalid policy: %v", err)
  201. }
  202. return nil
  203. }
  204. // initPolicyFromConfigMap initialize policy from configMap
  205. func initPolicyFromConfigMap(client clientset.Interface, policyRef *kubeschedulerconfig.SchedulerPolicyConfigMapSource, policy *schedulerapi.Policy) error {
  206. // Use a policy serialized in a config map value.
  207. policyConfigMap, err := client.CoreV1().ConfigMaps(policyRef.Namespace).Get(policyRef.Name, metav1.GetOptions{})
  208. if err != nil {
  209. return fmt.Errorf("couldn't get policy config map %s/%s: %v", policyRef.Namespace, policyRef.Name, err)
  210. }
  211. data, found := policyConfigMap.Data[kubeschedulerconfig.SchedulerPolicyConfigMapKey]
  212. if !found {
  213. return fmt.Errorf("missing policy config map value at key %q", kubeschedulerconfig.SchedulerPolicyConfigMapKey)
  214. }
  215. err = runtime.DecodeInto(latestschedulerapi.Codec, []byte(data), policy)
  216. if err != nil {
  217. return fmt.Errorf("invalid policy: %v", err)
  218. }
  219. return nil
  220. }
  221. // NewFromConfig returns a new scheduler using the provided Config.
  222. func NewFromConfig(config *factory.Config) *Scheduler {
  223. metrics.Register()
  224. return &Scheduler{
  225. config: config,
  226. }
  227. }
  228. //var timeout *time.Ticker
  229. // Run begins watching and scheduling. It waits for cache to be synced, then starts a goroutine and returns immediately.
  230. func (sched *Scheduler) Run() {
  231. if !sched.config.WaitForCacheSync() {
  232. return
  233. }
  234. //customcache.Timeout := time.NewTicker(time.Duration(10 * time.Second))
  235. go wait.Until(sched.scheduleOne, 0, sched.config.StopEverything)
  236. }
  237. // Config returns scheduler's config pointer. It is exposed for testing purposes.
  238. func (sched *Scheduler) Config() *factory.Config {
  239. return sched.config
  240. }
  241. // recordFailedSchedulingEvent records an event for the pod that indicates the
  242. // pod has failed to schedule.
  243. // NOTE: This function modifies "pod". "pod" should be copied before being passed.
  244. func (sched *Scheduler) recordSchedulingFailure(pod *v1.Pod, err error, reason string, message string) {
  245. sched.config.Error(pod, err)
  246. sched.config.Recorder.Event(pod, v1.EventTypeWarning, "FailedScheduling", message)
  247. sched.config.PodConditionUpdater.Update(pod, &v1.PodCondition{
  248. Type: v1.PodScheduled,
  249. Status: v1.ConditionFalse,
  250. Reason: reason,
  251. Message: err.Error(),
  252. })
  253. }
  254. // schedule implements the scheduling algorithm and returns the suggested result(host,
  255. // evaluated nodes number,feasible nodes number).
  256. func (sched *Scheduler) schedule(pod *v1.Pod) (core.ScheduleResult, error) {
  257. result, err := sched.config.Algorithm.Schedule(pod, sched.config.NodeLister)
  258. if err != nil {
  259. pod = pod.DeepCopy()
  260. sched.recordSchedulingFailure(pod, err, v1.PodReasonUnschedulable, err.Error())
  261. return core.ScheduleResult{}, err
  262. }
  263. return result, err
  264. }
  265. // preempt tries to create room for a pod that has failed to schedule, by preempting lower priority pods if possible.
  266. // If it succeeds, it adds the name of the node where preemption has happened to the pod spec.
  267. // It returns the node name and an error if any.
  268. func (sched *Scheduler) preempt(preemptor *v1.Pod, scheduleErr error) (string, error) {
  269. preemptor, err := sched.config.PodPreemptor.GetUpdatedPod(preemptor)
  270. if err != nil {
  271. klog.Errorf("Error getting the updated preemptor pod object: %v", err)
  272. return "", err
  273. }
  274. node, victims, nominatedPodsToClear, err := sched.config.Algorithm.Preempt(preemptor, sched.config.NodeLister, scheduleErr)
  275. if err != nil {
  276. klog.Errorf("Error preempting victims to make room for %v/%v.", preemptor.Namespace, preemptor.Name)
  277. return "", err
  278. }
  279. var nodeName = ""
  280. if node != nil {
  281. nodeName = node.Name
  282. // Update the scheduling queue with the nominated pod information. Without
  283. // this, there would be a race condition between the next scheduling cycle
  284. // and the time the scheduler receives a Pod Update for the nominated pod.
  285. sched.config.SchedulingQueue.UpdateNominatedPodForNode(preemptor, nodeName)
  286. // Make a call to update nominated node name of the pod on the API server.
  287. err = sched.config.PodPreemptor.SetNominatedNodeName(preemptor, nodeName)
  288. if err != nil {
  289. klog.Errorf("Error in preemption process. Cannot set 'NominatedPod' on pod %v/%v: %v", preemptor.Namespace, preemptor.Name, err)
  290. sched.config.SchedulingQueue.DeleteNominatedPodIfExists(preemptor)
  291. return "", err
  292. }
  293. for _, victim := range victims {
  294. if err := sched.config.PodPreemptor.DeletePod(victim); err != nil {
  295. klog.Errorf("Error preempting pod %v/%v: %v", victim.Namespace, victim.Name, err)
  296. return "", err
  297. }
  298. sched.config.Recorder.Eventf(victim, v1.EventTypeNormal, "Preempted", "by %v/%v on node %v", preemptor.Namespace, preemptor.Name, nodeName)
  299. }
  300. metrics.PreemptionVictims.Set(float64(len(victims)))
  301. }
  302. // Clearing nominated pods should happen outside of "if node != nil". Node could
  303. // be nil when a pod with nominated node name is eligible to preempt again,
  304. // but preemption logic does not find any node for it. In that case Preempt()
  305. // function of generic_scheduler.go returns the pod itself for removal of
  306. // the 'NominatedPod' field.
  307. for _, p := range nominatedPodsToClear {
  308. rErr := sched.config.PodPreemptor.RemoveNominatedNodeName(p)
  309. if rErr != nil {
  310. klog.Errorf("Cannot remove 'NominatedPod' field of pod: %v", rErr)
  311. // We do not return as this error is not critical.
  312. }
  313. }
  314. return nodeName, err
  315. }
  316. // assumeVolumes will update the volume cache with the chosen bindings
  317. //
  318. // This function modifies assumed if volume binding is required.
  319. func (sched *Scheduler) assumeVolumes(assumed *v1.Pod, host string) (allBound bool, err error) {
  320. allBound, err = sched.config.VolumeBinder.Binder.AssumePodVolumes(assumed, host)
  321. if err != nil {
  322. sched.recordSchedulingFailure(assumed, err, SchedulerError,
  323. fmt.Sprintf("AssumePodVolumes failed: %v", err))
  324. }
  325. return
  326. }
  327. // bindVolumes will make the API update with the assumed bindings and wait until
  328. // the PV controller has completely finished the binding operation.
  329. //
  330. // If binding errors, times out or gets undone, then an error will be returned to
  331. // retry scheduling.
  332. func (sched *Scheduler) bindVolumes(assumed *v1.Pod) error {
  333. klog.V(5).Infof("Trying to bind volumes for pod \"%v/%v\"", assumed.Namespace, assumed.Name)
  334. err := sched.config.VolumeBinder.Binder.BindPodVolumes(assumed)
  335. if err != nil {
  336. klog.V(1).Infof("Failed to bind volumes for pod \"%v/%v\": %v", assumed.Namespace, assumed.Name, err)
  337. // Unassume the Pod and retry scheduling
  338. if forgetErr := sched.config.SchedulerCache.ForgetPod(assumed); forgetErr != nil {
  339. klog.Errorf("scheduler cache ForgetPod failed: %v", forgetErr)
  340. }
  341. sched.recordSchedulingFailure(assumed, err, "VolumeBindingFailed", err.Error())
  342. return err
  343. }
  344. klog.V(5).Infof("Success binding volumes for pod \"%v/%v\"", assumed.Namespace, assumed.Name)
  345. return nil
  346. }
  347. // assume signals to the cache that a pod is already in the cache, so that binding can be asynchronous.
  348. // assume modifies `assumed`.
  349. func (sched *Scheduler) assume(assumed *v1.Pod, host string) error {
  350. // Optimistically assume that the binding will succeed and send it to apiserver
  351. // in the background.
  352. // If the binding fails, scheduler will release resources allocated to assumed pod
  353. // immediately.
  354. assumed.Spec.NodeName = host
  355. if err := sched.config.SchedulerCache.AssumePod(assumed); err != nil {
  356. klog.Errorf("scheduler cache AssumePod failed: %v", err)
  357. // This is most probably result of a BUG in retrying logic.
  358. // We report an error here so that pod scheduling can be retried.
  359. // This relies on the fact that Error will check if the pod has been bound
  360. // to a node and if so will not add it back to the unscheduled pods queue
  361. // (otherwise this would cause an infinite loop).
  362. sched.recordSchedulingFailure(assumed, err, SchedulerError,
  363. fmt.Sprintf("AssumePod failed: %v", err))
  364. return err
  365. }
  366. // if "assumed" is a nominated pod, we should remove it from internal cache
  367. if sched.config.SchedulingQueue != nil {
  368. sched.config.SchedulingQueue.DeleteNominatedPodIfExists(assumed)
  369. }
  370. return nil
  371. }
  372. // bind binds a pod to a given node defined in a binding object. We expect this to run asynchronously, so we
  373. // handle binding metrics internally.
  374. func (sched *Scheduler) bind(assumed *v1.Pod, b *v1.Binding) error {
  375. bindingStart := time.Now()
  376. // If binding succeeded then PodScheduled condition will be updated in apiserver so that
  377. // it's atomic with setting host.
  378. err := sched.config.GetBinder(assumed).Bind(b)
  379. if finErr := sched.config.SchedulerCache.FinishBinding(assumed); finErr != nil {
  380. klog.Errorf("scheduler cache FinishBinding failed: %v", finErr)
  381. }
  382. if err != nil {
  383. klog.V(1).Infof("Failed to bind pod: %v/%v", assumed.Namespace, assumed.Name)
  384. if err := sched.config.SchedulerCache.ForgetPod(assumed); err != nil {
  385. klog.Errorf("scheduler cache ForgetPod failed: %v", err)
  386. }
  387. sched.recordSchedulingFailure(assumed, err, SchedulerError,
  388. fmt.Sprintf("Binding rejected: %v", err))
  389. return err
  390. }
  391. metrics.BindingLatency.Observe(metrics.SinceInSeconds(bindingStart))
  392. metrics.DeprecatedBindingLatency.Observe(metrics.SinceInMicroseconds(bindingStart))
  393. metrics.SchedulingLatency.WithLabelValues(metrics.Binding).Observe(metrics.SinceInSeconds(bindingStart))
  394. metrics.DeprecatedSchedulingLatency.WithLabelValues(metrics.Binding).Observe(metrics.SinceInSeconds(bindingStart))
  395. sched.config.Recorder.Eventf(assumed, v1.EventTypeNormal, "Scheduled", "Successfully assigned %v/%v to %v", assumed.Namespace, assumed.Name, b.Target.Name)
  396. return nil
  397. }
  398. // scheduleOne does the entire scheduling workflow for a single pod. It is serialized on the scheduling algorithm's host fitting.
  399. func (sched *Scheduler) scheduleOne() {
  400. // // Check if the cache needs update
  401. // select {
  402. // // clean the cache if 10 seconds are passed
  403. // case <-customcache.LabCache.Timeout.C:
  404. // klog.Infof("Time to erase")
  405. // customcache.LabCache.CleanCache()
  406. // default:
  407. // }
  408. // Check if the cache needs update
  409. //klog.Infof("The value of the Ticker: %v", customcache.LabCache.Timeout.C)
  410. // select {
  411. // // clean the cache if 10 seconds are passed
  412. // case <-customcache.LabCache.Timeout.C:
  413. // klog.Infof("Time to erase: %v", time.Now())
  414. // //customcache.LabCache.Timeout.Stop()
  415. // customcache.LabCache.CleanCache()
  416. // klog.Infof("Cache: %v", customcache.LabCache.Cache)
  417. // default:
  418. // }
  419. fwk := sched.config.Framework
  420. pod := sched.config.NextPod()
  421. // pod could be nil when schedulerQueue is closed
  422. if pod == nil {
  423. return
  424. }
  425. if pod.DeletionTimestamp != nil {
  426. sched.config.Recorder.Eventf(pod, v1.EventTypeWarning, "FailedScheduling", "skip schedule deleting pod: %v/%v", pod.Namespace, pod.Name)
  427. klog.V(3).Infof("Skip schedule deleting pod: %v/%v", pod.Namespace, pod.Name)
  428. return
  429. }
  430. klog.V(3).Infof("Attempting to schedule pod: %v/%v", pod.Namespace, pod.Name)
  431. // Synchronously attempt to find a fit for the pod.
  432. start := time.Now()
  433. pluginContext := framework.NewPluginContext()
  434. scheduleResult, err := sched.schedule(pod)
  435. if err != nil {
  436. // schedule() may have failed because the pod would not fit on any host, so we try to
  437. // preempt, with the expectation that the next time the pod is tried for scheduling it
  438. // will fit due to the preemption. It is also possible that a different pod will schedule
  439. // into the resources that were preempted, but this is harmless.
  440. if fitError, ok := err.(*core.FitError); ok {
  441. if !util.PodPriorityEnabled() || sched.config.DisablePreemption {
  442. klog.V(3).Infof("Pod priority feature is not enabled or preemption is disabled by scheduler configuration." +
  443. " No preemption is performed.")
  444. } else {
  445. preemptionStartTime := time.Now()
  446. sched.preempt(pod, fitError)
  447. metrics.PreemptionAttempts.Inc()
  448. metrics.SchedulingAlgorithmPremptionEvaluationDuration.Observe(metrics.SinceInSeconds(preemptionStartTime))
  449. metrics.DeprecatedSchedulingAlgorithmPremptionEvaluationDuration.Observe(metrics.SinceInMicroseconds(preemptionStartTime))
  450. metrics.SchedulingLatency.WithLabelValues(metrics.PreemptionEvaluation).Observe(metrics.SinceInSeconds(preemptionStartTime))
  451. metrics.DeprecatedSchedulingLatency.WithLabelValues(metrics.PreemptionEvaluation).Observe(metrics.SinceInSeconds(preemptionStartTime))
  452. }
  453. // Pod did not fit anywhere, so it is counted as a failure. If preemption
  454. // succeeds, the pod should get counted as a success the next time we try to
  455. // schedule it. (hopefully)
  456. metrics.PodScheduleFailures.Inc()
  457. } else {
  458. klog.Errorf("error selecting node for pod: %v", err)
  459. metrics.PodScheduleErrors.Inc()
  460. }
  461. return
  462. }
  463. metrics.SchedulingAlgorithmLatency.Observe(metrics.SinceInSeconds(start))
  464. metrics.DeprecatedSchedulingAlgorithmLatency.Observe(metrics.SinceInMicroseconds(start))
  465. // Tell the cache to assume that a pod now is running on a given node, even though it hasn't been bound yet.
  466. // This allows us to keep scheduling without waiting on binding to occur.
  467. assumedPod := pod.DeepCopy()
  468. // Assume volumes first before assuming the pod.
  469. //
  470. // If all volumes are completely bound, then allBound is true and binding will be skipped.
  471. //
  472. // Otherwise, binding of volumes is started after the pod is assumed, but before pod binding.
  473. //
  474. // This function modifies 'assumedPod' if volume binding is required.
  475. allBound, err := sched.assumeVolumes(assumedPod, scheduleResult.SuggestedHost)
  476. if err != nil {
  477. klog.Errorf("error assuming volumes: %v", err)
  478. metrics.PodScheduleErrors.Inc()
  479. return
  480. }
  481. // Run "reserve" plugins.
  482. if sts := fwk.RunReservePlugins(pluginContext, assumedPod, scheduleResult.SuggestedHost); !sts.IsSuccess() {
  483. sched.recordSchedulingFailure(assumedPod, sts.AsError(), SchedulerError, sts.Message())
  484. metrics.PodScheduleErrors.Inc()
  485. return
  486. }
  487. // assume modifies `assumedPod` by setting NodeName=scheduleResult.SuggestedHost
  488. err = sched.assume(assumedPod, scheduleResult.SuggestedHost)
  489. if err != nil {
  490. klog.Errorf("error assuming pod: %v", err)
  491. metrics.PodScheduleErrors.Inc()
  492. // trigger un-reserve plugins to clean up state associated with the reserved Pod
  493. fwk.RunUnreservePlugins(pluginContext, assumedPod, scheduleResult.SuggestedHost)
  494. return
  495. }
  496. // bind the pod to its host asynchronously (we can do this b/c of the assumption step above).
  497. go func() {
  498. // Bind volumes first before Pod
  499. if !allBound {
  500. err := sched.bindVolumes(assumedPod)
  501. if err != nil {
  502. klog.Errorf("error binding volumes: %v", err)
  503. metrics.PodScheduleErrors.Inc()
  504. // trigger un-reserve plugins to clean up state associated with the reserved Pod
  505. fwk.RunUnreservePlugins(pluginContext, assumedPod, scheduleResult.SuggestedHost)
  506. return
  507. }
  508. }
  509. // Run "permit" plugins.
  510. permitStatus := fwk.RunPermitPlugins(pluginContext, assumedPod, scheduleResult.SuggestedHost)
  511. if !permitStatus.IsSuccess() {
  512. var reason string
  513. if permitStatus.Code() == framework.Unschedulable {
  514. reason = v1.PodReasonUnschedulable
  515. } else {
  516. metrics.PodScheduleErrors.Inc()
  517. reason = SchedulerError
  518. }
  519. if forgetErr := sched.Cache().ForgetPod(assumedPod); forgetErr != nil {
  520. klog.Errorf("scheduler cache ForgetPod failed: %v", forgetErr)
  521. }
  522. sched.recordSchedulingFailure(assumedPod, permitStatus.AsError(), reason, permitStatus.Message())
  523. // trigger un-reserve plugins to clean up state associated with the reserved Pod
  524. fwk.RunUnreservePlugins(pluginContext, assumedPod, scheduleResult.SuggestedHost)
  525. return
  526. }
  527. // Run "prebind" plugins.
  528. prebindStatus := fwk.RunPrebindPlugins(pluginContext, assumedPod, scheduleResult.SuggestedHost)
  529. if !prebindStatus.IsSuccess() {
  530. var reason string
  531. if prebindStatus.Code() == framework.Unschedulable {
  532. reason = v1.PodReasonUnschedulable
  533. } else {
  534. metrics.PodScheduleErrors.Inc()
  535. reason = SchedulerError
  536. }
  537. if forgetErr := sched.Cache().ForgetPod(assumedPod); forgetErr != nil {
  538. klog.Errorf("scheduler cache ForgetPod failed: %v", forgetErr)
  539. }
  540. sched.recordSchedulingFailure(assumedPod, prebindStatus.AsError(), reason, prebindStatus.Message())
  541. // trigger un-reserve plugins to clean up state associated with the reserved Pod
  542. fwk.RunUnreservePlugins(pluginContext, assumedPod, scheduleResult.SuggestedHost)
  543. return
  544. }
  545. err := sched.bind(assumedPod, &v1.Binding{
  546. ObjectMeta: metav1.ObjectMeta{Namespace: assumedPod.Namespace, Name: assumedPod.Name, UID: assumedPod.UID},
  547. Target: v1.ObjectReference{
  548. Kind: "Node",
  549. Name: scheduleResult.SuggestedHost,
  550. },
  551. })
  552. metrics.E2eSchedulingLatency.Observe(metrics.SinceInSeconds(start))
  553. metrics.DeprecatedE2eSchedulingLatency.Observe(metrics.SinceInMicroseconds(start))
  554. if err != nil {
  555. klog.Errorf("error binding pod: %v", err)
  556. metrics.PodScheduleErrors.Inc()
  557. // trigger un-reserve plugins to clean up state associated with the reserved Pod
  558. fwk.RunUnreservePlugins(pluginContext, assumedPod, scheduleResult.SuggestedHost)
  559. } else {
  560. klog.V(2).Infof("pod %v/%v is bound successfully on node %v, %d nodes evaluated, %d nodes were found feasible", assumedPod.Namespace, assumedPod.Name, scheduleResult.SuggestedHost, scheduleResult.EvaluatedNodes, scheduleResult.FeasibleNodes)
  561. metrics.PodScheduleSuccesses.Inc()
  562. // Run "postbind" plugins.
  563. fwk.RunPostbindPlugins(pluginContext, assumedPod, scheduleResult.SuggestedHost)
  564. }
  565. }()
  566. }