Skip to content

Bug 1916384: pkg/cvo: Use shutdownContext for final status synchronization #501

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 4, 2021
Merged

Bug 1916384: pkg/cvo: Use shutdownContext for final status synchronization #501

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
29 changes: 17 additions & 12 deletions pkg/cvo/cvo.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -292,38 +292,43 @@ func loadConfigMapVerifierDataFromUpdate(update *payload.Update, clientBuilder s
return verifier, persister, nil
}

// Run runs the cluster version operator until stopCh is completed. Workers is ignored for now.
func (optr *Operator) Run(ctx context.Context, workers int) error {
// Run runs the cluster version operator until runContext.Done() and
// then attempts a clean shutdown limited by shutdownContext.Done().
// Assumes runContext.Done() occurs before or simultaneously with
// shutdownContext.Done().
func (optr *Operator) Run(runContext context.Context, shutdownContext context.Context, workers int) error {
defer optr.queue.ShutDown()
stopCh := ctx.Done()
stopCh := runContext.Done()
workerStopCh := make(chan struct{})

klog.Infof("Starting ClusterVersionOperator with minimum reconcile period %s", optr.minimumUpdateCheckInterval)
defer klog.Info("Shutting down ClusterVersionOperator")

if !cache.WaitForCacheSync(stopCh, optr.cacheSynced...) {
return fmt.Errorf("caches never synchronized: %w", ctx.Err())
return fmt.Errorf("caches never synchronized: %w", runContext.Err())
}

// trigger the first cluster version reconcile always
optr.queue.Add(optr.queueKey())

// start the config sync loop, and have it notify the queue when new status is detected
go runThrottledStatusNotifier(ctx, optr.statusInterval, 2, optr.configSync.StatusCh(), func() { optr.queue.Add(optr.queueKey()) })
go optr.configSync.Start(ctx, 16, optr.name, optr.cvLister)
go wait.UntilWithContext(ctx, func(ctx context.Context) { optr.worker(ctx, optr.availableUpdatesQueue, optr.availableUpdatesSync) }, time.Second)
go wait.UntilWithContext(ctx, func(ctx context.Context) { optr.worker(ctx, optr.upgradeableQueue, optr.upgradeableSync) }, time.Second)
go wait.UntilWithContext(ctx, func(ctx context.Context) {
go runThrottledStatusNotifier(runContext, optr.statusInterval, 2, optr.configSync.StatusCh(), func() { optr.queue.Add(optr.queueKey()) })
go optr.configSync.Start(runContext, 16, optr.name, optr.cvLister)
go wait.UntilWithContext(runContext, func(runContext context.Context) {
optr.worker(runContext, optr.availableUpdatesQueue, optr.availableUpdatesSync)
}, time.Second)
go wait.UntilWithContext(runContext, func(runContext context.Context) { optr.worker(runContext, optr.upgradeableQueue, optr.upgradeableSync) }, time.Second)
go wait.UntilWithContext(runContext, func(runContext context.Context) {
defer close(workerStopCh)

// run the worker, then when the queue is closed sync one final time to flush any pending status
optr.worker(ctx, optr.queue, func(ctx context.Context, key string) error { return optr.sync(ctx, key) })
if err := optr.sync(ctx, optr.queueKey()); err != nil {
optr.worker(runContext, optr.queue, func(runContext context.Context, key string) error { return optr.sync(runContext, key) })
if err := optr.sync(shutdownContext, optr.queueKey()); err != nil {
Copy link
Member

@LalatenduMohanty LalatenduMohanty Jan 20, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can we add a unit test for this change?

Copy link
Member Author

@wking wking Jan 20, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can we add a unit test for this change?

I'm not sure... We can probably add an integration (e2e-agnostic-operator) test for it. Do we want to hold up this PR while I work that out, or can I defer it to follow-up work?

Copy link
Member

@LalatenduMohanty LalatenduMohanty Jan 27, 2021 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It depends how much time we need to add the test. If it is just one day then we should wait for the test. The issue with merging it without tests is that I am not confident if this fixes the issue or not and having a test for it will make me more confident to merge it.

utilruntime.HandleError(fmt.Errorf("unable to perform final sync: %v", err))
}
}, time.Second)
if optr.signatureStore != nil {
go optr.signatureStore.Run(ctx, optr.minimumUpdateCheckInterval*2)
go optr.signatureStore.Run(runContext, optr.minimumUpdateCheckInterval*2)
}

<-stopCh
Expand Down
2 changes: 1 addition & 1 deletion pkg/start/start.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -237,7 +237,7 @@ func (o *Options) run(ctx context.Context, controllerCtx *Context, lock *resourc
resultChannelCount++
go func() {
defer utilruntime.HandleCrash()
err := controllerCtx.CVO.Run(runContext, 2)
err := controllerCtx.CVO.Run(runContext, shutdownContext, 2)
resultChannel <- asyncResult{name: "main operator", error: err}
}()

Expand Down