Skip to content

Commit

Permalink
add first test
Browse files Browse the repository at this point in the history
  • Loading branch information
Ryan Zhang committed Sep 23, 2024
1 parent 19ed98c commit 9b8484f
Show file tree
Hide file tree
Showing 14 changed files with 1,228 additions and 266 deletions.
10 changes: 10 additions & 0 deletions apis/placement/v1alpha1/common.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -14,4 +14,14 @@ const (

// ResourceOverrideSnapshotKind is the kind of the ResourceOverrideSnapshotKind.
ResourceOverrideSnapshotKind = "ResourceOverrideSnapshot"

// StagedUpdateRunFinalizer is used by the staged update run controller to make sure that the stagedUpdateRun
// object is not deleted until all its dependent resources are deleted.
StagedUpdateRunFinalizer = fleetPrefix + "stagedupdaterun-finalizer"

// TargetUpdateRunLabel is the label that indicates the target update run of a staged update run.
TargetUpdateRunLabel = fleetPrefix + "targetupdaterun"

// The name of delete stage in the staged update run
UpdateRunDeleteStageName = fleetPrefix + "deleteStage"
)
11 changes: 9 additions & 2 deletions apis/placement/v1alpha1/stagedupdate_types.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -157,6 +157,12 @@ type StagedUpdateRunStatus struct {
// +kubebuilder:validation:Optional
PolicySnapshotIndexUsed string `json:"policySnapshotIndexUsed,omitempty"`

// PolicyObservedCRPGeneration is the generation of the CRP which the scheduler uses to perform the scheduling.
// It is computed at the beginning of the update run from the policy snapshot object. If the CRP is updated
// during the update run in a way that causes the policy snapshot condition to change, the update run is abandoned.
// +kubebuilder:validation:Optional
PolicyObservedNodeCount int `json:"policyObservedNodeCount,omitempty"`

// ApplyStrategy is the apply strategy that the stagedUpdateRun is using.
// It is the same as the apply strategy in the CRP when the staged update run starts.
// The apply strategy is not updated during the update run even if it changes in the CRP.
Expand Down Expand Up @@ -201,8 +207,9 @@ const (
// StagedUpdateRunConditionInitialized indicates whether the staged update run is initialized, meaning it
// has computed all the stages according to the referenced strategy and is ready to start the update.
// Its condition status can be one of the following:
// - "True": The staged update run is initialized.
// - "False": The staged update run encountered an error during initialization.
// - "True": The staged update run is initialized successfully.
// - "False": The staged update run encountered an error during initialization and aborted.
// - "Unknown": The staged update run initialization has started.
StagedUpdateRunConditionInitialized StagedUpdateRunConditionType = "Initialized"

// StagedUpdateRunConditionProgressing indicates whether the staged update run is making progress.
Expand Down
53 changes: 29 additions & 24 deletions apis/placement/v1beta1/commons.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -6,18 +6,25 @@ Licensed under the MIT license.
package v1beta1

const (
ClusterResourcePlacementKind = "ClusterResourcePlacement"
ClusterResourcePlacementResource = "clusterresourceplacements"
ClusterResourceBindingKind = "ClusterResourceBinding"
ClusterResourceSnapshotKind = "ClusterResourceSnapshot"
// ClusterResourcePlacementKind represents the kind of ClusterResourcePlacement.
ClusterResourcePlacementKind = "ClusterResourcePlacement"
// ClusterResourcePlacementResource represents the resource name for ClusterResourcePlacement.
ClusterResourcePlacementResource = "clusterresourceplacements"
// ClusterResourceBindingKind represents the kind of ClusterResourceBinding.
ClusterResourceBindingKind = "ClusterResourceBinding"
// ClusterResourceSnapshotKind represents the kind of ClusterResourceSnapshot.
ClusterResourceSnapshotKind = "ClusterResourceSnapshot"
// ClusterSchedulingPolicySnapshotKind represents the kind of ClusterSchedulingPolicySnapshot.
ClusterSchedulingPolicySnapshotKind = "ClusterSchedulingPolicySnapshot"
WorkKind = "Work"
AppliedWorkKind = "AppliedWork"
// WorkKind represents the kind of Work.
WorkKind = "Work"
// AppliedWorkKind represents the kind of AppliedWork.
AppliedWorkKind = "AppliedWork"
)

const (
// Unprefixed labels/annotations are reserved for end-users
// we will add a kubernetes-fleet.io to designate these labels/annotations as official fleet labels/annotations.
// fleetPrefix is the prefix used for official fleet labels/annotations.
// Unprefixed labels/annotations are reserved for end-users.
// See https://github.com/kubernetes/community/blob/master/contributors/devel/sig-architecture/api-conventions.md#label-selector-and-annotation-conventions
fleetPrefix = "kubernetes-fleet.io/"

Expand All @@ -29,25 +36,25 @@ const (
// cluster.
WorkFinalizer = fleetPrefix + "work-cleanup"

// CRPTrackingLabel is the label that points to the cluster resource policy that creates a resource binding.
// CRPTrackingLabel points to the cluster resource policy that creates this resource binding.
CRPTrackingLabel = fleetPrefix + "parent-CRP"

// IsLatestSnapshotLabel tells if the snapshot is the latest one.
// IsLatestSnapshotLabel indicates if the snapshot is the latest one.
IsLatestSnapshotLabel = fleetPrefix + "is-latest-snapshot"

// FleetResourceLabelKey is that label that indicates the resource is a fleet resource.
// FleetResourceLabelKey indicates that the resource is a fleet resource.
FleetResourceLabelKey = fleetPrefix + "is-fleet-resource"

// FirstWorkNameFmt is the format of the name of the work generated with first resource snapshot .
// FirstWorkNameFmt is the format of the name of the work generated with the first resource snapshot.
// The name of the first work is {crpName}-work.
FirstWorkNameFmt = "%s-work"

// WorkNameWithSubindexFmt is the format of the name of a work generated with resource snapshot with subindex.
// The name of the first work is {crpName}-{subindex}.
// WorkNameWithSubindexFmt is the format of the name of a work generated with a resource snapshot with a subindex.
// The name of the work is {crpName}-{subindex}.
WorkNameWithSubindexFmt = "%s-%d"

// WorkNameWithConfigEnvelopeFmt is the format of the name of a work generated with config envelop.
// The format is {workPrefix}-configMap-uuid
// WorkNameWithConfigEnvelopeFmt is the format of the name of a work generated with a config envelope.
// The format is {workPrefix}-configMap-uuid.
WorkNameWithConfigEnvelopeFmt = "%s-configmap-%s"

// ParentResourceSnapshotIndexLabel is the label applied to work that contains the index of the resource snapshot that generates the work.
Expand All @@ -56,25 +63,23 @@ const (
// ParentBindingLabel is the label applied to work that contains the name of the binding that generates the work.
ParentBindingLabel = fleetPrefix + "parent-resource-binding"

// CRPGenerationAnnotation is the annotation that indicates the generation of the CRP from
// which an object is derived or last updated.
// CRPGenerationAnnotation indicates the generation of the CRP from which an object is derived or last updated.
CRPGenerationAnnotation = fleetPrefix + "CRP-generation"

// EnvelopeConfigMapAnnotation is the annotation that indicates the configmap is an envelope configmap that contains resources
// we need to apply to the member cluster instead of the configMap itself.
// EnvelopeConfigMapAnnotation indicates that the configmap is an envelope configmap containing resources to be applied to the member cluster instead of the configMap itself.
EnvelopeConfigMapAnnotation = fleetPrefix + "envelope-configmap"

// EnvelopeTypeLabel is the label that marks the work object as generated from an envelope object.
// EnvelopeTypeLabel marks the work object as generated from an envelope object.
// The value of the annotation is the type of the envelope object.
EnvelopeTypeLabel = fleetPrefix + "envelope-work"

// EnvelopeNamespaceLabel is the label that contains the namespace of the envelope object that the work is generated from.
// EnvelopeNamespaceLabel contains the namespace of the envelope object that the work is generated from.
EnvelopeNamespaceLabel = fleetPrefix + "envelope-namespace"

// EnvelopeNameLabel is the label that contains the name of the envelope object that the work is generated from.
// EnvelopeNameLabel contains the name of the envelope object that the work is generated from.
EnvelopeNameLabel = fleetPrefix + "envelope-name"

// PreviousBindingStateAnnotation is the annotation that records the previous state of a binding.
// PreviousBindingStateAnnotation records the previous state of a binding.
// This is used to remember if an "unscheduled" binding was moved from a "bound" state or a "scheduled" state.
PreviousBindingStateAnnotation = fleetPrefix + "previous-binding-state"
)
Expand Down
2 changes: 1 addition & 1 deletion pkg/controllers/clusterresourceplacement/resource_selector.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -286,7 +286,7 @@ func (r *Reconciler) fetchAllResourcesInOneNamespace(namespaceName string, place
}
for _, obj := range objs {
uObj := obj.DeepCopyObject().(*unstructured.Unstructured)
shouldInclude, err := utils.ShouldPropagateObj(r.InformerManager, uObj)
shouldInclude, err := controller.ShouldPropagateObj(r.InformerManager, uObj)
if err != nil {
klog.ErrorS(err, "cannot determine if we should propagate an object", "object", klog.KObj(uObj))
return nil, err
Expand Down
184 changes: 2 additions & 182 deletions pkg/controllers/rollout/override.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -7,122 +7,15 @@ package rollout

import (
"context"
"errors"
"sort"
"strconv"

apierrors "k8s.io/apimachinery/pkg/api/errors"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/apimachinery/pkg/types"
"k8s.io/klog/v2"
"sigs.k8s.io/controller-runtime/pkg/client"

clusterv1beta1 "go.goms.io/fleet/apis/cluster/v1beta1"
placementv1alpha1 "go.goms.io/fleet/apis/placement/v1alpha1"
placementv1beta1 "go.goms.io/fleet/apis/placement/v1beta1"
"go.goms.io/fleet/pkg/utils"
"go.goms.io/fleet/pkg/utils/controller"
"go.goms.io/fleet/pkg/utils/overrider"
)

// fetchAllMatchingOverridesForResourceSnapshot fetches all the matching overrides which are attached to the selected resources.
func (r *Reconciler) fetchAllMatchingOverridesForResourceSnapshot(ctx context.Context, crp string, masterResourceSnapshot *placementv1beta1.ClusterResourceSnapshot) ([]*placementv1alpha1.ClusterResourceOverrideSnapshot, []*placementv1alpha1.ResourceOverrideSnapshot, error) {
// fetch the cro and ro snapshot list first before finding the matched ones.
latestSnapshotLabelMatcher := client.MatchingLabels{
placementv1beta1.IsLatestSnapshotLabel: strconv.FormatBool(true),
}
croList := &placementv1alpha1.ClusterResourceOverrideSnapshotList{}
if err := r.Client.List(ctx, croList, latestSnapshotLabelMatcher); err != nil {
klog.ErrorS(err, "Failed to list all the clusterResourceOverrideSnapshots")
return nil, nil, err
}
roList := &placementv1alpha1.ResourceOverrideSnapshotList{}
if err := r.Client.List(ctx, roList, latestSnapshotLabelMatcher); err != nil {
klog.ErrorS(err, "Failed to list all the resourceOverrideSnapshots")
return nil, nil, err
}

if len(croList.Items) == 0 && len(roList.Items) == 0 {
return nil, nil, nil // no overrides and nothing to do
}

resourceSnapshots, err := controller.FetchAllClusterResourceSnapshots(ctx, r.Client, crp, masterResourceSnapshot)
if err != nil {
return nil, nil, err
}

possibleCROs := make(map[placementv1beta1.ResourceIdentifier]bool)
possibleROs := make(map[placementv1beta1.ResourceIdentifier]bool)
// List all the possible CROs and ROs based on the selected resources.
for _, snapshot := range resourceSnapshots {
for _, res := range snapshot.Spec.SelectedResources {
var uResource unstructured.Unstructured
if err := uResource.UnmarshalJSON(res.Raw); err != nil {
klog.ErrorS(err, "Resource has invalid content", "snapshot", klog.KObj(snapshot), "selectedResource", res.Raw)
return nil, nil, controller.NewUnexpectedBehaviorError(err)
}
// If the resource is namespaced scope resource, the resource could be selected by the namespace or selected
// by the object itself.
if !r.InformerManager.IsClusterScopedResources(uResource.GroupVersionKind()) {
croKey := placementv1beta1.ResourceIdentifier{
Group: utils.NamespaceMetaGVK.Group,
Version: utils.NamespaceMetaGVK.Version,
Kind: utils.NamespaceMetaGVK.Kind,
Name: uResource.GetNamespace(),
}
possibleCROs[croKey] = true // selected by the namespace
roKey := placementv1beta1.ResourceIdentifier{
Group: uResource.GetObjectKind().GroupVersionKind().Group,
Version: uResource.GetObjectKind().GroupVersionKind().Version,
Kind: uResource.GetObjectKind().GroupVersionKind().Kind,
Namespace: uResource.GetNamespace(),
Name: uResource.GetName(),
}
possibleROs[roKey] = true // selected by the object itself
} else {
croKey := placementv1beta1.ResourceIdentifier{
Group: uResource.GetObjectKind().GroupVersionKind().Group,
Version: uResource.GetObjectKind().GroupVersionKind().Version,
Kind: uResource.GetObjectKind().GroupVersionKind().Kind,
Name: uResource.GetName(),
}
possibleCROs[croKey] = true // selected by the object itself
}
}
}

filteredCRO := make([]*placementv1alpha1.ClusterResourceOverrideSnapshot, 0, len(croList.Items))
filteredRO := make([]*placementv1alpha1.ResourceOverrideSnapshot, 0, len(roList.Items))
for i := range croList.Items {
for _, selector := range croList.Items[i].Spec.OverrideSpec.ClusterResourceSelectors {
croKey := placementv1beta1.ResourceIdentifier{
Group: selector.Group,
Version: selector.Version,
Kind: selector.Kind,
Name: selector.Name,
}
if possibleCROs[croKey] {
filteredCRO = append(filteredCRO, &croList.Items[i])
break
}
}
}
for i := range roList.Items {
for _, selector := range roList.Items[i].Spec.OverrideSpec.ResourceSelectors {
roKey := placementv1beta1.ResourceIdentifier{
Group: selector.Group,
Version: selector.Version,
Kind: selector.Kind,
Namespace: roList.Items[i].Namespace,
Name: selector.Name,
}
if possibleROs[roKey] {
filteredRO = append(filteredRO, &roList.Items[i])
break
}
}
}
return filteredCRO, filteredRO, nil
return controller.FetchAllMatchOverridesForResourceSnapshot(ctx, r.Client, r.InformerManager, crp, masterResourceSnapshot)
}

// pickFromResourceMatchedOverridesForTargetCluster will look for any overrides associated with the "Bound" or "Scheduled" binding.
Expand All @@ -131,78 +24,5 @@ func (r *Reconciler) fetchAllMatchingOverridesForResourceSnapshot(ctx context.Co
// It returns names of cro and ro attached to the target cluster, and they're ordered by its namespace (if present) and
// then name.
func (r *Reconciler) pickFromResourceMatchedOverridesForTargetCluster(ctx context.Context, binding *placementv1beta1.ClusterResourceBinding, croList []*placementv1alpha1.ClusterResourceOverrideSnapshot, roList []*placementv1alpha1.ResourceOverrideSnapshot) ([]string, []placementv1beta1.NamespacedName, error) {
if len(croList) == 0 && len(roList) == 0 {
return nil, nil, nil
}

cluster := clusterv1beta1.MemberCluster{}
if err := r.Client.Get(ctx, types.NamespacedName{Name: binding.Spec.TargetCluster}, &cluster); err != nil {
if apierrors.IsNotFound(err) {
klog.V(2).InfoS("MemberCluster has been deleted and we expect that scheduler will update the spec of binding to unscheduled", "memberCluster", binding.Spec.TargetCluster, "clusterResourceBinding", klog.KObj(binding))
return nil, nil, controller.NewExpectedBehaviorError(err)
}
klog.ErrorS(err, "Failed to get the memberCluster", "memberCluster", binding.Spec.TargetCluster, "clusterResourceBinding", klog.KObj(binding))
return nil, nil, controller.NewAPIServerError(true, err)
}

croFiltered := make([]*placementv1alpha1.ClusterResourceOverrideSnapshot, 0, len(croList))
for i, cro := range croList {
matched, err := isClusterMatched(cluster, cro.Spec.OverrideSpec.Policy)
if err != nil {
klog.ErrorS(err, "Invalid clusterResourceOverride", "clusterResourceOverride", klog.KObj(cro))
return nil, nil, controller.NewUnexpectedBehaviorError(err)
}
if matched {
croFiltered = append(croFiltered, croList[i])
}
}
// There are no priority for now and sort the cro list by its name.
sort.SliceStable(croFiltered, func(i, j int) bool {
return croFiltered[i].Name < croFiltered[j].Name
})

roFiltered := make([]*placementv1alpha1.ResourceOverrideSnapshot, 0, len(roList))
for i, ro := range roList {
matched, err := isClusterMatched(cluster, ro.Spec.OverrideSpec.Policy)
if err != nil {
klog.ErrorS(err, "Invalid resourceOverride", "resourceOverride", klog.KObj(ro))
return nil, nil, controller.NewUnexpectedBehaviorError(err)
}
if matched {
roFiltered = append(roFiltered, roList[i])
}
}
// There are no priority for now and sort the ro list by its namespace and then name.
sort.SliceStable(roFiltered, func(i, j int) bool {
if roFiltered[i].Namespace == roFiltered[j].Namespace {
return roFiltered[i].Name < roFiltered[j].Name
}
return roFiltered[i].Namespace < roFiltered[j].Namespace
})
croNames := make([]string, len(croFiltered))
for i, o := range croFiltered {
croNames[i] = o.Name
}
roNames := make([]placementv1beta1.NamespacedName, len(roFiltered))
for i, o := range roFiltered {
roNames[i] = placementv1beta1.NamespacedName{Name: o.Name, Namespace: o.Namespace}
}
klog.V(2).InfoS("Found matched overrides for the binding", "binding", klog.KObj(binding), "matchedCROCount", len(croNames), "matchedROCount", len(roNames))
return croNames, roNames, nil
}

func isClusterMatched(cluster clusterv1beta1.MemberCluster, policy *placementv1alpha1.OverridePolicy) (bool, error) {
if policy == nil {
return false, errors.New("policy is nil")
}
for _, rule := range policy.OverrideRules {
matched, err := overrider.IsClusterMatched(cluster, rule)
if err != nil {
return false, err
}
if matched {
return true, nil
}
}
return false, nil
return controller.PickFromResourceMatchedOverridesForTargetCluster(ctx, r.Client, binding.Spec.TargetCluster, croList, roList)
}
Loading

0 comments on commit 9b8484f

Please sign in to comment.