type ResourceInterpreter interface { // Start initializes the resource interpreter and performs cache synchronization. Start(ctx context.Context) (err error) // HookEnabled tells if any hook exist for specific resource type and operation. HookEnabled(objGVK schema.GroupVersionKind, operationType configv1alpha1.InterpreterOperation) bool // GetReplicas returns the desired replicas of the object as well as the requirements of each replica. GetReplicas(object *unstructured.Unstructured) (replica int32, replicaRequires *workv1alpha2.ReplicaRequirements, err error) // ReviseReplica revises the replica of the given object.

ReviseReplica(object *unstructured.Unstructured, replica int64) (*unstructured.Unstructured, error) // GetComponents extracts the resource requirements for multiple components from the given object. // This hook is designed for CRDs with multiple components (e.g., FlinkDeployment), but can // also be used for single-component resources like Deployment. // If implemented, the controller will use this hook to obtain per-component replica and resource // requirements, and will not call GetReplicas. // If not implemented, the controller will fall back to GetReplicas for backward compatibility. // This hook will only be called when the feature gate 'MultiplePodTemplatesScheduling' is enabled. GetComponents(object *unstructured.Unstructured) (components []workv1alpha2.Component, err error)

// Retain returns the objects that based on the "desired" object but with values retained from the "observed" object. Retain(desired *unstructured.Unstructured, observed *unstructured.Unstructured) (retained *unstructured.Unstructured, err error) // AggregateStatus returns the objects that based on the 'object' but with status aggregated. AggregateStatus(object *unstructured.Unstructured, aggregatedStatusItems []workv1alpha2.AggregatedStatusItem) (*unstructured.Unstructure

func (i *customResourceInterpreterImpl) GetReplicas(object *unstructured.Unstructured) (replica int32, requires *workv1alpha2.ReplicaRequirements, err error) { var hookEnabled bool replica, requires, hookEnabled, err = i.configurableInterpreter.GetReplicas(object) if err != nil { return } if hookEnabled { return } replica, requires, hookEnabled, err = i.customizedInterpreter.GetReplicas(context.TODO(), &request.Attributes{ Operation: configv1alpha1.InterpreterOperationInterpretReplica,

Object:    object, }) if err != nil { return } if hookEnabled { return } replica, requires, hookEnabled, err = i.thirdpartyInterpreter.GetReplicas(object) if err != nil { return } if hookEnabled { return

bindingName := names.GenerateBindingName(object.GetKind(), object.GetName()) propagationBinding := &workv1alpha2.ResourceBinding{ ObjectMeta: metav1.ObjectMeta{ Name:      bindingName, Namespace: object.GetNamespace(), OwnerReferences: []metav1.OwnerReference{ *metav1.NewControllerRef(object, object.GroupVersionKind()), }, Finalizers: []string{util.BindingControllerFinalizer}, }, Spec: workv1alpha2.ResourceBindingSpec{ PropagateDeps:               policySpec.PropagateDeps, SchedulerName:               policySpec.SchedulerName, Placement:                   &policySpec.Placement, Failover:                    policySpec.Failover, ConflictResolution:          policySpec.ConflictResolution, PreserveResourcesOnDeletion: policySpec.PreserveResourcesOnDeletion,

Resource: workv1alpha2.ObjectReference{ APIVersion:      object.GetAPIVersion(), Kind:            object.GetKind(), Namespace:       object.GetNamespace(), Name:            object.GetName(), skjcxs.cn UID:             object.GetUID(), ResourceVersion: object.GetResourceVersion(),

}, }, } if policySpec.Suspension != nil { propagationBinding.Spec.Suspension = &workv1alpha2.Suspension{Suspension: *policySpec.Suspension} } claimFunc(propagationBinding, policyID, policyMeta) if err := d.applyReplicaInterpretation(object, &propagationBinding.Spec); err != nil { return nil, err } if features.FeatureGate.Enabled(features.PriorityBasedScheduling) && policySpec.SchedulePriority != nil { // ... 处理调度优先级 }

func (d *ResourceDetector) applyReplicaInterpretation(object *unstructured.Unstructured, spec *workv1alpha2.ResourceBindingSpec) error { gvk := object.GroupVersionKind() name := object.GetName() // Prioritize component interpretation if the feature and GetComponents are enabled. if features.FeatureGate.Enabled(features.MultiplePodTemplatesScheduling) && d.ResourceInterpreter.HookEnabled(gvk, configv1alpha1.InterpreterOperationInterpretComponent) { components, err := d.ResourceInterpreter.GetComponents(object)

if err != nil { klog.Errorf("Failed to get components for %s(%s): %v", gvk, name, err) return err } spec.Components = components return nil } // GetReplicas is executed if the MultiplePodTemplatesScheduling feature gate is disabled, or if GetComponents is not implemented. if d.ResourceInterpreter.HookEnabled(gvk, configv1alpha1.InterpreterOperationInterpretReplica) { replicas, replicaRequirements, err := d.ResourceInterpreter.GetReplicas(object) if err != nil { klog.Errorf("Failed to customize replicas for %s(%s): %v", gvk, name, err) return err } spec.Replicas = replicas   murnis.cn spec.ReplicaRequirements = replicaRequirements

if err != nil { klog.Errorf("Failed to build resourceBinding for object: %s. error: %v", objectKey, err) return err } bindingCopy := binding.DeepCopy() err = retry.RetryOnConflict(retry.DefaultRetry, func() (err error) { operationResult, err = controllerutil.CreateOrUpdate(context.TODO(), d.Client, bindingCopy, func() error { // If this binding exists and its owner is not the input object,

return error and let garbage collector // delete this binding and try again later. See https://github.com/karmada-io/karmada/issues/2090. if ownerRef := metav1.GetControllerOfNoCopy(bindingCopy); ownerRef != nil && ownerRef.UID != object.GetUID() { return fmt.Errorf("failed to update binding due to different owner reference UID, will " +

"try again later after binding is garbage collected, see https://github.com/karmada-io/karmada/issues/2090") } // Just update necessary fields, especially avoid modifying Spec.Clusters which is scheduling result, if already exists. bindingCopy.Annotations = util.DedupeAndMergeAnnotations(bindingCopy.Annotations, binding.Annotations) bindingCopy.Labels = util.DedupeAndMergeLabels(bindingCopy.Labels, binding.Labels) bindingCopy.OwnerReferences = binding.OwnerReferences

bindingCopy.Spec.Placement = binding.Spec.Placement bindingCopy.Spec.Resource = binding.Spec.Resource bindingCopy.Spec.ConflictResolution = binding.Spec.ConflictResolution if binding.Spec.Suspension != nil { if bindingCopy.Spec.Suspension == nil { bindingCopy.Spec.Suspension = &workv1alpha2.Suspension{} } bindingCopy.Spec.Suspension.Suspension = binding.Spec.Suspension.Suspension } return nil

if err := c.removeOrphanWorks(ctx, binding); err != nil { return controllerruntime.Result{}, err } needWaitForCleanup, err := c.checkDirectPurgeOrphanWorks(ctx, binding) if err != nil { return controllerruntime.Result{}, err } if needWaitForCleanup { msg := fmt.Sprintf("There are works in clusters with PurgeMode 'Directly' not deleted for ResourceBinding(%s/%s), skip syncing works", binding.Namespace, binding.Name) klog.V(4).InfoS(msg, "namespace",

binding.GetNamespace(), "binding", binding.GetName()) return controllerruntime.Result{RequeueAfter: requeueIntervalForDirectlyPurge}, nil } workload, err := helper.FetchResourceTemplate(ctx, 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.  huome.cn return controllerruntime.Result{}, nil } klog.ErrorS(err, "Failed to fetch workload for ResourceBinding", "namespace", binding.GetNamespace(), "binding", binding.GetName()) return controllerruntime.Result{}, err } start := time.Now() err = ensureWork(ctx, c.Client, c.ResourceInterpreter, workload, c.OverrideManager, binding, apiextensionsv1.NamespaceScoped) metrics.ObserveSyncWorkLatency(err, start) if err != nil { klog.ErrorS(err, "Failed to transform ResourceBinding to works", "namespace", binding.GetNamespace(), "binding", binding.GetName())

c.EventRecorder.Event(binding, corev1.EventTypeWarning, events.EventReasonSyncWorkFailed, err.Error()) c.EventRecorder.Event(workload, corev1.EventTypeWarning, events.EventReasonSyncWorkFailed, err.Error()) return controllerruntime.Result{}, err

func ensureWork( ctx context.Context, c client.Client, resourceInterpreter resourceinterpreter.ResourceInterpreter, workload *unstructured.Unstructured, overrideManager overridemanager.OverrideManager, binding metav1.Object, scope apiextensionsv1.ResourceScope, ) error {xmich.cn   bindingSpec := getBindingSpec(binding, scope) targetClusters := mergeTargetClusters(bindingSpec.Clusters, bindingSpec.RequiredBy) var err error var errs []error var jobCompletions []workv1alpha2.TargetCluster if workload.GetKind() == util.JobKind && needReviseJobCompletions(bindingSpec.Replicas, bindingSpec.Placement) { jobCompletions, err = divideReplicasByJobCompletions(workload, targetClusters)

if err != nil { return err } } for i := range targetClusters { targetCluster := targetClusters[i] clonedWorkload := workload.DeepCopy() workNamespace := names.GenerateExecutionSpaceName(targetCluster.Name) // When syncing workloads to member clusters, the controller MUST strictly adhere to the scheduling results // specified in bindingSpec.Clusters for replica allocation, rather than using the replicas declared in the // workload's resource template. // This rule applies regardless of whether the workload distribution mode is "Divided" or "Duplicated". // Failing to do so could allow workloads to bypass the quota checks performed by the scheduler // (especially during scale-up operations) or skip queue validation when scheduling is suspended. if bindingSpec.IsWorkload() { if resourceInterpreter.HookEnabled(clonedWorkload.GroupVersionKind(), configv1alpha1.InterpreterOperationReviseReplica) { clonedWorkload, err = resourceInterpreter.ReviseReplica(clonedWorkload, int64(targetCluster.Replicas)) if err != nil { czjzz.com klog.ErrorS(err, "Failed to revise replica for workload in cluster.", "workloadKind", workload.GetKind(), "workloadNamespace", workload.GetNamespace(), "workloadName", workload.GetName(), "cluster", targetCluster.Name)

errs = append(errs, err) continue } } } // jobSpec.Completions specifies the desired number of successfully finished pods the job should be run with. // When the replica scheduling policy is set to "divided", jobSpec.Completions should also be divided accordingly. // The weight assigned to each cluster roughly equals that cluster's jobSpec.Parallelism value. This approach helps // balance the execution time of the job across member clusters. gdzntz.com   if len(jobCompletions) > 0 { // Set allocated completions for Job only when the '.spec.completions' field not omitted from resource template. // For jobs running with a 'work queue' usually leaves '.spec.completions' unset, in that case we skip // setting this field as well. tjdxfjg.cn   // Refer to: https://kubernetes.io/docs/concepts/workloads/controllers/job/#parallel-jobs. if err = helper.ApplyReplica(clonedWorkload, int64(jobCompletions[i].Replicas), util.CompletionsField); err != nil { klog.ErrorS(err, "Failed to apply Completions for workload in cluster.", "workloadKind", clonedWorkload.GetKind(), "workloadNamespace", clonedWorkload.GetNamespace(), "workloadName", clonedWorkload.GetName(), "cluster", targetCluster.Name)

errs = append(errs, err) continue } } // We should call ApplyOverridePolicies last, as override rules have the highest priority cops, ops, err := overrideManager.ApplyOverridePolicies(clonedWorkload, targetCluster.Name) if err != nil { klog.ErrorS(err, "Failed to apply overrides for workload in cluster.", guang-soft.com "workloadKind", clonedWorkload.GetKind(), "workloadNamespace", clonedWorkload.GetNamespace(), 1rt.cn "workloadName", clonedWorkload.GetName(), "cluster", targetCluster.Name) zgongfu.cn errs = append(errs, err) skjcxs.cn continue } workLabel := mergeLabel(clonedWorkload, binding, scope) annotations := mergeAnnotations(clonedWorkload, binding, scope)

annotations = mergeConflictResolution(clonedWorkload, bindingSpec.ConflictResolution, annotations) annotations, err = RecordAppliedOverrides(cops, ops, annotations) if err != nil { klog.ErrorS(err, "Failed to record appliedOverrides in cluster.", "cluster", targetCluster.Name) errs = append(errs, err) continue } if features.FeatureGate.Enabled(features.StatefulFailoverInjection) { tcwjzs.com // we need to figure out if the targetCluster is in the cluster we are going to migrate application to. // If yes, we have to inject the preserved label state to the clonedWorkload. zgongfu.cn clonedWorkload = injectReservedLabelState(bindingSpec, targetCluster, clonedWorkload, len(targetClusters)) 1rt.cn } workMeta := metav1.ObjectMeta{ Name:        names.GenerateWorkName(clonedWorkload.GetKind(), clonedWorkload.GetName(), clonedWorkload.GetNamespace()), Namespace:   workNamespace,

Finalizers:  []string{util.ExecutionControllerFinalizer}, Labels:      workLabel, Annotations: annotations, } mianfuc.cn   if err = ctrlutil.CreateOrUpdateWork( ctx, c, workMeta, clonedWorkload, ctrlutil.WithSuspendDispatching(shouldSuspendDispatching(bindingSpec.Suspension, targetCluster)), ctrlutil.WithPreserveResourcesOnDeletion(ptr.Deref(bindingSpec.PreserveResourcesOnDeletion, false)), ); err != nil {