diff --git a/apis/kubefleet.dev/placement/v1alpha1/interface.go b/apis/kubefleet.dev/placement/v1alpha1/interface.go new file mode 100644 index 000000000..bfc3100d4 --- /dev/null +++ b/apis/kubefleet.dev/placement/v1alpha1/interface.go @@ -0,0 +1,150 @@ +/* +Copyright 2026 The KubeFleet 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 v1alpha1 + +import ( + "sigs.k8s.io/controller-runtime/pkg/client" +) + +// Verify the implementation of the accessor interfaces for the placement policy, +// placement resource snapshot, and placement binding resources. +var _ PlacementPolicyAccessor = &PlacementPolicy{} +var _ PlacementPolicyAccessor = &ClusterPlacementPolicy{} +var _ PlacementResourceSnapshotAccessor = &PlacementResourceSnapshot{} +var _ PlacementResourceSnapshotAccessor = &ClusterPlacementResourceSnapshot{} +var _ PlacementBindingAccessor = &PlacementBinding{} +var _ PlacementBindingAccessor = &ClusterPlacementBinding{} + +// PlacementPolicyAccessor provides unified access to the spec and status of placement policy resources, +// namespace-scoped and cluster-scoped. +// +// +kubebuilder:object:generate=false +type PlacementPolicyAccessor interface { + client.Object + + GetSpec() *PlacementPolicySpec + GetStatus() *PlacementPolicyStatus + + SetSpec(PlacementPolicySpec) + SetStatus(PlacementPolicyStatus) +} + +func (p *PlacementPolicy) GetSpec() *PlacementPolicySpec { + return &p.Spec +} + +func (p *PlacementPolicy) GetStatus() *PlacementPolicyStatus { + return &p.Status +} + +func (p *PlacementPolicy) SetSpec(spec PlacementPolicySpec) { + p.Spec = spec +} + +func (p *PlacementPolicy) SetStatus(status PlacementPolicyStatus) { + p.Status = status +} + +func (p *ClusterPlacementPolicy) GetSpec() *PlacementPolicySpec { + return &p.Spec +} + +func (p *ClusterPlacementPolicy) GetStatus() *PlacementPolicyStatus { + return &p.Status +} + +func (p *ClusterPlacementPolicy) SetSpec(spec PlacementPolicySpec) { + p.Spec = spec +} + +func (p *ClusterPlacementPolicy) SetStatus(status PlacementPolicyStatus) { + p.Status = status +} + +// PlacementResourceSnapshotAccessor provides unified access to the spec of placement resource snapshot resources, +// namespace-scoped and cluster-scoped. +// +// +kubebuilder:object:generate=false +type PlacementResourceSnapshotAccessor interface { + client.Object + + GetSpec() *PlacementResourceSnapshotSpec + + SetSpec(PlacementResourceSnapshotSpec) +} + +func (p *PlacementResourceSnapshot) GetSpec() *PlacementResourceSnapshotSpec { + return &p.Spec +} + +func (p *PlacementResourceSnapshot) SetSpec(spec PlacementResourceSnapshotSpec) { + p.Spec = spec +} + +func (p *ClusterPlacementResourceSnapshot) GetSpec() *PlacementResourceSnapshotSpec { + return &p.Spec +} + +func (p *ClusterPlacementResourceSnapshot) SetSpec(spec PlacementResourceSnapshotSpec) { + p.Spec = spec +} + +// PlacementBindingAccessor provides unified access to the spec and status of placement binding resources, +// namespace-scoped and cluster-scoped. +// +// +kubebuilder:object:generate=false +type PlacementBindingAccessor interface { + client.Object + + GetSpec() *PlacementBindingSpec + GetStatus() *PlacementBindingStatus + + SetSpec(PlacementBindingSpec) + SetStatus(PlacementBindingStatus) +} + +func (p *PlacementBinding) GetSpec() *PlacementBindingSpec { + return &p.Spec +} + +func (p *PlacementBinding) GetStatus() *PlacementBindingStatus { + return &p.Status +} + +func (p *PlacementBinding) SetSpec(spec PlacementBindingSpec) { + p.Spec = spec +} + +func (p *PlacementBinding) SetStatus(status PlacementBindingStatus) { + p.Status = status +} + +func (p *ClusterPlacementBinding) GetSpec() *PlacementBindingSpec { + return &p.Spec +} + +func (p *ClusterPlacementBinding) GetStatus() *PlacementBindingStatus { + return &p.Status +} + +func (p *ClusterPlacementBinding) SetSpec(spec PlacementBindingSpec) { + p.Spec = spec +} + +func (p *ClusterPlacementBinding) SetStatus(status PlacementBindingStatus) { + p.Status = status +} diff --git a/apis/kubefleet.dev/placement/v1alpha1/placementbinding_types.go b/apis/kubefleet.dev/placement/v1alpha1/placementbinding_types.go index 39e2c21b8..e3976ceef 100644 --- a/apis/kubefleet.dev/placement/v1alpha1/placementbinding_types.go +++ b/apis/kubefleet.dev/placement/v1alpha1/placementbinding_types.go @@ -30,9 +30,11 @@ const ( const ( PlacementBindingSynchronizedCondReasonAllResourcesSynchronized = "AllResourcesSynchronized" PlacementBindingSynchronizedCondReasonFailedToSynchronizeSomeResources = "FailedToSynchronizeSomeResources" + PlacementBindingSynchronizedCondReasonWaitingForSynchronization = "WaitingForSynchronization" - PlacementBindingAvailableCondReasonAllResourcesAvailable = "AllResourcesAvailable" - PlacementBindingAvailableCondReasonSomeResourcesUnavailable = "SomeResourcesUnavailable" + PlacementBindingAvailableCondReasonAllResourcesAvailable = "AllResourcesAvailable" + PlacementBindingAvailableCondReasonSomeResourcesUnavailable = "SomeResourcesUnavailable" + PlacementBindingAvailableCondReasonWaitingForAvailabilityCheck = "WaitingForAvailabilityCheck" ) // PlacementBinding is the KubeFleet API that binds the resources selected by a placement @@ -163,6 +165,13 @@ type PlacementBindingStatus struct { // +kubebuilder:validation:Optional // +kubebuilder:validation:MaxItems=50 FailedResources []FailedResource `json:"failedResources,omitempty"` + + // The name of the placement resource snapshot that KubeFleet has last processed for this binding. + // This field helps KubeFleet track the processing progress; it also reveals whether the reported status + // is up to date. + // + // +kubebuilder:validation:Optional + LastProcessedResourceSnapshotName *string `json:"lastProcessedResourceSnapshotName,omitempty"` } type FailedResource struct { diff --git a/apis/kubefleet.dev/placement/v1alpha1/placementresourcesnapshot_types.go b/apis/kubefleet.dev/placement/v1alpha1/placementresourcesnapshot_types.go index 0f2d284d7..eef9816be 100644 --- a/apis/kubefleet.dev/placement/v1alpha1/placementresourcesnapshot_types.go +++ b/apis/kubefleet.dev/placement/v1alpha1/placementresourcesnapshot_types.go @@ -21,6 +21,50 @@ import ( runtime "k8s.io/apimachinery/pkg/runtime" ) +const ( + // When users create a placement policy to place resources across member clusters, KubeFleet will capture the + // resources selected by the placement policy at a specific point in time in the form of placement resource + // snapshots. This enables KubeFleet to roll out resources to member clusters in a consistent manner. + // + // As resources change over time, there might be a time series of placement resource snapshots associated + // with a placement policy. KubeFleet assigns these snapshots with a monotonically increasing index based on + // their creation timestamp, starting from 0 with a step of 1. + // + // Due to sizing limitations in Kubernetes, when there are too many resources being selected at a time, or + // when some resources are too large, KubeFleet will capture them using multiple placement resource snapshots. + // These snapshots share the same index (as they are snapshots from the same point in time), and KubeFleet + // will further assign them each with a sub-index to tell them apart, also starting from 0 with a step of 1. + // The snapshot of the sub-index 0 is considered the primary snapshot of the same index. + + // PlacementResourceSnapshotOwnedByLabelKey is a label key that denotes the owner placement policy of + // a placement resource snapshot. Its value is the name of the owner placement policy. + // + // This label is set on all placement resource snapshots. + PlacementResourceSnapshotOwnedByLabelKey = "placement.kubefleet.dev/placement-resource-snapshot-owned-by" + // PlacementResourceSnapshotIndexLabelKey is a label key that denotes the index of a placement resource snapshot. + // Its value is the index integer formatted as a string. + // + // This label is set on all placement resource snapshots. + PlacementResourceSnapshotIndexLabelKey = "placement.kubefleet.dev/placement-resource-snapshot-index" + // PlacementResourceSnapshotSubIndexLabelKey is a label key that denotes the sub-index of a placement resource snapshot. + // Its value is the sub-index integer formatted as a string. + // + // This label is set on all placement resource snapshots. + PlacementResourceSnapshotSubIndexLabelKey = "placement.kubefleet.dev/placement-resource-snapshot-sub-index" + // SubIndexedPlacementResourceSnapshotCountLabelKey is a label key that denotes the total number of sub-indexed + // placement resource snapshots associated with the same index. Its value is the count integer + // formatted as a string. + // + // This label is set only on resource placement snapshots with the sub-index of 0. + SubIndexedPlacementResourceSnapshotCountLabelKey = "placement.kubefleet.dev/sub-indexed-placement-resource-snapshot-count" + + // PlacementResourceSnapshotContentsHashAnnotationKey is an annotation key that denotes the hash of the contents + // of a placement resource snapshot. Its value is the hash string. + // + // This annotation is set on all placement resource snapshots. + PlacementResourceSnapshotContentsHashAnnotationKey = "placement.kubefleet.dev/placement-resource-snapshot-contents-hash" +) + // PlacementResourceSnapshot is the KubeFleet API that captures the resources selected by a placement policy // as seen on the hub cluster at a specific point in time. It is referenced by other KubeFleet APIs // to enable consistent rollouts of resources across multiple member clusters in the fleet. diff --git a/apis/kubefleet.dev/placement/v1alpha1/work_types.go b/apis/kubefleet.dev/placement/v1alpha1/work_types.go index f282ab694..5b12fe155 100644 --- a/apis/kubefleet.dev/placement/v1alpha1/work_types.go +++ b/apis/kubefleet.dev/placement/v1alpha1/work_types.go @@ -21,6 +21,16 @@ import ( "k8s.io/apimachinery/pkg/runtime" ) +const ( + WorkOwnerNamespaceLabelKey = "placement.kubefleet.dev/owner-namespace" + WorkOwnedByPlacementPolicyLabelKey = "placement.kubefleet.dev/owned-by-placement-policy" + WorkOwnedByPlacementBindingLabelKey = "placement.kubefleet.dev/owned-by-placement-binding" + + WorkLinkedToPrimaryPlacementResourceSnapshotAnnotationKey = "placement.kubefleet.dev/linked-to-primary-placement-resource-snapshot" + LinkedWorkCountAnnotationKey = "placement.kubefleet.dev/linked-work-count" + WorkDerivedFromSourceAnnotationKey = "placement.kubefleet.dev/derived-from" +) + const ( // The condition types for the Work API. WorkCondTypeApplied = "Applied" diff --git a/apis/kubefleet.dev/placement/v1alpha1/zz_generated.deepcopy.go b/apis/kubefleet.dev/placement/v1alpha1/zz_generated.deepcopy.go index 429b4e7fa..ad5cad7a4 100644 --- a/apis/kubefleet.dev/placement/v1alpha1/zz_generated.deepcopy.go +++ b/apis/kubefleet.dev/placement/v1alpha1/zz_generated.deepcopy.go @@ -710,6 +710,11 @@ func (in *PlacementBindingStatus) DeepCopyInto(out *PlacementBindingStatus) { (*in)[i].DeepCopyInto(&(*out)[i]) } } + if in.LastProcessedResourceSnapshotName != nil { + in, out := &in.LastProcessedResourceSnapshotName, &out.LastProcessedResourceSnapshotName + *out = new(string) + **out = **in + } } // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new PlacementBindingStatus. diff --git a/config/crd/bases/placement.kubefleet.dev_clusterplacementbindings.yaml b/config/crd/bases/placement.kubefleet.dev_clusterplacementbindings.yaml index a78e895bc..87dfb4ef9 100644 --- a/config/crd/bases/placement.kubefleet.dev_clusterplacementbindings.yaml +++ b/config/crd/bases/placement.kubefleet.dev_clusterplacementbindings.yaml @@ -560,6 +560,12 @@ spec: type: object maxItems: 50 type: array + lastProcessedResourceSnapshotName: + description: |- + The name of the placement resource snapshot that KubeFleet has last processed for this binding. + This field helps KubeFleet track the processing progress; it also reveals whether the reported status + is up to date. + type: string selectedResources: description: The number of resources that are included in the currently associated resource snapshot(s). diff --git a/config/crd/bases/placement.kubefleet.dev_placementbindings.yaml b/config/crd/bases/placement.kubefleet.dev_placementbindings.yaml index b45d48f4e..e6c135e67 100644 --- a/config/crd/bases/placement.kubefleet.dev_placementbindings.yaml +++ b/config/crd/bases/placement.kubefleet.dev_placementbindings.yaml @@ -560,6 +560,12 @@ spec: type: object maxItems: 50 type: array + lastProcessedResourceSnapshotName: + description: |- + The name of the placement resource snapshot that KubeFleet has last processed for this binding. + This field helps KubeFleet track the processing progress; it also reveals whether the reported status + is up to date. + type: string selectedResources: description: The number of resources that are included in the currently associated resource snapshot(s). diff --git a/pkg/v1/controllers/workgenerator/cleanup.go b/pkg/v1/controllers/workgenerator/cleanup.go new file mode 100644 index 000000000..4ef934dbc --- /dev/null +++ b/pkg/v1/controllers/workgenerator/cleanup.go @@ -0,0 +1,83 @@ +/* +Copyright 2026 The KubeFleet 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 workgenerator + +import ( + "context" + "fmt" + + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + "k8s.io/klog/v2" + "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" + + placementv1alpha1 "github.com/kubefleet-dev/kubefleet/apis/kubefleet.dev/placement/v1alpha1" + "github.com/kubefleet-dev/kubefleet/pkg/utils" + "github.com/kubefleet-dev/kubefleet/pkg/utils/errors" +) + +func (r *Reconciler) addPlacementBindingCleanupFinalizer(ctx context.Context, placementBinding placementv1alpha1.PlacementBindingAccessor) error { + if controllerutil.ContainsFinalizer(placementBinding, workGeneratorCleanupFinalizer) { + return nil + } + controllerutil.AddFinalizer(placementBinding, workGeneratorCleanupFinalizer) + if err := r.hubClient.Update(ctx, placementBinding); err != nil { + return errors.NewAPIServerError(err, "failed to add cleanup finalizer to placement binding", false) + } + return nil +} + +// cleanupWorks deletes the primary Work object owned by a placement binding in the reserved namespace of the +// target cluster; all the other Work objects are cleaned up via owner-reference cascade deletion. +func (r *Reconciler) cleanupWorks(ctx context.Context, placementBinding placementv1alpha1.PlacementBindingAccessor) error { + if !controllerutil.ContainsFinalizer(placementBinding, workGeneratorCleanupFinalizer) { + // The cleanup finalizer has been dropped; no cleanup is needed. + return nil + } + + derivedFromSourceFormatter := &placementResourceSnapshotDerivedFromSourceFormatter{ + snapshotNamespacedName: types.NamespacedName{ + Namespace: placementBinding.GetNamespace(), + Name: placementBinding.GetSpec().PlacementPolicyName, + }, + snapshotSubIdx: "0", + } + workName, err := uniqueNameForWorkDerivedFromPlacementResourceSnapshot(placementBinding, true, derivedFromSourceFormatter) + if err != nil { + return errors.Wraps(err, "failed to generate work name for primary placement resource snapshot") + } + workForPrimaryResSnapshot := &placementv1alpha1.Work{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: fmt.Sprintf(utils.NamespaceNameFormat, placementBinding.GetSpec().ClusterName), + Name: workName, + }, + } + if err := r.hubClient.Delete(ctx, workForPrimaryResSnapshot); err != nil && !apierrors.IsNotFound(err) { + return errors.NewAPIServerError(err, "failed to delete work object for primary placement resource snapshot", false, + "work", klog.KObj(workForPrimaryResSnapshot)) + } + // This work object is set as the owner of all other work objects created for this placement binding; + // no further cleanup is needed. + + // Remove the cleanup finalizer from the placement binding. + controllerutil.RemoveFinalizer(placementBinding, workGeneratorCleanupFinalizer) + if err := r.hubClient.Update(ctx, placementBinding); err != nil { + return errors.NewAPIServerError(err, "failed to remove cleanup finalizer from placement binding", false) + } + return nil +} diff --git a/pkg/v1/controllers/workgenerator/controller.go b/pkg/v1/controllers/workgenerator/controller.go new file mode 100644 index 000000000..b4da0b03e --- /dev/null +++ b/pkg/v1/controllers/workgenerator/controller.go @@ -0,0 +1,270 @@ +/* +Copyright 2026 The KubeFleet 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 workgenerator + +import ( + "context" + "time" + + "k8s.io/client-go/util/workqueue" + "k8s.io/klog/v2" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/builder" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/controller" + "sigs.k8s.io/controller-runtime/pkg/event" + "sigs.k8s.io/controller-runtime/pkg/handler" + "sigs.k8s.io/controller-runtime/pkg/predicate" + "sigs.k8s.io/controller-runtime/pkg/reconcile" + + "k8s.io/apimachinery/pkg/api/equality" + apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/types" + + placementv1alpha1 "github.com/kubefleet-dev/kubefleet/apis/kubefleet.dev/placement/v1alpha1" + "github.com/kubefleet-dev/kubefleet/pkg/utils/errors" + parallelizerutil "github.com/kubefleet-dev/kubefleet/pkg/utils/parallelizer" +) + +const ( + controllerName = "work-generator" + + workGeneratorCleanupFinalizer = "placement.kubefleet.dev/work-generator-cleanup" +) + +type Reconciler struct { + hubClient client.Client + + parallelizer parallelizerutil.Parallelizer +} + +func New(hubClient client.Client, workerCnt int) *Reconciler { + parallelizer := parallelizerutil.NewParallelizer(workerCnt) + + return &Reconciler{ + hubClient: hubClient, + parallelizer: parallelizer, + } +} + +// TO-DO (chenyu1): switch to field-based indexes for better performance when listing objects. + +func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) { + startTime := time.Now() + klog.V(2).InfoS("Reconciliation starts", "placementBinding", req.NamespacedName, "controller", controllerName) + defer func() { + latency := time.Since(startTime).Milliseconds() + klog.V(2).InfoS("Reconciliation ends", "placementBinding", req.NamespacedName, "controller", controllerName, "latency", latency) + }() + + // Retrieve the PlacementBinding object. + placementBinding, err := r.retrievePlacementBinding(ctx, req.NamespacedName) + if err != nil { + if apierrors.IsNotFound(err) { + klog.V(2).InfoS("placement binding is not found", "namespacedName", req.NamespacedName, "controller", controllerName) + return ctrl.Result{}, nil + } + klog.ErrorS(err, "", "namespacedName", req.NamespacedName, "controller", controllerName) + return ctrl.Result{}, errors.Wraps(err, "", "namespacedName", req.NamespacedName, "controller", controllerName) + } + + placementBindingSpec := placementBinding.GetSpec() + // Clean up the work objects for the placement binding if it has been marked for deletion or if it has been + // suspended. + if placementBinding.GetDeletionTimestamp() != nil || placementBindingSpec.Suspended { + if err := r.cleanupWorks(ctx, placementBinding); err != nil { + wrappedErr := errors.Wraps(err, "failed to clean up work objects for placement binding", + "placementBinding", klog.KObj(placementBinding), "controller", controllerName) + klog.ErrorS(wrappedErr, "failed to clean up work objects for placement binding", + errors.Args(wrappedErr)...) + return ctrl.Result{}, wrappedErr + } + return ctrl.Result{}, nil + } + // Add the cleanup finalizer if it is not already present. + if err := r.addPlacementBindingCleanupFinalizer(ctx, placementBinding); err != nil { + wrappedErr := errors.Wraps(err, "failed to add cleanup finalizer to placement binding", + "placementBinding", klog.KObj(placementBinding), "controller", controllerName) + klog.ErrorS(wrappedErr, "failed to add cleanup finalizer to placement binding", errors.Args(wrappedErr)...) + return ctrl.Result{}, wrappedErr + } + + // Do a sanity check; verify if a target cluster and a (primary) placement resource snapshot have been assigned. + if len(placementBindingSpec.ClusterName) == 0 || len(placementBindingSpec.ResourceSnapshotName) == 0 { + wrappedErr := errors.NewUnexpectedError(nil, "the placement binding does not have a target cluster or a placement resource snapshot assigned", + "placementBinding", klog.KObj(placementBinding), "controller", controllerName) + klog.ErrorS(wrappedErr, "failed to process placement binding", errors.Args(wrappedErr)...) + return ctrl.Result{}, wrappedErr + } + + // Retrieve the Work objects owned by the placement binding. + works, err := r.listWorksByOwnerBinding(ctx, placementBindingSpec.ClusterName, placementBinding.GetNamespace(), placementBinding.GetName()) + if err != nil { + wrappedErr := errors.Wraps(err, "", "placementBinding", klog.KObj(placementBinding), + "targetCluster", placementBindingSpec.ClusterName, "controller", controllerName) + klog.ErrorS(wrappedErr, "failed to list work objects owned by binding", errors.Args(wrappedErr)...) + return ctrl.Result{}, wrappedErr + } + + // Check if the Work objects are consistent with the assigned primary and secondary placement resource snapshots. + // If so, no need to update the spec of the Work objects; just sync the status back to the placement binding + // instead. + // + // Note (chenyu1): this check is intended as a shortcut to avoid constant re-generation and validation of + // work objects (which can be expensive when there are a large number of manifests to place); once the controller + // signals that it has completed processing a placement binding given a specific configuration (a specific + // set of placement resource snapshots) and generated all the needed work objects, the control loop will skip + // to status reporting. In general we do not try to guard against byzantine faults here, especially + // considering that work objects are KubeFleet internal API objects that reside in reserved namespaces; if a + // non-KubeFleet agent decides to tamper with work objects, the system is not guaranteed to auto-recover. + // The changes, however, will be overwritten upon rollouts. + upToDate, err := areWorksUpToDate(placementBinding, works) + if err != nil { + wrappedErr := errors.Wraps(err, "failed to check if work objects are up-to-date", + "placementBinding", klog.KObj(placementBinding), "targetCluster", placementBindingSpec.ClusterName, + "controller", controllerName) + klog.ErrorS(wrappedErr, "failed to check if work objects are up-to-date", errors.Args(wrappedErr)...) + return ctrl.Result{}, wrappedErr + } + if upToDate { + if err := r.refreshPlacementBindingStatus(ctx, placementBinding, works); err != nil { + wrappedErr := errors.Wraps(err, "failed to refresh placement binding status", + "placementBinding", klog.KObj(placementBinding), "targetCluster", placementBindingSpec.ClusterName, + "controller", controllerName) + klog.ErrorS(wrappedErr, "failed to refresh placement binding status", errors.Args(wrappedErr)...) + return ctrl.Result{}, wrappedErr + } + return ctrl.Result{}, nil + } + + // The Work objects are absent or not up-to-date. Retrieve the placement resource snapshots and create/update + // the Work objects accordingly. + + // Retrieve the assigned primary and secondary placement resource snapshots referenced by the placement binding. + placementResourceSnapshots, err := r.retrievePrimaryAndSecondaryPlacementResourceSnapshots(ctx, placementBinding) + if err != nil { + wrappedErr := errors.Wraps(err, "failed to retrieve placement resource snapshots", + "placementBinding", klog.KObj(placementBinding), "targetCluster", placementBindingSpec.ClusterName, + "controller", controllerName) + klog.ErrorS(wrappedErr, "failed to retrieve placement resource snapshots referenced by binding", + errors.Args(wrappedErr)...) + return ctrl.Result{}, wrappedErr + } + + // Create or update the work objects. + createdOrUpdatedWorks, writtenToStorage, err := r.refreshWorks(ctx, placementBinding, placementResourceSnapshots, works) + if err != nil { + wrappedErr := errors.Wraps(err, "failed to refresh work objects", + "placementBinding", klog.KObj(placementBinding), "targetCluster", placementBindingSpec.ClusterName, + "controller", controllerName) + klog.ErrorS(wrappedErr, "failed to refresh work objects for placement binding", + errors.Args(wrappedErr)...) + return ctrl.Result{}, wrappedErr + } + + // Report the processing progress via placement binding status. + if err := r.reportPlacementBindingProcessingProgress(ctx, placementBinding, placementResourceSnapshots[0], createdOrUpdatedWorks); err != nil { + wrappedErr := errors.Wraps(err, "failed to report placement binding processing progress", + "placementBinding", klog.KObj(placementBinding), "targetCluster", placementBindingSpec.ClusterName, + "controller", controllerName) + klog.ErrorS(wrappedErr, "failed to report placement binding processing progress", + errors.Args(wrappedErr)...) + return ctrl.Result{}, wrappedErr + } + + // The work objects have been refreshed. Normally the controller needs only to wait for the work objects + // to be processed by the KubeFleet member agent, then refresh the placement binding status upon receiving + // create/update events from the work objects, and there is no need to requeue manually. However, there exists + // a corner case in which a rollout attempt does not involve any change in the work objects; in this case there + // will not be any change events from the work objects and the work generator needs to requeue manually to + // have the placement binding status refreshed. + if !writtenToStorage { + // The work objects have not been created or updated; requeue manually. + return ctrl.Result{RequeueAfter: 1 * time.Second}, nil + } + // The work objects have been created or updated; wait for change events from the work objects to refresh + // the placement binding status. + return ctrl.Result{}, nil +} + +func (r *Reconciler) SetupWithManager(mgr ctrl.Manager, maxConcurrentReconciles int) error { + // enqueueOwnerBindingForWork resolves the owner placement binding from a work object's labels and enqueues it + // for reconciliation. eventType is used for logging only. + enqueueOwnerBindingForWork := func(work client.Object, eventType string, q workqueue.TypedRateLimitingInterface[reconcile.Request]) { + if work == nil { + wrappedErr := errors.NewUnexpectedError(nil, "received a nil work object", "eventType", eventType, "controller", controllerName) + klog.ErrorS(wrappedErr, "received a nil work object", errors.Args(wrappedErr)...) + return + } + labels := work.GetLabels() + ownerBindingNSName, nsNameFound := labels[placementv1alpha1.WorkOwnerNamespaceLabelKey] + ownerBindingName, bindingNameFound := labels[placementv1alpha1.WorkOwnedByPlacementBindingLabelKey] + if !nsNameFound || !bindingNameFound { + err := errors.NewUnexpectedError(nil, "work object is missing required labels", + "work", klog.KObj(work), "eventType", eventType, "controller", controllerName) + klog.ErrorS(err, "work object is missing required labels", errors.Args(err)...) + return + } + ownerBinding := types.NamespacedName{Namespace: ownerBindingNSName, Name: ownerBindingName} + klog.V(2).InfoS("Enqueue the owner placement binding for reconciliation", + "work", klog.KObj(work), "eventType", eventType, "placementBinding", ownerBinding) + q.Add(reconcile.Request{NamespacedName: ownerBinding}) + } + + workObjHandlerFuncs := handler.Funcs{ + // The controller needs to watch for work object create events as the client-side cache might + // lag under heavy load, i.e., it might learn about a work object only after its status has been updated. + CreateFunc: func(_ context.Context, e event.TypedCreateEvent[client.Object], q workqueue.TypedRateLimitingInterface[reconcile.Request]) { + enqueueOwnerBindingForWork(e.Object, "create", q) + }, + UpdateFunc: func(_ context.Context, e event.TypedUpdateEvent[client.Object], q workqueue.TypedRateLimitingInterface[reconcile.Request]) { + if e.ObjectOld == nil || e.ObjectNew == nil { + wrappedErr := errors.NewUnexpectedError(nil, "received nil work objects in update event", "controller", controllerName) + klog.ErrorS(wrappedErr, "received nil work objects in update event", errors.Args(wrappedErr)...) + return + } + + oldWork, canCastOldWork := e.ObjectOld.(*placementv1alpha1.Work) + newWork, canCastNewWork := e.ObjectNew.(*placementv1alpha1.Work) + if !canCastOldWork || !canCastNewWork { + wrappedErr := errors.NewUnexpectedError(nil, "failed to cast work objects in update event", "controller", controllerName) + klog.ErrorS(wrappedErr, "failed to cast work objects in update event", errors.Args(wrappedErr)...) + return + } + + // Only enqueue when the status has changed, so that status can be synced back to the owner binding. + if !equality.Semantic.DeepEqual(oldWork.Status, newWork.Status) { + enqueueOwnerBindingForWork(e.ObjectNew, "update", q) + } + }, + DeleteFunc: func(_ context.Context, e event.TypedDeleteEvent[client.Object], q workqueue.TypedRateLimitingInterface[reconcile.Request]) { + enqueueOwnerBindingForWork(e.Object, "delete", q) + }, + } + + return ctrl.NewControllerManagedBy(mgr). + Named(controllerName). + WithOptions(controller.Options{MaxConcurrentReconciles: maxConcurrentReconciles}). + // The controller watches placement binding objects (both namespace-scoped and cluster-scoped) for spec + // changes (generation predicate). + Watches(&placementv1alpha1.PlacementBinding{}, &handler.EnqueueRequestForObject{}, builder.WithPredicates(predicate.GenerationChangedPredicate{})). + Watches(&placementv1alpha1.ClusterPlacementBinding{}, &handler.EnqueueRequestForObject{}, builder.WithPredicates(predicate.GenerationChangedPredicate{})). + // The controller watches work objects for status changes, so that status can be synced back to their + // owner placement bindings. + Watches(&placementv1alpha1.Work{}, workObjHandlerFuncs). + Complete(r) +} diff --git a/pkg/v1/controllers/workgenerator/derivedfrom.go b/pkg/v1/controllers/workgenerator/derivedfrom.go new file mode 100644 index 000000000..d475ce67d --- /dev/null +++ b/pkg/v1/controllers/workgenerator/derivedfrom.go @@ -0,0 +1,61 @@ +/* +Copyright 2026 The KubeFleet 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 workgenerator + +import ( + "k8s.io/apimachinery/pkg/types" +) + +// Verify that all formatter implements the derivedFromSourceFormatter interface. +var _ derivedFromSourceFormatter = &placementResourceSnapshotDerivedFromSourceFormatter{} + +// derivedFromSourceFormatter is an interface that helps format the ID of a source that derives a work object for +// various use cases, primarily for preparing unique names for work objects. +type derivedFromSourceFormatter interface { + // StrictDNSLabel returns a string that is a valid DNS label (max. 63 chars, all lowercase, alphanumeric characters + // and hyphens, and must start and end with an alphanumeric character). + // + // This value is used as a sub-component of the unique name for a work object derived from a source. It is + // for informational purposes only and does not need to be unique across all work objects that are created/updated + // for the same placement binding. + StrictDNSLabel() string + // SourceType returns a string that identifies the type of the source that derives a work object, e.g., + // `placement-resource-snapshot` for placement resource snapshots. + SourceType() string + // SourceID returns a string that uniquely identifies the source (of the same type) that derives a + // work object. + SourceID() string +} + +// placementResourceSnapshotDerivedFromSourceFormatter is a formatter for placement resource snapshots that implements +// the derivedFromSourceFormatter interface. +type placementResourceSnapshotDerivedFromSourceFormatter struct { + snapshotNamespacedName types.NamespacedName + snapshotSubIdx string +} + +func (f *placementResourceSnapshotDerivedFromSourceFormatter) SourceID() string { + return f.snapshotSubIdx +} + +func (f *placementResourceSnapshotDerivedFromSourceFormatter) SourceType() string { + return "placement-resource-snapshot" +} + +func (f *placementResourceSnapshotDerivedFromSourceFormatter) StrictDNSLabel() string { + return f.snapshotSubIdx +} diff --git a/pkg/v1/controllers/workgenerator/retrieval.go b/pkg/v1/controllers/workgenerator/retrieval.go new file mode 100644 index 000000000..d3b1de2d8 --- /dev/null +++ b/pkg/v1/controllers/workgenerator/retrieval.go @@ -0,0 +1,184 @@ +/* +Copyright 2026 The KubeFleet 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 workgenerator + +import ( + "context" + "fmt" + "sort" + "strconv" + + "k8s.io/apimachinery/pkg/types" + "k8s.io/klog/v2" + "sigs.k8s.io/controller-runtime/pkg/client" + + placementv1alpha1 "github.com/kubefleet-dev/kubefleet/apis/kubefleet.dev/placement/v1alpha1" + "github.com/kubefleet-dev/kubefleet/pkg/utils" + "github.com/kubefleet-dev/kubefleet/pkg/utils/errors" +) + +func (r *Reconciler) retrievePlacementBinding(ctx context.Context, namespacedName types.NamespacedName) (placementv1alpha1.PlacementBindingAccessor, error) { + var placementBinding placementv1alpha1.PlacementBindingAccessor + if namespacedName.Namespace == "" { + // The placement binding is cluster-scoped. + placementBinding = &placementv1alpha1.ClusterPlacementBinding{} + } else { + // The placement binding is namespace-scoped. + placementBinding = &placementv1alpha1.PlacementBinding{} + } + + if err := r.hubClient.Get(ctx, namespacedName, placementBinding); err != nil { + return nil, errors.NewAPIServerError(err, "failed to retrieve placement binding", true) + } + return placementBinding, nil +} + +// listWorksByOwnerBinding lists the Work objects owned by a placement binding within a Fleet member cluster reserved +// namespace. +func (r *Reconciler) listWorksByOwnerBinding(ctx context.Context, clusterName, ownerBindingNSName, ownerBindingName string) ([]placementv1alpha1.Work, error) { + memberClusterNamespace := fmt.Sprintf(utils.NamespaceNameFormat, clusterName) + + workList := &placementv1alpha1.WorkList{} + listOptions := []client.ListOption{ + client.InNamespace(memberClusterNamespace), + client.MatchingLabels{ + placementv1alpha1.WorkOwnedByPlacementBindingLabelKey: ownerBindingName, + placementv1alpha1.WorkOwnerNamespaceLabelKey: ownerBindingNSName, + }, + } + if err := r.hubClient.List(ctx, workList, listOptions...); err != nil { + return nil, errors.NewAPIServerError(err, "failed to list work objects", true) + } + return workList.Items, nil +} + +// retrievePrimaryAndSecondaryPlacementResourceSnapshots retrieves the primary placement resource snapshot referenced +// by the placement binding, along with any secondary snapshots that share the same index. The returned snapshots are +// sorted in ascending order of their sub-indices (the primary, sub-index 0, comes first). +func (r *Reconciler) retrievePrimaryAndSecondaryPlacementResourceSnapshots( + ctx context.Context, + placementBinding placementv1alpha1.PlacementBindingAccessor, +) ([]placementv1alpha1.PlacementResourceSnapshotAccessor, error) { + namespace := placementBinding.GetNamespace() + primarySnapshotName := placementBinding.GetSpec().ResourceSnapshotName + + // Retrieve the primary placement resource snapshot referenced by the binding. + var primarySnapshot placementv1alpha1.PlacementResourceSnapshotAccessor + if namespace == "" { + // The placement binding is cluster-scoped. + primarySnapshot = &placementv1alpha1.ClusterPlacementResourceSnapshot{} + } else { + // The placement binding is namespace-scoped. + primarySnapshot = &placementv1alpha1.PlacementResourceSnapshot{} + } + if err := r.hubClient.Get(ctx, types.NamespacedName{Namespace: namespace, Name: primarySnapshotName}, primarySnapshot); err != nil { + return nil, errors.NewAPIServerError(err, "failed to retrieve the primary placement resource snapshot", true, + "primaryPlacementResourceSnapshotName", primarySnapshotName) + } + + // Determine how many snapshots share the same index via the count label on the primary snapshot. + countStr := primarySnapshot.GetLabels()[placementv1alpha1.SubIndexedPlacementResourceSnapshotCountLabelKey] + count, err := strconv.Atoi(countStr) + if err != nil || count < 1 { + return nil, errors.NewUnexpectedError(err, "invalid sub-indexed placement resource snapshot count label on the primary placement resource snapshot", + "primaryPlacementResourceSnapshot", klog.KObj(primarySnapshot), "countVal", countStr) + } + if count == 1 { + // The primary snapshot is the only snapshot associated with this index. + return []placementv1alpha1.PlacementResourceSnapshotAccessor{primarySnapshot}, nil + } + + // There are secondary snapshots; list all snapshots that share the same owner and index. + ownedBy := primarySnapshot.GetLabels()[placementv1alpha1.PlacementResourceSnapshotOwnedByLabelKey] + index := primarySnapshot.GetLabels()[placementv1alpha1.PlacementResourceSnapshotIndexLabelKey] + if ownedBy == "" || index == "" { + return nil, errors.NewUnexpectedError(nil, "the primary placement resource snapshot is missing required labels", + "primaryPlacementResourceSnapshot", klog.KObj(primarySnapshot)) + } + labelMatchers := client.MatchingLabels{ + placementv1alpha1.PlacementResourceSnapshotOwnedByLabelKey: ownedBy, + placementv1alpha1.PlacementResourceSnapshotIndexLabelKey: index, + } + + var snapshots []placementv1alpha1.PlacementResourceSnapshotAccessor + if namespace == "" { + snapshotList := &placementv1alpha1.ClusterPlacementResourceSnapshotList{} + if err := r.hubClient.List(ctx, snapshotList, labelMatchers); err != nil { + return nil, errors.NewAPIServerError(err, "failed to list cluster placement resource snapshots", true) + } + snapshots = make([]placementv1alpha1.PlacementResourceSnapshotAccessor, len(snapshotList.Items)) + for i := range snapshotList.Items { + snapshots[i] = &snapshotList.Items[i] + } + } else { + snapshotList := &placementv1alpha1.PlacementResourceSnapshotList{} + if err := r.hubClient.List(ctx, snapshotList, client.InNamespace(namespace), labelMatchers); err != nil { + return nil, errors.NewAPIServerError(err, "failed to list placement resource snapshots", true) + } + snapshots = make([]placementv1alpha1.PlacementResourceSnapshotAccessor, len(snapshotList.Items)) + for i := range snapshotList.Items { + snapshots[i] = &snapshotList.Items[i] + } + } + + // Sort the snapshots by their sub-indices in ascending order. + var sortErrs []error + sort.Slice(snapshots, func(i, j int) bool { + subIdxI, iErr := strconv.Atoi(snapshots[i].GetLabels()[placementv1alpha1.PlacementResourceSnapshotSubIndexLabelKey]) + subIdxJ, jErr := strconv.Atoi(snapshots[j].GetLabels()[placementv1alpha1.PlacementResourceSnapshotSubIndexLabelKey]) + if iErr != nil { + sortErrs = append(sortErrs, fmt.Errorf("failed to convert sub-index label to integer: %w (placementResourceSnapshot: %s)", iErr, snapshots[i].GetName())) + return false + } + if jErr != nil { + sortErrs = append(sortErrs, fmt.Errorf("failed to convert sub-index label to integer: %w (placementResourceSnapshot: %s)", jErr, snapshots[j].GetName())) + return false + } + return subIdxI < subIdxJ + }) + if len(sortErrs) > 0 { + return nil, errors.NewUnexpectedError(nil, "failed to sort placement resource snapshots by sub-index", "errs", sortErrs) + } + + // Do some sanity checks; verify that all snapshots dictated by the count label are present and they have + // the same snapshotted resource hash. + + if len(snapshots) < count { + // Normally this branch will never run, as the placement resource snapshot manager creates secondary + // snapshots first, then the primary snapshot with the count label. + return nil, errors.NewUnexpectedError(nil, "there are fewer placement resource snapshots than the count label indicates", + "primaryPlacementResourceSnapshot", klog.KObj(primarySnapshot), "expectedCount", count, "actualCount", len(snapshots)) + } + + primarySnapshottedResHash := primarySnapshot.GetAnnotations()[placementv1alpha1.PlacementResourceSnapshotContentsHashAnnotationKey] + for i := range snapshots[:count] { + snapshottedResHash := snapshots[i].GetAnnotations()[placementv1alpha1.PlacementResourceSnapshotContentsHashAnnotationKey] + if snapshottedResHash != primarySnapshottedResHash { + // Normally this branch will never run, as the placement resource snapshot manager uses ordered creation + // to make sure that hashes are consistent across all snapshots with the same index. + return nil, errors.NewUnexpectedError(nil, "the contents hash of a placement resource snapshot does not match the primary snapshot", + "primaryPlacementResourceSnapshot", klog.KObj(primarySnapshot), + "hashMismatchedPlacementResourceSnapshot", klog.KObj(snapshots[i]), + "hashOnPrimaryPlacementResourceSnapshot", primarySnapshottedResHash, + "mismatchedHash", snapshottedResHash) + } + } + + // Any snapshots beyond the count are orphans from an overwritten resource change; return only the ones + // dictated by the count label, which are guaranteed to be consistent. + return snapshots[:count], nil +} diff --git a/pkg/v1/controllers/workgenerator/status.go b/pkg/v1/controllers/workgenerator/status.go new file mode 100644 index 000000000..c05d6e16a --- /dev/null +++ b/pkg/v1/controllers/workgenerator/status.go @@ -0,0 +1,251 @@ +/* +Copyright 2026 The KubeFleet 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 workgenerator + +import ( + "context" + + "k8s.io/apimachinery/pkg/api/equality" + "k8s.io/apimachinery/pkg/api/meta" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/klog/v2" + "k8s.io/utils/ptr" + + placementv1alpha1 "github.com/kubefleet-dev/kubefleet/apis/kubefleet.dev/placement/v1alpha1" + "github.com/kubefleet-dev/kubefleet/pkg/utils/condition" + "github.com/kubefleet-dev/kubefleet/pkg/utils/errors" +) + +func (r *Reconciler) refreshPlacementBindingStatus( + ctx context.Context, + placementBinding placementv1alpha1.PlacementBindingAccessor, + works []placementv1alpha1.Work, +) error { + oldStatus := placementBinding.GetStatus().DeepCopy() + + refreshPlacementBindingSyncCond(placementBinding, works) + refreshPlacementBindingAvailableCond(placementBinding, works) + + total, synced, available, failed := countResourcesInWorksByProcessingResults(works) + placementBinding.GetStatus().SelectedResources = ptr.To(int32(total)) + placementBinding.GetStatus().SynchronizedResources = ptr.To(int32(synced)) + placementBinding.GetStatus().AvailableResources = ptr.To(int32(available)) + if len(failed) > 50 { + klog.V(2).InfoS("Too many failed resources to report in placement binding status; truncating the list to 50", + "placementBinding", klog.KObj(placementBinding), "totalFailedResources", len(failed)) + failed = failed[:50] + } + placementBinding.GetStatus().FailedResources = failed + + // Skip the update if the status has not changed. + if equality.Semantic.DeepEqual(oldStatus, placementBinding.GetStatus()) { + klog.V(2).InfoS("No need to update placement binding status as it has not changed", + "placementBinding", klog.KObj(placementBinding), + "selectedResources", total, "synchronizedResources", synced, "availableResources", available, + "failedResources", len(failed)) + return nil + } + + if err := r.hubClient.Status().Update(ctx, placementBinding); err != nil { + return errors.NewAPIServerError(err, "failed to update placement binding status", true) + } + klog.V(2).InfoS("Updated placement binding status", + "placementBinding", klog.KObj(placementBinding), + "selectedResources", total, "synchronizedResources", synced, "availableResources", available, + "failedResources", len(failed)) + return nil +} + +func (r *Reconciler) reportPlacementBindingProcessingProgress( + ctx context.Context, + placementBinding placementv1alpha1.PlacementBindingAccessor, + primaryPlacementResourceSnapshot placementv1alpha1.PlacementResourceSnapshotAccessor, + worksToCreateOrUpdate []*placementv1alpha1.Work, +) error { + placementBindingStatus := placementBinding.GetStatus() + // Set the last processed placement resource snapshot name on the placement binding status. + placementBindingStatus.LastProcessedResourceSnapshotName = ptr.To(primaryPlacementResourceSnapshot.GetName()) + + // Set a false Synchronized condition with the WaitingForSynchronization reason on the placement binding. + meta.SetStatusCondition(&placementBindingStatus.Conditions, metav1.Condition{ + Type: placementv1alpha1.PlacementBindingCondTypeSynchronized, + Status: metav1.ConditionFalse, + ObservedGeneration: placementBinding.GetGeneration(), + Reason: placementv1alpha1.PlacementBindingSynchronizedCondReasonWaitingForSynchronization, + Message: "Waiting for the resources to be synchronized to the target cluster", + }) + // Set an unknown Available condition with the WaitingForAvailabilityCheck reason on the placement binding. + meta.SetStatusCondition(&placementBindingStatus.Conditions, metav1.Condition{ + Type: placementv1alpha1.PlacementBindingCondTypeAvailable, + Status: metav1.ConditionUnknown, + ObservedGeneration: placementBinding.GetGeneration(), + Reason: placementv1alpha1.PlacementBindingAvailableCondReasonWaitingForAvailabilityCheck, + Message: "Waiting for the resources to be checked for availability in the target cluster", + }) + + // Count the number of manifests in all created/updated work objects. + total := 0 + for idx := range worksToCreateOrUpdate { + total += len(worksToCreateOrUpdate[idx].Spec.Manifests) + } + placementBindingStatus.SelectedResources = ptr.To(int32(total)) + + // Clear the other counters and failed resources as their previous values no longer apply. + placementBindingStatus.SynchronizedResources = nil + placementBindingStatus.AvailableResources = nil + placementBindingStatus.FailedResources = nil + + if err := r.hubClient.Status().Update(ctx, placementBinding); err != nil { + return errors.NewAPIServerError(err, "failed to update placement binding status", true) + } + klog.V(2).InfoS("Reported placement binding processing progress", + "placementBinding", klog.KObj(placementBinding), "selectedResources", total) + return nil +} + +func refreshPlacementBindingSyncCond(placementBinding placementv1alpha1.PlacementBindingAccessor, works []placementv1alpha1.Work) { + // The binding is synchronized only if every work has been applied and its applied condition is up-to-date. + synchronized := true + for idx := range works { + work := &works[idx] + appliedCond := meta.FindStatusCondition(work.Status.Conditions, placementv1alpha1.WorkCondTypeApplied) + if !condition.IsConditionStatusTrue(appliedCond, work.GetGeneration()) { + synchronized = false + break + } + } + + var syncCond metav1.Condition + if synchronized { + syncCond = metav1.Condition{ + Type: placementv1alpha1.PlacementBindingCondTypeSynchronized, + Status: metav1.ConditionTrue, + ObservedGeneration: placementBinding.GetGeneration(), + Reason: placementv1alpha1.PlacementBindingSynchronizedCondReasonAllResourcesSynchronized, + Message: "All resources have been synchronized to the target cluster", + } + } else { + syncCond = metav1.Condition{ + Type: placementv1alpha1.PlacementBindingCondTypeSynchronized, + Status: metav1.ConditionFalse, + ObservedGeneration: placementBinding.GetGeneration(), + Reason: placementv1alpha1.PlacementBindingSynchronizedCondReasonFailedToSynchronizeSomeResources, + Message: "Some resources might be out of sync in the target cluster", + } + } + meta.SetStatusCondition(&placementBinding.GetStatus().Conditions, syncCond) +} + +func refreshPlacementBindingAvailableCond(placementBinding placementv1alpha1.PlacementBindingAccessor, works []placementv1alpha1.Work) { + // The binding is available only if every work is available and its available condition is up-to-date. + available := true + for idx := range works { + work := &works[idx] + availableCond := meta.FindStatusCondition(work.Status.Conditions, placementv1alpha1.WorkCondTypeAvailable) + if !condition.IsConditionStatusTrue(availableCond, work.GetGeneration()) { + available = false + break + } + } + + var availableCond metav1.Condition + if available { + availableCond = metav1.Condition{ + Type: placementv1alpha1.PlacementBindingCondTypeAvailable, + Status: metav1.ConditionTrue, + ObservedGeneration: placementBinding.GetGeneration(), + Reason: placementv1alpha1.PlacementBindingAvailableCondReasonAllResourcesAvailable, + Message: "All resources are available in the target cluster", + } + } else { + availableCond = metav1.Condition{ + Type: placementv1alpha1.PlacementBindingCondTypeAvailable, + Status: metav1.ConditionFalse, + ObservedGeneration: placementBinding.GetGeneration(), + Reason: placementv1alpha1.PlacementBindingAvailableCondReasonSomeResourcesUnavailable, + Message: "Some resources might be unavailable in the target cluster", + } + } + meta.SetStatusCondition(&placementBinding.GetStatus().Conditions, availableCond) +} + +func countResourcesInWorksByProcessingResults(works []placementv1alpha1.Work) ( + total, synced, available int, + failed []placementv1alpha1.FailedResource, +) { + for i := range works { + work := &works[i] + total += len(work.Spec.Manifests) + for j := range work.Status.Manifests { + manifest := &work.Status.Manifests[j] + + appliedCond := meta.FindStatusCondition(manifest.Conditions, placementv1alpha1.ManifestCondTypeApplied) + // Note that the checks below do not take into account the condition's observed generation; this is + // because for manifest conditions KubeFleet uses the generation of the actual manifest object + // being applied, not the generation of the work object. + switch { + case appliedCond == nil: + // The Applied condition has not been set yet; the manifest has not been processed. + continue + case appliedCond.Status != metav1.ConditionTrue: + // The manifest has failed to be applied. + failed = append(failed, failedResourceFromManifestStatus(manifest, appliedCond)) + continue + default: + // The manifest has been applied. + synced++ + } + + availableCond := meta.FindStatusCondition(manifest.Conditions, placementv1alpha1.ManifestCondTypeAvailable) + switch { + case availableCond == nil: + // The Available condition has not been set yet; the manifest has not been processed. + continue + case availableCond.Status != metav1.ConditionTrue: + // The manifest is not available. + failed = append(failed, failedResourceFromManifestStatus(manifest, availableCond)) + continue + default: + // The manifest is available. + available++ + } + } + } + return total, synced, available, failed +} + +// failedResourceFromManifestStatus builds a FailedResource from a per-manifest status and the condition that +// is not true (nil if the condition is absent). +func failedResourceFromManifestStatus(manifest *placementv1alpha1.PerManifestStatus, falseCond *metav1.Condition) placementv1alpha1.FailedResource { + failedResource := placementv1alpha1.FailedResource{ + ObjectRef: placementv1alpha1.ObjectReference{ + Namespace: manifest.Identifier.Namespace, + Name: manifest.Identifier.Name, + APIGroup: manifest.Identifier.APIGroup, + APIVersion: manifest.Identifier.APIVersion, + Kind: manifest.Identifier.Kind, + }, + DiffDetails: manifest.DiffDetails, + } + if falseCond != nil { + // Note that per KubeFleet API semantics, the observed generation set in the copied condition is the generation + // of the actual manifest object being applied in the member cluster, not the generation of the work object + // nor the placement binding object. + failedResource.Conditions = []metav1.Condition{*falseCond} + } + return failedResource +} diff --git a/pkg/v1/controllers/workgenerator/uniquename.go b/pkg/v1/controllers/workgenerator/uniquename.go new file mode 100644 index 000000000..20fa8d845 --- /dev/null +++ b/pkg/v1/controllers/workgenerator/uniquename.go @@ -0,0 +1,133 @@ +/* +Copyright 2026 The KubeFleet 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 workgenerator + +import ( + "crypto/sha256" + "fmt" + "strings" + + placementv1alpha1 "github.com/kubefleet-dev/kubefleet/apis/kubefleet.dev/placement/v1alpha1" +) + +const ( + nameLenLimit = 251 + hashSegLen = 12 +) + +const ( + // The name format for work objects when they are derived from placement resource snapshots. + // Typically, these objects are named using the format: + // + // `[PLACEMENT-POLICY-NAMESPACED-NAME]-work-[HASH]`, if the work is derived from the primary + // placement resource snapshot, or + // `[PLACEMENT-POLICY-NAMESPACED-NAME]-work-[DERIVED-FROM-SOURCE-LABEL]-[HASH]`, if the work is + // derived from other sources (e.g., a secondary placement resource snapshot). + // + // where + // + // * `[PLACEMENT-POLICY-NAMESPACED-NAME]` is the namespace and name of the placement policy that owns the + // placement resource snapshot (and indirectly owns the work objects via placement binding), in the + // format `[NAMESPACE]-[NAME]` (if the placement policy is cluster-scoped, the namespaced name segment is + // simply the placement policy name); + // * `[DERIVED-FROM-SOURCE-LABEL]` is a label that helps identify the source where the work object is derived + // from; it is not guaranteed to be unique and is added for informational purposes only. + // * `[HASH]` is the first few characters of the hash of the value + // `[PLACEMENT-POLICY-NAMESPACE]/[PLACEMENT-POLICY-NAME]-work` or + // `[PLACEMENT-POLICY-NAMESPACE]/[PLACEMENT-POLICY-NAME]-work-[DERIVED-FROM-SOURCE-TYPE]/[DERIVED-FROM-SOURCE-ID]` + // respectively, where `[DERIVED-FROM-SOURCE-TYPE]` is the type of the source where the work object is + // derived from (e.g., `placement-resource-snapshot` for placement resource snapshots), and + // `[DERIVED-FROM-SOURCE-ID]` is an identifier of the source object. Together the segment uniquely identifies + // the source where the work object is derived from (among all the work objects that are created/updated + // for the placement binding). + // + // The slash is used here instead of a dash to avoid collisions between different namespace/name combinations, + // e.g., to make sure that a placement policy named `red` in namespace `team-a` and a placement policy named + // `a-red` in namespace `team` do not produce the same hash. + // + // If the name becomes too long (> 251 characters), KubeFleet will truncate the placement policy namespaced name + // segment and the derived from source marker segment as appropriate. + workDerivedFromPrimarySnapshotSourceNameFmt = "%s-work-%s" + workDerivedFromOtherSourcesNameFmt = "%s-work-%s-%s" +) + +// uniqueNameForWorkDerivedFromPlacementResourceSnapshot generates a unique name for a work object derived from a +// placement resource snapshot, given the owner placement binding and the snapshot sub-index (0 = primary). +func uniqueNameForWorkDerivedFromPlacementResourceSnapshot( + placementBinding placementv1alpha1.PlacementBindingAccessor, + isFromPrimarySnapshot bool, + derivedFromSrcFormatter derivedFromSourceFormatter, +) (string, error) { + namespace := placementBinding.GetNamespace() + policyName := placementBinding.GetSpec().PlacementPolicyName + + // The namespaced name of the owner placement policy, in the format `[NAMESPACE]-[NAME]`; for cluster-scoped + // placement policies, it is simply the placement policy name. + namespacedName := policyName + if namespace != "" { + namespacedName = fmt.Sprintf("%s-%s", namespace, policyName) + } + + // The hash is computed over the namespace and name (separated by a slash) plus the derived from source marker, + // so that different namespace/name combinations never collide, and so that a hash suffix is always present. + hashInput := fmt.Sprintf("%s/%s-work", namespace, policyName) + if !isFromPrimarySnapshot { + hashInput = fmt.Sprintf("%s/%s-work-%s/%s", namespace, policyName, derivedFromSrcFormatter.SourceType(), derivedFromSrcFormatter.SourceID()) + } + hash := fmt.Sprintf("%x", sha256.Sum256([]byte(hashInput)))[:hashSegLen] + + // Remove all dots from the namespaced name segment so that truncation cannot leave a trailing dot, + // which would produce an invalid DNS subdomain label. + namespacedName = strings.ReplaceAll(namespacedName, ".", "") + + if isFromPrimarySnapshot { + // The work is derived from the primary placement resource snapshot; the name omits the source marker segment. + name := fmt.Sprintf(workDerivedFromPrimarySnapshotSourceNameFmt, namespacedName, hash) + if len(name) <= nameLenLimit { + return name, nil + } + + // The name is too long; truncate the namespaced name segment. The hash suffix always disambiguates. + reservedLen := len(fmt.Sprintf(workDerivedFromPrimarySnapshotSourceNameFmt, "", hash)) + availableLen := nameLenLimit - reservedLen + if len(namespacedName) > availableLen { + namespacedName = namespacedName[:availableLen] + } + return fmt.Sprintf(workDerivedFromPrimarySnapshotSourceNameFmt, namespacedName, hash), nil + } + + // The work is derived from another source (e.g., a secondary placement resource snapshot); the name carries + // the source marker segment. + derivedFromSrcLabel := derivedFromSrcFormatter.StrictDNSLabel() + name := fmt.Sprintf(workDerivedFromOtherSourcesNameFmt, namespacedName, derivedFromSrcLabel, hash) + if len(name) <= nameLenLimit { + return name, nil + } + + // The name is too long; truncate the namespaced name and source marker segments, splitting the available + // space evenly between them. The hash suffix always disambiguates. + reservedLen := len(fmt.Sprintf(workDerivedFromOtherSourcesNameFmt, "", "", hash)) + availableLen := nameLenLimit - reservedLen + availablePerSeg := availableLen / 2 + if len(namespacedName) > availablePerSeg { + namespacedName = namespacedName[:availablePerSeg] + } + if len(derivedFromSrcLabel) > availablePerSeg { + derivedFromSrcLabel = derivedFromSrcLabel[:availablePerSeg] + } + return fmt.Sprintf(workDerivedFromOtherSourcesNameFmt, namespacedName, derivedFromSrcLabel, hash), nil +} diff --git a/pkg/v1/controllers/workgenerator/works.go b/pkg/v1/controllers/workgenerator/works.go new file mode 100644 index 000000000..88725d62c --- /dev/null +++ b/pkg/v1/controllers/workgenerator/works.go @@ -0,0 +1,389 @@ +/* +Copyright 2026 The KubeFleet 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 workgenerator + +import ( + "context" + "fmt" + "strconv" + "sync/atomic" + + "k8s.io/apimachinery/pkg/api/equality" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/types" + "k8s.io/apimachinery/pkg/util/sets" + "k8s.io/klog/v2" + "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" + + placementv1alpha1 "github.com/kubefleet-dev/kubefleet/apis/kubefleet.dev/placement/v1alpha1" + "github.com/kubefleet-dev/kubefleet/pkg/utils" + "github.com/kubefleet-dev/kubefleet/pkg/utils/errors" + "github.com/kubefleet-dev/kubefleet/pkg/utils/parallelizer" +) + +var ( + workGVK = schema.GroupVersionKind{ + Group: placementv1alpha1.GroupVersion.Group, + Version: placementv1alpha1.GroupVersion.Version, + Kind: "Work", + } +) + +// areWorksUpToDate checks if the work objects for a placement binding are up-to-date, i.e., if all the work objects +// needed given the current placement binding spec have been created/updated. If so, the work generation can skip +// the work object create/update ops and skip to refreshing the placement binding status. +func areWorksUpToDate(placementBinding placementv1alpha1.PlacementBindingAccessor, works []placementv1alpha1.Work) (bool, error) { + // Check if the last processed placement resource snapshot name in the placement binding status matches the + // primary placement resource snapshot name in the placement binding spec. If so, it ensures that all the needed + // work objects have been created/updated for the placement binding. + lastProcessedSnapshotName := "" + if placementBinding.GetStatus().LastProcessedResourceSnapshotName != nil { + lastProcessedSnapshotName = *placementBinding.GetStatus().LastProcessedResourceSnapshotName + } + primarySnapshotName := placementBinding.GetSpec().ResourceSnapshotName + + if lastProcessedSnapshotName != primarySnapshotName { + return false, nil + } + + // Do some sanity checks, just to make sure that the cache is up to date. + if len(works) == 0 { + // No work objects exist for the placement binding. The cache might not have caught up yet. + return false, errors.NewTransientError(nil, "no work objects are found; the cache might be stale") + } + + // Check if all the work objects have been linked to the expected placement resource snapshot (in the spec), + // and verify that the linked work count recorded on the primary work matches the number of listed works. + linkedWorkCount := -1 + for idx := range works { + work := &works[idx] + annotations := work.GetAnnotations() + if linked := annotations[placementv1alpha1.WorkLinkedToPrimaryPlacementResourceSnapshotAnnotationKey]; linked != primarySnapshotName { + // The work object is linked to a different placement resource snapshot than the one in the placement binding spec. + // This might happen if the cache is stale. + return false, errors.NewTransientError(nil, "found a work object that is not linked to the expected primary placement resource snapshot", + "work", klog.KObj(work), "linkedPlacementResourceSnapshotName", linked, "expectedPlacementResourceSnapshotName", primarySnapshotName) + } + + linkedWorkCountStr, found := annotations[placementv1alpha1.LinkedWorkCountAnnotationKey] + if !found { + continue + } + if linkedWorkCount != -1 { + // At any time there should be exactly one primary placement resource snapshot that has the + // linked work count annotation. + return false, errors.NewUnexpectedError(nil, "multiple primary placement resource snapshots have the linked work count annotation", + "work", klog.KObj(work), "linkedWorkCount", linkedWorkCountStr) + } + var err error + linkedWorkCount, err = strconv.Atoi(linkedWorkCountStr) + if err != nil || linkedWorkCount < 1 { + return false, errors.NewUnexpectedError(err, "invalid linked work count annotation on work object", + "work", klog.KObj(work), "linkedWorkCount", linkedWorkCountStr) + } + } + if linkedWorkCount != len(works) { + return false, errors.NewTransientError(nil, "the number of work objects is not as expected", + "expectedWorkCount", linkedWorkCount, "actualWorkCount", len(works)) + } + + // Check if the sync strategy of the placement binding still matches that on the work objects. + syncStrategy := placementBinding.GetSpec().SyncStrategy + for idx := range works { + work := &works[idx] + if !equality.Semantic.DeepEqual(work.Spec.SyncStrategy, syncStrategy) { + return false, nil + } + } + + return true, nil +} + +func (r *Reconciler) refreshWorks(ctx context.Context, + placementBinding placementv1alpha1.PlacementBindingAccessor, + sortedPlacementResourceSnapshots []placementv1alpha1.PlacementResourceSnapshotAccessor, + works []placementv1alpha1.Work, +) ([]*placementv1alpha1.Work, bool, error) { + writtenToStorage := false + worksToDelete := []*placementv1alpha1.Work{} + + // Build an index of work objects by their names. + existingWorksByName := make(map[string]*placementv1alpha1.Work, len(works)) + for idx := range works { + work := &works[idx] + existingWorksByName[work.GetName()] = work + } + + seenWorkNames := sets.Set[string]{} + // First, build a work object for the primary placement resource snapshot. This is considered to be the + // primary work object for the placement binding. + // + // This work object serves as the owner of all other work objects created for this placement binding. KubeFleet + // leverages this setup to ensure that if a placement binding is deleted, all the work objects created for it + // will be cleaned up automatically by K8s' built-in GC process. + // + // We cannot set the placement binding itself as the owner of the work objects as they might reside in + // different namespaces, and cross-namespace ownership is not allowed in K8s. The list-then-delete loop has + // limitations as well, as stale cache might leave some work objects behind. + primaryPlacementResourceSnapshot := sortedPlacementResourceSnapshots[0] + primaryWorkToCreateOrUpdate, err := buildWorkObjectFor(primaryPlacementResourceSnapshot, placementBinding, primaryPlacementResourceSnapshot.GetName()) + if err != nil { + return nil, false, errors.Wraps(err, "failed to build work object for primary placement resource snapshot", + "primaryPlacementResourceSnapshot", klog.KObj(primaryPlacementResourceSnapshot)) + } + seenWorkNames.Insert(primaryWorkToCreateOrUpdate.GetName()) + + // Then build work objects for any secondary placement resource snapshots. These work objects are considered to + // be secondary work objects for the placement binding. + var additionalWorksToCreateOrUpdate []*placementv1alpha1.Work + for idx := 1; idx < len(sortedPlacementResourceSnapshots); idx++ { + snapshot := sortedPlacementResourceSnapshots[idx] + work, err := buildWorkObjectFor(snapshot, placementBinding, primaryPlacementResourceSnapshot.GetName()) + if err != nil { + return nil, false, errors.Wraps(err, "failed to build work object for placement resource snapshot", + "placementResourceSnapshot", klog.KObj(snapshot)) + } + if seenWorkNames.Has(work.GetName()) { + return nil, false, errors.NewUnexpectedError(nil, "duplicate work object built for placement resource snapshot", + "work", klog.KObj(work), "placementResourceSnapshot", klog.KObj(snapshot)) + } + additionalWorksToCreateOrUpdate = append(additionalWorksToCreateOrUpdate, work) + seenWorkNames.Insert(work.GetName()) + } + + // Add the linked work object count annotation on the primary work object. The count is the total number of + // work objects created for the placement binding, including the primary work object itself. + primaryWorkToCreateOrUpdate.GetAnnotations()[placementv1alpha1.LinkedWorkCountAnnotationKey] = fmt.Sprintf("%d", len(additionalWorksToCreateOrUpdate)+1) + + // Check for dangling work objects (those that are no longer linked with any source) and add them to the + // deletion list. + for _, work := range existingWorksByName { + if seenWorkNames.Has(work.GetName()) { + continue + } + klog.V(2).InfoS("A work object is no longer needed; mark it for deletion", "work", klog.KObj(work)) + worksToDelete = append(worksToDelete, work) + } + + // Issue the delete ops in parallel. The control loop deletes the dangling work objects first to avoid + // potential conflicts (e.g., creating the same object twice). This is a best-effort attempt as we cannot + // create/update/delete work objects in a transactional manner. + if err := r.deleteWorkObjects(ctx, worksToDelete, placementBinding); err != nil { + return nil, false, errors.Wraps(err, "failed to delete dangling work objects") + } + + // Create the primary work object first. This is needed as the controller needs its object UID to set + // owner references on the secondary work objects. + createdOrUpdatedWorks, primaryWorkObjWrittenToStorage, err := r.createOrUpdateWorkObjects(ctx, + []*placementv1alpha1.Work{primaryWorkToCreateOrUpdate}, placementBinding) + if err != nil { + return nil, false, errors.Wraps(err, "failed to create or update work object for primary placement resource snapshot", + "primaryPlacementResourceSnapshot", klog.KObj(primaryPlacementResourceSnapshot)) + } + ownerWorkObjRef := metav1.NewControllerRef(createdOrUpdatedWorks[0], workGVK) + writtenToStorage = primaryWorkObjWrittenToStorage + + // Set the owner reference on all secondary work objects. + for idx := range additionalWorksToCreateOrUpdate { + work := additionalWorksToCreateOrUpdate[idx] + work.SetOwnerReferences([]metav1.OwnerReference{*ownerWorkObjRef}) + } + + // Issue the create or update ops for the secondary work objects in parallel. + additionalCreatedOrUpdatedWorks, additionalCreatedOrUpdated, err := r.createOrUpdateWorkObjects(ctx, additionalWorksToCreateOrUpdate, placementBinding) + if err != nil { + return nil, false, errors.Wraps(err, "failed to create or update additional work objects for secondary placement resource snapshots") + } + createdOrUpdatedWorks = append(createdOrUpdatedWorks, additionalCreatedOrUpdatedWorks...) + if !writtenToStorage { + writtenToStorage = additionalCreatedOrUpdated + } + + return createdOrUpdatedWorks, writtenToStorage, nil +} + +func buildWorkObjectFor( + placementResourceSnapshot placementv1alpha1.PlacementResourceSnapshotAccessor, + placementBinding placementv1alpha1.PlacementBindingAccessor, + primaryPlacementResourceSnapshotName string, +) (*placementv1alpha1.Work, error) { + snapshotSubIdx := placementResourceSnapshot.GetLabels()[placementv1alpha1.PlacementResourceSnapshotSubIndexLabelKey] + if len(snapshotSubIdx) == 0 { + return nil, errors.NewUnexpectedError(nil, "no sub-index label found on the placement resource snapshot") + } + derivedFromSnapshotSrcFormatter := &placementResourceSnapshotDerivedFromSourceFormatter{ + snapshotNamespacedName: types.NamespacedName{ + Namespace: placementResourceSnapshot.GetNamespace(), + Name: placementResourceSnapshot.GetName(), + }, + snapshotSubIdx: snapshotSubIdx, + } + placementBindingSpec := placementBinding.GetSpec() + placementResourceSnapshotSpec := placementResourceSnapshot.GetSpec() + + workName, err := uniqueNameForWorkDerivedFromPlacementResourceSnapshot(placementBinding, snapshotSubIdx == "0", derivedFromSnapshotSrcFormatter) + if err != nil { + return nil, errors.Wraps(err, "failed to generate unique name for the work object", "snapshotSubIdx", snapshotSubIdx) + } + + work := &placementv1alpha1.Work{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: fmt.Sprintf(utils.NamespaceNameFormat, placementBinding.GetSpec().ClusterName), + Name: workName, + }, + } + updateWorkObjectMetadataAndSpec( + work, + placementBinding.GetNamespace(), + placementBindingSpec.PlacementPolicyName, + placementBinding.GetName(), + primaryPlacementResourceSnapshotName, + derivedFromSnapshotSrcFormatter, + placementResourceSnapshotSpec.Resources, + placementBindingSpec.SyncStrategy.DeepCopy(), + ) + return work, nil +} + +func updateWorkObjectMetadataAndSpec( + work *placementv1alpha1.Work, + ownerNamespace, ownerPlacementPolicy, ownerPlacementBinding string, + primaryPlacementResourceSnapshotName string, + derivedFromSrcFormatter derivedFromSourceFormatter, + resources []placementv1alpha1.SnapshottedResource, + syncStrategy *placementv1alpha1.SyncStrategy, +) { + // Set annotations on the work object. + annotations := work.GetAnnotations() + if annotations == nil { + annotations = make(map[string]string) + } + // Set the linked to primary placement resource snapshot annotation on the work object. + annotations[placementv1alpha1.WorkLinkedToPrimaryPlacementResourceSnapshotAnnotationKey] = primaryPlacementResourceSnapshotName + + // Set the derived from source annotation on the work object. + // + // For work objects derived from placement resource snapshots, the annotation is set with the value + // `placement-resource-snapshot/[SUB-INDEX]`, where `[SUB-INDEX]` is the sub-index of the placement + // resource snapshot that the work object is derived from. + // + // Sub-indices are used here instead of indices to avoid any fluctuations caused by the progression + // of placement resource snapshots over rollouts. + annotations[placementv1alpha1.WorkDerivedFromSourceAnnotationKey] = fmt.Sprintf("%s/%s", + derivedFromSrcFormatter.SourceType(), derivedFromSrcFormatter.SourceID()) + work.SetAnnotations(annotations) + + // Set the owner labels on the work object. + labels := work.GetLabels() + if labels == nil { + labels = make(map[string]string) + } + labels[placementv1alpha1.WorkOwnerNamespaceLabelKey] = ownerNamespace + labels[placementv1alpha1.WorkOwnedByPlacementPolicyLabelKey] = ownerPlacementPolicy + labels[placementv1alpha1.WorkOwnedByPlacementBindingLabelKey] = ownerPlacementBinding + work.SetLabels(labels) + + // Set the snapshotted resources on the work object. + manifests := make([]placementv1alpha1.Manifest, len(resources)) + for i := range resources { + manifests[i] = placementv1alpha1.Manifest{RawExtension: resources[i].Manifest} + } + work.Spec.Manifests = manifests + + // Set the sync strategy on the work object. + work.Spec.SyncStrategy = syncStrategy +} + +func (r *Reconciler) createOrUpdateWorkObjects( + ctx context.Context, + worksToCreateOrUpdate []*placementv1alpha1.Work, + placementBinding placementv1alpha1.PlacementBindingAccessor, +) ([]*placementv1alpha1.Work, bool, error) { + childCtx, childCancel := context.WithCancel(ctx) + defer childCancel() + + createdOrUpdatedWorks := make([]*placementv1alpha1.Work, len(worksToCreateOrUpdate)) + errFlag := parallelizer.NewErrorFlag() + createdOrUpdated := atomic.Bool{} + r.parallelizer.ParallelizeUntil(childCtx, len(worksToCreateOrUpdate), func(idx int) { + work := worksToCreateOrUpdate[idx] + + createdOrUpdatedWork := &placementv1alpha1.Work{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: work.GetNamespace(), + Name: work.GetName(), + }, + } + resOp, err := controllerutil.CreateOrUpdate(childCtx, r.hubClient, createdOrUpdatedWork, func() error { + // Work objects are considered to be fully internal KubeFleet resources; for this reason + // here the control loop chooses to overwrite the spec, labels, annotations, and owner references of + // the work object with the latest values instead of attempting to do a merge. + createdOrUpdatedWork.Spec = work.Spec + createdOrUpdatedWork.SetLabels(work.GetLabels()) + createdOrUpdatedWork.SetAnnotations(work.GetAnnotations()) + createdOrUpdatedWork.SetOwnerReferences(work.GetOwnerReferences()) + return nil + }) + if err != nil { + wrappedErr := errors.Wraps(err, "failed to create or update work object", + "work", klog.KObj(work), "resOp", resOp) + errFlag.Raise(wrappedErr) + childCancel() + return + } + + createdOrUpdatedWorks[idx] = createdOrUpdatedWork + if resOp != controllerutil.OperationResultNone { + // The work object has been created or updated. + createdOrUpdated.CompareAndSwap(false, true) + } + klog.V(2).InfoS("Successfully created or updated work object", + "work", klog.KObj(createdOrUpdatedWork), "resOp", resOp, + "placementBinding", klog.KObj(placementBinding)) + }, "createOrUpdateWorkObjects") + if err := errFlag.Lower(); err != nil { + return nil, false, err + } + return createdOrUpdatedWorks, createdOrUpdated.Load(), nil +} + +func (r *Reconciler) deleteWorkObjects( + ctx context.Context, + worksToDelete []*placementv1alpha1.Work, + placementBinding placementv1alpha1.PlacementBindingAccessor, +) error { + childCtx, childCancel := context.WithCancel(ctx) + defer childCancel() + + errFlag := parallelizer.NewErrorFlag() + r.parallelizer.ParallelizeUntil(childCtx, len(worksToDelete), func(idx int) { + work := worksToDelete[idx] + + if err := r.hubClient.Delete(childCtx, work); err != nil && !apierrors.IsNotFound(err) { + wrappedErr := errors.Wraps(err, "failed to delete work object", "work", klog.KObj(work)) + errFlag.Raise(wrappedErr) + childCancel() + return + } + klog.V(2).InfoS("Successfully deleted work object", + "work", klog.KObj(work), + "placementBinding", klog.KObj(placementBinding)) + }, "deleteWorkObjects") + return errFlag.Lower() +}