Skip to content

Bug 1822752: pkg/cvo: Fix ups from separating load from apply #683 #745

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
openshift-merge-robot merged 1 commit into from
Feb 22, 2022
Merged

Bug 1822752: pkg/cvo: Fix ups from separating load from apply #683 #745

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion pkg/cvo/cvo.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -575,7 +575,7 @@ func (optr *Operator) sync(ctx context.Context, key string) error {
}

// inform the config sync loop about our desired state
status := optr.configSync.Update(ctx, config.Generation, desired, config.Spec.Overrides, state, optr.name, optr.cvLister)
status := optr.configSync.Update(ctx, config.Generation, desired, config.Spec.Overrides, state, optr.name)

// write cluster version status
return optr.syncStatus(ctx, original, config, status, errs)
Expand Down
32 changes: 15 additions & 17 deletions pkg/cvo/cvo_scenarios_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -204,12 +204,12 @@ func TestCVO_StartupAndSync(t *testing.T) {
t.Fatal(err)
}
actions = client.Actions()
if len(actions) != 3 {
if len(actions) != 2 {
t.Fatalf("%s", spew.Sdump(actions))
}
expectGet(t, actions[0], "clusterversions", "", "version")
actual = cvs["version"].(*configv1.ClusterVersion)
expectUpdateStatus(t, actions[2], "clusterversions", "", &configv1.ClusterVersion{
expectUpdateStatus(t, actions[1], "clusterversions", "", &configv1.ClusterVersion{
ObjectMeta: metav1.ObjectMeta{
Name: "version",
Generation: 1,
Expand Down Expand Up @@ -593,12 +593,12 @@ func TestCVO_StartupAndSyncUnverifiedPayload(t *testing.T) {
t.Fatal(err)
}
actions = client.Actions()
if len(actions) != 3 {
if len(actions) != 2 {
t.Fatalf("%s", spew.Sdump(actions))
}
expectGet(t, actions[0], "clusterversions", "", "version")
actual = cvs["version"].(*configv1.ClusterVersion)
expectUpdateStatus(t, actions[2], "clusterversions", "", &configv1.ClusterVersion{
expectUpdateStatus(t, actions[1], "clusterversions", "", &configv1.ClusterVersion{
ObjectMeta: metav1.ObjectMeta{
Name: "version",
Generation: 1,
Expand Down Expand Up @@ -958,12 +958,12 @@ func TestCVO_StartupAndSyncPreconditionFailing(t *testing.T) {
t.Fatal(err)
}
actions = client.Actions()
if len(actions) != 3 {
if len(actions) != 2 {
t.Fatalf("%s", spew.Sdump(actions))
}
expectGet(t, actions[0], "clusterversions", "", "version")
actual = cvs["version"].(*configv1.ClusterVersion)
expectUpdateStatus(t, actions[2], "clusterversions", "", &configv1.ClusterVersion{
expectUpdateStatus(t, actions[1], "clusterversions", "", &configv1.ClusterVersion{
ObjectMeta: metav1.ObjectMeta{
Name: "version",
Generation: 1,
Expand Down Expand Up @@ -1115,7 +1115,6 @@ func TestCVO_StartupAndSyncPreconditionFailing(t *testing.T) {
},
VersionHash: "DL-FFQ2Uem8=",
History: []configv1.UpdateHistory{
//{State: configv1.PartialUpdate, Image: "image/image:1", Version: "1.0.0-abc", StartedTime: defaultStartedTime},
{State: configv1.CompletedUpdate, Image: "image/image:1", Version: "1.0.0-abc", StartedTime: defaultStartedTime, CompletionTime: &defaultCompletionTime},
},
Conditions: []configv1.ClusterOperatorStatusCondition{
Expand Down Expand Up @@ -1313,12 +1312,12 @@ func TestCVO_UpgradeUnverifiedPayload(t *testing.T) {
},
)
actions = client.Actions()
if len(actions) != 3 {
if len(actions) != 2 {
t.Fatalf("%s", spew.Sdump(actions))
}
expectGet(t, actions[0], "clusterversions", "", "version")
actual := cvs["version"].(*configv1.ClusterVersion)
expectUpdateStatus(t, actions[2], "clusterversions", "", &configv1.ClusterVersion{
expectUpdateStatus(t, actions[1], "clusterversions", "", &configv1.ClusterVersion{
ObjectMeta: metav1.ObjectMeta{
Name: "version",
ResourceVersion: "1",
Expand Down Expand Up @@ -1561,12 +1560,12 @@ func TestCVO_UpgradeUnverifiedPayloadRetrieveOnce(t *testing.T) {
},
)
actions = client.Actions()
if len(actions) != 3 {
if len(actions) != 2 {
t.Fatalf("%s", spew.Sdump(actions))
}
expectGet(t, actions[0], "clusterversions", "", "version")
actual := cvs["version"].(*configv1.ClusterVersion)
expectUpdateStatus(t, actions[2], "clusterversions", "", &configv1.ClusterVersion{
expectUpdateStatus(t, actions[1], "clusterversions", "", &configv1.ClusterVersion{
ObjectMeta: metav1.ObjectMeta{
Name: "version",
ResourceVersion: "1",
Expand Down Expand Up @@ -1660,7 +1659,6 @@ func TestCVO_UpgradeUnverifiedPayloadRetrieveOnce(t *testing.T) {
Image: "image/image:1",
URL: configv1.URL("https://example.com/v1.0.1-abc"),
},
//LastProgress: time.Unix(1, 0),
Generation: 1,
loadPayloadStatus: LoadPayloadStatus{
Step: "PayloadLoaded",
Expand Down Expand Up @@ -2116,11 +2114,11 @@ func TestCVO_UpgradeVerifiedPayload(t *testing.T) {
t.Fatal(err)
}
actions := client.Actions()
if len(actions) != 3 {
if len(actions) != 2 {
t.Fatalf("%s", spew.Sdump(actions))
}
expectGet(t, actions[0], "clusterversions", "", "version")
expectUpdateStatus(t, actions[2], "clusterversions", "", &configv1.ClusterVersion{
expectUpdateStatus(t, actions[1], "clusterversions", "", &configv1.ClusterVersion{
ObjectMeta: metav1.ObjectMeta{
Name: "version",
ResourceVersion: "1",
Expand Down Expand Up @@ -2211,7 +2209,7 @@ func TestCVO_RestartAndReconcile(t *testing.T) {
t.Fatal(err)
}
actions := client.Actions()
if len(actions) != 3 {
if len(actions) != 2 {
t.Fatalf("%s", spew.Sdump(actions))
}
expectGet(t, actions[0], "clusterversions", "", "version")
Expand Down Expand Up @@ -2473,7 +2471,7 @@ func TestCVO_ErrorDuringReconcile(t *testing.T) {
t.Fatal(err)
}
actions := client.Actions()
if len(actions) != 3 {
if len(actions) != 2 {
t.Fatalf("%s", spew.Sdump(actions))
}
expectGet(t, actions[0], "clusterversions", "", "version")
Expand Down Expand Up @@ -2730,7 +2728,7 @@ func TestCVO_ParallelError(t *testing.T) {
t.Fatal(err)
}
actions := client.Actions()
if len(actions) != 3 {
if len(actions) != 2 {
t.Fatalf("%s", spew.Sdump(actions))
}
expectGet(t, actions[0], "clusterversions", "", "version")
Expand Down
2 changes: 1 addition & 1 deletion pkg/cvo/sync_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -413,7 +413,7 @@ func (r *fakeSyncRecorder) StatusCh() <-chan SyncWorkerStatus {
func (r *fakeSyncRecorder) Start(ctx context.Context, maxWorkers int, cvoOptrName string, lister configlistersv1.ClusterVersionLister) {
}

func (r *fakeSyncRecorder) Update(ctx context.Context, generation int64, desired configv1.Update, overrides []configv1.ComponentOverride, state payload.State, cvoOptrName string, lister configlistersv1.ClusterVersionLister) *SyncWorkerStatus {
func (r *fakeSyncRecorder) Update(ctx context.Context, generation int64, desired configv1.Update, overrides []configv1.ComponentOverride, state payload.State, cvoOptrName string) *SyncWorkerStatus {
r.Updates = append(r.Updates, desired)
return r.Returns
}
Expand Down
18 changes: 7 additions & 11 deletions pkg/cvo/sync_worker.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@ import (
// ConfigSyncWorker abstracts how the image is synchronized to the server. Introduced for testing.
type ConfigSyncWorker interface {
Start(ctx context.Context, maxWorkers int, cvoOptrName string, lister configlistersv1.ClusterVersionLister)
Update(ctx context.Context, generation int64, desired configv1.Update, overrides []configv1.ComponentOverride, state payload.State, cvoOptrName string, lister configlistersv1.ClusterVersionLister) *SyncWorkerStatus
Update(ctx context.Context, generation int64, desired configv1.Update, overrides []configv1.ComponentOverride, state payload.State, cvoOptrName string) *SyncWorkerStatus
StatusCh() <-chan SyncWorkerStatus
}

Expand Down Expand Up @@ -214,7 +214,7 @@ func (w *SyncWorker) StatusCh() <-chan SyncWorkerStatus {
return w.report
}

func (w *SyncWorker) syncPayload(ctx context.Context, work *SyncWork, reporter StatusReporter, clusterVersion *configv1.ClusterVersion) error {
func (w *SyncWorker) syncPayload(ctx context.Context, work *SyncWork, reporter StatusReporter) error {
desired := configv1.Release{
Version: work.Desired.Version,
Image: work.Desired.Image,
Expand Down Expand Up @@ -346,15 +346,11 @@ func (w *SyncWorker) syncPayload(ctx context.Context, work *SyncWork, reporter S
}

// loadUpdatedPayload retrieves the image. If successfully retrieved it updates payload otherwise it returns an error.
func (w *SyncWorker) loadUpdatedPayload(ctx context.Context, work *SyncWork, cvoOptrName string, lister configlistersv1.ClusterVersionLister) error {
config, err := lister.Get(cvoOptrName)
if err != nil {
return err
}
func (w *SyncWorker) loadUpdatedPayload(ctx context.Context, work *SyncWork, cvoOptrName string) error {
// reporter hides status updates that occur earlier than the previous failure,
// so that we don't fail, then immediately start reporting an earlier status
reporter := &statusWrapper{w: w, previousStatus: w.status.DeepCopy()}
if err := w.syncPayload(ctx, work, reporter, config); err != nil {
if err := w.syncPayload(ctx, work, reporter); err != nil {
klog.V(2).Infof("loadUpdatedPayload syncPayload err=%v", err)
return err
}
Expand All @@ -366,7 +362,7 @@ func (w *SyncWorker) loadUpdatedPayload(ctx context.Context, work *SyncWork, cvo
// the initial state or whatever the last recorded status was.
// TODO: in the future it may be desirable for changes that alter desired to wait briefly before returning,
// giving the sync loop the opportunity to observe our change and begin working towards it.
func (w *SyncWorker) Update(ctx context.Context, generation int64, desired configv1.Update, overrides []configv1.ComponentOverride, state payload.State, cvoOptrName string, lister configlistersv1.ClusterVersionLister) *SyncWorkerStatus {
func (w *SyncWorker) Update(ctx context.Context, generation int64, desired configv1.Update, overrides []configv1.ComponentOverride, state payload.State, cvoOptrName string) *SyncWorkerStatus {
w.lock.Lock()
defer w.lock.Unlock()

Expand Down Expand Up @@ -423,7 +419,7 @@ func (w *SyncWorker) Update(ctx context.Context, generation int64, desired confi
}

w.lock.Unlock()
err := w.loadUpdatedPayload(ctx, work, cvoOptrName, lister)
err := w.loadUpdatedPayload(ctx, work, cvoOptrName)
w.lock.Lock()
if err != nil {
return w.status.DeepCopy()
Expand Down Expand Up @@ -558,7 +554,7 @@ type statusWrapper struct {
}

func (w *statusWrapper) ValidPayloadStatus(release configv1.Release) bool {
return w.previousStatus.loadPayloadStatus.Release.Image == release.Image
return equalDigest(w.previousStatus.loadPayloadStatus.Release.Image, release.Image)
}

func (w *statusWrapper) ReportPayload(payloadStatus LoadPayloadStatus) {
Expand Down