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/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/pkg/utils/informer/informermanager.go b/pkg/utils/informer/informermanager.go index 07aed02fb..b7d151910 100644 --- a/pkg/utils/informer/informermanager.go +++ b/pkg/utils/informer/informermanager.go @@ -30,6 +30,14 @@ import ( ctrlcache "sigs.k8s.io/controller-runtime/pkg/cache" ) +// Note (chenyu1): many methods in this utility, such as IsInformerSynced and Lister, will implicitly create an informer +// for the queried resource if one does not exist already. This might have side effects as such informers +// will not start until the manager's Start() method is called, provided that such resources have support +// for LIST/WATCH ops. Normally this is fine as the resource watcher is configured to periodically register +// all applicable resources in the informer manager, but the gaps between the synchronization might lead to +// unexpected behaviors (hopefully temporary). For newer code that needs to integrate with the informer manager, +// consider calling IsInformerSet first to check if an informer has been set up, before calling other methods. + // InformerManager manages dynamic shared informer for all resources, include Kubernetes resource and // custom resources defined by CustomResourceDefinition. type Manager interface { @@ -42,6 +50,9 @@ type Manager interface { // IsInformerSynced checks if the resource's informer is synced. IsInformerSynced(resource schema.GroupVersionResource) bool + // IsInformerSet returns if an informer has been set up for the given resource. + IsInformerSet(gvk schema.GroupVersionKind) bool + // Start will run all informers, the informers will keep running until the channel closed. // It is intended to be called after create new informer(s), and it's safe to call multi times. Start() @@ -153,6 +164,14 @@ func (s *informerManagerImpl) IsInformerSynced(resource schema.GroupVersionResou return s.informerFactory.ForResource(resource).Informer().HasSynced() } +func (s *informerManagerImpl) IsInformerSet(gvk schema.GroupVersionKind) bool { + s.resourcesLock.RLock() + defer s.resourcesLock.RUnlock() + + _, ok := s.apiResources[gvk] + return ok +} + func (s *informerManagerImpl) Lister(resource schema.GroupVersionResource) cache.GenericLister { return s.informerFactory.ForResource(resource).Lister() } diff --git a/pkg/v1/managers/placementresourcesnapshot/manager.go b/pkg/v1/managers/placementresourcesnapshot/manager.go new file mode 100644 index 000000000..da78ad155 --- /dev/null +++ b/pkg/v1/managers/placementresourcesnapshot/manager.go @@ -0,0 +1,236 @@ +/* +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 placementresourcesnapshot + +import ( + "context" + "fmt" + "hash/fnv" + "sync" + + "k8s.io/apimachinery/pkg/api/meta" + "k8s.io/client-go/dynamic" + "k8s.io/klog/v2" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + + placementv1alpha1 "github.com/kubefleet-dev/kubefleet/apis/kubefleet.dev/placement/v1alpha1" + errors "github.com/kubefleet-dev/kubefleet/pkg/utils/errors" + "github.com/kubefleet-dev/kubefleet/pkg/utils/informer" +) + +const ( + managerName = "placementresourcesnapshot" +) + +const ( + // The two custom fields that are used to index placement resource snapshots in the cache. + // + // Important: the placement resource snapshot manager runs under the assumption that proper custom fields + // have been added and indexed in the cache when running. Failure to complete such prior setup **before + // the manager starts** will result in unexpected behaviors. Make sure that the manager uses the client + // provided by the hub controller manager, and `SetupWithManager` is called before the manager starts. + + // ownedByAndSubIndexedCustomFieldName is the name of the custom field that indexes placement resource + // snapshots by their owner placement policies and their sub-indices. + // + // This is added to help the manager retrieve all primary placement resource snapshots (i.e., those + // with a sub-index of 0) associated with a placement policy. + ownedByAndSubIndexedCustomFieldName = "ownedByWithSubIndex" + // ownedByAndIndexedCustomFieldName is the name of the custom field that indexes placement resource snapshots by + // their owner placement policies and their indices. + // + // This is added to help the manager retrieve all placement resource snapshots of a specific index + // associated with a placement policy. + ownedByAndIndexedCustomFieldName = "ownedByWithIndex" + + // The format of the custom field values for the two custom fields above. + // + // Note that slashes are used to avoid unexpected collisions. + ownedByAndSubIndexedCustomFieldFmt = "%s/%s" + ownedByAndIndexedCustomFieldFmt = "%s/%s" +) + +const ( + // The format of the key used to find the mutex for a placement policy in the mutex array. + // + // Note that slashes are used to avoid unexpected collisions. + placementPolicyKeyFmt = "%s/%s" + + minSlotCnt = 16 +) + +type Manager struct { + hubClient client.Client + hubDynamicClient dynamic.Interface + hubDynamicInformerManager informer.Manager + + restMapper meta.RESTMapper + + mus []sync.Mutex + muSlotCnt int32 +} + +// New returns a new Manager. +func New(mgr ctrl.Manager, + hubDynamicClient dynamic.Interface, + hubDynamicInformerManager informer.Manager, + restMapper meta.RESTMapper, + muSlotCnt int32, +) (*Manager, error) { + if muSlotCnt < minSlotCnt { + return nil, errors.NewUserError(nil, "mu slot size must be greater than or equal to the minimum limit", + "manager", managerName, "limit", minSlotCnt, "actual", muSlotCnt) + } + + return &Manager{ + hubClient: mgr.GetClient(), + hubDynamicClient: hubDynamicClient, + hubDynamicInformerManager: hubDynamicInformerManager, + restMapper: restMapper, + mus: make([]sync.Mutex, muSlotCnt), + muSlotCnt: muSlotCnt, + }, nil +} + +// SetupWithManager sets up the indices the placement resource snapshot manager needs to run properly. +// It must be called before the manager starts. +func (m *Manager) SetupWithManager(ctx context.Context, mgr ctrl.Manager) error { + fieldIdxer := mgr.GetFieldIndexer() + + // Index placement resource snapshots by their owner placement policies and their sub-indices. + if err := fieldIdxer.IndexField(ctx, &placementv1alpha1.PlacementResourceSnapshot{}, ownedByAndSubIndexedCustomFieldName, func(rawObj client.Object) []string { + snapshot, ok := rawObj.(*placementv1alpha1.PlacementResourceSnapshot) + if !ok { + wrappedErr := errors.NewUnexpectedError(nil, "failed to convert object to placement resource snapshot", + "object", klog.KObj(rawObj)) + klog.ErrorS(wrappedErr, "failed to index placement resource snapshot by owner and sub-index", errors.Args(wrappedErr)...) + return nil + } + + ownedBy := snapshot.GetLabels()[placementv1alpha1.PlacementResourceSnapshotOwnedByLabelKey] + subIndex := snapshot.GetLabels()[placementv1alpha1.PlacementResourceSnapshotSubIndexLabelKey] + if ownedBy == "" || subIndex == "" { + wrappedErr := errors.NewUnexpectedError(nil, "placement resource snapshot is missing required labels", + "placementResourceSnapshot", klog.KObj(snapshot)) + klog.ErrorS(wrappedErr, "failed to index placement resource snapshot by owner and sub-index", errors.Args(wrappedErr)...) + return nil + } + + v := fmt.Sprintf(ownedByAndSubIndexedCustomFieldFmt, ownedBy, subIndex) + return []string{v} + }); err != nil { + return errors.NewUnexpectedError(err, "failed to index placement resource snapshots by owner and sub-index", "manager", managerName) + } + + // Index placement resource snapshots by their owner placement policies and their indices. + if err := fieldIdxer.IndexField(ctx, &placementv1alpha1.PlacementResourceSnapshot{}, ownedByAndIndexedCustomFieldName, func(rawObj client.Object) []string { + snapshot, ok := rawObj.(*placementv1alpha1.PlacementResourceSnapshot) + if !ok { + wrappedErr := errors.NewUnexpectedError(nil, "failed to convert object to placement resource snapshot", + "object", klog.KObj(rawObj)) + klog.ErrorS(wrappedErr, "failed to index placement resource snapshot by owner and index", errors.Args(wrappedErr)...) + return nil + } + + ownedBy := snapshot.GetLabels()[placementv1alpha1.PlacementResourceSnapshotOwnedByLabelKey] + index := snapshot.GetLabels()[placementv1alpha1.PlacementResourceSnapshotIndexLabelKey] + if ownedBy == "" || index == "" { + wrappedErr := errors.NewUnexpectedError(nil, "placement resource snapshot is missing required labels", + "placementResourceSnapshot", klog.KObj(snapshot)) + klog.ErrorS(wrappedErr, "failed to index placement resource snapshot by owner and index", errors.Args(wrappedErr)...) + return nil + } + + v := fmt.Sprintf(ownedByAndIndexedCustomFieldFmt, ownedBy, index) + return []string{v} + }); err != nil { + return errors.NewUnexpectedError(err, "failed to index placement resource snapshots by owner and index", "manager", managerName) + } + + // Index cluster placement resource snapshots by their owner placement policies and their sub-indices. + if err := fieldIdxer.IndexField(ctx, &placementv1alpha1.ClusterPlacementResourceSnapshot{}, ownedByAndSubIndexedCustomFieldName, func(rawObj client.Object) []string { + snapshot, ok := rawObj.(*placementv1alpha1.ClusterPlacementResourceSnapshot) + if !ok { + wrappedErr := errors.NewUnexpectedError(nil, "failed to convert object to cluster placement resource snapshot", + "object", klog.KObj(rawObj)) + klog.ErrorS(wrappedErr, "failed to index cluster placement resource snapshot by owner and sub-index", errors.Args(wrappedErr)...) + return nil + } + + ownedBy := snapshot.GetLabels()[placementv1alpha1.PlacementResourceSnapshotOwnedByLabelKey] + subIndex := snapshot.GetLabels()[placementv1alpha1.PlacementResourceSnapshotSubIndexLabelKey] + if ownedBy == "" || subIndex == "" { + wrappedErr := errors.NewUnexpectedError(nil, "cluster placement resource snapshot is missing required labels", + "clusterPlacementResourceSnapshot", klog.KObj(snapshot)) + klog.ErrorS(wrappedErr, "failed to index cluster placement resource snapshot by owner and sub-index", errors.Args(wrappedErr)...) + return nil + } + + v := fmt.Sprintf(ownedByAndSubIndexedCustomFieldFmt, ownedBy, subIndex) + return []string{v} + }); err != nil { + return errors.NewUnexpectedError(err, "failed to index cluster placement resource snapshots by owner and sub-index", "manager", managerName) + } + + // Index cluster placement resource snapshots by their owner placement policies and their indices. + if err := fieldIdxer.IndexField(ctx, &placementv1alpha1.ClusterPlacementResourceSnapshot{}, ownedByAndIndexedCustomFieldName, func(rawObj client.Object) []string { + snapshot, ok := rawObj.(*placementv1alpha1.ClusterPlacementResourceSnapshot) + if !ok { + wrappedErr := errors.NewUnexpectedError(nil, "failed to convert object to cluster placement resource snapshot", + "object", klog.KObj(rawObj)) + klog.ErrorS(wrappedErr, "failed to index cluster placement resource snapshot by owner and index", errors.Args(wrappedErr)...) + return nil + } + + ownedBy := snapshot.GetLabels()[placementv1alpha1.PlacementResourceSnapshotOwnedByLabelKey] + index := snapshot.GetLabels()[placementv1alpha1.PlacementResourceSnapshotIndexLabelKey] + if ownedBy == "" || index == "" { + wrappedErr := errors.NewUnexpectedError(nil, "cluster placement resource snapshot is missing required labels", + "clusterPlacementResourceSnapshot", klog.KObj(snapshot)) + klog.ErrorS(wrappedErr, "failed to index cluster placement resource snapshot by owner and index", errors.Args(wrappedErr)...) + return nil + } + + v := fmt.Sprintf(ownedByAndIndexedCustomFieldFmt, ownedBy, index) + return []string{v} + }); err != nil { + return errors.NewUnexpectedError(err, "failed to index cluster placement resource snapshots by owner and index", "manager", managerName) + } + + return nil +} + +func (m *Manager) acquireLock(placementPolicy placementv1alpha1.PlacementPolicyAccessor) { + placementPolicyKey := fmt.Sprintf(placementPolicyKeyFmt, placementPolicy.GetNamespace(), placementPolicy.GetName()) + + hasher := fnv.New32a() + hasher.Write([]byte(placementPolicyKey)) + + slot := int(hasher.Sum32() % uint32(m.muSlotCnt)) + m.mus[slot].Lock() +} + +func (m *Manager) releaseLock(placementPolicy placementv1alpha1.PlacementPolicyAccessor) { + placementPolicyKey := fmt.Sprintf(placementPolicyKeyFmt, placementPolicy.GetNamespace(), placementPolicy.GetName()) + + hasher := fnv.New32a() + hasher.Write([]byte(placementPolicyKey)) + + slot := int(hasher.Sum32() % uint32(m.muSlotCnt)) + m.mus[slot].Unlock() +} diff --git a/pkg/v1/managers/placementresourcesnapshot/ops.go b/pkg/v1/managers/placementresourcesnapshot/ops.go new file mode 100644 index 000000000..def4f2f53 --- /dev/null +++ b/pkg/v1/managers/placementresourcesnapshot/ops.go @@ -0,0 +1,556 @@ +/* +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 placementresourcesnapshot + +import ( + "context" + "fmt" + "sort" + "strconv" + + 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/client" + + placementv1alpha1 "github.com/kubefleet-dev/kubefleet/apis/kubefleet.dev/placement/v1alpha1" + errors "github.com/kubefleet-dev/kubefleet/pkg/utils/errors" +) + +const ( + dryRunAnnotationKey = "kubefleet.dev/dry-run" +) + +// SnapshotResourcesIfNoLatestSnapshot checks for the latest placement resource snapshot(s) associated +// with a placement policy. If no snapshot exists at all, it creates a new placement resource snapshot; otherwise, +// it returns the current latest placement resource snapshot(s) and a flag that signals whether the +// latest snapshot is up-to-date. +// +// This method is exposed for controllers such as the placement policy controller, which needs to check +// placement resource snapshots for status reporting and generally do not need to manipulate the snapshots themselves. +func (m *Manager) SnapshotResourcesIfNoLatestSnapshot( + ctx context.Context, + placementPolicy placementv1alpha1.PlacementPolicyAccessor, +) ([]placementv1alpha1.PlacementResourceSnapshotAccessor, bool, error) { + // Do a sanity check. + if placementPolicy == nil { + return nil, false, errors.NewUnexpectedError(nil, "placement policy accessor is nil", "manager", managerName) + } + + // Acquire the mutex for the placement policy. + m.acquireLock(placementPolicy) + defer m.releaseLock(placementPolicy) + + // Retrieve the latest placement resource snapshot(s) associated with the placement policy. + snapshots, err := m.retrieveLatestSnapshot(ctx, placementPolicy) + if err != nil { + return nil, false, errors.Wraps(err, "failed to retrieve the latest placement resource snapshot(s)") + } + + // Retrieve the currently selected resources and their hash based on the placement policy. + currentResources, currentHash, err := m.retrieveAndHashSelectedResources(ctx, placementPolicy) + if err != nil { + return nil, false, errors.Wraps(err, "failed to retrieve and hash the selected resources") + } + + if len(snapshots) > 0 { + // A latest placement resource snapshot exists; check if it is up-to-date. + primarySnapshot := snapshots[0] + + isUpToDate, err := m.isSnapshotUpToDate(ctx, placementPolicy, primarySnapshot, currentHash) + if err != nil { + return nil, false, errors.Wraps(err, "failed to check if the latest placement resource snapshot is up-to-date", + "primaryPlacementResourceSnapshot", klog.KObj(primarySnapshot)) + } + return snapshots, isUpToDate, nil + } + + // No latest placement resource snapshot exists; create one. + createdSnapshots, err := m.createResourceSnapshotAnyway(ctx, placementPolicy, nil, currentResources, currentHash) + if err != nil { + return nil, false, errors.Wraps(err, "failed to create a placement resource snapshot", "manager", managerName) + } + // A freshly created placement resource snapshot is up-to-date. + return createdSnapshots, true, nil +} + +// SnapshotResourcesIfLatestSnapshotIsNotUpToDate checks for the latest placement resource snapshot(s) associated +// with a placement policy. If the latest snapshot is not up-to-date or no snapshots are present, it creates a +// new placement resource snapshot; otherwise, it returns the current latest placement resource snapshot(s). +// +// This method is exposed for rollout purposes, where a controller may need to create a new placement resource +// snapshot per user requests to complete a rollout. +func (m *Manager) SnapshotResourcesIfLatestSnapshotIsNotUpToDate( + ctx context.Context, + placementPolicy placementv1alpha1.PlacementPolicyAccessor, +) ([]placementv1alpha1.PlacementResourceSnapshotAccessor, error) { + // Do a sanity check. + if placementPolicy == nil { + return nil, errors.NewUnexpectedError(nil, "placement policy accessor is nil", "manager", managerName) + } + + // Acquire the mutex for the placement policy. + m.acquireLock(placementPolicy) + defer m.releaseLock(placementPolicy) + + // Retrieve the latest placement resource snapshot(s) associated with the placement policy. + snapshots, err := m.retrieveLatestSnapshot(ctx, placementPolicy) + if err != nil { + return nil, errors.Wraps(err, "failed to retrieve the latest placement resource snapshot(s)") + } + + // Retrieve the currently selected resources and their hash based on the placement policy. + currentResources, currentHash, err := m.retrieveAndHashSelectedResources(ctx, placementPolicy) + if err != nil { + return nil, errors.Wraps(err, "failed to retrieve and hash the selected resources") + } + + if len(snapshots) > 0 { + // A latest placement resource snapshot exists; check if it is up-to-date. + latestPrimarySnapshot := snapshots[0] + isUpToDate, err := m.isSnapshotUpToDate(ctx, placementPolicy, latestPrimarySnapshot, currentHash) + if err != nil { + return nil, errors.Wraps(err, "failed to check if the latest placement resource snapshot is up-to-date", + "primaryPlacementResourceSnapshot", klog.KObj(latestPrimarySnapshot)) + } + // If the latest snapshot is up-to-date, return it. + if isUpToDate { + return snapshots, nil + } + } + + // No latest placement resource snapshot exists, or the latest one is not up-to-date; create a new one. + var latestPrimarySnapshot placementv1alpha1.PlacementResourceSnapshotAccessor + if len(snapshots) > 0 { + latestPrimarySnapshot = snapshots[0] + } + createdSnapshots, err := m.createResourceSnapshotAnyway(ctx, placementPolicy, latestPrimarySnapshot, currentResources, currentHash) + if err != nil { + return nil, errors.Wraps(err, "failed to create a placement resource snapshot", "manager", managerName) + } + return createdSnapshots, nil +} + +// retrieveLatestSnapshot retrieves the latest placement resource snapshot(s) associated with a placement policy. +// +// If there are multiple placement resource snapshots with the same index, they will be returned in the ascending order +// of their sub-indices. +// +// Note that this method assumes that the corresponding mutex for the placement policy has been acquired before +// calling this method. +func (m *Manager) retrieveLatestSnapshot(ctx context.Context, placementPolicy placementv1alpha1.PlacementPolicyAccessor) ( + placementResourceSnapshot []placementv1alpha1.PlacementResourceSnapshotAccessor, err error) { + // Retrieve the primary resource snapshots associated with the placement policy. + var snapshots []placementv1alpha1.PlacementResourceSnapshotAccessor + fieldMatchers := client.MatchingFields{ + ownedByAndSubIndexedCustomFieldName: fmt.Sprintf(ownedByAndSubIndexedCustomFieldFmt, placementPolicy.GetName(), "0"), + } + if placementPolicy.GetNamespace() == "" { + // The placement policy is cluster-scoped; list cluster placement resource snapshots. + placementResourceSnapshotList := &placementv1alpha1.ClusterPlacementResourceSnapshotList{} + if err := m.hubClient.List(ctx, placementResourceSnapshotList, fieldMatchers); err != nil { + return nil, errors.NewAPIServerError(err, "failed to list cluster placement resource snapshots", true) + } + snapshots = make([]placementv1alpha1.PlacementResourceSnapshotAccessor, len(placementResourceSnapshotList.Items)) + for i := range placementResourceSnapshotList.Items { + snapshots[i] = &placementResourceSnapshotList.Items[i] + } + } else { + // The placement policy is namespace-scoped; list placement resource snapshots in the same namespace. + placementResourceSnapshotList := &placementv1alpha1.PlacementResourceSnapshotList{} + if err := m.hubClient.List(ctx, placementResourceSnapshotList, + client.InNamespace(placementPolicy.GetNamespace()), fieldMatchers); err != nil { + return nil, errors.NewAPIServerError(err, "failed to list placement resource snapshots", true) + } + snapshots = make([]placementv1alpha1.PlacementResourceSnapshotAccessor, len(placementResourceSnapshotList.Items)) + for i := range placementResourceSnapshotList.Items { + snapshots[i] = &placementResourceSnapshotList.Items[i] + } + } + + if len(snapshots) == 0 { + // No placement resource snapshot exists for the placement policy. + return nil, nil + } + + // Sort the primary snapshots by their indices. + var sortErrs []error + sort.Slice(snapshots, func(i, j int) bool { + indexIStr := snapshots[i].GetLabels()[placementv1alpha1.PlacementResourceSnapshotIndexLabelKey] + indexJStr := snapshots[j].GetLabels()[placementv1alpha1.PlacementResourceSnapshotIndexLabelKey] + indexI, iErr := strconv.Atoi(indexIStr) + indexJ, jErr := strconv.Atoi(indexJStr) + if iErr != nil { + sortErrs = append(sortErrs, fmt.Errorf("failed to convert index label to integer: %w (placementResourceSnapshot: %v)", + iErr, klog.KObj(snapshots[i]))) + return false + } + if jErr != nil { + sortErrs = append(sortErrs, fmt.Errorf("failed to convert index label to integer: %w (placementResourceSnapshot: %v)", + jErr, klog.KObj(snapshots[j]))) + return false + } + return indexI < indexJ + }) + if len(sortErrs) > 0 { + return nil, errors.NewUnexpectedError(nil, "failed to sort primary placement resource snapshots", "errs", sortErrs) + } + + latestPrimarySnapshot := snapshots[len(snapshots)-1] + // Check if there are snapshots with the same index. + subIndexedSnapshotCntStr := latestPrimarySnapshot.GetLabels()[placementv1alpha1.SubIndexedPlacementResourceSnapshotCountLabelKey] + if subIndexedSnapshotCntStr == "1" { + // The primary placement resource snapshot is the only snapshot with the latest index; return it. + return []placementv1alpha1.PlacementResourceSnapshotAccessor{latestPrimarySnapshot}, nil + } + subIndexedSnapshotCnt, err := strconv.Atoi(subIndexedSnapshotCntStr) + if err != nil { + return nil, errors.NewUnexpectedError(err, "failed to convert sub-indexed placement resource snapshot count label to integer", + "placementResourceSnapshot", klog.KObj(latestPrimarySnapshot)) + } + + if subIndexedSnapshotCnt < 1 { + // Do a sanity check. + return nil, errors.NewUnexpectedError(nil, "sub-indexed placement resource snapshot count label is less than 1", + "placementResourceSnapshot", klog.KObj(latestPrimarySnapshot), "subIndexedSnapshotCount", subIndexedSnapshotCnt) + } + + // There are sub-indexed placement resource snapshots with the same index; retrieve them. + latestIndex := latestPrimarySnapshot.GetLabels()[placementv1alpha1.PlacementResourceSnapshotIndexLabelKey] + fieldMatchers = client.MatchingFields{ + ownedByAndIndexedCustomFieldName: fmt.Sprintf(ownedByAndIndexedCustomFieldFmt, placementPolicy.GetName(), latestIndex), + } + var subIndexedSnapshots []placementv1alpha1.PlacementResourceSnapshotAccessor + if placementPolicy.GetNamespace() == "" { + // The placement policy is cluster-scoped; list cluster placement resource snapshots. + placementResourceSnapshotList := &placementv1alpha1.ClusterPlacementResourceSnapshotList{} + if err := m.hubClient.List(ctx, placementResourceSnapshotList, fieldMatchers); err != nil { + return nil, errors.NewAPIServerError(err, "failed to list cluster placement resource snapshots", true) + } + subIndexedSnapshots = make([]placementv1alpha1.PlacementResourceSnapshotAccessor, len(placementResourceSnapshotList.Items)) + for i := range placementResourceSnapshotList.Items { + subIndexedSnapshots[i] = &placementResourceSnapshotList.Items[i] + } + } else { + // The placement policy is namespace-scoped; list placement resource snapshots in the same namespace. + placementResourceSnapshotList := &placementv1alpha1.PlacementResourceSnapshotList{} + if err := m.hubClient.List(ctx, placementResourceSnapshotList, + client.InNamespace(placementPolicy.GetNamespace()), fieldMatchers); err != nil { + return nil, errors.NewAPIServerError(err, "failed to list placement resource snapshots", true) + } + subIndexedSnapshots = make([]placementv1alpha1.PlacementResourceSnapshotAccessor, len(placementResourceSnapshotList.Items)) + for i := range placementResourceSnapshotList.Items { + subIndexedSnapshots[i] = &placementResourceSnapshotList.Items[i] + } + } + // Sort the sub-indexed snapshots by their sub-indices. + sortErrs = nil + sort.Slice(subIndexedSnapshots, func(i, j int) bool { + subIndexIStr := subIndexedSnapshots[i].GetLabels()[placementv1alpha1.PlacementResourceSnapshotSubIndexLabelKey] + subIndexJStr := subIndexedSnapshots[j].GetLabels()[placementv1alpha1.PlacementResourceSnapshotSubIndexLabelKey] + subIndexI, iErr := strconv.Atoi(subIndexIStr) + subIndexJ, jErr := strconv.Atoi(subIndexJStr) + if iErr != nil { + sortErrs = append(sortErrs, fmt.Errorf("failed to convert sub-index label to integer: %w (placementResourceSnapshot: %v)", + iErr, klog.KObj(subIndexedSnapshots[i]))) + return false + } + if jErr != nil { + sortErrs = append(sortErrs, fmt.Errorf("failed to convert sub-index label to integer: %w (placementResourceSnapshot: %v)", + jErr, klog.KObj(subIndexedSnapshots[j]))) + return false + } + return subIndexI < subIndexJ + }) + if len(sortErrs) > 0 { + return nil, errors.NewUnexpectedError(nil, "failed to sort sub-indexed placement resource snapshots", "errs", sortErrs) + } + + // Verify that there are enough sub-indexed placement resource snapshots as dictated by the count label. + if len(subIndexedSnapshots) < subIndexedSnapshotCnt { + // Normally this would never happen, as the manager creates secondary placement resource snapshots first + // before creating the primary placement resource snapshot with the count label. + return nil, errors.NewUnexpectedError(nil, "there are fewer sub-indexed placement resource snapshots than the count label indicates", + "expectedCount", subIndexedSnapshotCnt, "actualCount", len(subIndexedSnapshots)) + } + + // As there is no way to create multiple placement resource snapshots with the same index in a transactional + // manner, there exists a corner case where the manager, when going through several snapshot creation passes, + // created more placement resource snapshots than the count label indicates. The extra snapshots are orphans from + // resource changes that have been overwritten. + // + // This is not registered as an error, and here the manager returns only the number of placement resource snapshots + // dictated by the count label, which is guaranteed to be consistent. The orphaned snapshots will eventually + // be cleaned up. + if len(subIndexedSnapshots) > subIndexedSnapshotCnt { + // There are more snapshots than expected; log a warning and only return the ones dictated by the count. + klog.Warningf("found more sub-indexed placement resource snapshots (%d) than the count label indicates (%d) for placement policy %v; only returning the first %d", + len(subIndexedSnapshots), subIndexedSnapshotCnt, klog.KObj(placementPolicy), subIndexedSnapshotCnt) + } + + return subIndexedSnapshots[:subIndexedSnapshotCnt], nil +} + +// isSnapshotUpToDate checks if the given placement resource snapshot is up-to-date, i.e., the snapshot is +// consistent with the current state of the resources as selected by the placement policy. +// +// Note that this method assumes that the corresponding mutex for the placement policy has been acquired before +// calling this method. +func (m *Manager) isSnapshotUpToDate( + ctx context.Context, + placementPolicy placementv1alpha1.PlacementPolicyAccessor, + primaryPlacementResourceSnapshot placementv1alpha1.PlacementResourceSnapshotAccessor, + currentHash string, +) (bool, error) { + // Get the contents hash annotation from the given primary placement resource snapshot. + snapshotHash := primaryPlacementResourceSnapshot.GetAnnotations()[placementv1alpha1.PlacementResourceSnapshotContentsHashAnnotationKey] + + if snapshotHash != currentHash { + // The hashes do not match; the placement resource snapshot is not up-to-date. + // + // Note that due to the check being carried out using a cached client, false negatives are possible, i.e., + // a newer snapshot with matching hash might have been created, yet the cache has not been updated yet. + // However, this is considered OK as any attempt to create a new snapshot based on the false negative + // will lead to a failure (`AlreadyExists` error). Eventually the cache will catch up, and consistency + // will be restored. + return false, nil + } + + // The hashes do match. + // + // Note that due to the check being carried out using a cached client, false positives can occur due to the + // situation where the user does an A -> B -> A type of resource change, and in this scenario the false positive + // might lead to side effects, e.g., empty rollouts, inconsistent status reporting. Here KubeFleet does a + // dry-run to verify that the snapshot is indeed up-to-date. + + // Compute the index of the snapshot that would be created next. + currentIdxStr := primaryPlacementResourceSnapshot.GetLabels()[placementv1alpha1.PlacementResourceSnapshotIndexLabelKey] + currentIdx, err := strconv.Atoi(currentIdxStr) + if err != nil { + return false, errors.NewUnexpectedError(err, "failed to convert index label to integer") + } + nextIdx := currentIdx + 1 + + nextName, err := uniqueNameForPrimaryPlacementResourceSnapshot(placementPolicy.GetName(), nextIdx) + if err != nil { + return false, errors.Wraps(err, "failed to generate the unique name for the next placement resource snapshot", + "nextSnapshotIndex", nextIdx) + } + + // Build the patch target for the next-index snapshot without fetching it. + var nextSnapshot client.Object + if placementPolicy.GetNamespace() == "" { + nextSnapshot = &placementv1alpha1.ClusterPlacementResourceSnapshot{ + ObjectMeta: metav1.ObjectMeta{Name: nextName}, + } + } else { + nextSnapshot = &placementv1alpha1.PlacementResourceSnapshot{ + ObjectMeta: metav1.ObjectMeta{Name: nextName, Namespace: placementPolicy.GetNamespace()}, + } + } + + // Send a dry-run JSON merge patch that adds the dry-run annotation to the next-index snapshot. + patchData := fmt.Appendf(nil, `{"metadata":{"annotations":{%q:%q}}}`, dryRunAnnotationKey, "true") + err = m.hubClient.Patch(ctx, nextSnapshot, client.RawPatch(types.MergePatchType, patchData), client.DryRunAll) + switch { + case err == nil: + // The dry-run patch succeeded; a newer snapshot already exists. Report this as an error; the caller + // should requeue and wait until the cache catches up. + return false, errors.NewTransientError(nil, "a newer snapshot already exists (found via dry-run ops); the client cache might be stale", "nextSnapshotName", nextName) + case apierrors.IsNotFound(err): + // The dry-run patch failed with a NotFound error; no newer snapshot exists. + return true, nil + default: + // The dry-run patch failed with an unexpected error; report it. + return false, errors.NewAPIServerError(err, "failed to perform dry-run patch on the next placement resource snapshot", false, + "nextSnapshotName", nextName) + } +} + +// createResourceSnapshotAnyway creates a new placement resource snapshot for the given placement policy. +// +// Snapshot creation spans multiple objects (secondaries then the primary) and is not transactional; it relies on +// the mutex plus the hub controller manager's leader election for serialization. Because the listing/cleanup steps +// read from a cached client, a stale cache can lead to `AlreadyExists` (on create) or `NotFound` (on the up-to-date +// dry-run) errors. These are expected and surfaced to the caller so that it requeues; each retry re-runs the orphan +// cleanup from a clean slate, and the operation converges once the cache catches up. +// +// Note that this method assumes that the corresponding mutex for the placement policy has been acquired before +// calling this method. +func (m *Manager) createResourceSnapshotAnyway( + ctx context.Context, + placementPolicy placementv1alpha1.PlacementPolicyAccessor, + latestPrimaryPlacementResourceSnapshot placementv1alpha1.PlacementResourceSnapshotAccessor, + currentResources []placementv1alpha1.SnapshottedResource, + currentHash string, +) ([]placementv1alpha1.PlacementResourceSnapshotAccessor, error) { + // Compute the index of the snapshot that would be created next. + nextSnapshotIdx := 0 + if latestPrimaryPlacementResourceSnapshot != nil { + lastSeenSnapshotIdxStr := latestPrimaryPlacementResourceSnapshot.GetLabels()[placementv1alpha1.PlacementResourceSnapshotIndexLabelKey] + lastSeenSnapshotIdx, err := strconv.Atoi(lastSeenSnapshotIdxStr) + if err != nil { + return nil, errors.NewUnexpectedError(err, + "failed to convert last seen primary placement resource snapshot index label to integer") + } + nextSnapshotIdx = lastSeenSnapshotIdx + 1 + } + + // Clean up orphaned secondary placement resource snapshots (if any). + // + // Due to the inability to create multiple placement resource snapshots with the same index in a + // transactional manner, it is possible that the manager has already created a few secondary placement + // resource snapshot in a previous pass. In this case, the manager should delete the existing snapshots + // (its content might be outdated, and the object spec is immutable anyway) before creating new ones. + acted, err := m.cleanUpOrphanedSecondarySnapshots(ctx, placementPolicy, nextSnapshotIdx) + if err != nil { + return nil, errors.Wraps(err, "failed to clean up orphaned secondary placement resource snapshots") + } + if acted { + // Ask the caller to requeue when there are orphaned secondary placement resource snapshots to be cleaned up. + // This helps avoid oscillation issues where a not fully deleted snapshot blocks later creation. + return nil, errors.NewTransientError(nil, "cleaned up orphaned secondary placement resource snapshots; requeue before creating new snapshots", "snapshotIndex", nextSnapshotIdx) + } + + // Split the resources into size-controlled groups. Each group corresponds to a placement resource snapshot + // that will be created. + resGroups, err := splitResourcesIntoSizeControlledGroups(currentResources) + if err != nil { + return nil, errors.Wraps(err, "failed to split the selected resources into size-controlled groups") + } + + // createdSnapshots holds the created snapshots for the new index, keyed by their sub-indices. + createdSnapshots := make([]placementv1alpha1.PlacementResourceSnapshotAccessor, len(resGroups)) + + // Note (chenyu1): evaluate if parallelization is needed here. In most cases the number of secondary + // placement resource snapshots is small, so the overhead of parallelization might not be worth it. + if len(resGroups) > 1 { + // Create the secondary placement resource snapshots first. Start with the last resource group and work + // backwards, so that the primary snapshot (which carries the count label) is created last. + for subIdx := len(resGroups) - 1; subIdx >= 1; subIdx-- { + secondaryName, err := uniqueNameForSecondaryPlacementResourceSnapshot(placementPolicy.GetName(), nextSnapshotIdx, subIdx) + if err != nil { + return nil, errors.Wraps(err, "failed to generate the unique name for a secondary placement resource snapshot", + "snapshotIndex", nextSnapshotIdx, "snapshotSubIndex", subIdx) + } + + secondarySnapshot, err := secondaryPlacementResourceSnapshot( + placementPolicy.GetNamespace(), secondaryName, placementPolicy, nextSnapshotIdx, subIdx, resGroups[subIdx], currentHash, m.hubClient.Scheme()) + if err != nil { + return nil, errors.Wraps(err, "failed to build a secondary placement resource snapshot", + "secondaryPlacementResourceSnapshotName", secondaryName, + "snapshotIndex", nextSnapshotIdx, "snapshotSubIndex", subIdx) + } + + if err := m.hubClient.Create(ctx, secondarySnapshot); err != nil { + return nil, errors.NewAPIServerError(err, "failed to create a secondary placement resource snapshot", false, + "secondaryPlacementResourceSnapshot", klog.KObj(secondarySnapshot), + "snapshotIndex", nextSnapshotIdx, "snapshotSubIndex", subIdx) + } + + createdSnapshots[subIdx] = secondarySnapshot + } + } + + // Create the primary placement resource snapshot last, with the count label. + primaryName, err := uniqueNameForPrimaryPlacementResourceSnapshot(placementPolicy.GetName(), nextSnapshotIdx) + if err != nil { + return nil, errors.Wraps(err, "failed to generate the unique name for the primary placement resource snapshot", + "snapshotIndex", nextSnapshotIdx) + } + + primarySnapshot, err := primaryPlacementResourceSnapshot( + placementPolicy.GetNamespace(), primaryName, placementPolicy, nextSnapshotIdx, resGroups[0], currentHash, len(resGroups), m.hubClient.Scheme()) + if err != nil { + return nil, errors.Wraps(err, "failed to build the primary placement resource snapshot", + "primaryPlacementResourceSnapshotName", primaryName, "snapshotIndex", nextSnapshotIdx) + } + + if err := m.hubClient.Create(ctx, primarySnapshot); err != nil { + // Note that if the primary placement resource snapshot already exists, no deletion will be attempted. The + // caller must retry and create the next placement resource snapshot with a new index. + return nil, errors.NewAPIServerError(err, "failed to create the primary placement resource snapshot", false, + "primaryPlacementResourceSnapshot", klog.KObj(primarySnapshot), "snapshotIndex", nextSnapshotIdx) + } + + createdSnapshots[0] = primarySnapshot + return createdSnapshots, nil +} + +// cleanUpOrphanedSecondarySnapshots deletes all secondary placement resource snapshots at the given index. +// +// Note that this method assumes that the corresponding mutex for the placement policy has been acquired before +// calling this method. +func (m *Manager) cleanUpOrphanedSecondarySnapshots( + ctx context.Context, + placementPolicy placementv1alpha1.PlacementPolicyAccessor, + nextSnapshotIdx int, +) (bool, error) { + // List all placement resource snapshots at the given index. + fieldMatchers := client.MatchingFields{ + ownedByAndIndexedCustomFieldName: fmt.Sprintf(ownedByAndIndexedCustomFieldFmt, placementPolicy.GetName(), strconv.Itoa(nextSnapshotIdx)), + } + + var snapshots []placementv1alpha1.PlacementResourceSnapshotAccessor + if placementPolicy.GetNamespace() == "" { + // The placement policy is cluster-scoped; list cluster placement resource snapshots. + placementResourceSnapshotList := &placementv1alpha1.ClusterPlacementResourceSnapshotList{} + if err := m.hubClient.List(ctx, placementResourceSnapshotList, fieldMatchers); err != nil { + return false, errors.NewAPIServerError(err, "failed to list cluster placement resource snapshots", true) + } + snapshots = make([]placementv1alpha1.PlacementResourceSnapshotAccessor, len(placementResourceSnapshotList.Items)) + for i := range placementResourceSnapshotList.Items { + snapshots[i] = &placementResourceSnapshotList.Items[i] + } + } else { + // The placement policy is namespace-scoped; list placement resource snapshots in the same namespace. + placementResourceSnapshotList := &placementv1alpha1.PlacementResourceSnapshotList{} + if err := m.hubClient.List(ctx, placementResourceSnapshotList, + client.InNamespace(placementPolicy.GetNamespace()), fieldMatchers); err != nil { + return false, errors.NewAPIServerError(err, "failed to list placement resource snapshots", true) + } + snapshots = make([]placementv1alpha1.PlacementResourceSnapshotAccessor, len(placementResourceSnapshotList.Items)) + for i := range placementResourceSnapshotList.Items { + snapshots[i] = &placementResourceSnapshotList.Items[i] + } + } + + // Do a sanity check; verify that there is no primary placement resource snapshot at the given index. + for idx := range snapshots { + snapshot := snapshots[idx] + subIdxStr := snapshot.GetLabels()[placementv1alpha1.PlacementResourceSnapshotSubIndexLabelKey] + if subIdxStr == "0" { + // This normally should never occur. + return false, errors.NewUnexpectedError(nil, + "found a primary placement resource snapshot at the given index while cleaning up orphaned secondary snapshots", + "primaryPlacementResourceSnapshot", klog.KObj(snapshot)) + } + } + + // Delete all the secondary placement resource snapshots at the given index. + acted := false + for idx := range snapshots { + snapshot := snapshots[idx] + if err := m.hubClient.Delete(ctx, snapshot); err != nil { + return false, errors.NewAPIServerError(err, "failed to delete an orphaned secondary placement resource snapshot", + false, "secondaryPlacementResourceSnapshot", klog.KObj(snapshot)) + } + acted = true + } + return acted, nil +} diff --git a/pkg/v1/managers/placementresourcesnapshot/resources.go b/pkg/v1/managers/placementresourcesnapshot/resources.go new file mode 100644 index 000000000..69a9e1bd2 --- /dev/null +++ b/pkg/v1/managers/placementresourcesnapshot/resources.go @@ -0,0 +1,418 @@ +/* +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 placementresourcesnapshot + +import ( + "context" + "fmt" + "sort" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/util/sets" + "k8s.io/klog/v2" + "k8s.io/kubectl/pkg/util/deployment" + + placementv1alpha1 "github.com/kubefleet-dev/kubefleet/apis/kubefleet.dev/placement/v1alpha1" + errors "github.com/kubefleet-dev/kubefleet/pkg/utils/errors" + hasher "github.com/kubefleet-dev/kubefleet/pkg/utils/resource" +) + +const ( + // etcd has a 1.5 MiB limit for objects by default, and Kubernetes clients might + // reject request entities too large (~2/~3 MiB, depending on the protocol in use). + // + // With these factors considered, we set the maximum size of all resource data in a single placement + // resource snapshot to be ~1.2 MiB, or ~1.26 MB, which should be safe in most cases. Note that the padding + // space is not just reserved for safety reasons, but also to accommodate the additional fields in + // the placement resource snapshot object, such as metadata and labels. + maxPerSnapshotResourceDataSizeBytes = 1258291 // 1.2 MiB, or ~1.26 MB. + maxPerSnapshotResourceCnt = 50 +) + +const ( + // The format in use to generate a unique identifier for each selected resource. + // + // The format is `[API-GROUP]/[API-VERSION]/[KIND]/[NAMESPACE]/[NAME]`, where + // `[API-GROUP]` is the API group of the resource, `[API-VERSION]` is the API version of the resource, + // `[KIND]` is the kind of the resource, `[NAMESPACE]` is the namespace of the resource, + // and `[NAME]` is the name of the resource. + // + // Note that for cluster-scoped resources, the `[NAMESPACE]` segment will be empty. + resourceUniqueIdStrFmt = "%s/%s/%s/%s/%s" +) + +func (m *Manager) retrieveAndHashSelectedResources( + ctx context.Context, + placementPolicyAccessor placementv1alpha1.PlacementPolicyAccessor, +) ( + resources []placementv1alpha1.SnapshottedResource, + hash string, + err error, +) { + placementPolicySpec := placementPolicyAccessor.GetSpec() + resources = make([]placementv1alpha1.SnapshottedResource, 0, len(placementPolicySpec.ResourceSelectors)) + seen := sets.Set[string]{} + + if len(placementPolicySpec.ResourceSelectors) == 0 { + // KubeFleet does not consider the absence of resource selectors to be an error; however, this special + // case should be handled by the caller (i.e., the placement resource snapshot manager should not be + // called at all if there are no resource selectors), hence the unexpected error returned here. + return nil, "", errors.NewUnexpectedError(nil, "no resource selectors are present") + } + + for idx := range placementPolicySpec.ResourceSelectors { + selector := placementPolicySpec.ResourceSelectors[idx] + + switch { + case len(selector.Name) != 0: + // Retrieve the resource by name. + resourceFromSelector, err := m.retrieveResourceByName(ctx, placementPolicyAccessor, selector) + if err != nil { + return nil, "", errors.Wraps(err, "failed to retrieve a selected resource (name-based selector)", + "resourceSelectorIndex", idx) + } + resourceId := resourceUniqueId(resourceFromSelector) + if !seen.Has(resourceId) { + // The resource has not been seen before; add it to the list of selected resources. + seen.Insert(resourceId) + resources = append(resources, resourceFromSelector) + } else { + // The resource has already been seen; skip it and log a message. + klog.V(2).InfoS("Found duplicate selected resource; skipping it", "resourceId", resourceId, "resourceSelectorIndex", idx) + } + case selector.LabelSelector != nil: + // Retrieve the resources by label selector. + resourcesFromSelector, err := m.retrieveResourcesByLabelSelector(ctx, placementPolicyAccessor, selector) + if err != nil { + return nil, "", errors.Wraps(err, "failed to retrieve selected resources (label selector-based selector)", + "manager", managerName, "resourceSelectorIndex", idx) + } + for idx := range resourcesFromSelector { + resourceFromSelector := resourcesFromSelector[idx] + resourceId := resourceUniqueId(resourceFromSelector) + if !seen.Has(resourceId) { + // The resource has not been seen before; add it to the list of selected resources. + seen.Insert(resourceId) + resources = append(resources, resourceFromSelector) + } else { + // The resource has already been seen; skip it and log a message. + klog.V(2).InfoS("Found duplicate selected resource; skipping it", "resourceId", resourceId, "resourceSelectorIndex", idx) + } + } + default: + return nil, "", errors.NewUserError(nil, "invalid resource selector: neither name nor label selector is specified", + "manager", managerName, "resourceSelectorIndex", idx) + } + } + + // Sort the selected resources to ensure deterministic outcomes. + sort.Slice(resources, func(i, j int) bool { + return resourceUniqueId(resources[i]) < resourceUniqueId(resources[j]) + }) + + hash, err = hasher.HashOf(resources) + if err != nil { + return nil, "", errors.Wraps(err, "failed to compute the hash of the selected resources", + "manager", managerName) + } + return resources, hash, nil +} + +func (m *Manager) retrieveResourceByName( + ctx context.Context, + placementPolicyAccessor placementv1alpha1.PlacementPolicyAccessor, + resourceSelector placementv1alpha1.ResourceSelector, +) (placementv1alpha1.SnapshottedResource, error) { + gvk := schema.GroupVersionKind{ + Group: resourceSelector.APIGroup, + Version: resourceSelector.APIVersion, + Kind: resourceSelector.Kind, + } + + // Convert the GVK to a GVR using the REST mapper. + mapping, err := m.restMapper.RESTMapping(gvk.GroupKind(), gvk.Version) + if err != nil { + return placementv1alpha1.SnapshottedResource{}, errors.NewUnexpectedError(err, "failed to map GVK to GVR", + "manager", managerName, "gvk", gvk) + } + gvr := mapping.Resource + + // Determine the namespace of the selected resource. + // + // If the placement policy is namespace-scoped, the selected resource is assumed to be from the same namespace; + // if the placement policy is cluster-scoped, the namespace is taken from the resource selector. + namespace := resourceSelector.Namespace + if placementPolicyAccessor.GetNamespace() != "" { + namespace = placementPolicyAccessor.GetNamespace() + } + + var resource *unstructured.Unstructured + // Before retrieving the resource via cache, verify if the informer has been synced. + if m.hubDynamicInformerManager.IsInformerSet(gvk) && m.hubDynamicInformerManager.IsInformerSynced(gvr) { + // An informer for the selected resource has been set up and synced; proceed to retrieve the resource from the cache. + var obj runtime.Object + if namespace == "" { + obj, err = m.hubDynamicInformerManager.Lister(gvr).Get(resourceSelector.Name) + } else { + obj, err = m.hubDynamicInformerManager.Lister(gvr).ByNamespace(namespace).Get(resourceSelector.Name) + } + if err != nil { + return placementv1alpha1.SnapshottedResource{}, errors.NewAPIServerError(err, "failed to get selected resource", true, + "manager", managerName, "gvr", gvr, "namespace", namespace, "name", resourceSelector.Name) + } + var ok bool + resource, ok = obj.(*unstructured.Unstructured) + if !ok { + return placementv1alpha1.SnapshottedResource{}, errors.NewUnexpectedError(nil, "failed to convert the retrieved resource to unstructured", + "manager", managerName, "gvr", gvr, "namespace", namespace, "name", resourceSelector.Name) + } + } else { + // No informer is set up for the selected resource, or the informer has not been synced yet. + // + // As a fallback, retrieve the resource directly from the API server. + klog.V(2).InfoS("Informer for the selected resource is not set up or not synced; retrieving the resource directly from the API server", + "manager", managerName, "gvr", gvr) + if namespace == "" { + resource, err = m.hubDynamicClient.Resource(gvr).Get(ctx, resourceSelector.Name, metav1.GetOptions{}) + } else { + resource, err = m.hubDynamicClient.Resource(gvr).Namespace(namespace).Get(ctx, resourceSelector.Name, metav1.GetOptions{}) + } + if err != nil { + return placementv1alpha1.SnapshottedResource{}, errors.NewAPIServerError(err, "failed to get selected resource directly from the API server", false, + "manager", managerName, "gvr", gvr, "namespace", namespace, "name", resourceSelector.Name) + } + } + + snapshottedResource, err := snapshotResource(resource) + if err != nil { + return placementv1alpha1.SnapshottedResource{}, + errors.Wraps(err, "failed to snapshot selected resource", "manager", managerName, + "gvr", gvr, "namespace", namespace, "name", resourceSelector.Name) + } + return snapshottedResource, nil +} + +func (m *Manager) retrieveResourcesByLabelSelector( + ctx context.Context, + placementPolicyAccessor placementv1alpha1.PlacementPolicyAccessor, + resourceSelector placementv1alpha1.ResourceSelector, +) ([]placementv1alpha1.SnapshottedResource, error) { + gvk := schema.GroupVersionKind{ + Group: resourceSelector.APIGroup, + Version: resourceSelector.APIVersion, + Kind: resourceSelector.Kind, + } + + // Convert the GVK to a GVR using the REST mapper. + mapping, err := m.restMapper.RESTMapping(gvk.GroupKind(), gvk.Version) + if err != nil { + return nil, errors.NewUnexpectedError(err, "failed to map GVK to GVR", + "manager", managerName, "gvk", gvk) + } + gvr := mapping.Resource + + // Convert the label selector into a selector string. + selector, err := metav1.LabelSelectorAsSelector(resourceSelector.LabelSelector) + if err != nil { + return nil, errors.NewUserError(err, "invalid label selector", + "manager", managerName, "gvk", gvk, "labelSelector", resourceSelector.LabelSelector) + } + + // Determine the namespace of the selected resources. + // + // If the placement policy is namespace-scoped, the selected resources are assumed to be from the same namespace; + // if the placement policy is cluster-scoped, the namespace is taken from the resource selector. + namespace := resourceSelector.Namespace + if placementPolicyAccessor.GetNamespace() != "" { + namespace = placementPolicyAccessor.GetNamespace() + } + + var resources []*unstructured.Unstructured + if m.hubDynamicInformerManager.IsInformerSet(gvk) && m.hubDynamicInformerManager.IsInformerSynced(gvr) { + // An informer for the selected resources has been set up and synced; proceed to retrieve the resources from the cache. + var objList []runtime.Object + if namespace == "" { + objList, err = m.hubDynamicInformerManager.Lister(gvr).List(selector) + } else { + objList, err = m.hubDynamicInformerManager.Lister(gvr).ByNamespace(namespace).List(selector) + } + if err != nil { + return nil, errors.NewAPIServerError(err, "failed to list the selected resources", true, + "manager", managerName, "gvr", gvr, "namespace", namespace, "labelSelector", selector.String()) + } + + for idx := range objList { + obj := objList[idx] + resource, ok := obj.(*unstructured.Unstructured) + if !ok { + return nil, errors.NewUnexpectedError(nil, "failed to convert the retrieved resource to unstructured", + "manager", managerName, "gvr", gvr, "namespace", namespace) + } + resources = append(resources, resource) + } + } else { + // No informer is set up for the selected resources, or the informer has not been synced yet. + // + // As a fallback, retrieve the resources directly from the API server. + klog.V(2).InfoS("Informer for the selected resources is not set up or not synced; retrieving the resources directly from the API server", + "manager", managerName, "gvr", gvr) + var resourceList *unstructured.UnstructuredList + if namespace == "" { + resourceList, err = m.hubDynamicClient.Resource(gvr).List(ctx, metav1.ListOptions{ + LabelSelector: selector.String(), + }) + } else { + resourceList, err = m.hubDynamicClient.Resource(gvr).Namespace(namespace).List(ctx, metav1.ListOptions{ + LabelSelector: selector.String(), + }) + } + if err != nil { + return nil, errors.NewAPIServerError(err, "failed to list the selected resources", false, + "manager", managerName, "gvr", gvr, "namespace", namespace, "labelSelector", selector.String()) + } + + for idx := range resourceList.Items { + resource := &resourceList.Items[idx] + resources = append(resources, resource) + } + } + + snapshottedResources := make([]placementv1alpha1.SnapshottedResource, len(resources)) + for idx := range resources { + resource := resources[idx] + snapshottedResources[idx], err = snapshotResource(resource) + if err != nil { + return nil, errors.Wraps(err, "failed to snapshot selected resource", "manager", managerName, + "gvr", gvr, "namespace", namespace, "name", resource.GetName()) + } + } + return snapshottedResources, nil +} + +// snapshotResource removes fields that are not needed in a snapshot from an unstructured resource and converts it +// into a SnapshottedResource. +func snapshotResource(resource *unstructured.Unstructured) (placementv1alpha1.SnapshottedResource, error) { + // Create a deep copy of the resource. + resourceCopy := resource.DeepCopy() + + // Remove certain labels and annotations. + if annotations := resourceCopy.GetAnnotations(); annotations != nil { + // Remove the last applied configuration set by kubectl. + delete(annotations, corev1.LastAppliedConfigAnnotation) + + // Remove the revision annotation set by deployment controller. + delete(annotations, deployment.RevisionAnnotation) + + if len(annotations) == 0 { + resourceCopy.SetAnnotations(nil) + } else { + resourceCopy.SetAnnotations(annotations) + } + } + + // Remove certain system-managed fields. + resourceCopy.SetOwnerReferences(nil) + resourceCopy.SetManagedFields(nil) + + // Remove the read-only fields. + resourceCopy.SetCreationTimestamp(metav1.Time{}) + resourceCopy.SetDeletionTimestamp(nil) + resourceCopy.SetDeletionGracePeriodSeconds(nil) + resourceCopy.SetGeneration(0) + resourceCopy.SetResourceVersion("") + resourceCopy.SetSelfLink("") + resourceCopy.SetUID("") + + // Remove the status field. + unstructured.RemoveNestedField(resourceCopy.Object, "status") + + resourceCopyRawData, err := resourceCopy.MarshalJSON() + if err != nil { + return placementv1alpha1.SnapshottedResource{}, errors.NewUnexpectedError(err, "failed to marshal the resource copy to JSON", + "manager", managerName, "resource", klog.KObj(resourceCopy)) + } + + gvk := resource.GroupVersionKind() + + // Note that for regular Kubernetes resources, the additional information field is always left empty. + return placementv1alpha1.SnapshottedResource{ + Identifier: placementv1alpha1.ObjectReference{ + Namespace: resourceCopy.GetNamespace(), + Name: resourceCopy.GetName(), + APIGroup: gvk.Group, + APIVersion: gvk.Version, + Kind: gvk.Kind, + }, + Manifest: runtime.RawExtension{Raw: resourceCopyRawData}, + }, nil +} + +func resourceUniqueId(resource placementv1alpha1.SnapshottedResource) string { + return fmt.Sprintf(resourceUniqueIdStrFmt, + resource.Identifier.APIGroup, + resource.Identifier.APIVersion, + resource.Identifier.Kind, + resource.Identifier.Namespace, + resource.Identifier.Name) +} + +func splitResourcesIntoSizeControlledGroups(resources []placementv1alpha1.SnapshottedResource) ([][]placementv1alpha1.SnapshottedResource, error) { + if len(resources) == 0 { + // Return one single empty group. + return [][]placementv1alpha1.SnapshottedResource{{}}, nil + } + + var groups [][]placementv1alpha1.SnapshottedResource + var currentGroup []placementv1alpha1.SnapshottedResource + currentSize := 0 + + for i := range resources { + resource := resources[i] + resourceSize := len(resource.Manifest.Raw) + for _, info := range resource.AdditionalInfo { + resourceSize += len(info) + } + + if resourceSize > maxPerSnapshotResourceDataSizeBytes { + // A single resource exceeds the per-snapshot size limit; it can never fit into any group. + return nil, errors.NewUserError(nil, "a single selected resource is too large to fit in a placement resource snapshot", + "manager", managerName, "resource", resource.Identifier, + "resourceSizeBytes", resourceSize, "maxPerSnapshotResourceDataSizeBytes", maxPerSnapshotResourceDataSizeBytes) + } + + // Start a new group if adding this resource would exceed either the size or the count limit. + if len(currentGroup) > 0 && + (currentSize+resourceSize > maxPerSnapshotResourceDataSizeBytes || len(currentGroup) >= maxPerSnapshotResourceCnt) { + groups = append(groups, currentGroup) + currentGroup = nil + currentSize = 0 + } + + currentGroup = append(currentGroup, resource) + currentSize += resourceSize + } + + if len(currentGroup) > 0 { + groups = append(groups, currentGroup) + } + + return groups, nil +} diff --git a/pkg/v1/managers/placementresourcesnapshot/snapshots.go b/pkg/v1/managers/placementresourcesnapshot/snapshots.go new file mode 100644 index 000000000..8f2acf0b1 --- /dev/null +++ b/pkg/v1/managers/placementresourcesnapshot/snapshots.go @@ -0,0 +1,130 @@ +/* +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 placementresourcesnapshot + +import ( + "strconv" + + placementv1alpha1 "github.com/kubefleet-dev/kubefleet/apis/kubefleet.dev/placement/v1alpha1" + errors "github.com/kubefleet-dev/kubefleet/pkg/utils/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/klog/v2" + "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" +) + +func primaryPlacementResourceSnapshot( + namespace, name string, + ownerPlacementPolicy placementv1alpha1.PlacementPolicyAccessor, + idx int, + resources []placementv1alpha1.SnapshottedResource, + resourceHash string, + snapshotCount int, + scheme *runtime.Scheme, +) (placementv1alpha1.PlacementResourceSnapshotAccessor, error) { + labels := map[string]string{ + placementv1alpha1.PlacementResourceSnapshotOwnedByLabelKey: ownerPlacementPolicy.GetName(), + placementv1alpha1.PlacementResourceSnapshotIndexLabelKey: strconv.Itoa(idx), + placementv1alpha1.PlacementResourceSnapshotSubIndexLabelKey: "0", + placementv1alpha1.SubIndexedPlacementResourceSnapshotCountLabelKey: strconv.Itoa(snapshotCount), + } + annotations := map[string]string{ + placementv1alpha1.PlacementResourceSnapshotContentsHashAnnotationKey: resourceHash, + } + + var primarySnapshot placementv1alpha1.PlacementResourceSnapshotAccessor + if namespace == "" { + primarySnapshot = &placementv1alpha1.ClusterPlacementResourceSnapshot{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Labels: labels, + Annotations: annotations, + }, + Spec: placementv1alpha1.PlacementResourceSnapshotSpec{ + Resources: resources, + }, + } + } else { + primarySnapshot = &placementv1alpha1.PlacementResourceSnapshot{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Namespace: namespace, + Labels: labels, + Annotations: annotations, + }, + Spec: placementv1alpha1.PlacementResourceSnapshotSpec{ + Resources: resources, + }, + } + } + + if err := controllerutil.SetControllerReference(ownerPlacementPolicy, primarySnapshot, scheme); err != nil { + return nil, errors.NewUnexpectedError(err, "failed to set controller reference on the primary placement resource snapshot", + "manager", managerName, "primaryPlacementResourceSnapshot", klog.KObj(primarySnapshot)) + } + return primarySnapshot, nil +} + +func secondaryPlacementResourceSnapshot( + namespace, name string, + ownerPlacementPolicy placementv1alpha1.PlacementPolicyAccessor, + idx, subIdx int, + resources []placementv1alpha1.SnapshottedResource, + resourceHash string, + scheme *runtime.Scheme, +) (placementv1alpha1.PlacementResourceSnapshotAccessor, error) { + labels := map[string]string{ + placementv1alpha1.PlacementResourceSnapshotOwnedByLabelKey: ownerPlacementPolicy.GetName(), + placementv1alpha1.PlacementResourceSnapshotIndexLabelKey: strconv.Itoa(idx), + placementv1alpha1.PlacementResourceSnapshotSubIndexLabelKey: strconv.Itoa(subIdx), + } + annotations := map[string]string{ + placementv1alpha1.PlacementResourceSnapshotContentsHashAnnotationKey: resourceHash, + } + + var secondarySnapshot placementv1alpha1.PlacementResourceSnapshotAccessor + if namespace == "" { + secondarySnapshot = &placementv1alpha1.ClusterPlacementResourceSnapshot{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Labels: labels, + Annotations: annotations, + }, + Spec: placementv1alpha1.PlacementResourceSnapshotSpec{ + Resources: resources, + }, + } + } else { + secondarySnapshot = &placementv1alpha1.PlacementResourceSnapshot{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Namespace: namespace, + Labels: labels, + Annotations: annotations, + }, + Spec: placementv1alpha1.PlacementResourceSnapshotSpec{ + Resources: resources, + }, + } + } + + if err := controllerutil.SetControllerReference(ownerPlacementPolicy, secondarySnapshot, scheme); err != nil { + return nil, errors.NewUnexpectedError(err, "failed to set controller reference on a secondary placement resource snapshot", + "manager", managerName, "secondaryPlacementResourceSnapshot", klog.KObj(secondarySnapshot)) + } + return secondarySnapshot, nil +} diff --git a/pkg/v1/managers/placementresourcesnapshot/uniquename.go b/pkg/v1/managers/placementresourcesnapshot/uniquename.go new file mode 100644 index 000000000..f689e7f5e --- /dev/null +++ b/pkg/v1/managers/placementresourcesnapshot/uniquename.go @@ -0,0 +1,145 @@ +/* +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 placementresourcesnapshot + +import ( + "crypto/sha256" + "fmt" + "strconv" + "strings" +) + +const ( + nameLenLimit = 251 + hashSegLen = 12 +) + +const ( + // The name format for primary placement resource snapshots. Typically, these snapshots are named using the + // format: + // + // `[PLACEMENT-POLICY-NAME]-resource-snapshot-[SNAPSHOT-INDEX]`, + // + // where `[PLACEMENT-POLICY-NAME]` is the name of the owner placement policy, and + // `[SNAPSHOT-INDEX]` is the monotonically increasing index of the snapshot. + // + // If the name becomes too long (> 251 characters), KubeFleet will truncate the placement policy name segment + // and the snapshot index segment as appropriate and add a hash suffix to the name, i.e., + // + // `[PLACEMENT-POLICY-NAME-TRUNCATED]-resource-snapshot-[SNAPSHOT-INDEX-TRUNCATED]-[HASH]`, + // + // where `[HASH]` is the first few characters of the hash of the value + // `[PLACEMENT-POLICY-NAME]-resource-snapshot-[SNAPSHOT-INDEX]`. + PrimaryPlacementResourceSnapshotNameFmt = "%s-resource-snapshot-%d" + PrimaryPlacementResourceSnapshotNameWithHashFmt = "%s-resource-snapshot-%s-%s" + + // The name format for secondary placement resource snapshots. Typically, these snapshots are named using the + // format: + // + // `[PLACEMENT-POLICY-NAME]-resource-snapshot-[SNAPSHOT-INDEX]-[SNAPSHOT-SUB-INDEX]`, + // + // where `[PLACEMENT-POLICY-NAME]` is the name of the owner placement policy, + // `[SNAPSHOT-INDEX]` is the monotonically increasing index of the snapshot, and + // `[SNAPSHOT-SUB-INDEX]` is the monotonically increasing sub-index of the snapshot. + // + // If the name becomes too long (> 251 characters), KubeFleet will truncate the placement policy name segment + // and the `[SNAPSHOT-INDEX]-[SNAPSHOT-SUB-INDEX]` segment as appropriate and add a hash suffix to the name, i.e., + // + // `[PLACEMENT-POLICY-NAME-TRUNCATED]-resource-snapshot-[INDEX-TRUNCATED]-[HASH]`, + // + // where `[HASH]` is the first few characters of the hash of the value + // `[PLACEMENT-POLICY-NAME-TRUNCATED]-resource-snapshot-[SNAPSHOT-INDEX]-[SNAPSHOT-SUB-INDEX]`. + SecondaryPlacementResourceSnapshotNameFmt = "%s-resource-snapshot-%d-%d" + SecondaryPlacementResourceSnapshotNameWithHashFmt = "%s-resource-snapshot-%s-%s" +) + +// uniqueNameForPrimaryPlacementResourceSnapshot generates a unique name for a primary placement resource snapshot. +func uniqueNameForPrimaryPlacementResourceSnapshot(placementPolicyName string, idx int) (string, error) { + name := fmt.Sprintf(PrimaryPlacementResourceSnapshotNameFmt, placementPolicyName, idx) + if len(name) <= nameLenLimit { + return name, nil + } + + // The name is too long; truncate the placement policy name segment and append a hash suffix. + // The hash is computed over the full (untruncated) name. + // + // Note that here only the first few (12) characters are kept. This does lead to increased risk of name + // collisions, but the chances still remain extremely low. If such a collision does occur, manual intervention + // is needed for resolution. + hash := fmt.Sprintf("%x", sha256.Sum256([]byte(name)))[:hashSegLen] + + // Compute how many characters are left for the two variable segments (the placement policy name and the + // snapshot index), then split the available space evenly between them. + // + // reservedLen accounts for the static decoration and the hash suffix only. + // + // The offset 1 is the length of placeholder index (0). + reservedLen := len(fmt.Sprintf(PrimaryPlacementResourceSnapshotNameWithHashFmt, "", "0", hash)) - 1 + availableLen := nameLenLimit - reservedLen + availablePerSeg := availableLen / 2 + + // Remove all dots from the placement policy name segment so that truncation cannot leave a trailing dot, + // which would produce an invalid DNS subdomain label. + truncatedPlacementPolicyName := strings.ReplaceAll(placementPolicyName, ".", "") + if len(truncatedPlacementPolicyName) > availablePerSeg { + truncatedPlacementPolicyName = truncatedPlacementPolicyName[:availablePerSeg] + } + + truncatedIdxStr := strconv.Itoa(idx) + if len(truncatedIdxStr) > availablePerSeg { + truncatedIdxStr = truncatedIdxStr[:availablePerSeg] + } + + return fmt.Sprintf(PrimaryPlacementResourceSnapshotNameWithHashFmt, truncatedPlacementPolicyName, truncatedIdxStr, hash), nil +} + +func uniqueNameForSecondaryPlacementResourceSnapshot(placementPolicyName string, idx int, subIdx int) (string, error) { + name := fmt.Sprintf(SecondaryPlacementResourceSnapshotNameFmt, placementPolicyName, idx, subIdx) + if len(name) <= nameLenLimit { + return name, nil + } + + // The name is too long; truncate the placement policy name segment and append a hash suffix. + // The hash is computed over the full (untruncated) name. + // + // Note that here only the first few (12) characters are kept. This does lead to increased risk of name + // collisions, but the chances still remain extremely low. If such a collision does occur, manual intervention + // is needed for resolution. + hash := fmt.Sprintf("%x", sha256.Sum256([]byte(name)))[:hashSegLen] + + // Compute how many characters are left for the two variable segments (the placement policy name and the + // combined snapshot index/sub-index segment), then split the available space evenly between them. + // + // reservedLen accounts for the static decoration and the hash suffix only. + reservedLen := len(fmt.Sprintf(SecondaryPlacementResourceSnapshotNameWithHashFmt, "", "", hash)) + availableLen := nameLenLimit - reservedLen + availablePerSeg := availableLen / 2 + + // Remove all dots from the placement policy name segment so that truncation cannot leave a trailing dot, + // which would produce an invalid DNS subdomain label. + truncatedPlacementPolicyName := strings.ReplaceAll(placementPolicyName, ".", "") + if len(truncatedPlacementPolicyName) > availablePerSeg { + truncatedPlacementPolicyName = truncatedPlacementPolicyName[:availablePerSeg] + } + + truncatedIdxStr := fmt.Sprintf("%d-%d", idx, subIdx) + if len(truncatedIdxStr) > availablePerSeg { + truncatedIdxStr = truncatedIdxStr[:availablePerSeg] + } + + return fmt.Sprintf(SecondaryPlacementResourceSnapshotNameWithHashFmt, truncatedPlacementPolicyName, truncatedIdxStr, hash), nil +} diff --git a/test/utils/informer/manager.go b/test/utils/informer/manager.go index 004ef9364..2caf94d3e 100644 --- a/test/utils/informer/manager.go +++ b/test/utils/informer/manager.go @@ -192,3 +192,8 @@ func (m *FakeManager) AddEventHandlerToInformer(_ schema.GroupVersionResource, _ func (m *FakeManager) CreateInformerForResource(_ informer.APIResourceMeta) { // No-op for testing } + +func (m *FakeManager) IsInformerSet(_ schema.GroupVersionKind) bool { + // For testing, we can assume that the informer is always set for the given resource. + return true +}