From a80b1798c45f88e6b38261a1657c0dfa787de136 Mon Sep 17 00:00:00 2001 From: Filip Grzadkowski Date: Mon, 18 Apr 2016 14:26:16 +0200 Subject: [PATCH] Add pod condition PodScheduled to detect situation when scheduler tried to schedule a Pod, but failed. Ref #24404 --- pkg/api/resource_helpers.go | 39 +++++++++++++++++++++++-- pkg/api/types.go | 2 ++ pkg/api/v1/types.go | 2 ++ pkg/kubelet/kubelet.go | 10 +++++++ pkg/registry/pod/etcd/etcd.go | 4 +++ plugin/pkg/scheduler/factory/factory.go | 20 +++++++++++-- plugin/pkg/scheduler/scheduler.go | 22 ++++++++++++++ plugin/pkg/scheduler/scheduler_test.go | 7 +++++ test/e2e/scheduler_predicates.go | 6 ++++ 9 files changed, 106 insertions(+), 6 deletions(-) diff --git a/pkg/api/resource_helpers.go b/pkg/api/resource_helpers.go index 7a98a4c2f0701..3ebd80df90363 100644 --- a/pkg/api/resource_helpers.go +++ b/pkg/api/resource_helpers.go @@ -18,6 +18,7 @@ package api import ( "k8s.io/kubernetes/pkg/api/resource" + "k8s.io/kubernetes/pkg/api/unversioned" ) // Returns string version of ResourceName. @@ -80,12 +81,44 @@ func IsPodReadyConditionTrue(status PodStatus) bool { // Extracts the pod ready condition from the given status and returns that. // Returns nil if the condition is not present. func GetPodReadyCondition(status PodStatus) *PodCondition { + _, condition := GetPodCondition(&status, PodReady) + return condition +} + +func GetPodCondition(status *PodStatus, conditionType PodConditionType) (int, *PodCondition) { for i, c := range status.Conditions { - if c.Type == PodReady { - return &status.Conditions[i] + if c.Type == conditionType { + return i, &status.Conditions[i] + } + } + return -1, nil +} + +// Updates existing pod condition or creates a new one. Sets LastTransitionTime to now if the +// status has changed. +// Returns true if pod condition has changed or has been added. +func UpdatePodCondition(status *PodStatus, condition *PodCondition) bool { + condition.LastTransitionTime = unversioned.Now() + // Try to find this pod condition. + conditionIndex, oldCondition := GetPodCondition(status, condition.Type) + + if oldCondition == nil { + // We are adding new pod condition. + status.Conditions = append(status.Conditions, *condition) + return true + } else { + // We are updating an existing condition, so we need to check if it has changed. + if condition.Status == oldCondition.Status { + condition.LastTransitionTime = oldCondition.LastTransitionTime } + status.Conditions[conditionIndex] = *condition + // Return true if one of the fields have changed. + return condition.Status != oldCondition.Status || + condition.Reason != oldCondition.Reason || + condition.Message != oldCondition.Message || + !condition.LastProbeTime.Equal(oldCondition.LastProbeTime) || + !condition.LastTransitionTime.Equal(oldCondition.LastTransitionTime) } - return nil } // IsNodeReady returns true if a node is ready; false otherwise. diff --git a/pkg/api/types.go b/pkg/api/types.go index 09eff349810e9..425810c233f22 100644 --- a/pkg/api/types.go +++ b/pkg/api/types.go @@ -1058,6 +1058,8 @@ type PodConditionType string // These are valid conditions of pod. const ( + // PodScheduled represents status of the scheduling process for this pod. + PodScheduled PodConditionType = "PodScheduled" // PodReady means the pod is able to service requests and should be added to the // load balancing pools of all matching services. PodReady PodConditionType = "Ready" diff --git a/pkg/api/v1/types.go b/pkg/api/v1/types.go index 9f53d8b628093..49ec71dddd27b 100644 --- a/pkg/api/v1/types.go +++ b/pkg/api/v1/types.go @@ -1288,6 +1288,8 @@ type PodConditionType string // These are valid conditions of pod. const ( + // PodScheduled represents status of the scheduling process for this pod. + PodScheduled PodConditionType = "PodScheduled" // PodReady means the pod is able to service requests and should be added to the // load balancing pools of all matching services. PodReady PodConditionType = "Ready" diff --git a/pkg/kubelet/kubelet.go b/pkg/kubelet/kubelet.go index c52eb6710288b..2592270aee756 100644 --- a/pkg/kubelet/kubelet.go +++ b/pkg/kubelet/kubelet.go @@ -3286,6 +3286,16 @@ func (kl *Kubelet) generateAPIPodStatus(pod *api.Pod, podStatus *kubecontainer.P s.Phase = GetPhase(spec, s.ContainerStatuses) kl.probeManager.UpdatePodStatus(pod.UID, s) s.Conditions = append(s.Conditions, status.GeneratePodReadyCondition(spec, s.ContainerStatuses, s.Phase)) + // s (the PodStatus we are creating) will not have a PodScheduled condition yet, because converStatusToAPIStatus() + // does not create one. If the existing PodStatus has a PodScheduled condition, then copy it into s and make sure + // it is set to true. If the existing PodStatus does not have a PodScheduled condition, then create one that is set to true. + if _, oldPodScheduled := api.GetPodCondition(&pod.Status, api.PodScheduled); oldPodScheduled != nil { + s.Conditions = append(s.Conditions, *oldPodScheduled) + } + api.UpdatePodCondition(&pod.Status, &api.PodCondition{ + Type: api.PodScheduled, + Status: api.ConditionTrue, + }) if !kl.standaloneMode { hostIP, err := kl.GetHostIP() diff --git a/pkg/registry/pod/etcd/etcd.go b/pkg/registry/pod/etcd/etcd.go index 33dfb949bf471..dee4807c71a6f 100644 --- a/pkg/registry/pod/etcd/etcd.go +++ b/pkg/registry/pod/etcd/etcd.go @@ -166,6 +166,10 @@ func (r *BindingREST) setPodHostAndAnnotations(ctx api.Context, podID, oldMachin for k, v := range annotations { pod.Annotations[k] = v } + api.UpdatePodCondition(&pod.Status, &api.PodCondition{ + Type: api.PodScheduled, + Status: api.ConditionTrue, + }) finalPod = pod return pod, nil })) diff --git a/plugin/pkg/scheduler/factory/factory.go b/plugin/pkg/scheduler/factory/factory.go index 55e2b51610e10..1c45c53c884a3 100644 --- a/plugin/pkg/scheduler/factory/factory.go +++ b/plugin/pkg/scheduler/factory/factory.go @@ -286,9 +286,10 @@ func (f *ConfigFactory) CreateFromKeys(predicateKeys, priorityKeys sets.String, return &scheduler.Config{ SchedulerCache: f.schedulerCache, // The scheduler only needs to consider schedulable nodes. - NodeLister: f.NodeLister.NodeCondition(getNodeConditionPredicate()), - Algorithm: algo, - Binder: &binder{f.Client}, + NodeLister: f.NodeLister.NodeCondition(getNodeConditionPredicate()), + Algorithm: algo, + Binder: &binder{f.Client}, + PodConditionUpdater: &podConditionUpdater{f.Client}, NextPod: func() *api.Pod { return f.getNextPod() }, @@ -448,6 +449,19 @@ func (b *binder) Bind(binding *api.Binding) error { // return b.Pods(binding.Namespace).Bind(binding) } +type podConditionUpdater struct { + *client.Client +} + +func (p *podConditionUpdater) Update(pod *api.Pod, condition *api.PodCondition) error { + glog.V(2).Infof("Updating pod condition for %s/%s to (%s==%s)", pod.Namespace, pod.Name, condition.Type, condition.Status) + if api.UpdatePodCondition(&pod.Status, condition) { + _, err := p.Pods(pod.Namespace).UpdateStatus(pod) + return err + } + return nil +} + type clock interface { Now() time.Time } diff --git a/plugin/pkg/scheduler/scheduler.go b/plugin/pkg/scheduler/scheduler.go index b4f0f0c2e7f41..49b81c5832380 100644 --- a/plugin/pkg/scheduler/scheduler.go +++ b/plugin/pkg/scheduler/scheduler.go @@ -37,6 +37,10 @@ type Binder interface { Bind(binding *api.Binding) error } +type PodConditionUpdater interface { + Update(pod *api.Pod, podCondition *api.PodCondition) error +} + // Scheduler watches for new unscheduled pods. It attempts to find // nodes that they fit on and writes bindings back to the api server. type Scheduler struct { @@ -50,6 +54,10 @@ type Config struct { NodeLister algorithm.NodeLister Algorithm algorithm.ScheduleAlgorithm Binder Binder + // PodConditionUpdater is used only in case of scheduling errors. If we succeed + // with scheduling, PodScheduled condition will be updated in apiserver in /bind + // handler so that binding and setting PodCondition it is atomic. + PodConditionUpdater PodConditionUpdater // NextPod should be a function that blocks until the next pod // is available. We don't use a channel for this, because scheduling @@ -92,6 +100,12 @@ func (s *Scheduler) scheduleOne() { glog.V(1).Infof("Failed to schedule: %+v", pod) s.config.Error(pod, err) s.config.Recorder.Eventf(pod, api.EventTypeWarning, "FailedScheduling", "%v", err) + s.config.PodConditionUpdater.Update(pod, &api.PodCondition{ + Type: api.PodScheduled, + Status: api.ConditionFalse, + Reason: "Unschedulable", + Message: err.Error(), + }) return } metrics.SchedulingAlgorithmLatency.Observe(metrics.SinceInMicroseconds(start)) @@ -120,11 +134,19 @@ func (s *Scheduler) scheduleOne() { } bindingStart := time.Now() + // If binding succeded then PodScheduled condition will be updated in apiserver so that + // it's atomic with setting host. err := s.config.Binder.Bind(b) if err != nil { glog.V(1).Infof("Failed to bind pod: %+v", err) s.config.Error(pod, err) s.config.Recorder.Eventf(pod, api.EventTypeNormal, "FailedScheduling", "Binding rejected: %v", err) + s.config.PodConditionUpdater.Update(pod, &api.PodCondition{ + Type: api.PodScheduled, + Status: api.ConditionFalse, + Reason: "BindingRejected", + Message: err.Error(), + }) return } metrics.BindingLatency.Observe(metrics.SinceInMicroseconds(bindingStart)) diff --git a/plugin/pkg/scheduler/scheduler_test.go b/plugin/pkg/scheduler/scheduler_test.go index 01843e16e5e3e..614a9d81f6a30 100644 --- a/plugin/pkg/scheduler/scheduler_test.go +++ b/plugin/pkg/scheduler/scheduler_test.go @@ -43,6 +43,12 @@ type fakeBinder struct { func (fb fakeBinder) Bind(binding *api.Binding) error { return fb.b(binding) } +type fakePodConditionUpdater struct{} + +func (fc fakePodConditionUpdater) Update(pod *api.Pod, podCondition *api.PodCondition) error { + return nil +} + func podWithID(id, desiredHost string) *api.Pod { return &api.Pod{ ObjectMeta: api.ObjectMeta{Name: id, SelfLink: testapi.Default.SelfLink("pods", id)}, @@ -128,6 +134,7 @@ func TestScheduler(t *testing.T) { gotBinding = b return item.injectBindError }}, + PodConditionUpdater: fakePodConditionUpdater{}, Error: func(p *api.Pod, err error) { gotPod = p gotError = err diff --git a/test/e2e/scheduler_predicates.go b/test/e2e/scheduler_predicates.go index 6a094de484ff5..fdb29e1994d47 100644 --- a/test/e2e/scheduler_predicates.go +++ b/test/e2e/scheduler_predicates.go @@ -45,8 +45,14 @@ func getPodsScheduled(pods *api.PodList) (scheduledPods, notScheduledPods []api. for _, pod := range pods.Items { if !masterNodes.Has(pod.Spec.NodeName) { if pod.Spec.NodeName != "" { + _, scheduledCondition := api.GetPodCondition(&pod.Status, api.PodScheduled) + Expect(scheduledCondition != nil).To(Equal(true)) + Expect(scheduledCondition.Status).To(Equal(api.ConditionTrue)) scheduledPods = append(scheduledPods, pod) } else { + _, scheduledCondition := api.GetPodCondition(&pod.Status, api.PodScheduled) + Expect(scheduledCondition != nil).To(Equal(true)) + Expect(scheduledCondition.Status).To(Equal(api.ConditionFalse)) notScheduledPods = append(notScheduledPods, pod) } }