forked from juju/juju
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add watcher to upgrade that notifies when upgrade can start
Instead of, as we are with mongo, watching for any changes and then requiring the consumer to verify it's an actionable change, create a new watcher extending NamespaceWatcher which does this verification for you This combines Watch and AllProvisionedControllersReady from state/upgrade Test this by mocking the underlying watcher. Ideally, we would have some tests watching a real database, but we are not able to do this at the moment.
- Loading branch information
1 parent
ee5d6dc
commit 930e2a5
Showing
6 changed files
with
262 additions
and
34 deletions.
There are no files selected for viewing
Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,105 @@ | ||
| // Copyright 2023 Canonical Ltd. | ||
| // Licensed under the AGPLv3, see LICENCE file for details. | ||
|
|
||
| package service | ||
|
|
||
| import ( | ||
| "context" | ||
|
|
||
| "github.com/juju/errors" | ||
| "github.com/juju/worker/v3/catacomb" | ||
|
|
||
| "github.com/juju/juju/core/changestream" | ||
| "github.com/juju/juju/core/watcher" | ||
| ) | ||
|
|
||
| type upgradeReadyWatcher struct { | ||
| ctx context.Context | ||
| catacomb catacomb.Catacomb | ||
|
|
||
| st State | ||
| upgradeUUID string | ||
|
|
||
| in <-chan []string | ||
| out chan struct{} | ||
| } | ||
|
|
||
| // NewUpgradeReadyWatcher creates a watcher which notifies when all controller | ||
| // nodes have been registered, meaning the upgrade is ready to start | ||
| func NewUpgradeReadyWatcher(ctx context.Context, st State, wf WatcherFactory, upgradeUUID string) (watcher.NotifyWatcher, error) { | ||
| namespaceWatcher, err := wf.NewNamespaceWatcher("upgrade_info_controller_node", changestream.Create|changestream.Update, "") | ||
| if err != nil { | ||
| return nil, errors.Trace(err) | ||
| } | ||
|
|
||
| w := &upgradeReadyWatcher{ | ||
| ctx: ctx, | ||
| st: st, | ||
| upgradeUUID: upgradeUUID, | ||
| out: make(chan struct{}), | ||
| in: namespaceWatcher.Changes(), | ||
| } | ||
|
|
||
| if err := catacomb.Invoke(catacomb.Plan{ | ||
| Site: &w.catacomb, | ||
| Work: w.loop, | ||
| }); err != nil { | ||
| return nil, errors.Trace(err) | ||
| } | ||
|
|
||
| if err := w.catacomb.Add(namespaceWatcher); err != nil { | ||
| return nil, errors.Trace(err) | ||
| } | ||
|
|
||
| return w, nil | ||
| } | ||
|
|
||
| func (w *upgradeReadyWatcher) Kill() { | ||
| w.catacomb.Kill(nil) | ||
| } | ||
|
|
||
| func (w *upgradeReadyWatcher) Wait() error { | ||
| return w.catacomb.Wait() | ||
| } | ||
|
|
||
| func (w *upgradeReadyWatcher) Changes() <-chan struct{} { | ||
| return w.out | ||
| } | ||
|
|
||
| func (w *upgradeReadyWatcher) loop() error { | ||
| defer close(w.out) | ||
|
|
||
| // By reassigning the in and out channels, we effectively ticktock between | ||
| // read mode and dispatch mode. This ensures we always dispatch deltas that | ||
| // we received before reading more, and every channel read/write is guarded | ||
| // by checks of the tomb and subscription liveness. | ||
| // Start in read mode so we don't send an erroneous initial message | ||
| var out chan struct{} | ||
| in := w.in | ||
|
|
||
| for { | ||
| select { | ||
| case <-w.catacomb.Dying(): | ||
| return w.catacomb.ErrDying() | ||
| case <-w.ctx.Done(): | ||
| w.catacomb.Kill(context.Cause(w.ctx)) | ||
| case _, ok := <-in: | ||
| if !ok { | ||
| return nil | ||
| } | ||
| ready, err := w.st.AllProvisionedControllersReady(w.ctx, w.upgradeUUID) | ||
| if err != nil { | ||
| return errors.Trace(err) | ||
| } | ||
| if ready { | ||
| // Tick over to dispatch mode. | ||
| in = nil | ||
| out = w.out | ||
| } | ||
| case out <- struct{}{}: | ||
| // We have dispatched. Tick over to read mode. | ||
| in = w.in | ||
| out = nil | ||
| } | ||
| } | ||
| } |
Oops, something went wrong.