Skip to content

Commit 6727c44

Browse files
jwttyjwtty
committed
fix comments
1 parent 3919885 commit 6727c44

File tree

4 files changed

+152
-73
lines changed

4 files changed

+152
-73
lines changed

‎pkg/controllers/updaterun/controller_integration_test.go‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -259,6 +259,9 @@ func getTestClusterSchedulingPolicySnapshot(idx int) *placementv1beta1.ClusterSc
259259
},
260260
},
261261
Spec: placementv1beta1.SchedulingPolicySnapshotSpec{
262+
Policy: &placementv1beta1.PlacementPolicy{
263+
PlacementType: placementv1beta1.PickNPlacementType,
264+
},
262265
PolicyHash: []byte("hash"),
263266
},
264267
}

‎pkg/controllers/updaterun/initialization.go‎

Lines changed: 39 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -117,21 +117,34 @@ func (r *Reconciler) determinePolicySnapshot(
117117

118118
latestPolicySnapshot := policySnapshotList.Items[0]
119119
updateRun.Status.PolicySnapshotIndexUsed = latestPolicySnapshot.Name
120+
120121
// Get the cluster count from the policy snapshot.
121-
clusterCount, err := annotations.ExtractNumOfClustersFromPolicySnapshot(&latestPolicySnapshot)
122-
if err != nil {
123-
annErr := fmt.Errorf("%w: the policySnapshot `%s` doesn't have valid cluster count annotation", err, latestPolicySnapshot.Name)
124-
klog.ErrorS(controller.NewUnexpectedBehaviorError(annErr), "Failed to get the cluster count from the latestPolicySnapshot", "clusterResourcePlacement", placementName, "latestPolicySnapshot", latestPolicySnapshot.Name, "clusterStagedUpdateRun", updateRunRef)
122+
if latestPolicySnapshot.Spec.Policy == nil {
123+
nopolicyErr := fmt.Errorf("policy snapshot `%s` does not have a policy", latestPolicySnapshot.Name)
124+
klog.ErrorS(controller.NewUnexpectedBehaviorError(nopolicyErr), "Failed to get the policy from the latestPolicySnapshot", "clusterResourcePlacement", placementName, "latestPolicySnapshot", latestPolicySnapshot.Name, "clusterStagedUpdateRun", updateRunRef)
125125
// no more retries here.
126-
return nil, -1, fmt.Errorf("%w, %s", errInitializedFailed, annErr.Error())
126+
return nil, -1, fmt.Errorf("%w: %s", errInitializedFailed, nopolicyErr.Error())
127+
}
128+
// for pickAll policy, the observed cluster count is not included in the policy snapshot. We set it to -1. It will be validated in the binding stages.
129+
clusterCount := -1
130+
if latestPolicySnapshot.Spec.Policy.PlacementType == placementv1beta1.PickNPlacementType {
131+
count, err := annotations.ExtractNumOfClustersFromPolicySnapshot(&latestPolicySnapshot)
132+
if err != nil {
133+
annErr := fmt.Errorf("%w: the policy snapshot `%s` doesn't have valid cluster count annotation", err, latestPolicySnapshot.Name)
134+
klog.ErrorS(controller.NewUnexpectedBehaviorError(annErr), "Failed to get the cluster count from the latestPolicySnapshot", "clusterResourcePlacement", placementName, "latestPolicySnapshot", latestPolicySnapshot.Name, "clusterStagedUpdateRun", updateRunRef)
135+
// no more retries here.
136+
return nil, -1, fmt.Errorf("%w, %s", errInitializedFailed, annErr.Error())
137+
}
138+
clusterCount = count
139+
} else if latestPolicySnapshot.Spec.Policy.PlacementType == placementv1beta1.PickFixedPlacementType {
140+
clusterCount = len(latestPolicySnapshot.Spec.Policy.ClusterNames)
127141
}
128142
updateRun.Status.PolicyObservedClusterCount = clusterCount
129143
klog.V(2).InfoS("Found the latest policy snapshot", "latestPolicySnapshot", latestPolicySnapshot.Name, "observed cluster count", updateRun.Status.PolicyObservedClusterCount, "clusterStagedUpdateRun", updateRunRef)
130144

131145
if !condition.IsConditionStatusTrue(latestPolicySnapshot.GetCondition(string(placementv1beta1.PolicySnapshotScheduled)), latestPolicySnapshot.Generation) {
132146
scheduleErr := fmt.Errorf("policy snapshot `%s` not fully scheduled yet", latestPolicySnapshot.Name)
133147
klog.ErrorS(scheduleErr, "The policy snapshot is not scheduled successfully", "clusterResourcePlacement", placementName, "latestPolicySnapshot", latestPolicySnapshot.Name, "clusterStagedUpdateRun", updateRunRef)
134-
// hmmmm, should we retry and see if the policy snapshot is scheduled later?
135148
return nil, -1, fmt.Errorf("%w: %s", errInitializedFailed, scheduleErr.Error())
136149
}
137150
return &latestPolicySnapshot, clusterCount, nil
@@ -159,8 +172,8 @@ func (r *Reconciler) collectScheduledClusters(
159172
var toBeDeletedBindings, selectedBindings []*placementv1beta1.ClusterResourceBinding
160173
for i, binding := range bindingList.Items {
161174
if binding.Spec.SchedulingPolicySnapshotName == latestPolicySnapshot.Name {
162-
if binding.Spec.State != placementv1beta1.BindingStateScheduled {
163-
return nil, nil, controller.NewUnexpectedBehaviorError(fmt.Errorf("binding `%s`'s state %s is not scheduled", binding.Name, binding.Spec.State))
175+
if binding.Spec.State != placementv1beta1.BindingStateScheduled && binding.Spec.State != placementv1beta1.BindingStateBound {
176+
return nil, nil, controller.NewUnexpectedBehaviorError(fmt.Errorf("binding `%s`'s state %s is not scheduled or bound", binding.Name, binding.Spec.State))
164177
}
165178
klog.V(2).InfoS("Found a scheduled binding", "binding", binding.Name, "clusterResourcePlacement", placementName, "latestPolicySnapshot", latestPolicySnapshot.Name, "clusterStagedUpdateRun", updateRunRef)
166179
selectedBindings = append(selectedBindings, &bindingList.Items[i])
@@ -170,10 +183,9 @@ func (r *Reconciler) collectScheduledClusters(
170183
}
171184
}
172185

173-
// should this be a valid case?
174-
if len(selectedBindings) == 0 {
175-
nobindingErr := fmt.Errorf("no scheduled clusterResourceBindings found for the latest policy snapshot %s", latestPolicySnapshot.Name)
176-
klog.ErrorS(nobindingErr, "Failed to find the scheduled clusterResourceBindings", "clusterResourcePlacement", placementName, "latestPolicySnapshot", latestPolicySnapshot.Name, "clusterStagedUpdateRun", updateRunRef)
186+
if len(selectedBindings) == 0 && len(toBeDeletedBindings) == 0 {
187+
nobindingErr := fmt.Errorf("no scheduled or to-be-deleted clusterResourceBindings found for the latest policy snapshot %s", latestPolicySnapshot.Name)
188+
klog.ErrorS(nobindingErr, "Failed to collect clusterResourceBindings", "clusterResourcePlacement", placementName, "latestPolicySnapshot", latestPolicySnapshot.Name, "clusterStagedUpdateRun", updateRunRef)
177189
// no more retries here.
178190
return nil, nil, fmt.Errorf("%w: %s", errInitializedFailed, nobindingErr.Error())
179191
}
@@ -288,24 +300,22 @@ func (r *Reconciler) computeRunStageStatus(
288300

289301
// Check if the stage is empty.
290302
if len(curStageClusters) == 0 {
291-
emptyErr := fmt.Errorf("stage `%s` has no clusters selected", stage.Name)
292-
klog.ErrorS(emptyErr, "No cluster is selected for the stage", "clusterStagedUpdateStrategy", updateStrategyName, "stage name", stage.Name, "clusterStagedUpdateRun", updateRunRef)
293-
// no more retries.
294-
return fmt.Errorf("%w: %s", errInitializedFailed, emptyErr.Error())
295-
}
296-
297-
// Sort the clusters in the stage based on the SortingLabelKey and cluster name.
298-
sort.Slice(curStageClusters, func(i, j int) bool {
299-
if stage.SortingLabelKey == nil {
303+
// since we allow no selected bindings, a stage can be empty.
304+
klog.InfoS("No cluster is selected for the stage", "clusterStagedUpdateStrategy", updateStrategyName, "stage name", stage.Name, "clusterStagedUpdateRun", updateRunRef)
305+
} else {
306+
// Sort the clusters in the stage based on the SortingLabelKey and cluster name.
307+
sort.Slice(curStageClusters, func(i, j int) bool {
308+
if stage.SortingLabelKey == nil {
309+
return curStageClusters[i].Name < curStageClusters[j].Name
310+
}
311+
labelI, _ := strconv.Atoi(curStageClusters[i].Labels[*stage.SortingLabelKey])
312+
labelJ, _ := strconv.Atoi(curStageClusters[j].Labels[*stage.SortingLabelKey])
313+
if labelI != labelJ {
314+
return labelI < labelJ
315+
}
300316
return curStageClusters[i].Name < curStageClusters[j].Name
301-
}
302-
labelI, _ := strconv.Atoi(curStageClusters[i].Labels[*stage.SortingLabelKey])
303-
labelJ, _ := strconv.Atoi(curStageClusters[j].Labels[*stage.SortingLabelKey])
304-
if labelI != labelJ {
305-
return labelI < labelJ
306-
}
307-
return curStageClusters[i].Name < curStageClusters[j].Name
308-
})
317+
})
318+
}
309319

310320
// Record the clusters in the stage.
311321
curStageUpdatingStatus.Clusters = make([]placementv1alpha1.ClusterUpdatingStatus, len(curStageClusters))

‎pkg/controllers/updaterun/initialization_integration_test.go‎

Lines changed: 96 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -35,7 +35,7 @@ var (
3535
}
3636
)
3737

38-
var _ = Describe("Updaterun initialization tests", func() {
38+
var _ = FDescribe("Updaterun initialization tests", func() {
3939
var updateRun *placementv1alpha1.ClusterStagedUpdateRun
4040
var crp *placementv1beta1.ClusterResourcePlacement
4141
var policySnapshot *placementv1beta1.ClusterSchedulingPolicySnapshot
@@ -210,6 +210,18 @@ var _ = Describe("Updaterun initialization tests", func() {
210210
Expect(k8sClient.Delete(ctx, snapshot2)).Should(Succeed())
211211
})
212212

213+
It("Should failt to initialize if the latest policy snapshot has a nil policy", func() {
214+
By("Creating scheduling policy snapshot with nil policy")
215+
policySnapshot.Spec.Policy = nil
216+
Expect(k8sClient.Create(ctx, policySnapshot)).To(Succeed())
217+
218+
By("Creating a new clusterStagedUpdateRun")
219+
Expect(k8sClient.Create(ctx, updateRun)).To(Succeed())
220+
221+
By("Validating the initialization failed")
222+
validateFailedInitCondition(ctx, updateRun, "does not have a policy")
223+
})
224+
213225
It("Should fail to initialize if the latest policy snapshot does not have valid cluster count annotation", func() {
214226
By("Creating scheduling policy snapshot with invalid cluster count annotation")
215227
delete(policySnapshot.Annotations, placementv1beta1.NumberOfClustersAnnotation)
@@ -233,8 +245,8 @@ var _ = Describe("Updaterun initialization tests", func() {
233245
validateFailedInitCondition(ctx, updateRun, "not fully scheduled yet")
234246
})
235247

236-
It("Should copy the latest policy snapshot details to the updateRun status", func() {
237-
By("Creating scheduling policy snapshot")
248+
It("Should copy the latest policy snapshot details to the updateRun status -- pickN policy", func() {
249+
By("Creating scheduling policy snapshot with pickN policy")
238250
Expect(k8sClient.Create(ctx, policySnapshot)).To(Succeed())
239251

240252
By("Set the latest policy snapshot condition as fully scheduled")
@@ -263,6 +275,71 @@ var _ = Describe("Updaterun initialization tests", func() {
263275
return nil
264276
}, timeout, interval).Should(Succeed(), "failed to update the updateRun status with policy snapshot details")
265277
})
278+
279+
It("Should copy the latest policy snapshot details to the updateRun status -- pickFixed policy", func() {
280+
By("Creating scheduling policy snapshot with pickFixed policy")
281+
policySnapshot.Spec.Policy.PlacementType = placementv1beta1.PickFixedPlacementType
282+
policySnapshot.Spec.Policy.ClusterNames = []string{"cluster-0", "cluster-1"}
283+
Expect(k8sClient.Create(ctx, policySnapshot)).To(Succeed())
284+
285+
By("Set the latest policy snapshot condition as fully scheduled")
286+
meta.SetStatusCondition(&policySnapshot.Status.Conditions, metav1.Condition{
287+
Type: string(placementv1beta1.PolicySnapshotScheduled),
288+
Status: metav1.ConditionTrue,
289+
ObservedGeneration: policySnapshot.Generation,
290+
Reason: "scheduled",
291+
})
292+
Expect(k8sClient.Status().Update(ctx, policySnapshot)).Should(Succeed(), "failed to update the policy snapshot condition")
293+
294+
By("Creating a new clusterStagedUpdateRun")
295+
Expect(k8sClient.Create(ctx, updateRun)).To(Succeed())
296+
297+
By("Validating the initialization failed")
298+
Eventually(func() error {
299+
if err := k8sClient.Get(ctx, updateRunNamespacedName, updateRun); err != nil {
300+
return err
301+
}
302+
if updateRun.Status.PolicySnapshotIndexUsed != policySnapshot.Name {
303+
return fmt.Errorf("updateRun status `PolicySnapshotIndexUsed` mismatch: got %s, want %s", updateRun.Status.PolicySnapshotIndexUsed, policySnapshot.Name)
304+
}
305+
if updateRun.Status.PolicyObservedClusterCount != 2 {
306+
return fmt.Errorf("updateRun status `PolicyObservedClusterCount` mismatch: got %d, want %d", updateRun.Status.PolicyObservedClusterCount, 2)
307+
}
308+
return nil
309+
}, timeout, interval).Should(Succeed(), "failed to update the updateRun status with policy snapshot details")
310+
})
311+
312+
It("Should copy the latest policy snapshot details to the updateRun status -- pickAll policy", func() {
313+
By("Creating scheduling policy snapshot with pickAll policy")
314+
policySnapshot.Spec.Policy.PlacementType = placementv1beta1.PickAllPlacementType
315+
Expect(k8sClient.Create(ctx, policySnapshot)).To(Succeed())
316+
317+
By("Set the latest policy snapshot condition as fully scheduled")
318+
meta.SetStatusCondition(&policySnapshot.Status.Conditions, metav1.Condition{
319+
Type: string(placementv1beta1.PolicySnapshotScheduled),
320+
Status: metav1.ConditionTrue,
321+
ObservedGeneration: policySnapshot.Generation,
322+
Reason: "scheduled",
323+
})
324+
Expect(k8sClient.Status().Update(ctx, policySnapshot)).Should(Succeed(), "failed to update the policy snapshot condition")
325+
326+
By("Creating a new clusterStagedUpdateRun")
327+
Expect(k8sClient.Create(ctx, updateRun)).To(Succeed())
328+
329+
By("Validating the initialization failed")
330+
Eventually(func() error {
331+
if err := k8sClient.Get(ctx, updateRunNamespacedName, updateRun); err != nil {
332+
return err
333+
}
334+
if updateRun.Status.PolicySnapshotIndexUsed != policySnapshot.Name {
335+
return fmt.Errorf("updateRun status `PolicySnapshotIndexUsed` mismatch: got %s, want %s", updateRun.Status.PolicySnapshotIndexUsed, policySnapshot.Name)
336+
}
337+
if updateRun.Status.PolicyObservedClusterCount != -1 {
338+
return fmt.Errorf("updateRun status `PolicyObservedClusterCount` mismatch: got %d, want %d", updateRun.Status.PolicyObservedClusterCount, -1)
339+
}
340+
return nil
341+
}, timeout, interval).Should(Succeed(), "failed to update the updateRun status with policy snapshot details")
342+
})
266343
})
267344

268345
Context("Test collectScheduledClusters", func() {
@@ -283,15 +360,28 @@ var _ = Describe("Updaterun initialization tests", func() {
283360
Expect(k8sClient.Status().Update(ctx, policySnapshot)).Should(Succeed(), "failed to update the policy snapshot condition")
284361
})
285362

286-
It("Should fail to initialize if there is no selected cluster", func() {
363+
It("Should fail to initialize if there is no selected or to-be-deleted cluster", func() {
287364
By("Creating a new clusterStagedUpdateRun")
288365
Expect(k8sClient.Create(ctx, updateRun)).To(Succeed())
289366

290367
By("Validating the initialization failed")
291-
validateFailedInitCondition(ctx, updateRun, "no scheduled clusterResourceBindings found")
368+
validateFailedInitCondition(ctx, updateRun, "no scheduled or to-be-deleted clusterResourceBindings found")
369+
})
370+
371+
It("Should not report error if there are only to-be-deleted clusters", func() {
372+
By("Creating a to-be-deleted clusterResourceBinding")
373+
binding := getTestClusterResourceBinding(policySnapshot.Name+"a", "cluster-0")
374+
Expect(k8sClient.Create(ctx, binding)).To(Succeed())
375+
376+
By("Creating a new clusterStagedUpdateRun")
377+
Expect(k8sClient.Create(ctx, updateRun)).To(Succeed())
378+
379+
By("Validating the initialization not failed due to no selected cluster")
380+
// it should fail due to strategy not found
381+
validateFailedInitCondition(ctx, updateRun, "referenced clusterStagedUpdateStrategy not found")
292382
})
293383

294-
It("Should retry if the bindings are not in Scheduled state", func() {
384+
It("Should retry if the bindings are not in Scheduled or Bound state", func() {
295385
By("Creating a not scheduled clusterResourceBinding")
296386
binding := getTestClusterResourceBinding(policySnapshot.Name, "cluster-1")
297387
binding.Spec.State = ""
@@ -410,20 +500,6 @@ var _ = Describe("Updaterun initialization tests", func() {
410500
validateFailedInitCondition(ctx, updateRun, "the sorting label `not-exist-label:`")
411501
})
412502

413-
It("Should fail to initialize if some stage is completely empty", func() {
414-
By("Creating a clusterStagedUpdateStrategy stage that can select no clusters")
415-
updateStrategy.Spec.Stages[0].LabelSelector = &metav1.LabelSelector{
416-
MatchLabels: map[string]string{"group": "not-exist"},
417-
}
418-
Expect(k8sClient.Create(ctx, updateStrategy)).To(Succeed())
419-
420-
By("Creating a new clusterStagedUpdateRun")
421-
Expect(k8sClient.Create(ctx, updateRun)).To(Succeed())
422-
423-
By("Validating the initialization failed")
424-
validateFailedInitCondition(ctx, updateRun, "stage `stage1` has no clusters selected")
425-
})
426-
427503
It("Should fail to initialize if some cluster appears in multiple stages", func() {
428504
By("Creating a clusterStagedUpdateStrategy with overlapping stages")
429505
updateStrategy.Spec.Stages[1].LabelSelector = &metav1.LabelSelector{

‎pkg/controllers/updaterun/suite_test.go‎

Lines changed: 14 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,7 @@ import (
1414
. "github.com/onsi/ginkgo/v2"
1515
. "github.com/onsi/gomega"
1616

17-
"k8s.io/apimachinery/pkg/runtime/schema"
17+
"k8s.io/client-go/dynamic"
1818
"k8s.io/client-go/kubernetes/scheme"
1919
"k8s.io/client-go/rest"
2020
"k8s.io/klog/v2"
@@ -29,7 +29,8 @@ import (
2929
clusterv1beta1 "go.goms.io/fleet/apis/cluster/v1beta1"
3030
placementv1alpha1 "go.goms.io/fleet/apis/placement/v1alpha1"
3131
fleetv1beta1 "go.goms.io/fleet/apis/placement/v1beta1"
32-
"go.goms.io/fleet/test/utils/informer"
32+
"go.goms.io/fleet/pkg/utils"
33+
"go.goms.io/fleet/pkg/utils/informer"
3334
)
3435

3536
var (
@@ -88,31 +89,20 @@ var _ = BeforeSuite(func() {
8889
By("set k8s client same as the controller manager")
8990
k8sClient = mgr.GetClient()
9091

91-
// setup our main reconciler
92-
fakeInformer := &informer.FakeManager{
93-
APIResources: map[schema.GroupVersionKind]bool{
94-
{
95-
Group: "",
96-
Version: "v1",
97-
Kind: "Service",
98-
}: true,
99-
{
100-
Group: "",
101-
Version: "v1",
102-
Kind: "Deployment",
103-
}: true,
104-
{
105-
Group: "",
106-
Version: "v1",
107-
Kind: "Secret",
108-
}: true,
109-
},
110-
IsClusterScopedResource: false,
111-
}
92+
// setup informer manager for the reconciler
93+
dynamicClient, err := dynamic.NewForConfig(cfg)
94+
Expect(err).Should(Succeed())
95+
dynamicInformerManager := informer.NewInformerManager(dynamicClient, 0, ctx.Done())
96+
dynamicInformerManager.AddStaticResource(informer.APIResourceMeta{
97+
GroupVersionKind: utils.NamespaceGVK,
98+
GroupVersionResource: utils.NamespaceGVR,
99+
IsClusterScoped: true,
100+
}, nil)
112101

102+
// setup our main reconciler
113103
err = (&Reconciler{
114104
Client: k8sClient,
115-
InformerManager: fakeInformer,
105+
InformerManager: dynamicInformerManager,
116106
}).SetupWithManager(mgr)
117107
Expect(err).Should(Succeed())
118108

0 commit comments

Comments
 (0)