Karmada 资源对象部署的设计与实现

// pkg/detector/detector.go                                                                                                                          
func (d *ResourceDetector) Start(ctx context.Context) error {
    klog.Infof("Starting resource detector.")




    // 用户创建的资源对象查找是否有匹配的 propagation policy
    // 如果匹配不到，如果无法找到，则暂时将该对象放入 resource detector的 waitingObjects map 成员中
    d.waitingObjects = make(map[keys.ClusterWideKey]struct{})
    d.stopCh = ctx.Done()                                                                                                                             




    // 运行一个名为"propagationPolicy reconciler"的 AsyncWorker                                                        
    policyWorkerOptions := util.Options{
        Name:          "propagationPolicy reconciler",
        KeyFunc:       ClusterWideKeyFunc,           // 生成一个 cluster key
        ReconcileFunc: d.ReconcilePropagationPolicy, // 解决 propagation policy 发生变化时
    }
    ......
    // 运行一个名为"resource detector"的 AsyncWorker                                                                           
    detectorWorkerOptions := util.Options{
        Name:               "resource detector",
        KeyFunc:            ClusterWideKeyFunc,
        ReconcileFunc:      d.Reconcile,          // 主要作用: 资源对象和 propagation policy 匹配，生成 resource binding
        RateLimiterOptions: d.RateLimiterOptions, // 队列限速设置
    }
    ......                                                                                                                                                          
}




// pkg/detector/detector.go
// propagation policy 去匹配资源对象 object
func (d *ResourceDetector) ReconcilePropagationPolicy(key util.QueueKey) error {
    ckey, ok := key.(keys.ClusterWideKey)
    ......
    // 根据 namespace 去获取 propagation policy
    unstructuredObj, err := d.propagationPolicyLister.Get(ckey.NamespaceKey())
    // 如果不存在, 解决 propagation policy 被删除的情况
    if err != nil {
        if apierrors.IsNotFound(err) {
            klog.Infof("PropagationPolicy(%s) has been removed.", ckey.NamespaceKey())
            return d.HandlePropagationPolicyDeletion(ckey.Namespace, ckey.Name)
        }
        ......
    }
    ......
    // 当 propagation policy 被创建，会去 d.waitingObjects 中寻找是否有合适的资源对象 object 与之匹配
    return d.HandlePropagationPolicyCreation(propagationObject)
}




// pkg/detector/detector.go
// 资源对象 object 去匹配 propagation policy
func (d *ResourceDetector) Reconcile(key util.QueueKey) error {
    // 获取在一个集群下资源的所有信息，包括 group version kind name namespace
    // 通过这些信息可以在一个集群下唯一定位一个资源
    clusterWideKey, ok := key.(keys.ClusterWideKey)
    ......




    // 获取具体资源对象 object
    object, err := d.GetUnstructuredObject(clusterWideKey)
    if err != nil {
        if apierrors.IsNotFound(err) {
            // 如果找不到，就将资源对象 object 从 d.waitingObjects 删除
            d.RemoveWaiting(clusterWideKey)
            ......
        }
        return err
    }




    // 资源对象 object 尝试去匹配
    propagationPolicy, err := d.LookForMatchedPolicy(object, clusterWideKey)
    
    if propagationPolicy != nil {
        // 判断 propagation policy.Spec.DependentOverrides 是否存在
        if present, err := helper.IsDependentOverridesPresent(d.Client, propagationPolicy); err != nil || !present {
            return fmt.Errorf("waiting for dependent overrides")                                                                    
        }
        // 将资源对象 object 从 d.waitingObjects 删除
        d.RemoveWaiting(clusterWideKey)
        // ApplyPolicy - 将资源对象 object 和 propagation policy 绑定，生成 resource binding
        return d.ApplyPolicy(object, clusterWideKey, propagationPolicy)
    }




    if d.isWaiting(clusterWideKey) {
        // 程序到这里，就说明资源对象 object 找不到对应的 propagation policy
        d.EventRecorder.Event(object, corev1.EventTypeWarning, workv1alpha2.EventReasonApplyPolicyFailed, "No policy match for resource")
        return nil
    }
    // 如果找不到，就继续放入 d.waitingObjects 中，知道找到合适的 propagation policy
    d.AddWaiting(clusterWideKey)
    ......
}

// pkg/scheduler/scheduler.go                                           
func (s *Scheduler) doScheduleBinding(namespace, name string) (err error) {
    rb, err := s.bindingLister.ResourceBindings(namespace).Get(name)
    ......
    policyPlacement, policyPlacementStr, err := s.getPlacement(rb)




    // 1. 调和调度（ReconcileSchedule） 
    if appliedPlacement := util.GetLabelValue(rb.Annotations, util.PolicyPlacementAnnotation); policyPlacementStr != appliedPlacement {
        .......
    }
    // 2. 扩缩容调度（ScaleSchedule） 
    if policyPlacement.ReplicaScheduling != nil && util.IsBindingReplicasChanged(&rb.Spec, policyPlacement.ReplicaScheduling) {
        ......
    
    // 3. 首次调度（FirstSchedule） 
    if rb.Spec.Replicas == 0 ||
        policyPlacement.ReplicaScheduling == nil ||
        policyPlacement.ReplicaScheduling.ReplicaSchedulingType == policyv1alpha1.ReplicaSchedulingTypeDuplicated {
        klog.V(3).Infof("Start to schedule ResourceBinding(%s/%s) as scheduling type is duplicated", namespace, name)
        err = s.scheduleResourceBinding(rb)
        metrics.BindingSchedule(string(ReconcileSchedule), utilmetrics.DurationInSeconds(start), err)
        return err
    }
    // 4. 无需调度（AvoidSchedule）
    klog.V(3).Infof("Don't need to schedule ResourceBinding(%s/%s)", namespace, name)
    return nil
}




// pkg/scheduler/scheduler.go
func (s *Scheduler) scheduleResourceBinding(resourceBinding *workv1alpha2.ResourceBinding) (err error) {
    klog.V(4).InfoS("Begin scheduling resource binding", "resourceBinding", klog.KObj(resourceBinding))
    defer klog.V(4).InfoS("End scheduling resource binding", "resourceBinding", klog.KObj(resourceBinding))
    
    // 1. 获取 resource binding 相关 policy.placement
    placement, placementStr, err := s.getPlacement(resourceBinding)
    ......
    // 2. 根据算法进行调度
    scheduleResult, err := s.Algorithm.Schedule(context.TODO(), &placement, &resourceBinding.Spec, &core.ScheduleAlgorithmOption{EnableEmptyWorkloadPropagation: s.enableEmptyWorkloadPropagation})
    ......
    // 3. 更新 rb.spec.clusters 以及 rb.annotations
    klog.V(4).Infof("ResourceBinding %s/%s scheduled to clusters %v", resourceBinding.Namespace, resourceBinding.Name, scheduleResult.SuggestedClusters)
    return s.patchScheduleResultForResourceBinding(resourceBinding, placementStr, scheduleResult.SuggestedClusters)
}

// pkg/controllers/binding/binding_controller.go
func (c *ResourceBindingController) syncBinding(binding *workv1alpha2.ResourceBinding) (controllerruntime.Result, error) {
    .......
    // 获取 workloads - 位于 resource binding.Spec.Resource
    workload, err := helper.FetchWorkload(c.DynamicClient, c.InformerManager, c.RESTMapper, binding.Spec.Resource)
    ......
    // 生成 work(具体流程见下面函数)
    err = ensureWork(c.Client, c.ResourceInterpreter, workload, c.OverrideManager, binding, apiextensionsv1.NamespaceScoped)
    ......
    // 聚合资源绑定工作状态将收集当前资源绑定对象的所有工作状态，然后将状态信息聚合到 resource binding 状态
    err = helper.AggregateResourceBindingWorkStatus(c.Client, binding, workload)
    .......
    // 更新 resource binding.status
    err = c.updateResourceStatus(binding)
    .......
}


// pkg/controllers/binding/common.go
func ensureWork(
    c client.Client, resourceInterpreter resourceinterpreter.ResourceInterpreter, workload *unstructured.Unstructured,
    overrideManager overridemanager.OverrideManager, binding metav1.Object, scope apiextensionsv1.ResourceScope,
) error {
    ......
    // 遍历经过 schedule 调度后的结果 targetClusters
    for i := range targetClusters {
        targetCluster := targetClusters[i]
        clonedWorkload := workload.DeepCopy()


        // 根据集群名字 targetCluster.Name 生成 work namespace 的名字 - 例如：karmada-es-member1
        workNamespace, err := names.GenerateExecutionSpaceName(targetCluster.Name)
        ......
        if hasScheduledReplica {
            // HookEnabled 查看资源类型对象是否存在需要修改副本数的工作，如果有就执行
            if resourceInterpreter.HookEnabled(clonedWorkload.GroupVersionKind(), configv1alpha1.InterpreterOperationReviseReplica) {
                // ReviseReplica 在目标集群中修改资源对象副本数量
                clonedWorkload, err = resourceInterpreter.ReviseReplica(clonedWorkload, desireReplicaInfos[targetCluster.Name])
                ......
            }
        ......
        // ApplyOverridePolicies 在特定目标集群上去执行 overrides policy
        // 如果有多个 overrides policy 存在，则按照以下的规则：
        // 对于 cluster scoped 资源，按策略名称升序去执行 cluster override policy
        // 对于 namespaced scoped 资源，先执行 cluster override policy，然后再执行 override policy
        cops, ops, err := overrideManager.ApplyOverridePolicies(clonedWorkload, targetCluster.Name)
        ......
        // 更新 work 对象的 labels 和 annotations
        workLabel := mergeLabel(clonedWorkload, workNamespace, binding, scope)
        annotations := mergeAnnotations(clonedWorkload, binding, scope)
        annotations, err = RecordAppliedOverrides(cops, ops, annotations)
        ......
        // 更新 work
        if err = helper.CreateOrUpdateWork(c, workMeta, clonedWorkload); err != nil {
            return err
        }
    }
    return nil
}

// pkg/controllers/execution/execution_congtoller.go
func (c *Controller) syncToClusters(clusterName string, work *workv1alpha1.Work) error {
    var errs []error
    syncSucceedNum := 0
    for _, manifest := range work.Spec.Workload.Manifests {
        // 从 work 对象的 spec.workload.manifests 获取要在集群中部署的资源对象 workload
        workload := &unstructured.Unstructured{}
        err := workload.UnmarshalJSON(manifest.Raw)
        ......
        // 判断是否已经在该成员集群中部署过了
        applied := helper.IsResourceApplied(&work.Status)
        if applied {
            // 如果部署过了就更新
            err = c.tryUpdateWorkload(clusterName, workload)
            ......
        } else {
            // 没有部署过就重新创建
            err = c.tryCreateWorkload(clusterName, workload)
            .......
        }
        // 生成相对应的事件：成功
        c.eventf(workload, corev1.EventTypeNormal, workv1alpha1.EventReasonSyncWorkSucceed, "Successfully applied resource(%v/%v) to cluster %s", workload.GetNamespace(), workload.GetName(), clusterName)
        syncSucceedNum++
    }
    ......
    // 更新 work status conditions
    err := c.updateAppliedCondition(work, metav1.ConditionTrue, "AppliedSuccessful", "Manifest has been successfully applied")
    ......
    return nil
}