Skip to content

Commit

Permalink
update Scheduled conditon when failed scheduling
Browse files Browse the repository at this point in the history 
Signed-off-by: lihanbo <[email protected]>
  • Loading branch information
@mrlihanbo
mrlihanbo committed Nov 30, 2021
1 parent 6dfa010 commit 3d1218c
Showing 1 changed file with 75 additions and 64 deletions.
139 changes: 75 additions & 64 deletions pkg/scheduler/scheduler.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@ import (
"k8s.io/client-go/dynamic"
"k8s.io/client-go/kubernetes"
"k8s.io/client-go/tools/cache"
"k8s.io/client-go/util/retry"
"k8s.io/client-go/util/workqueue"
"k8s.io/klog/v2"

Expand All @@ -37,16 +38,12 @@ import (
"github.com/karmada-io/karmada/pkg/scheduler/framework/plugins/tainttoleration"
"github.com/karmada-io/karmada/pkg/scheduler/metrics"
"github.com/karmada-io/karmada/pkg/util"
"github.com/karmada-io/karmada/pkg/util/helper"
)

// ScheduleType defines the schedule type of a binding object should be performed.
type ScheduleType string

const (
// FirstSchedule means the binding object hasn't been scheduled.
FirstSchedule ScheduleType = "FirstSchedule"

// ReconcileSchedule means the binding object associated policy has been changed.
ReconcileSchedule ScheduleType = "ReconcileSchedule"

Expand All @@ -58,8 +55,8 @@ const (
)

const (
scheduleSuccessReason = "BindingScheduled"

scheduleSuccessReason = "BindingScheduled"
scheduleFailedReason = "BindingFailedScheduling"
scheduleSuccessMessage = "the binding has been scheduled"
)

Expand Down Expand Up @@ -398,22 +395,24 @@ func (s *Scheduler) doScheduleBinding(namespace, name string) error {
return err
}

defer func() {
if err != nil && rb != nil {
failedSchedulingCondition := util.NewCondition(workv1alpha2.Scheduled, scheduleFailedReason, err.Error(), metav1.ConditionFalse)
if updateErr := s.updateBindingScheduledConditionIfNeeded(rb, failedSchedulingCondition); updateErr != nil {
klog.Errorf("failed to set failed scheduling condition for binding(%s/%s): %v", rb.Namespace, rb.Name, updateErr)
}
}
}()

start := time.Now()
if !helper.IsBindingReady(&rb.Status) {
// the binding has not been scheduled, need schedule
klog.Infof("Start scheduling ResourceBinding(%s/%s)", namespace, name)
err = s.scheduleResourceBinding(rb)
metrics.BindingSchedule(string(FirstSchedule), metrics.SinceInSeconds(start), err)
return err
}
policyPlacement, policyPlacementStr, err := s.getPlacement(rb)
if err != nil {
return err
}
appliedPlacement := util.GetLabelValue(rb.Annotations, util.PolicyPlacementAnnotation)
if policyPlacementStr != appliedPlacement {
// policy placement changed, need reschedule
klog.Infof("Reschedule ResourceBinding(%s/%s) as placement changed", namespace, name)
// policy placement changed, need schedule
klog.Infof("Start to schedule ResourceBinding(%s/%s) as placement changed", namespace, name)
err = s.scheduleResourceBinding(rb)
metrics.BindingSchedule(string(ReconcileSchedule), metrics.SinceInSeconds(start), err)
return err
Expand Down Expand Up @@ -449,22 +448,24 @@ func (s *Scheduler) doScheduleClusterBinding(name string) error {
return err
}

defer func() {
if err != nil && crb != nil {
failedSchedulingCondition := util.NewCondition(workv1alpha2.Scheduled, scheduleFailedReason, err.Error(), metav1.ConditionFalse)
if updateErr := s.updateClusterBindingScheduledConditionIfNeeded(crb, failedSchedulingCondition); updateErr != nil {
klog.Errorf("failed to set failed scheduling condition for cluster resource binding(%s): %v", crb.Name, updateErr)
}
}
}()

start := time.Now()
if !helper.IsBindingReady(&crb.Status) {
// the binding has not been scheduled, need schedule
klog.Infof("Start scheduling ClusterResourceBinding(%s)", name)
err = s.scheduleClusterResourceBinding(crb)
metrics.BindingSchedule(string(FirstSchedule), metrics.SinceInSeconds(start), err)
return err
}
policyPlacement, policyPlacementStr, err := s.getClusterPlacement(crb)
if err != nil {
return err
}
appliedPlacement := util.GetLabelValue(crb.Annotations, util.PolicyPlacementAnnotation)
if policyPlacementStr != appliedPlacement {
// policy placement changed, need reschedule
klog.Infof("Reschedule ClusterResourceBinding(%s) as placement changed", name)
// policy placement changed, need schedule
klog.Infof("Start to schedule ClusterResourceBinding(%s) as placement changed", name)
err = s.scheduleClusterResourceBinding(crb)
metrics.BindingSchedule(string(ReconcileSchedule), metrics.SinceInSeconds(start), err)
return err
Expand Down Expand Up @@ -517,7 +518,8 @@ func (s *Scheduler) scheduleResourceBinding(resourceBinding *workv1alpha2.Resour
if err != nil {
return err
}
return s.updateBindingStatusIfNeeded(binding)

return s.updateBindingScheduledConditionIfNeeded(binding, util.NewCondition(workv1alpha2.Scheduled, scheduleSuccessReason, scheduleSuccessMessage, metav1.ConditionTrue))
}

func (s *Scheduler) scheduleClusterResourceBinding(clusterResourceBinding *workv1alpha2.ClusterResourceBinding) (err error) {
Expand Down Expand Up @@ -554,7 +556,8 @@ func (s *Scheduler) scheduleClusterResourceBinding(clusterResourceBinding *workv
if err != nil {
return err
}
return s.updateClusterBindingStatusIfNeeded(binding)

return s.updateClusterBindingScheduledConditionIfNeeded(binding, util.NewCondition(workv1alpha2.Scheduled, scheduleSuccessReason, scheduleSuccessMessage, metav1.ConditionTrue))
}

func (s *Scheduler) handleErr(err error, key interface{}) {
Expand Down Expand Up @@ -694,7 +697,8 @@ func (s *Scheduler) rescheduleClusterResourceBinding(clusterResourceBinding *wor
if err != nil {
return err
}
return s.updateClusterBindingStatusIfNeeded(clusterResourceBinding)

return s.updateClusterBindingScheduledConditionIfNeeded(clusterResourceBinding, util.NewCondition(workv1alpha2.Scheduled, scheduleSuccessReason, scheduleSuccessMessage, metav1.ConditionTrue))
}

func (s *Scheduler) rescheduleResourceBinding(resourceBinding *workv1alpha2.ResourceBinding) error {
Expand All @@ -721,7 +725,8 @@ func (s *Scheduler) rescheduleResourceBinding(resourceBinding *workv1alpha2.Reso
if err != nil {
return err
}
return s.updateBindingStatusIfNeeded(resourceBinding)

return s.updateBindingScheduledConditionIfNeeded(resourceBinding, util.NewCondition(workv1alpha2.Scheduled, scheduleSuccessReason, scheduleSuccessMessage, metav1.ConditionTrue))
}

func (s *Scheduler) scaleScheduleResourceBinding(resourceBinding *workv1alpha2.ResourceBinding) (err error) {
Expand Down Expand Up @@ -752,7 +757,8 @@ func (s *Scheduler) scaleScheduleResourceBinding(resourceBinding *workv1alpha2.R
if err != nil {
return err
}
return s.updateBindingStatusIfNeeded(binding)

return s.updateBindingScheduledConditionIfNeeded(binding, util.NewCondition(workv1alpha2.Scheduled, scheduleSuccessReason, scheduleSuccessMessage, metav1.ConditionTrue))
}

func (s *Scheduler) scaleScheduleClusterResourceBinding(clusterResourceBinding *workv1alpha2.ClusterResourceBinding) (err error) {
Expand Down Expand Up @@ -790,7 +796,8 @@ func (s *Scheduler) scaleScheduleClusterResourceBinding(clusterResourceBinding *
if err != nil {
return err
}
return s.updateClusterBindingStatusIfNeeded(binding)

return s.updateClusterBindingScheduledConditionIfNeeded(binding, util.NewCondition(workv1alpha2.Scheduled, scheduleSuccessReason, scheduleSuccessMessage, metav1.ConditionTrue))
}

func (s *Scheduler) allClustersInReadyState(tcs []workv1alpha2.TargetCluster) bool {
Expand Down Expand Up @@ -838,46 +845,50 @@ func (s *Scheduler) establishEstimatorConnections() {
}
}

// updateBindingStatusIfNeeded sets the scheduled condition of ResourceBinding to true if needed
func (s *Scheduler) updateBindingStatusIfNeeded(rb *workv1alpha2.ResourceBinding) error {
oldScheduledCondition := meta.FindStatusCondition(rb.Status.Conditions, workv1alpha2.Scheduled)
newScheduledCondition := metav1.Condition{
Type: workv1alpha2.Scheduled,
Status: metav1.ConditionTrue,
Reason: scheduleSuccessReason,
Message: scheduleSuccessMessage,
}
if equality.Semantic.DeepEqual(oldScheduledCondition, newScheduledCondition) {
// updateBindingScheduledConditionIfNeeded sets the scheduled condition of ResourceBinding if needed
func (s *Scheduler) updateBindingScheduledConditionIfNeeded(rb *workv1alpha2.ResourceBinding, newScheduledCondition metav1.Condition) error {
if meta.IsStatusConditionPresentAndEqual(rb.Status.Conditions, workv1alpha2.Scheduled, newScheduledCondition.Status) {
return nil
}

meta.SetStatusCondition(&rb.Status.Conditions, newScheduledCondition)
_, err := s.KarmadaClient.WorkV1alpha2().ResourceBindings(rb.Namespace).UpdateStatus(context.TODO(), rb, metav1.UpdateOptions{})
if err != nil {
klog.Errorf("Failed to update ResourceBinding status(%s/%s): %v", rb.Namespace, rb.Name, err)
return err
}
return nil
return retry.RetryOnConflict(retry.DefaultRetry, func() error {
meta.SetStatusCondition(&rb.Status.Conditions, newScheduledCondition)
_, updateErr := s.KarmadaClient.WorkV1alpha2().ResourceBindings(rb.Namespace).UpdateStatus(context.TODO(), rb, metav1.UpdateOptions{})
if updateErr == nil {
return nil
}

if updated, err := s.KarmadaClient.WorkV1alpha2().ResourceBindings(rb.Namespace).Get(context.TODO(), rb.Name, metav1.GetOptions{}); err == nil {
// make a copy so we don't mutate the shared cache
rb = updated.DeepCopy()
} else {
klog.Errorf("failed to get updated resource binding %s/%s: %v", rb.Namespace, rb.Name, err)
}

return updateErr
})
}

// updateClusterBindingStatusIfNeeded sets the scheduled condition of ClusterResourceBinding to true if needed
func (s *Scheduler) updateClusterBindingStatusIfNeeded(crb *workv1alpha2.ClusterResourceBinding) error {
oldScheduledCondition := meta.FindStatusCondition(crb.Status.Conditions, workv1alpha2.Scheduled)
newScheduledCondition := metav1.Condition{
Type: workv1alpha2.Scheduled,
Status: metav1.ConditionTrue,
Reason: scheduleSuccessReason,
Message: scheduleSuccessMessage,
}
if equality.Semantic.DeepEqual(oldScheduledCondition, newScheduledCondition) {
// updateClusterBindingScheduledConditionIfNeeded sets the scheduled condition of ClusterResourceBinding if needed
func (s *Scheduler) updateClusterBindingScheduledConditionIfNeeded(crb *workv1alpha2.ClusterResourceBinding, newScheduledCondition metav1.Condition) error {
if meta.IsStatusConditionPresentAndEqual(crb.Status.Conditions, workv1alpha2.Scheduled, newScheduledCondition.Status) {
return nil
}

meta.SetStatusCondition(&crb.Status.Conditions, newScheduledCondition)
_, err := s.KarmadaClient.WorkV1alpha2().ClusterResourceBindings().UpdateStatus(context.TODO(), crb, metav1.UpdateOptions{})
if err != nil {
klog.Errorf("Failed to update ClusterResourceBinding status(%s): %v", crb.Name, err)
return err
}
return nil
return retry.RetryOnConflict(retry.DefaultBackoff, func() error {
meta.SetStatusCondition(&crb.Status.Conditions, newScheduledCondition)
_, updateErr := s.KarmadaClient.WorkV1alpha2().ClusterResourceBindings().UpdateStatus(context.TODO(), crb, metav1.UpdateOptions{})
if updateErr == nil {
return nil
}

if updated, err := s.KarmadaClient.WorkV1alpha2().ClusterResourceBindings().Get(context.TODO(), crb.Name, metav1.GetOptions{}); err == nil {
// make a copy so we don't mutate the shared cache
crb = updated.DeepCopy()
} else {
klog.Errorf("failed to get updated cluster resource binding %s: %v", crb.Name, err)
}

return updateErr
})
}

0 comments on commit 3d1218c

Please sign in to comment.