K8S Internals 系列：第四期

容器编排之争在 Kubernetes 一统天下局面形成后，K8S 成为了云原生时代的新一代操作系统。K8S 让一切变得简单了，但自身逐渐变得越来越复杂。【K8S Internals 系列专栏】围绕 K8S 生态的诸多方面，将由博云容器云研发团队定期分享有关调度、安全、网络、性能、存储、应用场景等热点话题。希望大家在享受 K8S 带来的高效便利的同时，又可以如庖丁解牛般领略其内核运行机制的魅力。

在上一期文章《K8S 多集群管理很难？试试 Karmada | K8S Internals 系列第 3 期》中，我们主要介绍了 karmada 架构及相关资源概念，了解了 karmada 的发展历史。

众所周知，karmada 组件中有三个组件与调度密切相关，分别是 karmada-scheduler，karmada-descheduler 以及 karmada-scheduler-estimator，这三个组件的协同作用可以实现 karmada 的多种调度策略。本文我们将主要介绍下这三个组件。

karmada调度组件

karmada-scheduler

karmada-scheduler的主要作用就是将k8s原生API资源对象（包含CRD资源）调度到成员集群上。

我们先看一下其与k8s集群的kube-scheduler的对比：

	kube-scheduler	karmada-scheduler
工作层面	单集群层面	多集群层面
支持扩展点	pre filter、filter、post filter、pre score、score、post score等	filter和score
考虑内容	会考虑虑容器端口是否冲突，volume是否已经到位等细节	不会考虑

调度插件

karmada scheduler在调度每个k8s原生API资源对象（包含CRD资源）时，会逐个调用各扩展点上的插件：

1. filter扩展点上的调度算法插件将不满足propagation policy的成员集群过滤掉
   karmada scheduler对每个考察中的成员集群调用每个插件的Filter方法，该方法都能返回一个Result对象表示该插件的调度结果，其中的code代表待下发资源是否能调度到某个成员集群上，reason用来解释这个结果，err包含调度算法插件执行过程中遇到的错误。

   // Filter checks if the API(CRD) of the resource is installed in the target cluster.
   func (p *APIInstalled) Filter(ctx context.Context, placement *policyv1alpha1.Placement, resource *workv1alpha2.ObjectReference, cluster *clusterv1alpha1.Cluster) *framework.Result {
       if !helper.IsAPIEnabled(cluster.Status.APIEnablements, resource.APIVersion, resource.Kind) {
           klog.V(2).Infof("Cluster(%s) not fit as missing API(%s, kind=%s)", cluster.Name, resource.APIVersion, resource.Kind)
           return framework.NewResult(framework.Unschedulable, "no such API resource")
       }
   
       return framework.NewResult(framework.Success)
   }
   

   // NewResult makes a result out of the given arguments and returns its pointer.
   func NewResult(code Code, reasons ...string) *Result {
       s := &Result{
           code:    code,
           reasons: reasons,
       }
       if code == Error {
           s.err = errors.New(strings.Join(reasons, ","))
       }
       return s
   }
   

2. score扩展点上的调度算法插件为每个经过上一步过滤的集群计算评分
   karmada scheduler对每个经过上一步过滤的成员集群调用每个插件的Score方法，该方法都能返回一个int64类型的评分结果。

const (
    // MinClusterScore is the minimum score a Score plugin is expected to return.
    MinClusterScore int64 = 0

    // MaxClusterScore is the maximum score a Score plugin is expected to return.
    MaxClusterScore int64 = 100
)

// Score calculates the score on the candidate cluster.
// if cluster object is exist in resourceBinding.Spec.Clusters, Score is 100, otherwise it is 0.
func (p *ClusterLocality) Score(ctx context.Context, placement *policyv1alpha1.Placement,
    spec *workv1alpha2.ResourceBindingSpec, cluster *clusterv1alpha1.Cluster) (int64, *framework.Result) {
    if len(spec.Clusters) == 0 {
        return framework.MinClusterScore, framework.NewResult(framework.Success)
    }

    replicas := util.GetSumOfReplicas(spec.Clusters)
    if replicas <= 0 {
        return framework.MinClusterScore, framework.NewResult(framework.Success)
    }
    // 再次触发调度时，已存在副本的集群得分更高
    if spec.TargetContains(cluster.Name) {
        return framework.MaxClusterScore, framework.NewResult(framework.Success)
    }

    return framework.MinClusterScore, framework.NewResult(framework.Success)
}

最终按照第二步的评分高低选择成员集群作为调度结果。

目前karmada的调度算法插件：

• APIInstalled: 用于检查资源的API(CRD)是否安装在目标集群中。

• ClusterAffinity: 用于检查资源选择器是否与集群标签匹配。

• SpreadConstraint: 用于检查 Cluster.Spec 中的 spread 属性即Provider/Zone/Region字段。

• TaintToleration: 用于检查传播策略是否容忍集群的污点。

• ClusterLocality 是一个评分插件，为目标集群进行评分。

调度场景

karmada scheduler的输入是resource detector的输出：resource binding 和cluster resource bingding。这里这里涉及到doScheduleBinding和doScheduleClusterBinding，两者流程类似我们单看doScheduleBinding：

func (s *Scheduler) doScheduleBinding(namespace, name string) (err error) {
    rb, err := s.bindingLister.ResourceBindings(namespace).Get(name)
    if err != nil {
        if apierrors.IsNotFound(err) {
            // the binding does not exist, do nothing
            return nil
        }
        return err
    }

    // Update "Scheduled" condition according to schedule result.
    defer func() {
        s.recordScheduleResultEventForResourceBinding(rb, err)
        var condition metav1.Condition
        if err == nil {
            condition = util.NewCondition(workv1alpha2.Scheduled, scheduleSuccessReason, scheduleSuccessMessage, metav1.ConditionTrue)
        } else {
            condition = util.NewCondition(workv1alpha2.Scheduled, scheduleFailedReason, err.Error(), metav1.ConditionFalse)
        }
        if updateErr := s.updateBindingScheduledConditionIfNeeded(rb, condition); updateErr != nil {
            klog.Errorf("Failed update condition(%s) for ResourceBinding(%s/%s)", workv1alpha2.Scheduled, rb.Namespace, rb.Name)
            if err == nil {
                // schedule succeed but update condition failed, return err in order to retry in next loop.
                err = updateErr
            }
        }
    }()

    start := time.Now()
    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 schedule
        klog.Infof("Start to schedule ResourceBinding(%s/%s) as placement changed", namespace, name)
        err = s.scheduleResourceBinding(rb)
        metrics.BindingSchedule(string(ReconcileSchedule), utilmetrics.DurationInSeconds(start), err)
        return err
    }
    if policyPlacement.ReplicaScheduling != nil && util.IsBindingReplicasChanged(&rb.Spec, policyPlacement.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
    }
    // TODO(dddddai): reschedule bindings on cluster change
    if s.allClustersInReadyState(rb.Spec.Clusters) {
        klog.Infof("Don't need to schedule ResourceBinding(%s/%s)", namespace, name)
        return nil
    }

    if features.FeatureGate.Enabled(features.Failover) {
        klog.Infof("Reschedule ResourceBinding(%s/%s) as cluster failure or deletion", namespace, name)
        err = s.scheduleResourceBinding(rb)
        metrics.BindingSchedule(string(FailoverSchedule), utilmetrics.DurationInSeconds(start), err)
        return err
    }
    return nil
}

可以看到这里实现了三种场景的调度：

• 分发资源时选择目标集群的规则变了

• 副本数变了，即扩缩容调度

• 故障恢复调度，当被调度的成员集群状态不正常时会触发重新调度

调度策略

在创建分发策略的时候，需要指定调度策略,举个例子：

apiVersion: policy.karmada.io/v1alpha1
kind: PropagationPolicy
metadata:
  name: nginx-propagation
spec:
  resourceSelectors:
    - apiVersion: apps/v1
      kind: Deployment
      name: nginx
  placement:
    clusterAffinity:
      clusterNames:
        - member1  #分发到成员集群member1和member2
        - member2
    replicaScheduling:
      replicaDivisionPreference: Weighted #划分副本策略
      replicaSchedulingType: Divided  #调度副本策略
      weightPreference:  
        staticWeightList:  #目标集群静态权重
          - targetCluster:
              clusterNames:
                - member1
            weight: 1
          - targetCluster:
              clusterNames:
                - member2
            weight: 1

相关字段设置关系如图所示：

• ReplicaSchedulingType 可设置为 Duplicated 或者 Divided，如果设置为Divided则需要进一步设置ReplicaDivisionPreference。

• ReplicaDivisionPreference 可设置为Aggregated或者Weighted,如果设置为Weighted，则需要进一步设置weightPreference。

• weightPreference中可设置为静态权重（staticWeightList）或者动态权重（DynamicWeight），目前动态权重的动态因素只有AvailableReplicas，即根据集群资源计算出的可调度副本数作为权重指标。

Duplicated策略

Duplicated策略表示将要分发资源的replicas相同数量的复制到所有的目标成员集群中，例如：分发deployment资源时，deployment资源设定的副本数为10，则分发到所有目标集群的deployment的副本数都是10，如图所示：

Divided策略

Divided策略表示将要分发资源的replicas划分到多个目标成员集群中，例如：分发deployment资源时，deployment资源设定的副本数为10，则分发到所有目标集群的deployment的副本总数是10。具体如何划分需要根据ReplicaDivisionPreference的值来决定，而ReplicaDivisionPreference的值可以设定为Aggregated或者Weighted。

Aggregated策略

Aggregated策略表示将副本调度到尽可能少的目标集群上，例如目标集群有三个，但是第一个目标集群拥有足够的资源调度到所有的副本，则副本会全部调度到第一个目标集群上，如图所示：

Weighted策略

Weighted策略具体设置通过weightPreference字段设置，如果weightPreference不设置，则默认给所有目标集群加相同的静态权重1。

当weightPreference设置为静态权重时，举个例子：有三个目标集群 A、 B、 C, 权重分别为 1、2、 3；需要分发副本数为12，则 A 集群分发副本数为： 12 * 1/（1+2+3）= 2 , B 集群分发副本数为： 12 * 2/（1+2+3）= 4 , A 集群分发副本数为： 12 * 3/（1+2+3）= 6 ;如图所示：

当weightPreference设置为动态权重时，静态权重的设置会被忽略，动态权重策略调度副本时需要根据 karmada-estimator计算出的各个目标集群可调度最大副本数进行计算各个目标集群调度的具体副本数，举个例子：

有三个目标集群 A、 B、 C, 最大可调度副本数分别为 6、12、 18；需要分发副本数为12，则 A 集群分发副本数为： 12 * 6/（6+12+18）= 2 , B 集群分发副本数为： 12 * 12/（6+12+18）= 4 , A 集群分发副本数为： 12 * 18/（6+12+18）= 6 ;如图所示：

karmada-descheduler

karmada-descheduler在调度策略为动态划分时（dynamic division）时才会生效；karmada-descheduler 将每隔一段时间检测一次所有部署，默认情况下每 2 分钟检测一次。在每个周期中，它会通过调用 karmada-scheduler-estimator 找出部署在目标调度集群中有多少不可调度的副本，然后更新ResourceBinding资源的Clusters[i].Replicas字段，并根据当前情况触发 karmada-scheduler 执行“Scale Schedule”。

karmada scheduler-estimator

当调度策略是动态权重调度或者Aggregated策略时，karmada-scheduler通过调用 karmada-scheduler-estimator不会将过多的副本分配到资源不足的集群中。karmada-scheduler-estimator用来计算集群中CPU，Memory，EphemeralStorage 或者请求创建工作负载中的其他资源是否满足调度需求，对集群中每个节点可调度副本数进行计算，最终计算出该集群可调度的最大副本数。

评估服务评估哪些指标

karmada scheduler-estimator评估了以下资源：

• cpu

• memory

• ephemeral-storage

• 其他标量资源：（1）扩展资源，例如： requests.nvidia.com/gpu: 4（2）kubernetes.io/下原生资源（3）hugepages- 资源（4）attachable-volumes- 资源

评估服务如何和调度器协调工作的

karmada scheduler-estimator是一个GRPC服务，当调度策略是动态权重调度或者Aggregated策略时，karmada-scheduler会遍历所有启动的karmada scheduler-estimator,调用其客户端方法MaxAvailableReplicas向karmada scheduler-estimator发送请求获取评估结果，即该集群可调度的最大副本数。

调度场景演示

karmada-descheduler再调度演示

1. 创建一个deployment,副本数设置为3，并divide它们到三个成员集群:

   apiVersion: policy.karmada.io/v1alpha1
   kind: PropagationPolicy
   metadata:
    name: nginx-propagation
   spec:
    resourceSelectors:
      - apiVersion: apps/v1
        kind: Deployment
        name: nginx
    placement:
      clusterAffinity:
        clusterNames:
          - member1
          - member2
          - member3
      replicaScheduling:
        replicaDivisionPreference: Weighted
        replicaSchedulingType: Divided
        weightPreference:
          dynamicWeight: AvailableReplicas
   ---
   apiVersion: apps/v1
   kind: Deployment
   metadata:
    name: nginx
    labels:
      app: nginx
   spec:
    replicas: 3
    selector:
      matchLabels:
        app: nginx
    template:
      metadata:
        labels:
          app: nginx
      spec:
        containers:
        - image: nginx
          name: nginx
          resources:
            requests:
              cpu: "1"
   

export KUBECONFIG="$HOME/.kube/karmada.config"
kubectl config use-context karmada-apiserver
kubectl apply -f test-deploy-0629.yaml

查看创建的结果：

kubectl karmada get pods

2. 我们设置member集群不可调度

   export KUBECONFIG="$HOME/.kube/members.config"
   kubectl config use-context member1
   kubectl cordon member1-control-plane
   kubectl delete pod nginx-68b895fcbd-jgwz6
   # member1集群上pod变得不可调度
   kubectl get pod
   

3. 大约 5 到 7 分钟后,查看pod，可以看到副本已经被调度到member2集群上

   export KUBECONFIG="$HOME/.kube/karmada.config"
   kubectl config use-context karmada-apiserver
   kubectl karmada get pods
   
   NAME                     CLUSTER   READY   STATUS    RESTARTS   AGE
   nginx-6cd649d446-k72rf   member3   1/1     Running   0          8m
   nginx-6cd649d446-hb4jk   member2   1/1     Running   0          8m
   nginx-6cd649d446-qmckr   member2   1/1     Running   0          1m
   

综述

可以看到karmada拥有较丰富调度策略，来满足工作负载在多集群间的调度，从而满足多种场景下的使用，例如双活，远程容灾，故障转移等等。

下一篇我们将为大家分享，基于 Karmada 和博云自研网络组件 Fabric，实现多集群业务跨集群服务发现、无损通信等高阶能力，助力业务单元化部署能力。