scheduler.go 24 KB

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