274 lines
12 KiB
Go
274 lines
12 KiB
Go
/*
|
|
Copyright 2021 The Karmada Authors.
|
|
|
|
Licensed under the Apache License, Version 2.0 (the "License");
|
|
you may not use this file except in compliance with the License.
|
|
You may obtain a copy of the License at
|
|
|
|
http://www.apache.org/licenses/LICENSE-2.0
|
|
|
|
Unless required by applicable law or agreed to in writing, software
|
|
distributed under the License is distributed on an "AS IS" BASIS,
|
|
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
See the License for the specific language governing permissions and
|
|
limitations under the License.
|
|
*/
|
|
|
|
package binding
|
|
|
|
import (
|
|
apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
|
|
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
|
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
|
|
"k8s.io/klog/v2"
|
|
"sigs.k8s.io/controller-runtime/pkg/client"
|
|
|
|
configv1alpha1 "github.com/karmada-io/karmada/pkg/apis/config/v1alpha1"
|
|
policyv1alpha1 "github.com/karmada-io/karmada/pkg/apis/policy/v1alpha1"
|
|
workv1alpha1 "github.com/karmada-io/karmada/pkg/apis/work/v1alpha1"
|
|
workv1alpha2 "github.com/karmada-io/karmada/pkg/apis/work/v1alpha2"
|
|
"github.com/karmada-io/karmada/pkg/resourceinterpreter"
|
|
"github.com/karmada-io/karmada/pkg/util"
|
|
"github.com/karmada-io/karmada/pkg/util/helper"
|
|
"github.com/karmada-io/karmada/pkg/util/names"
|
|
"github.com/karmada-io/karmada/pkg/util/overridemanager"
|
|
)
|
|
|
|
// ensureWork ensure Work to be created or updated.
|
|
func ensureWork(
|
|
c client.Client, resourceInterpreter resourceinterpreter.ResourceInterpreter, workload *unstructured.Unstructured,
|
|
overrideManager overridemanager.OverrideManager, binding metav1.Object, scope apiextensionsv1.ResourceScope,
|
|
) error {
|
|
var targetClusters []workv1alpha2.TargetCluster
|
|
var placement *policyv1alpha1.Placement
|
|
var requiredByBindingSnapshot []workv1alpha2.BindingSnapshot
|
|
var replicas int32
|
|
var conflictResolutionInBinding policyv1alpha1.ConflictResolution
|
|
switch scope {
|
|
case apiextensionsv1.NamespaceScoped:
|
|
bindingObj := binding.(*workv1alpha2.ResourceBinding)
|
|
targetClusters = bindingObj.Spec.Clusters
|
|
requiredByBindingSnapshot = bindingObj.Spec.RequiredBy
|
|
placement = bindingObj.Spec.Placement
|
|
replicas = bindingObj.Spec.Replicas
|
|
conflictResolutionInBinding = bindingObj.Spec.ConflictResolution
|
|
case apiextensionsv1.ClusterScoped:
|
|
bindingObj := binding.(*workv1alpha2.ClusterResourceBinding)
|
|
targetClusters = bindingObj.Spec.Clusters
|
|
requiredByBindingSnapshot = bindingObj.Spec.RequiredBy
|
|
placement = bindingObj.Spec.Placement
|
|
replicas = bindingObj.Spec.Replicas
|
|
conflictResolutionInBinding = bindingObj.Spec.ConflictResolution
|
|
}
|
|
|
|
targetClusters = mergeTargetClusters(targetClusters, requiredByBindingSnapshot)
|
|
|
|
var jobCompletions []workv1alpha2.TargetCluster
|
|
var err error
|
|
if workload.GetKind() == util.JobKind {
|
|
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)
|
|
|
|
// If and only if the resource template has replicas, and the replica scheduling policy is divided,
|
|
// we need to revise replicas.
|
|
if needReviseReplicas(replicas, placement) {
|
|
if resourceInterpreter.HookEnabled(clonedWorkload.GroupVersionKind(), configv1alpha1.InterpreterOperationReviseReplica) {
|
|
clonedWorkload, err = resourceInterpreter.ReviseReplica(clonedWorkload, int64(targetCluster.Replicas))
|
|
if err != nil {
|
|
klog.Errorf("Failed to revise replica for %s/%s/%s in cluster %s, err is: %v",
|
|
workload.GetKind(), workload.GetNamespace(), workload.GetName(), targetCluster.Name, err)
|
|
return err
|
|
}
|
|
}
|
|
|
|
// 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.
|
|
// Refer to: https://kubernetes.io/docs/concepts/workloads/controllers/job/#parallel-jobs.
|
|
if len(jobCompletions) > 0 {
|
|
if err = helper.ApplyReplica(clonedWorkload, int64(jobCompletions[i].Replicas), util.CompletionsField); err != nil {
|
|
klog.Errorf("Failed to apply Completions for %s/%s/%s in cluster %s, err is: %v",
|
|
clonedWorkload.GetKind(), clonedWorkload.GetNamespace(), clonedWorkload.GetName(), targetCluster.Name, err)
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
|
|
// We should call ApplyOverridePolicies last, as override rules have the highest priority
|
|
cops, ops, err := overrideManager.ApplyOverridePolicies(clonedWorkload, targetCluster.Name)
|
|
if err != nil {
|
|
klog.Errorf("Failed to apply overrides for %s/%s/%s, err is: %v", clonedWorkload.GetKind(), clonedWorkload.GetNamespace(), clonedWorkload.GetName(), err)
|
|
return err
|
|
}
|
|
workLabel := mergeLabel(clonedWorkload, workNamespace, binding, scope)
|
|
|
|
annotations := mergeAnnotations(clonedWorkload, workNamespace, binding, scope)
|
|
annotations = mergeConflictResolution(clonedWorkload, conflictResolutionInBinding, annotations)
|
|
annotations, err = RecordAppliedOverrides(cops, ops, annotations)
|
|
if err != nil {
|
|
klog.Errorf("Failed to record appliedOverrides, Error: %v", err)
|
|
return err
|
|
}
|
|
|
|
workMeta := metav1.ObjectMeta{
|
|
Name: names.GenerateWorkName(clonedWorkload.GetKind(), clonedWorkload.GetName(), clonedWorkload.GetNamespace()),
|
|
Namespace: workNamespace,
|
|
Finalizers: []string{util.ExecutionControllerFinalizer},
|
|
Labels: workLabel,
|
|
Annotations: annotations,
|
|
}
|
|
|
|
if err = helper.CreateOrUpdateWork(c, workMeta, clonedWorkload); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func mergeTargetClusters(targetClusters []workv1alpha2.TargetCluster, requiredByBindingSnapshot []workv1alpha2.BindingSnapshot) []workv1alpha2.TargetCluster {
|
|
if len(requiredByBindingSnapshot) == 0 {
|
|
return targetClusters
|
|
}
|
|
|
|
scheduledClusterNames := util.ConvertToClusterNames(targetClusters)
|
|
|
|
for _, requiredByBinding := range requiredByBindingSnapshot {
|
|
for _, targetCluster := range requiredByBinding.Clusters {
|
|
if !scheduledClusterNames.Has(targetCluster.Name) {
|
|
scheduledClusterNames.Insert(targetCluster.Name)
|
|
targetClusters = append(targetClusters, targetCluster)
|
|
}
|
|
}
|
|
}
|
|
|
|
return targetClusters
|
|
}
|
|
|
|
func mergeLabel(workload *unstructured.Unstructured, workNamespace string, binding metav1.Object, scope apiextensionsv1.ResourceScope) map[string]string {
|
|
var workLabel = make(map[string]string)
|
|
util.MergeLabel(workload, workv1alpha1.WorkNamespaceLabel, workNamespace)
|
|
util.MergeLabel(workload, workv1alpha1.WorkNameLabel, names.GenerateWorkName(workload.GetKind(), workload.GetName(), workload.GetNamespace()))
|
|
util.MergeLabel(workload, util.ManagedByKarmadaLabel, util.ManagedByKarmadaLabelValue)
|
|
if scope == apiextensionsv1.NamespaceScoped {
|
|
util.RemoveLabels(workload, workv1alpha2.ResourceBindingUIDLabel)
|
|
|
|
bindingID := util.GetLabelValue(binding.GetLabels(), workv1alpha2.ResourceBindingPermanentIDLabel)
|
|
util.MergeLabel(workload, workv1alpha2.ResourceBindingReferenceKey, names.GenerateBindingReferenceKey(binding.GetNamespace(), binding.GetName()))
|
|
util.MergeLabel(workload, workv1alpha2.ResourceBindingPermanentIDLabel, bindingID)
|
|
|
|
workLabel[workv1alpha2.ResourceBindingReferenceKey] = names.GenerateBindingReferenceKey(binding.GetNamespace(), binding.GetName())
|
|
workLabel[workv1alpha2.ResourceBindingPermanentIDLabel] = bindingID
|
|
} else {
|
|
util.RemoveLabels(workload, workv1alpha2.ClusterResourceBindingUIDLabel)
|
|
|
|
bindingID := util.GetLabelValue(binding.GetLabels(), workv1alpha2.ClusterResourceBindingPermanentIDLabel)
|
|
util.MergeLabel(workload, workv1alpha2.ClusterResourceBindingReferenceKey, names.GenerateBindingReferenceKey("", binding.GetName()))
|
|
util.MergeLabel(workload, workv1alpha2.ClusterResourceBindingPermanentIDLabel, bindingID)
|
|
|
|
workLabel[workv1alpha2.ClusterResourceBindingReferenceKey] = names.GenerateBindingReferenceKey("", binding.GetName())
|
|
workLabel[workv1alpha2.ClusterResourceBindingPermanentIDLabel] = bindingID
|
|
}
|
|
return workLabel
|
|
}
|
|
|
|
func mergeAnnotations(workload *unstructured.Unstructured, workNamespace string, binding metav1.Object, scope apiextensionsv1.ResourceScope) map[string]string {
|
|
annotations := make(map[string]string)
|
|
util.MergeAnnotation(workload, workv1alpha2.WorkNameAnnotation, names.GenerateWorkName(workload.GetKind(), workload.GetName(), workload.GetNamespace()))
|
|
util.MergeAnnotation(workload, workv1alpha2.WorkNamespaceAnnotation, workNamespace)
|
|
|
|
if scope == apiextensionsv1.NamespaceScoped {
|
|
util.MergeAnnotation(workload, workv1alpha2.ResourceBindingNamespaceAnnotationKey, binding.GetNamespace())
|
|
util.MergeAnnotation(workload, workv1alpha2.ResourceBindingNameAnnotationKey, binding.GetName())
|
|
annotations[workv1alpha2.ResourceBindingNamespaceAnnotationKey] = binding.GetNamespace()
|
|
annotations[workv1alpha2.ResourceBindingNameAnnotationKey] = binding.GetName()
|
|
} else {
|
|
util.MergeAnnotation(workload, workv1alpha2.ClusterResourceBindingAnnotationKey, binding.GetName())
|
|
annotations[workv1alpha2.ClusterResourceBindingAnnotationKey] = binding.GetName()
|
|
}
|
|
|
|
return annotations
|
|
}
|
|
|
|
// RecordAppliedOverrides record applied (cluster) overrides to annotations
|
|
func RecordAppliedOverrides(cops *overridemanager.AppliedOverrides, ops *overridemanager.AppliedOverrides,
|
|
annotations map[string]string) (map[string]string, error) {
|
|
if annotations == nil {
|
|
annotations = make(map[string]string)
|
|
}
|
|
|
|
if cops != nil {
|
|
appliedBytes, err := cops.MarshalJSON()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if appliedBytes != nil {
|
|
annotations[util.AppliedClusterOverrides] = string(appliedBytes)
|
|
}
|
|
}
|
|
|
|
if ops != nil {
|
|
appliedBytes, err := ops.MarshalJSON()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if appliedBytes != nil {
|
|
annotations[util.AppliedOverrides] = string(appliedBytes)
|
|
}
|
|
}
|
|
|
|
return annotations, nil
|
|
}
|
|
|
|
// mergeConflictResolution determine the conflictResolution annotation of Work: preferentially inherit from RT, then RB
|
|
func mergeConflictResolution(workload *unstructured.Unstructured, conflictResolutionInBinding policyv1alpha1.ConflictResolution,
|
|
annotations map[string]string) map[string]string {
|
|
// conflictResolutionInRT refer to the annotation in ResourceTemplate
|
|
conflictResolutionInRT := util.GetAnnotationValue(workload.GetAnnotations(), workv1alpha2.ResourceConflictResolutionAnnotation)
|
|
|
|
// the final conflictResolution annotation value of Work inherit from RT preferentially
|
|
// so if conflictResolution annotation is defined in RT already, just copy the value and return
|
|
if conflictResolutionInRT == workv1alpha2.ResourceConflictResolutionOverwrite || conflictResolutionInRT == workv1alpha2.ResourceConflictResolutionAbort {
|
|
annotations[workv1alpha2.ResourceConflictResolutionAnnotation] = conflictResolutionInRT
|
|
return annotations
|
|
} else if conflictResolutionInRT != "" {
|
|
// ignore its value and add logs if conflictResolutionInRT is neither abort nor overwrite.
|
|
klog.Warningf("ignore the invalid conflict-resolution annotation in ResourceTemplate %s/%s/%s: %s",
|
|
workload.GetKind(), workload.GetNamespace(), workload.GetName(), conflictResolutionInRT)
|
|
}
|
|
|
|
if conflictResolutionInBinding == policyv1alpha1.ConflictOverwrite {
|
|
annotations[workv1alpha2.ResourceConflictResolutionAnnotation] = workv1alpha2.ResourceConflictResolutionOverwrite
|
|
return annotations
|
|
}
|
|
|
|
annotations[workv1alpha2.ResourceConflictResolutionAnnotation] = workv1alpha2.ResourceConflictResolutionAbort
|
|
return annotations
|
|
}
|
|
|
|
func divideReplicasByJobCompletions(workload *unstructured.Unstructured, clusters []workv1alpha2.TargetCluster) ([]workv1alpha2.TargetCluster, error) {
|
|
var targetClusters []workv1alpha2.TargetCluster
|
|
completions, found, err := unstructured.NestedInt64(workload.Object, util.SpecField, util.CompletionsField)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if found {
|
|
targetClusters = helper.SpreadReplicasByTargetClusters(int32(completions), clusters, nil)
|
|
}
|
|
|
|
return targetClusters, nil
|
|
}
|
|
|
|
func needReviseReplicas(replicas int32, placement *policyv1alpha1.Placement) bool {
|
|
return replicas > 0 && placement != nil && placement.ReplicaSchedulingType() == policyv1alpha1.ReplicaSchedulingTypeDivided
|
|
}
|