Karmada Failover 详解¶

func getClusterHealthStatus(clusterClient *util.ClusterClient) (online, healthy bool) {
    healthStatus, err := healthEndpointCheck(clusterClient.KubeClient, "/readyz")
    if err != nil && healthStatus == http.StatusNotFound {
        // do health check with healthz endpoint if the readyz endpoint is not installed in member cluster
        healthStatus, err = healthEndpointCheck(clusterClient.KubeClient, "/healthz")
    }

    if err != nil {
        klog.Errorf("Failed to do cluster health check for cluster %v, err is : %v ", clusterClient.ClusterName, err)
        return false, false
    }

    if healthStatus != http.StatusOK {
        klog.Infof("Member cluster %v isn't healthy", clusterClient.ClusterName)
        return true, false
    }

    return true, true
}

func generateReadyCondition(imagesalthy bool) metav1.Condition {
    if !online {
        return util.NewCondition(clusterv1alpha1.ClusterConditionReady, clusterNotReachableReason, clusterNotReachableMsg, metav1.ConditionFalse)
    }
    if !healthy {
        return util.NewCondition(clusterv1alpha1.ClusterConditionReady, clusterNotReady, clusterUnhealthy, metav1.ConditionFalse)
    }

    return util.NewCondition(clusterv1alpha1.ClusterConditionReady, clusterReady, clusterHealthy, metav1.ConditionTrue)
}

online, healthy := getClusterHealthStatus(clusterClient)
observedReadyCondition := generateReadyCondition(online, healthy)
readyCondition := c.clusterConditionCache.thresholdAdjustedReadyCondition(cluster, &observedReadyCondition)

// cluster is offline after retry timeout, update cluster status immediately and return.
if !online && readyCondition.Status != metav1.ConditionTrue {
    klog.V(2).Infof("Cluster(%s) still offline after %s, ensuring offline is set.",
        cluster.Name, c.ClusterFailureThreshold.Duration)
    setTransitionTime(cluster.Status.Conditions, readyCondition)
    meta.SetStatusCondition(¤tClusterStatus.Conditions, *readyCondition)
    return c.updateStatusIfNeeded(cluster, currentClusterStatus)
}

func (c *Controller) taintClusterByCondition(ctx context.Context, cluster *clusterv1alpha1.Cluster) error {
    currentReadyCondition := meta.FindStatusCondition(cluster.Status.Conditions, clusterv1alpha1.ClusterConditionReady)
    var err error
    if currentReadyCondition != nil {
        switch currentReadyCondition.Status {
        case metav1.ConditionFalse:
            // Add NotReadyTaintTemplateForSched taint immediately.
            if err = utilhelper.UpdateClusterControllerTaint(ctx, c.Client, []*corev1.Taint{NotReadyTaintTemplateForSched}, []*corev1.Taint{UnreachableTaintTemplateForSched}, cluster); err != nil {
                klog.ErrorS(err, "Failed to instantly update UnreachableTaintForSched to NotReadyTaintForSched, will try again in the next cycle.", "cluster", cluster.Name)
            }
        case metav1.ConditionUnknown:
            // Add UnreachableTaintTemplateForSched taint immediately.
            if err = utilhelper.UpdateClusterControllerTaint(ctx, c.Client, []*corev1.Taint{UnreachableTaintTemplateForSched}, []*corev1.Taint{NotReadyTaintTemplateForSched}, cluster); err != nil {
                klog.ErrorS(err, "Failed to instantly swap NotReadyTaintForSched to UnreachableTaintForSched, will try again in the next cycle.", "cluster", cluster.Name)
            }
        case metav1.ConditionTrue:
            if err = utilhelper.UpdateClusterControllerTaint(ctx, c.Client, nil, []*corev1.Taint{NotReadyTaintTemplateForSched, UnreachableTaintTemplateForSched}, cluster); err != nil {
                klog.ErrorS(err, "Failed to remove schedule taints from cluster, will retry in next iteration.", "cluster", cluster.Name)
            }
        }
    } else {
        // Add NotReadyTaintTemplateForSched taint immediately.
        if err = utilhelper.UpdateClusterControllerTaint(ctx, c.Client, []*corev1.Taint{NotReadyTaintTemplateForSched}, nil, cluster); err != nil {
            klog.ErrorS(err, "Failed to add a NotReady taint to the newly added cluster, will try again in the next cycle.", "cluster", cluster.Name)
        }
    }
    return err
}

needEviction, tolerationTime, err := tc.needEviction(cluster, binding.Annotations)
if err != nil {
    klog.ErrorS(err, "Failed to check if binding needs eviction", "binding", fedKey.ClusterWideKey.NamespaceKey())
    return err
}

// Case 1: Need eviction now.
// Case 2: Need eviction after toleration time. If time is up, do eviction right now.
// Case 3: Tolerate forever, we do nothing.
if needEviction || tolerationTime == 0 {
    // update final result to evict the target cluster
    if features.FeatureGate.Enabled(features.GracefulEviction) {
        binding.Spec.GracefulEvictCluster(cluster, workv1alpha2.EvictionProducerTaintManager, workv1alpha2.EvictionReasonTaintUntolerated, "")
    } else {
        binding.Spec.RemoveCluster(cluster)
    }
    if err = tc.Update(context.TODO(), binding); err != nil {
        helper.EmitClusterEvictionEventForResourceBinding(binding, cluster, tc.EventRecorder, err)
        klog.ErrorS(err, "Failed to update binding", "binding", klog.KObj(binding))
        return err
    }
    if !features.FeatureGate.Enabled(features.GracefulEviction) {
        helper.EmitClusterEvictionEventForResourceBinding(binding, cluster, tc.EventRecorder, nil)
    }
} else if tolerationTime > 0 {
    tc.bindingEvictionWorker.AddAfter(fedKey, tolerationTime)
}

// This function no-(images cluster does not exist.
func (s *ResourceBindingSpec) GracefulEvictCluster(name, producer, reason, message string) {
   // find the cluster index
   var i int
   for i = 0; i < len(s.Clusters); i++ {
      if s.Clusters[i].Name == name {
         break
      }
   }
   // not found, do nothing
   if i >= len(s.Clusters) {
      return
   }

   // build eviction task
   evictingCluster := s.Clusters[i].DeepCopy()
   evictionTask := GracefulEvictionTask{
      FromCluster: evictingCluster.Name,
      Reason:      reason,
      Message:     message,
      Producer:    producer,
   }
   if evictingCluster.Replicas > 0 {
      evictionTask.Replicas = &evictingCluster.Replicas
   }

   s.GracefulEvictionTasks = append(s.GracefulEvictionTasks, evictionTask)
   s.Clusters = append(s.Clusters[:i], s.Clusters[i+1:]...)
}

// IsBindingReplicasChanged will check if the sum of replicas is different from the replicas of object
func IsBindingReplicasChanged(bindingSpec *workv1alpha2.ResourceBindingSpec, strategy *policyv1alpha1.ReplicaSchedulingStrategy) bool {
    if strategy == nil {
        return false
    }
    if strategy.ReplicaSchedulingType == policyv1alpha1.ReplicaSchedulingTypeDuplicated {
        for _, targetCluster := range bindingSpec.Clusters {
            if targetCluster.Replicas != bindingSpec.Replicas {
                return true
            }
        }
        return false
    }
    if strategy.ReplicaSchedulingType == policyv1alpha1.ReplicaSchedulingTypeDivided {
        replicasSum := GetSumOfReplicas(bindingSpec.Clusters)
        return replicasSum != bindingSpec.Replicas
    }
    return false
}

if rb.Spec.Placement.ReplicaScheduling != nil && util.IsBindingReplicasChanged(&rb.Spec, rb.Spec.Placement.ReplicaScheduling) {
    // binding replicas changed, need reschedule
    klog.Infof("Reschedule ResourceBinding(%s/%s) as replicas scaled down or scaled up", namespace, name)
    err = s.scheduleResourceBinding(rb)
    metrics.BindingSchedule(string(ScaleSchedule), utilmetrics.DurationInSeconds(start), err)
    return err
}

// Filter checks if the given tolerations in placement tolerate cluster's taints.
func (p *TaintToleration) Filter(
    _ context.Context,
    bindingSpec *workv1alpha2.ResourceBindingSpec,
    _ *workv1alpha2.ResourceBindingStatus,
    cluster *clusterv1alpha1.Cluster,
) *framework.Result {
    // skip the filter if the cluster is already in the list of scheduling results,
    // if the workload referencing by the binding can't tolerate the taint,
    // the taint-manager will evict it after a graceful period.
    if bindingSpec.TargetContains(cluster.Name) {
        return framework.NewResult(framework.Success)
    }

    filterPredicate := func(t *corev1.Taint) bool {
        return t.Effect == corev1.TaintEffectNoSchedule || t.Effect == corev1.TaintEffectNoExecute
    }

    taint, isUntolerated := v1helper.FindMatchingUntoleratedTaint(cluster.Spec.Taints, bindingSpec.Placement.ClusterTolerations, filterPredicate)
    if !isUntolerated {
        return framework.NewResult(framework.Success)
    }

    return framework.NewResult(framework.Unschedulable, fmt.Sprintf("cluster(s) had untolerated taint {%s}", taint.ToString()))
}

// TakeByWeight divide replicas by a weight list and merge the result into previous result.
func (a *Dispenser) TakeByWeight(w ClusterWeightInfoList) {
    if a.Done() {
        return
    }
    sum := w.GetWeightSum()
    if sum == 0 {
        return
    }

    sort.Sort(w)

    result := make([]workv1alpha2.TargetCluster, 0, w.Len())
    remain := a.NumReplicas
    for _, info := range w {
        replicas := int32(info.Weight * int64(a.NumReplicas) / sum)
        result = append(result, workv1alpha2.TargetCluster{
            Name:     info.ClusterName,
            Replicas: replicas,
        })
        remain -= replicas
    }
    // TODO(Garrybest): take rest replicas by fraction part
    for i := range result {
        if remain == 0 {(image(images
            break
        }
        result[i].Replicas++
        remain--
    }

    a.NumReplicas = remain
    a.Result = util.MergeTargetClusters(a.Result, result)
}

I0217 06:46:53.057843       1 scheduler.go:390] Reschedule ResourceBinding(default/nginx-deployment) as replicas scaled down or scaled up
I0217 06:46:53.057888       1 scheduler.go:473] "Begin scheduling resource binding" resourceBinding="default/nginx-deployment"
I0217 06:46:53.083460       1 generic_scheduler.go:119] cluster "member3" is not fit, reason: cluster(s) didn't match the placement cluster affinity constraint
I0217 06:46:53.083524       1 generic_scheduler.go:119] cluster "member1" is not fit, reason: cluster(s) had untolerated taint {cluster.karmada.io/not-ready: }
I0217 06:46:53.083542       1 generic_scheduler.go:73] feasible clusters found: [member2]
I0217 06:46:53.083567       1 generic_scheduler.go:146] Plugin ClusterLocality scores on default/nginx => [{member2 100}]
I0217 06:46:53.083590       1 generic_scheduler.go:146] Plugin ClusterAffinity scores on default/nginx => [{member2 0}]
I0217 06:46:53.083601       1 generic_scheduler.go:79] feasible clusters scores: [{member2 100}]
I0217 06:46:53.085488       1 util.go:72] Target cluster: [{member2 99}]
I0217 06:46:53.085532       1 select_clusters.go:19] select all clusters
I0217 06:46:53.085548       1 generic_scheduler.go:85] selected clusters: [member2]
I0217 06:46:53.085571       1 scheduler.go:489] ResourceBinding default/nginx-deployment scheduled to clusters [{member2 3}]
I0217 06:46:53.151403       1 scheduler.go:491] "End scheduling resource binding" resourceBinding="default/nginx-deployment"

// assessEvictionTasks assesses each task according to graceful eviction rules and
// returns the tasks that should be kept.
func assessEvictionTasks(bindingSpec workv1alpha2.ResourceBindingSpec,
    obse(images[]workv1alpha2.AggregatedStatusItem,
    timeout time.Duration,
    now metav1.Time,
) ([]workv1alpha2.GracefulEvictionTask, []string) {
    var keptTasks []workv1alpha2.GracefulEvictionTask
    var evictedClusters []string

    for _, task := range bindingSpec.GracefulEvictionTasks {
        // set creation timestamp for new task
        if task.CreationTimestamp.IsZero() {
            task.CreationTimestamp = now
            keptTasks = append(keptTasks, task)
            continue
        }

        // assess task according to observed status
        kt := assessSingleTask(task, assessmentOption{
            scheduleResult: bindingSpec.Clusters,
            timeout:        timeout,
            observedStatus: observedStatus,
        })
        if kt != nil {
            keptTasks = append(keptTasks, *kt)
        } else {
            evictedClusters = append(evictedClusters, task.FromCluster)
        }
    }
    return keptTasks, evictedClusters
}

kubectl -n karmada-es-member1 get work nginx-687f7fb96f -oyaml

...
apiVersion: work.karmada.io/v1alpha1
kind: Work
metadata:
  annotations:
    resourcebinding.karmada.io/name: nginx-deployment
    resourcebinding.karmada.io/namespace: default
  creationTimestamp: "2023-02-17T08:53:08Z"
  deletionGracePeriodSeconds: 0
  deletionTimestamp: "2023-02-17T08:55:39Z"
  finalizers:
  - karmada.io/execution-controller
  generation: 2
  labels:
    resourcebinding.karmada.io/key: 687f7fb96f
  name: nginx-687f7fb96f
  namespace: karmada-es-member1
  resourceVersion: "28134"
  uid: d215a03c-522d-4a89-9c5d-e87e27338a03
...

// syncBinding will sync resourceBinding to Works.
func (c *ResourceBindingController) syncBinding(binding *workv1alpha2.ResourceBinding) (controllerruntime.Result, error) {
    if err := c.removeOrphanWorks(binding); err != nil {
        return controllerruntime.Result{Requeue: true}, err
    }

    workload, err := helper.FetchWorkload(c.DynamicClient, c.InformerManager, c.RESTMapper, binding.Spec.Resource)
    if err != nil {
        if apierrors.IsNotFound(err) {
            // It might happen when the resource template has been removed but the garbage collector hasn't removed
            // the ResourceBinding which dependent on resource template.
            // So, just return without retry(requeue) would save unnecessary loop.
            return controllerruntime.Result{}, nil
        }
        klog.Errorf("Failed to fetch workload for resourceBinding(%s/%s). Error: %v.",
            binding.GetNamespace(), binding.GetName(), err)
        return controllerruntime.Result{Requeue: true}, err
    }
    var errs []error
    start := time.Now()
    err = ensureWork(c.(imagesesourceInterpreter, workload, c.OverrideManager, binding, apiextensionsv1.NamespaceScoped)
    metrics.ObserveSyncWorkLatency(binding.ObjectMeta, err, start)
    if err != nil {
        klog.Errorf("Failed to transform resourceBinding(%s/%s) to works. Error: %v.",
            binding.GetNamespace(), binding.GetName(), err)
        c.EventRecorder.Event(binding, corev1.EventTypeWarning, events.EventReasonSyncWorkFailed, err.Error())
        c.EventRecorder.Event(workload, corev1.EventTypeWarning, events.EventReasonSyncWorkFailed, err.Error())
        errs = append(errs, err)
    } else {
        msg := fmt.Sprintf("Sync work of resourceBinding(%s/%s) successful.", binding.Namespace, binding.Name)
        klog.V(4).Infof(msg)
        c.EventRecorder.Event(binding, corev1.EventTypeNormal, events.EventReasonSyncWorkSucceed, msg)
        c.EventRecorder.Event(workload, corev1.EventTypeNormal, events.EventReasonSyncWorkSucceed, msg)
    }
    if err = helper.AggregateResourceBindingWorkStatus(c.Client, binding, workload, c.EventRecorder); err != nil {
        klog.Errorf("Failed to aggregate workStatuses to resourceBinding(%s/%s). Error: %v.",
            binding.GetNamespace(), binding.GetName(), err)
        errs = append(errs, err)
    }
    if len(errs) > 0 {
        return controllerruntime.Result{Requeue: true}, errors.NewAggregate(errs)
    }

    err = c.updateResourceStatus(binding)
    if err != nil {
        return controllerruntime.Result{Requeue: true}, err
    }

    return controllerruntime.Result{}, nil
}

if !work.DeletionTimestamp.IsZero() {
    // Abort deleting workload if cluster is unready when unjoining cluster, otherwise the unjoin process will be failed.
    if util.IsClusterReady(&cluster.Status) {
        err := c.tryDeleteWorkload(clusterName, work)
        if err != nil {
            klog.Errorf("Failed to delete work %v, namespace is %v, err is %v", work.Name, work.Namespace, err)
            return controllerruntime.Result{Requeue: true}, err
        }
    } else if cluster.DeletionTimestamp.IsZero() { // cluster is unready, but not terminating
        return controllerruntime.Result{Requeue: true}, fmt.Errorf("cluster(%s) not ready", cluster.Name)
    }

    return c.removeFinalizer(work)
}