Permalink
May 2, 2013
Apr 27, 2012
Oct 18, 2016
Apr 27, 2012
Aug 29, 2013
Aug 29, 2013
Aug 29, 2013
Jul 3, 2013
Jul 9, 2013
Sep 4, 2013
Oct 18, 2016
Sep 19, 2012
Aug 29, 2013
Aug 29, 2013
Sep 19, 2012
Jun 29, 2013
Aug 20, 2013
Jun 29, 2013
Jun 29, 2013
Oct 10, 2012
Nov 28, 2012
Jun 29, 2013
Feb 17, 2015
Jun 29, 2013
Dec 8, 2016
Dec 8, 2016
Oct 18, 2016
Oct 18, 2016
Mar 11, 2015
Mar 11, 2015
Mar 19, 2015
Mar 11, 2015
Mar 16, 2015
Mar 19, 2015
Oct 18, 2016
Mar 16, 2015
Mar 14, 2015
Mar 19, 2015
Mar 19, 2015
Mar 14, 2015
Oct 18, 2016
Feb 17, 2015
Mar 16, 2017
Nov 2, 2014
Mar 16, 2017
Nov 19, 2015
Oct 18, 2016
Feb 28, 2017
Oct 24, 2014
Oct 18, 2016
Jan 21, 2015
Nov 19, 2013
Oct 10, 2014
Nov 19, 2013
Jan 21, 2015
Feb 17, 2015
Feb 28, 2017
Feb 21, 2017
Jul 8, 2014
Jul 3, 2013
Jul 21, 2014
Jun 29, 2013
Aug 29, 2013
Aug 29, 2013
Aug 29, 2013
Oct 18, 2016
Feb 17, 2015
Jun 29, 2013
Jun 26, 2013
Jun 24, 2013
Oct 18, 2016
Oct 18, 2016
Oct 18, 2016
Jun 24, 2013
Aug 20, 2013
Dec 17, 2012
Aug 20, 2013
Feb 28, 2017
Sep 19, 2012
Sep 19, 2012
Aug 20, 2013
Aug 20, 2013
Aug 20, 2013
Jul 8, 2014
Jul 8, 2014
Nov 2, 2014
Aug 20, 2013
Oct 28, 2014
Oct 18, 2016
Aug 20, 2013
Sep 25, 2012
Sep 24, 2012
Nov 21, 2016
Nov 18, 2015
Nov 18, 2015
Feb 28, 2017
Sep 24, 2012
Oct 18, 2016
Oct 18, 2016
Sep 18, 2015
Sep 18, 2015
Sep 25, 2012
Oct 18, 2016
Sep 25, 2012
Sep 24, 2012
May 14, 2014
Nov 30, 2012
Mar 25, 2013
Jul 9, 2013
Nov 26, 2012
Nov 30, 2012
Feb 28, 2017
Nov 26, 2012
Nov 28, 2012
Nov 2, 2014
Nov 26, 2012
Nov 26, 2012
Nov 26, 2012
Nov 26, 2012
Nov 26, 2012
Nov 26, 2012
Nov 26, 2012
Jun 24, 2013
Oct 18, 2016
Jul 3, 2013
Mar 16, 2017
Aug 3, 2015
Mar 16, 2017
Oct 18, 2016
Jul 23, 2014
Oct 17, 2014
Oct 18, 2016
Nov 4, 2015
May 12, 2015
Mar 3, 2015
Mar 4, 2015
Jun 10, 2013
Jun 10, 2013
Mar 2, 2015
Feb 28, 2017
Mar 10, 2015
Sep 24, 2015
Sep 24, 2015
Oct 18, 2016
Jun 29, 2013
Oct 2, 2012
Jul 9, 2013
Jul 9, 2013
Sep 26, 2012
Jul 9, 2013
Sep 23, 2014
Sep 23, 2014
Nov 21, 2012
Sep 23, 2014
Sep 23, 2014
Sep 23, 2014
Sep 23, 2014
Oct 18, 2016
Sep 27, 2012
Jun 20, 2014
Jun 20, 2014
Jun 20, 2014
Jun 20, 2014
Oct 18, 2016
Jun 20, 2014
Nov 17, 2014
Feb 28, 2017
Nov 17, 2014
Nov 17, 2014
Nov 17, 2014
Nov 17, 2014
Nov 17, 2014
Nov 17, 2014
Nov 17, 2014
Nov 17, 2014
Oct 18, 2016
Nov 17, 2014
Nov 17, 2014
Oct 18, 2016
Nov 17, 2014
Nov 4, 2015
Nov 4, 2015
Dec 8, 2016
Dec 8, 2016
Dec 8, 2016
Jul 22, 2014
Oct 18, 2016
Jun 19, 2014
Jul 22, 2014
Feb 28, 2017
Jul 8, 2014
Jul 8, 2014
Jul 22, 2014
Oct 30, 2014
Oct 18, 2016
Dec 8, 2016
Nov 4, 2015
Nov 15, 2015
Nov 4, 2015
Nov 20, 2014
Oct 30, 2014
Sep 24, 2014
Nov 16, 2014
Nov 4, 2015
Oct 18, 2016
Oct 18, 2016
Oct 18, 2016
Oct 18, 2016
Oct 18, 2016
Oct 18, 2016
Sep 11, 2014
Oct 18, 2016
Sep 11, 2014
Sep 11, 2014
Oct 18, 2016
Sep 11, 2014
Sep 23, 2014
Sep 16, 2014
Sep 16, 2014
Sep 22, 2014
Aug 29, 2014
Oct 18, 2016
Feb 28, 2017
Jan 21, 2015
Feb 28, 2017
Oct 18, 2016
Jan 26, 2015
Oct 18, 2016
Oct 18, 2016
Feb 28, 2017
Oct 18, 2016
Feb 13, 2017
Feb 13, 2017
Feb 13, 2017
Feb 23, 2017
Feb 28, 2017
Newer
100644
2709 lines (2462 sloc)
76.4 KB
1
// Copyright 2012, 2013 Canonical Ltd.
2
// Licensed under the AGPLv3, see LICENCE file for details.
3
4
package state
5
6
import (
25
26
// TODO(fwereade): 2015-11-18 lp:1517428
27
//
28
// This gets an import block of its own because it's such staggeringly bad
29
// practice. It's here because (1) it always has been, just not quite so
30
// explicitly and (2) even if we had the state watchers implemented as
31
// juju/watcher~s rather than juju/state/watcher~s -- which we don't, so
32
// it's misleading to use those *Chan types etc -- we don't yet have any
33
// ability to transform watcher output in the apiserver layer, so we're
34
// kinda stuck producing what we always have.
35
//
36
// See RelationUnitsWatcher below.
37
"github.com/juju/juju/apiserver/params"
38
)
39
42
// Watcher is implemented by all watchers; the actual
43
// changes channel is returned by a watcher-specific
45
type Watcher interface {
46
// Kill asks the watcher to stop without waiting for it do so.
48
// Wait waits for the watcher to die and returns any
49
// error encountered when it was running.
58
// NotifyWatcher generates signals when something changes, but it does not
59
// return any content for those changes
65
// StringsWatcher generates signals when something changes, returning
66
// the changes as a list of strings.
67
type StringsWatcher interface {
72
// RelationUnitsWatcher generates signals when units enter or leave
73
// the scope of a RelationUnit, and changes to the settings of those
74
// units known to have entered.
75
type RelationUnitsWatcher interface {
76
Watcher
77
78
// Note that it's not very nice exposing a params type directly here. This
79
// is a continuation of existing bad behaviour and not good practice; do
80
// not use this as a model. (FWIW, it used to be in multiwatcher; which is
81
// also api-ey; and the multiwatcher type was used directly in params
82
// anyway.)
83
Changes() <-chan params.RelationUnitsChange
84
}
85
86
// newCommonWatcher exists so that all embedders have a place from which
87
// to get a single TxnLogWatcher that will not be replaced in the lifetime
88
// of the embedder (and also to restrict the width of the interface by
89
// which they can access the rest of State, by storing st as a
90
// modelBackend).
107
// Stop stops the watcher, and returns any error encountered while running
108
// or shutting down.
109
func (w *commonWatcher) Stop() error {
110
w.Kill()
111
return w.Wait()
112
}
113
114
// Kill kills the watcher without waiting for it to shut down.
115
func (w *commonWatcher) Kill() {
117
}
118
119
// Wait waits for the watcher to die and returns any
120
// error encountered when it was running.
121
func (w *commonWatcher) Wait() error {
122
return w.tomb.Wait()
123
}
124
125
// Err returns any error encountered while running or shutting down, or
126
// tomb.ErrStillAlive if the watcher is still running.
127
func (w *commonWatcher) Err() error {
128
return w.tomb.Err()
129
}
130
131
// collect combines the effects of the one change, and any further changes read
132
// from more in the next 10ms. The result map describes the existence, or not,
133
// of every id observed to have changed. If a value is read from the supplied
134
// stop chan, collect returns false immediately.
135
func collect(one watcher.Change, more <-chan watcher.Change, stop <-chan struct{}) (map[interface{}]bool, bool) {
146
select {
147
case <-stop:
148
return nil, false
149
case another := <-more:
150
handle(another)
151
case <-timeout:
155
watchLogger.Tracef("read %d events for %d documents", count, len(result))
156
return result, true
157
}
158
159
func hasString(changes []string, name string) bool {
160
for _, v := range changes {
161
if v == name {
162
return true
163
}
164
}
165
return false
166
}
167
172
// entities; subsequent events are emitted whenever one or more entities are
173
// added, or change their lifecycle state. After an entity is found to be
186
// filter is used to exclude events not affecting interesting entities.
187
filter func(interface{}) bool
188
// transform, if non-nil, is used to transform a document ID immediately
189
// prior to emitting to the out channel.
190
transform func(string) string
191
// life holds the most recent known life states of interesting entities.
192
life map[string]Life
193
}
194
201
// WatchModels returns a StringsWatcher that notifies of changes to
202
// any models. If a model is removed this *won't* signal that the
203
// model has gone away - it's based on a collectionWatcher which omits
204
// these events.
207
col: modelsC,
208
global: true,
209
})
210
}
211
212
// WatchModelLives returns a StringsWatcher that notifies of changes
213
// to any model life values. The most important difference between
214
// this and WatchModels is that this will signal one last time if a
215
// model is removed.
216
func (st *State) WatchModelLives() StringsWatcher {
220
// WatchModelVolumes returns a StringsWatcher that notifies of changes to
221
// the lifecycles of all model-scoped volumes.
222
func (st *State) WatchModelVolumes() StringsWatcher {
223
return st.watchModelMachinestorage(volumesC)
226
// WatchModelFilesystems returns a StringsWatcher that notifies of changes
227
// to the lifecycles of all model-scoped filesystems.
228
func (st *State) WatchModelFilesystems() StringsWatcher {
229
return st.watchModelMachinestorage(filesystemsC)
233
pattern := fmt.Sprintf("^%s$", st.docID(names.NumberSnippet))
234
members := bson.D{{"_id", bson.D{{"$regex", pattern}}}}
235
filter := func(id interface{}) bool {
236
k, err := st.strictLocalID(id.(string))
237
if err != nil {
238
return false
239
}
240
return !strings.Contains(k, "/")
241
}
243
}
244
245
// WatchMachineVolumes returns a StringsWatcher that notifies of changes to
246
// the lifecycles of all volumes scoped to the specified machine.
247
func (st *State) WatchMachineVolumes(m names.MachineTag) StringsWatcher {
248
return st.watchMachineStorage(m, volumesC)
249
}
250
251
// WatchMachineFilesystems returns a StringsWatcher that notifies of changes
252
// to the lifecycles of all filesystems scoped to the specified machine.
253
func (st *State) WatchMachineFilesystems(m names.MachineTag) StringsWatcher {
254
return st.watchMachineStorage(m, filesystemsC)
255
}
256
257
func (st *State) watchMachineStorage(m names.MachineTag, collection string) StringsWatcher {
258
pattern := fmt.Sprintf("^%s/%s$", st.docID(m.Id()), names.NumberSnippet)
259
members := bson.D{{"_id", bson.D{{"$regex", pattern}}}}
260
prefix := m.Id() + "/"
261
filter := func(id interface{}) bool {
262
k, err := st.strictLocalID(id.(string))
263
if err != nil {
264
return false
265
}
266
return strings.HasPrefix(k, prefix)
267
}
271
// WatchEnvironVolumeAttachments returns a StringsWatcher that notifies of
272
// changes to the lifecycles of all volume attachments related to environ-
273
// scoped volumes.
274
func (st *State) WatchEnvironVolumeAttachments() StringsWatcher {
276
}
277
278
// WatchEnvironFilesystemAttachments returns a StringsWatcher that notifies
279
// of changes to the lifecycles of all filesystem attachments related to
280
// environ-scoped filesystems.
281
func (st *State) WatchEnvironFilesystemAttachments() StringsWatcher {
285
func (st *State) watchModelMachinestorageAttachments(collection string) StringsWatcher {
286
pattern := fmt.Sprintf("^%s.*:%s$", st.docID(""), names.NumberSnippet)
287
members := bson.D{{"_id", bson.D{{"$regex", pattern}}}}
288
filter := func(id interface{}) bool {
289
k, err := st.strictLocalID(id.(string))
290
if err != nil {
291
return false
292
}
293
colon := strings.IndexRune(k, ':')
294
if colon == -1 {
295
return false
296
}
297
return !strings.Contains(k[colon+1:], "/")
298
}
305
func (st *State) WatchMachineVolumeAttachments(m names.MachineTag) StringsWatcher {
306
return st.watchMachineStorageAttachments(m, volumeAttachmentsC)
307
}
308
309
// WatchMachineFilesystemAttachments returns a StringsWatcher that notifies of
310
// changes to the lifecycles of all filesystem attachments related to the specified
312
func (st *State) WatchMachineFilesystemAttachments(m names.MachineTag) StringsWatcher {
313
return st.watchMachineStorageAttachments(m, filesystemAttachmentsC)
314
}
315
316
func (st *State) watchMachineStorageAttachments(m names.MachineTag, collection string) StringsWatcher {
320
filter := func(id interface{}) bool {
321
k, err := st.strictLocalID(id.(string))
322
if err != nil {
323
return false
324
}
325
return strings.HasPrefix(k, prefix)
326
}
336
// WatchRemoteApplications returns a StringsWatcher that notifies of changes to
337
// the lifecycles of the remote applications in the model.
338
func (st *State) WatchRemoteApplications() StringsWatcher {
339
return newLifecycleWatcher(st, remoteApplicationsC, nil, isLocalID(st), nil)
342
// WatchStorageAttachments returns a StringsWatcher that notifies of
343
// changes to the lifecycles of all storage instances attached to the
344
// specified unit.
345
func (st *State) WatchStorageAttachments(unit names.UnitTag) StringsWatcher {
346
members := bson.D{{"unitid", unit.Id()}}
347
prefix := unitGlobalKey(unit.Id()) + "#"
348
filter := func(id interface{}) bool {
349
k, err := st.strictLocalID(id.(string))
350
if err != nil {
351
return false
352
}
353
return strings.HasPrefix(k, prefix)
354
}
355
tr := func(id string) string {
356
// Transform storage attachment document ID to storage ID.
357
return id[len(prefix):]
358
}
359
return newLifecycleWatcher(st, storageAttachmentsC, members, filter, tr)
364
func (a *Application) WatchUnits() StringsWatcher {
365
members := bson.D{{"application", a.doc.Name}}
366
prefix := a.doc.Name + "/"
369
if err != nil {
370
return false
371
}
372
return strings.HasPrefix(unitName, prefix)
379
func (a *Application) WatchRelations() StringsWatcher {
380
return watchApplicationRelations(a.st, a.doc.Name)
381
}
382
383
// WatchRelations returns a StringsWatcher that notifies of changes to the
384
// lifecycles of relations involving s.
385
func (s *RemoteApplication) WatchRelations() StringsWatcher {
386
return watchApplicationRelations(s.st, s.doc.Name)
389
func watchApplicationRelations(backend modelBackend, applicationName string) StringsWatcher {
395
return false
396
}
397
out := strings.HasPrefix(k, prefix) || strings.Contains(k, infix)
398
return out
405
// WatchModelMachines returns a StringsWatcher that notifies of changes to
406
// the lifecycles of the machines (but not containers) in the model.
407
func (st *State) WatchModelMachines() StringsWatcher {
413
k, err := st.strictLocalID(id.(string))
414
if err != nil {
415
return false
416
}
417
return !strings.Contains(k, "/")
426
return m.containersWatcher(isChild)
427
}
428
429
// WatchAllContainers returns a StringsWatcher that notifies of changes to the
430
// lifecycles of all containers on a machine.
431
func (m *Machine) WatchAllContainers() StringsWatcher {
432
isChild := fmt.Sprintf("^%s/%s/%s$", m.doc.DocID, names.ContainerTypeSnippet, names.NumberSnippet)
433
return m.containersWatcher(isChild)
434
}
435
436
func (m *Machine) containersWatcher(isChildRegexp string) StringsWatcher {
440
k := key.(string)
441
_, err := m.st.strictLocalID(k)
442
if err != nil {
443
return false
444
}
445
return compiled.MatchString(k)
452
collName string,
453
members bson.D,
454
filter func(key interface{}) bool,
455
transform func(id string) string,
456
) StringsWatcher {
458
commonWatcher: newCommonWatcher(backend),
459
coll: collFactory(backend.db(), collName),
495
// If no members criteria is specified, use the filter
496
// to reject any unsuitable initial elements.
497
if w.members == nil && w.filter != nil && !w.filter(doc.Id) {
498
continue
499
}
509
func (w *lifecycleWatcher) merge(ids set.Strings, updates map[interface{}]bool) error {
530
// exist are ignored (we'll hear about them in the next set of updates --
531
// all that's actually happened in that situation is that the watcher
532
// events have lagged a little behind reality).
533
iter := coll.Find(bson.D{{"_id", bson.D{{"$in", changed}}}}).Select(lifeFields).Iter()
541
542
// Add to ids any whose life state is known to have changed.
543
for id, newLife := range latest {
544
gone := newLife == Dead
545
oldLife, known := w.life[id]
546
switch {
547
case known && gone:
548
delete(w.life, id)
549
case !known && !gone:
550
w.life[id] = newLife
551
case known && newLife != oldLife:
552
w.life[id] = newLife
553
default:
554
continue
555
}
556
ids.Add(id)
557
}
558
return nil
561
// ErrStateClosed is returned from watchers if their underlying
562
// state connection has been closed.
563
var ErrStateClosed = fmt.Errorf("state has been closed")
564
565
// stateWatcherDeadError processes the error received when the watcher
566
// inside a state connection dies. If the State has been closed, the
567
// watcher will have been stopped and error will be nil, so we ensure
568
// that higher level watchers return a non-nil error in that case, as
569
// watchers are not expected to die unexpectedly without an error.
570
func stateWatcherDeadError(err error) error {
571
if err != nil {
572
return err
573
}
574
return ErrStateClosed
575
}
576
577
func (w *lifecycleWatcher) loop() error {
579
w.watcher.WatchCollectionWithFilter(w.collName, in, w.filter)
580
defer w.watcher.UnwatchCollection(w.collName, in)
587
values := ids.Values()
588
if w.transform != nil {
589
for i, v := range values {
590
values[i] = w.transform(v)
591
}
592
}
599
updates, ok := collect(ch, in, w.tomb.Dying())
600
if !ok {
601
return tomb.ErrDying
619
// Subsequent events are generated when a service increases MinUnits, or when
620
// one or more units belonging to a service are destroyed.
632
known: make(map[string]int),
633
out: make(chan []string),
634
}
635
go func() {
636
defer w.tomb.Done()
637
defer close(w.out)
638
w.tomb.Kill(w.loop())
639
}()
640
return w
641
}
642
656
w.known[doc.ApplicationName] = doc.Revno
657
applicationnames.Add(doc.ApplicationName)
662
func (w *minUnitsWatcher) merge(applicationnames set.Strings, change watcher.Change) error {
686
defer w.watcher.UnwatchCollection(minUnitsC, ch)
687
applicationnames, err := w.initial()
688
if err != nil {
689
return err
690
}
691
out := w.out
692
for {
693
select {
694
case <-w.tomb.Dying():
695
return tomb.ErrDying
716
// scopeInfo holds a RelationScopeWatcher's last-delivered state, and any
717
// known but undelivered changes thereto.
718
type scopeInfo struct {
719
base map[string]bool
720
diff map[string]bool
721
}
722
723
func (info *scopeInfo) add(name string) {
724
if info.base[name] {
725
delete(info.diff, name)
726
} else {
727
info.diff[name] = true
728
}
729
}
730
731
func (info *scopeInfo) remove(name string) {
732
if info.base[name] {
733
info.diff[name] = false
734
} else {
735
delete(info.diff, name)
736
}
737
}
738
739
func (info *scopeInfo) commit() {
740
for name, change := range info.diff {
741
if change {
742
info.base[name] = true
743
} else {
744
delete(info.base, name)
745
}
746
}
747
info.diff = map[string]bool{}
748
}
749
750
func (info *scopeInfo) hasChanges() bool {
751
return len(info.diff) > 0
752
}
753
754
func (info *scopeInfo) changes() *RelationScopeChange {
755
ch := &RelationScopeChange{}
756
for name, change := range info.diff {
757
if change {
758
ch.Entered = append(ch.Entered, name)
759
} else {
760
ch.Left = append(ch.Left, name)
761
}
762
}
763
return ch
768
// RelationScopeChange contains information about units that have
769
// entered or left a particular scope.
770
type RelationScopeChange struct {
771
Entered []string
772
Left []string
775
// RelationScopeWatcher observes changes to the set of units
776
// in a particular relation scope.
777
type RelationScopeWatcher struct {
778
commonWatcher
779
prefix string
780
ignore string
781
out chan *RelationScopeChange
782
}
783
784
func newRelationScopeWatcher(backend modelBackend, scope, ignore string) *RelationScopeWatcher {
799
// Changes returns a channel that will receive changes when units enter and
800
// leave a relation scope. The Entered field in the first event on the channel
801
// holds the initial state.
802
func (w *RelationScopeWatcher) Changes() <-chan *RelationScopeChange {
806
// initialInfo returns an uncommitted scopeInfo with the current set of units.
807
func (w *RelationScopeWatcher) initialInfo() (info *scopeInfo, err error) {
823
for _, doc := range docs {
824
if name := doc.unitName(); name != w.ignore {
825
info.add(name)
826
}
831
// mergeChanges updates info with the contents of the changes in ids. False
832
// values are always treated as removed; true values cause the associated
833
// document to be read, and whether it's treated as added or removed depends
834
// on the value of the document's Departing field.
835
func (w *RelationScopeWatcher) mergeChanges(info *scopeInfo, ids map[interface{}]bool) error {
839
var existIds []string
840
for id, exists := range ids {
841
switch id := id.(type) {
842
case string:
843
if exists {
844
existIds = append(existIds, id)
845
} else {
847
if err != nil {
848
return errors.Trace(err)
849
}
850
info.remove(unitNameFromScopeKey(key))
862
name := doc.unitName()
863
if doc.Departing {
864
info.remove(name)
865
} else if name != w.ignore {
866
info.add(name)
875
filter := func(id interface{}) bool {
876
return strings.HasPrefix(id.(string), fullPrefix)
878
w.watcher.WatchCollectionWithFilter(relationScopesC, in, filter)
879
defer w.watcher.UnwatchCollection(relationScopesC, in)
892
case ch := <-in:
893
latest, ok := collect(ch, in, w.tomb.Dying())
894
if !ok {
895
return tomb.ErrDying
896
}
897
if err := w.mergeChanges(info, latest); err != nil {
914
// scope of a RelationUnit, and changes to the settings of those units known
915
// to have entered.
924
// Watch returns a watcher that notifies of changes to conterpart units in
925
// the relation.
930
// WatchUnits returns a watcher that notifies of changes to the units of the
931
// specified application endpoint in the relation. This method will return an error
932
// if the endpoint is not globally scoped.
933
func (r *Relation) WatchUnits(serviceName string) (RelationUnitsWatcher, error) {
934
return r.watchUnits(serviceName, false)
935
}
936
937
func (r *Relation) watchUnits(applicationName string, counterpart bool) (RelationUnitsWatcher, error) {
939
if err != nil {
940
return nil, err
941
}
942
if ep.Scope != charm.ScopeGlobal {
943
return nil, errors.Errorf("%q endpoint is not globally scoped", ep.Name)
944
}
949
rsw := watchRelationScope(r.st, r.globalScope(), role, "")
950
return newRelationUnitsWatcher(r.st, rsw), nil
951
}
952
953
func newRelationUnitsWatcher(backend modelBackend, sw *RelationScopeWatcher) RelationUnitsWatcher {
960
}
961
go func() {
962
defer w.finish()
963
w.tomb.Kill(w.loop())
964
}()
965
return w
966
}
967
968
// Changes returns a channel that will receive the changes to
969
// counterpart units in a relation. The first event on the
970
// channel holds the initial state of the relation in its
980
func setRelationUnitChangeVersion(changes *params.RelationUnitsChange, key string, version int64) {
992
func (w *relationUnitsWatcher) mergeSettings(changes *params.RelationUnitsChange, key string) (int64, error) {
993
var doc struct {
994
TxnRevno int64 `bson:"txn-revno"`
995
Version int64 `bson:"version"`
996
}
1000
setRelationUnitChangeVersion(changes, key, doc.Version)
1001
return doc.TxnRevno, nil
1004
// mergeScope starts and stops settings watches on the units entering and
1005
// leaving the scope in the supplied RelationScopeChange event, and applies
1006
// the expressed changes to the supplied RelationUnitsChange event.
1007
func (w *relationUnitsWatcher) mergeScope(changes *params.RelationUnitsChange, c *RelationScopeChange) error {
1032
// remove removes s from strs and returns the modified slice.
1033
func remove(strs []string, s string) []string {
1034
for i, v := range strs {
1035
if s == v {
1036
strs[i] = strs[len(strs)-1]
1037
return strs[:len(strs)-1]
1063
case <-w.tomb.Dying():
1064
return tomb.ErrDying
1065
case c, ok := <-w.sw.Changes():
1066
if !ok {
1078
id, ok := c.Id.(string)
1079
if !ok {
1080
logger.Warningf("ignoring bad relation scope id: %#v", c.Id)
1096
// their lifecycle status. The initial event contains all units in the set,
1097
// regardless of lifecycle status; once a unit observed to be Dead or removed
1098
// has been reported, it will not be reported again.
1101
tag string
1102
getUnits func() ([]string, error)
1103
life map[string]Life
1104
in chan watcher.Change
1105
out chan []string
1110
// WatchSubordinateUnits returns a StringsWatcher tracking the unit's subordinate units.
1111
func (u *Unit) WatchSubordinateUnits() StringsWatcher {
1114
getUnits := func() ([]string, error) {
1115
if err := u.Refresh(); err != nil {
1116
return nil, err
1117
}
1118
return u.doc.Subordinates, nil
1119
}
1128
getUnits := func() ([]string, error) {
1129
if err := m.Refresh(); err != nil {
1130
return nil, err
1131
}
1132
return m.doc.Principals, nil
1133
}
1137
func newUnitsWatcher(backend modelBackend, tag names.Tag, getUnits func() ([]string, error), coll, id string) StringsWatcher {
1141
getUnits: getUnits,
1142
life: map[string]Life{},
1143
in: make(chan watcher.Change),
1144
out: make(chan []string),
1145
}
1146
go func() {
1147
defer w.tomb.Done()
1148
defer close(w.out)
1164
// lifeWatchDoc holds the fields used in starting and maintaining a watch
1165
// on a entity's lifecycle.
1166
type lifeWatchDoc struct {
1191
if err != nil {
1192
return nil, errors.Trace(err)
1193
}
1194
changes = append(changes, unitName)
1198
}
1199
}
1200
return changes, nil
1201
}
1202
1203
// update adds to and returns changes, such that it contains the names of any
1204
// non-Dead units to have entered or left the tracked set.
1206
latest, err := w.getUnits()
1207
if err != nil {
1208
return nil, err
1209
}
1210
for _, name := range latest {
1211
if _, known := w.life[name]; !known {
1212
changes, err = w.merge(changes, name)
1213
if err != nil {
1214
return nil, err
1215
}
1216
}
1217
}
1218
for name := range w.life {
1219
if hasString(latest, name) {
1220
continue
1221
}
1222
if !hasString(changes, name) {
1223
changes = append(changes, name)
1224
}
1225
delete(w.life, name)
1227
}
1228
return changes, nil
1229
}
1230
1231
// merge adds to and returns changes, such that it contains the supplied unit
1232
// name if that unit is unknown and non-Dead, or has changed lifecycle status.
1240
gone := false
1241
if err == mgo.ErrNotFound {
1242
gone = true
1243
} else if err != nil {
1244
return nil, err
1245
} else if doc.Life == Dead {
1246
gone = true
1247
}
1248
life, known := w.life[name]
1259
return changes, nil
1260
}
1261
if !hasString(changes, name) {
1262
changes = append(changes, name)
1263
}
1264
return changes, nil
1265
}
1266
1280
}
1281
}()
1282
changes, err := w.initial()
1283
if err != nil {
1284
return err
1285
}
1300
}
1301
if err != nil {
1302
return err
1303
}
1304
if len(changes) > 0 {
1305
out = w.out
1306
}
1307
case out <- changes:
1308
out = nil
1309
changes = nil
1310
}
1311
}
1312
}
1313
1314
// WatchHardwareCharacteristics returns a watcher for observing changes to a machine's hardware characteristics.
1319
// WatchControllerInfo returns a NotifyWatcher for the controllers collection
1320
func (st *State) WatchControllerInfo() NotifyWatcher {
1321
return newEntityWatcher(st, controllersC, modelGlobalKey)
1324
// Watch returns a watcher for observing changes to a machine.
1325
func (m *Machine) Watch() NotifyWatcher {
1330
func (a *Application) Watch() NotifyWatcher {
1331
return newEntityWatcher(a.st, applicationsC, a.doc.DocID)
1334
// WatchLeaderSettings returns a watcher for observing changed to a service's
1335
// leader settings.
1336
func (a *Application) WatchLeaderSettings() NotifyWatcher {
1337
docId := a.st.docID(leadershipSettingsKey(a.Name()))
1338
return newEntityWatcher(a.st, settingsC, docId)
1346
// Watch returns a watcher for observing changes to an model.
1347
func (e *Model) Watch() NotifyWatcher {
1348
return newEntityWatcher(e.st, modelsC, e.doc.UUID)
1351
// WatchUpgradeInfo returns a watcher for observing changes to upgrade
1352
// synchronisation state.
1353
func (st *State) WatchUpgradeInfo() NotifyWatcher {
1354
return newEntityWatcher(st, upgradeInfoC, currentUpgradeId)
1355
}
1356
1357
// WatchRestoreInfoChanges returns a NotifyWatcher that will inform
1358
// when the restore status changes.
1359
func (st *State) WatchRestoreInfoChanges() NotifyWatcher {
1360
return newEntityWatcher(st, restoreInfoC, currentRestoreId)
1361
}
1362
1363
// WatchForModelConfigChanges returns a NotifyWatcher waiting for the Model
1364
// Config to change.
1365
func (st *State) WatchForModelConfigChanges() NotifyWatcher {
1366
return newEntityWatcher(st, settingsC, st.docID(modelGlobalKey))
1369
// WatchForUnitAssignment watches for new services that request units to be
1370
// assigned to machines.
1371
func (st *State) WatchForUnitAssignment() StringsWatcher {
1381
// WatchStorageAttachment returns a watcher for observing changes
1382
// to a storage attachment.
1383
func (st *State) WatchStorageAttachment(s names.StorageTag, u names.UnitTag) NotifyWatcher {
1384
id := storageAttachmentId(u.Id(), s.Id())
1385
return newEntityWatcher(st, storageAttachmentsC, st.docID(id))
1386
}
1387
1388
// WatchVolumeAttachment returns a watcher for observing changes
1389
// to a volume attachment.
1390
func (st *State) WatchVolumeAttachment(m names.MachineTag, v names.VolumeTag) NotifyWatcher {
1391
id := volumeAttachmentId(m.Id(), v.Id())
1395
// WatchFilesystemAttachment returns a watcher for observing changes
1396
// to a filesystem attachment.
1397
func (st *State) WatchFilesystemAttachment(m names.MachineTag, f names.FilesystemTag) NotifyWatcher {
1398
id := filesystemAttachmentId(m.Id(), f.Id())
1399
return newEntityWatcher(st, filesystemAttachmentsC, st.docID(id))
1400
}
1401
1402
// WatchConfigSettings returns a watcher for observing changes to the
1403
// unit's service configuration settings. The unit must have a charm URL
1404
// set before this method is called, and the returned watcher will be
1405
// valid only while the unit's charm URL is not changed.
1406
// TODO(fwereade): this could be much smarter; if it were, uniter.Filter
1407
// could be somewhat simpler.
1409
if u.doc.CharmURL == nil {
1410
return nil, fmt.Errorf("unit charm not set")
1411
}
1416
// WatchMeterStatus returns a watcher observing changes that affect the meter status
1417
// of a unit.
1430
func newEntityWatcher(backend modelBackend, collName string, key interface{}) NotifyWatcher {
1431
return newDocWatcher(backend, []docKey{{collName, key}})
1435
// across collections.
1436
type docWatcher struct {
1437
commonWatcher
1438
out chan struct{}
1439
}
1440
1443
// docKey identifies a single item in a single collection.
1444
// It's used as a parameter to newDocWatcher to specify
1445
// which documents should be watched.
1451
// newDocWatcher returns a new docWatcher.
1452
// docKeys identifies the documents that should be watched (their id and which collection they are in)
1497
w.watcher.Watch(coll.Name(), k.docId, txnRevno, in)
1498
defer w.watcher.Unwatch(coll.Name(), k.docId, in)
1507
case ch := <-in:
1508
if _, ok := collect(ch, in, w.tomb.Dying()); !ok {
1509
return tomb.ErrDying
1510
}
1521
// The first event emitted contains the unit names of all units currently
1522
// assigned to the machine, irrespective of their life state. From then on,
1523
// a new event is emitted whenever a unit is assigned to or unassigned from
1524
// the machine, or the lifecycle of a unit that is currently assigned to
1525
// the machine changes.
1538
// WatchUnits returns a new StringsWatcher watching m's units.
1539
func (m *Machine) WatchUnits() StringsWatcher {
1543
func newMachineUnitsWatcher(m *Machine) StringsWatcher {
1544
w := &machineUnitsWatcher{
1550
}
1551
go func() {
1552
defer w.tomb.Done()
1553
defer close(w.out)
1554
w.tomb.Kill(w.loop())
1555
}()
1556
return w
1557
}
1558
1564
func (w *machineUnitsWatcher) updateMachine(pending []string) (new []string, err error) {
1569
for _, unitName := range w.machine.doc.Principals {
1570
if _, ok := w.known[unitName]; !ok {
1571
pending, err = w.merge(pending, unitName)
1580
func (w *machineUnitsWatcher) merge(pending []string, unitName string) (new []string, err error) {
1589
if err == mgo.ErrNotFound || doc.Principal == "" && (doc.MachineId == "" || doc.MachineId != w.machine.doc.Id) {
1594
if life != Dead && !hasString(pending, unitName) {
1595
pending = append(pending, unitName)
1597
for _, subunitName := range doc.Subordinates {
1598
if sublife, subknown := w.known[subunitName]; subknown {
1599
delete(w.known, subunitName)
1601
if sublife != Dead && !hasString(pending, subunitName) {
1602
pending = append(pending, subunitName)
1611
pending = append(pending, unitName)
1612
} else if life != doc.Life && !hasString(pending, unitName) {
1613
pending = append(pending, unitName)
1614
}
1615
w.known[unitName] = doc.Life
1616
for _, subunitName := range doc.Subordinates {
1617
if _, ok := w.known[subunitName]; !ok {
1618
pending, err = w.merge(pending, subunitName)
1641
w.watcher.Watch(machinesC, w.machine.doc.DocID, revno, machineCh)
1642
defer w.watcher.Unwatch(machinesC, w.machine.doc.DocID, machineCh)
1667
if len(changes) > 0 {
1668
out = w.out
1669
}
1670
case out <- changes:
1671
out = nil
1672
changes = nil
1677
// machineAddressesWatcher notifies about changes to a machine's addresses.
1678
//
1679
// The first event emitted contains the addresses currently assigned to the
1680
// machine. From then on, a new event is emitted whenever the machine's
1681
// addresses change.
1682
type machineAddressesWatcher struct {
1683
commonWatcher
1684
machine *Machine
1685
out chan struct{}
1686
}
1687
1688
var _ Watcher = (*machineAddressesWatcher)(nil)
1689
1691
func (m *Machine) WatchAddresses() NotifyWatcher {
1692
return newMachineAddressesWatcher(m)
1693
}
1694
1695
func newMachineAddressesWatcher(m *Machine) NotifyWatcher {
1696
w := &machineAddressesWatcher{
1698
out: make(chan struct{}),
1699
machine: &Machine{st: m.st, doc: m.doc}, // Copy so it may be freely refreshed
1700
}
1701
go func() {
1702
defer w.tomb.Done()
1703
defer close(w.out)
1704
w.tomb.Kill(w.loop())
1705
}()
1706
return w
1707
}
1708
1709
// Changes returns the event channel for w.
1710
func (w *machineAddressesWatcher) Changes() <-chan struct{} {
1711
return w.out
1712
}
1713
1714
func (w *machineAddressesWatcher) loop() error {
1718
if err != nil {
1719
return err
1720
}
1721
machineCh := make(chan watcher.Change)
1722
w.watcher.Watch(machinesC, w.machine.doc.DocID, revno, machineCh)
1723
defer w.watcher.Unwatch(machinesC, w.machine.doc.DocID, machineCh)
1730
case <-w.tomb.Dying():
1731
return tomb.ErrDying
1732
case <-machineCh:
1733
if err := w.machine.Refresh(); err != nil {
1734
return err
1735
}
1736
newAddresses := w.machine.Addresses()
1737
if !addressesEqual(newAddresses, addresses) {
1738
addresses = newAddresses
1739
out = w.out
1740
}
1741
case out <- struct{}{}:
1742
out = nil
1743
}
1744
}
1745
}
1746
1752
// actionStatusWatcher is a StringsWatcher that filters notifications
1753
// to Action Id's that match the ActionReceiver and ActionStatus set
1754
// provided.
1755
type actionStatusWatcher struct {
1756
commonWatcher
1757
source chan watcher.Change
1758
sink chan []string
1759
receiverFilter bson.D
1760
statusFilter bson.D
1761
}
1762
1763
var _ StringsWatcher = (*actionStatusWatcher)(nil)
1764
1765
// newActionStatusWatcher returns the StringsWatcher that will notify
1766
// on changes to Actions with the given ActionReceiver and ActionStatus
1767
// filters.
1768
func newActionStatusWatcher(backend modelBackend, receivers []ActionReceiver, statusSet ...ActionStatus) StringsWatcher {
1769
watchLogger.Debugf("newActionStatusWatcher receivers:'%+v', statuses'%+v'", receivers, statusSet)
1770
w := &actionStatusWatcher{
1772
source: make(chan watcher.Change),
1773
sink: make(chan []string),
1774
receiverFilter: actionReceiverInCollectionOp(receivers...),
1775
statusFilter: statusInCollectionOp(statusSet...),
1776
}
1777
1778
go func() {
1779
defer w.tomb.Done()
1780
defer close(w.sink)
1781
w.tomb.Kill(w.loop())
1782
}()
1783
1784
return w
1785
}
1786
1787
// Changes returns the channel that sends the ids of any
1788
// Actions that change in the actionsC collection, if they
1789
// match the ActionReceiver and ActionStatus filters on the
1790
// watcher.
1791
func (w *actionStatusWatcher) Changes() <-chan []string {
1792
watchLogger.Tracef("actionStatusWatcher Changes()")
1793
return w.sink
1794
}
1795
1796
// loop performs the main event loop cycle, polling for changes and
1797
// responding to Changes requests
1798
func (w *actionStatusWatcher) loop() error {
1799
watchLogger.Tracef("actionStatusWatcher loop()")
1800
var (
1801
changes []string
1807
1808
changes, err := w.initial()
1809
if err != nil {
1810
return err
1811
}
1812
1813
for {
1814
select {
1815
case <-w.tomb.Dying():
1816
return tomb.ErrDying
1819
case ch := <-in:
1820
updates, ok := collect(ch, in, w.tomb.Dying())
1821
if !ok {
1822
return tomb.ErrDying
1823
}
1825
return err
1826
}
1827
if len(changes) > 0 {
1828
out = w.sink
1829
}
1830
case out <- changes:
1832
out = nil
1833
}
1834
}
1835
}
1836
1837
// initial pre-loads the id's that have already been added to the
1838
// collection that would otherwise not normally trigger the watcher
1839
func (w *actionStatusWatcher) initial() ([]string, error) {
1840
watchLogger.Tracef("actionStatusWatcher initial()")
1841
return w.matchingIds()
1842
}
1843
1844
// matchingIds is a helper function that filters the actionsC collection
1845
// on the ActionReceivers and ActionStatus set defined on the watcher.
1846
// If ids are passed in the collection is further filtered to only
1847
// Actions that also have one of the supplied _id's.
1848
func (w *actionStatusWatcher) matchingIds(ids ...string) ([]string, error) {
1849
watchLogger.Tracef("actionStatusWatcher matchingIds() ids:'%+v'", ids)
1850
1855
query := bson.D{{"$and", []bson.D{idFilter, w.receiverFilter, w.statusFilter}}}
1856
iter := coll.Find(query).Iter()
1861
}
1862
watchLogger.Debugf("actionStatusWatcher matchingIds() ids:'%+v', found:'%+v'", ids, found)
1863
return found, iter.Close()
1864
}
1865
1866
// filterAndMergeIds combines existing pending changes along with
1867
// updates from the upstream watcher, and updates the changes set.
1868
// If the upstream changes do not match the ActionReceivers and
1869
// ActionStatus set filters defined on the watcher, they are silently
1870
// dropped.
1871
func (w *actionStatusWatcher) filterAndMergeIds(changes *[]string, updates map[interface{}]bool) error {
1872
watchLogger.Tracef("actionStatusWatcher filterAndMergeIds(changes:'%+v', updates:'%+v')", changes, updates)
1878
chIx, idAlreadyInChangeset := indexOf(localId, *changes)
1879
if exists {
1880
if !idAlreadyInChangeset {
1881
adds = append(adds, localId)
1882
}
1883
} else {
1884
if idAlreadyInChangeset {
1885
// remove id from changes
1886
*changes = append([]string(*changes)[:chIx], []string(*changes)[chIx+1:]...)
1887
}
1888
}
1889
default:
1890
return errors.Errorf("id is not of type string, got %T", id)
1891
}
1892
}
1893
if len(adds) > 0 {
1894
ids, err := w.matchingIds(adds...)
1895
if err != nil {
1896
return errors.Trace(err)
1897
}
1898
*changes = append(*changes, ids...)
1899
}
1900
return nil
1901
}
1902
1903
// inCollectionOp takes a key name and a list of potential values and
1904
// returns a bson.D Op that will match on the supplied key and values.
1905
func inCollectionOp(key string, ids ...string) bson.D {
1906
ret := bson.D{}
1907
switch len(ids) {
1908
case 0:
1909
case 1:
1910
ret = append(ret, bson.DocElem{key, ids[0]})
1911
default:
1912
ret = append(ret, bson.DocElem{key, bson.D{{"$in", ids}}})
1913
}
1914
return ret
1915
}
1916
1917
// localIdInCollectionOp is a special form of inCollectionOp that just
1918
// converts id's to their model-uuid prefixed form.
1919
func localIdInCollectionOp(st modelBackend, localIds ...string) bson.D {
1920
ids := make([]string, len(localIds))
1921
for i, id := range localIds {
1922
ids[i] = st.docID(id)
1923
}
1924
return inCollectionOp("_id", ids...)
1925
}
1926
1927
// actionReceiverInCollectionOp is a special form of inCollectionOp
1928
// that just converts []ActionReceiver to a []string containing the
1929
// ActionReceiver Name() values.
1930
func actionReceiverInCollectionOp(receivers ...ActionReceiver) bson.D {
1931
ids := make([]string, len(receivers))
1932
for i, r := range receivers {
1933
ids[i] = r.Tag().Id()
1934
}
1935
return inCollectionOp("receiver", ids...)
1936
}
1937
1938
// statusInCollectionOp is a special form of inCollectionOp that just
1939
// converts []ActionStatus to a []string with the same values.
1940
func statusInCollectionOp(statusSet ...ActionStatus) bson.D {
1941
ids := make([]string, len(statusSet))
1942
for i, s := range statusSet {
1943
ids[i] = string(s)
1944
}
1945
return inCollectionOp("status", ids...)
1946
}
1947
1948
// collectionWatcher is a StringsWatcher that watches for changes on the
1949
// specified collection that match a filter on the id.
1950
type collectionWatcher struct {
1957
// colWCfg contains the parameters for watching a collection.
1958
type colWCfg struct {
1959
col string
1960
filter func(interface{}) bool
1961
idconv func(string) string
1962
1963
// If global is true the watcher won't be limited to this model.
1964
global bool
1970
if cfg.global {
1971
if cfg.filter == nil {
1972
cfg.filter = func(x interface{}) bool {
1973
return true
1974
}
1975
}
1980
if cfg.filter == nil {
1981
cfg.filter = backstop
1982
} else {
1983
innerFilter := cfg.filter
1984
cfg.filter = func(id interface{}) bool {
1985
if !backstop(id) {
1986
return false
1987
}
1988
return innerFilter(id)
2015
// loop performs the main event loop cycle, polling for changes and
2016
// responding to Changes requests
2024
w.watcher.WatchCollectionWithFilter(w.col, w.source, w.filter)
2025
defer w.watcher.UnwatchCollection(w.col, w.source)
2038
case ch := <-in:
2039
updates, ok := collect(ch, in, w.tomb.Dying())
2040
if !ok {
2041
return tomb.ErrDying
2042
}
2056
// makeIdFilter constructs a predicate to filter keys that have the
2057
// prefix matching one of the passed in ActionReceivers, or returns nil
2058
// if tags is empty
2059
func makeIdFilter(backend modelBackend, marker string, receivers ...ActionReceiver) func(interface{}) bool {
2064
prefixes := make([]string, len(receivers))
2065
for ix, receiver := range receivers {
2067
}
2068
2069
return func(key interface{}) bool {
2070
switch key.(type) {
2071
case string:
2072
for _, prefix := range prefixes {
2073
if strings.HasPrefix(key.(string), prefix) {
2074
return true
2075
}
2076
}
2077
default:
2078
watchLogger.Errorf("key is not type string, got %T", key)
2079
}
2080
return false
2081
}
2082
}
2083
2109
// mergeIds is used for merging actionId's and actionResultId's that
2110
// come in via the updates map. It cleans up the pending changes to
2111
// account for id's being removed before the watcher consumes them,
2112
// and to account for the potential overlap between the id's that were
2113
// pending before the watcher started, and the new id's detected by the
2114
// watcher.
2117
func (w *collectionWatcher) mergeIds(changes *[]string, updates map[interface{}]bool) error {
2127
if err != nil {
2128
return "", errors.Trace(err)
2129
}
2130
}
2131
if w.idconv != nil {
2132
id = w.idconv(id)
2133
}
2134
return id, nil
2135
}
2136
2137
func mergeIds(st modelBackend, changes *[]string, updates map[interface{}]bool, idconv func(string) (string, error)) error {
2138
for val, idExists := range updates {
2139
id, ok := val.(string)
2140
if !ok {
2141
return errors.Errorf("id is not of type string, got %T", val)
2142
}
2145
if err != nil {
2146
return errors.Annotatef(err, "collection watcher")
2147
}
2148
2149
chIx, idAlreadyInChangeset := indexOf(id, *changes)
2150
if idExists {
2151
if !idAlreadyInChangeset {
2152
*changes = append(*changes, id)
2153
}
2154
} else {
2155
if idAlreadyInChangeset {
2156
// remove id from changes
2157
*changes = append([]string(*changes)[:chIx], []string(*changes)[chIx+1:]...)
2164
func actionNotificationIdToActionId(id string) string {
2165
ix := strings.Index(id, actionMarker)
2166
if ix == -1 {
2167
return id
2168
}
2169
return id[ix+len(actionMarker):]
2170
}
2171
2172
func indexOf(find string, in []string) (int, bool) {
2173
for ix, cur := range in {
2174
if cur == find {
2175
return ix, true
2176
}
2177
}
2178
return -1, false
2179
}
2180
2181
// ensureSuffixFn returns a function that will make sure the passed in
2182
// string has the marker token at the end of it
2183
func ensureSuffixFn(marker string) func(string) string {
2184
return func(p string) string {
2185
if !strings.HasSuffix(p, marker) {
2186
p = p + marker
2187
}
2188
return p
2189
}
2190
}
2191
2192
// watchEnqueuedActionsFilteredBy starts and returns a StringsWatcher
2193
// that notifies on new Actions being enqueued on the ActionRecevers
2194
// being watched.
2195
func (st *State) watchEnqueuedActionsFilteredBy(receivers ...ActionReceiver) StringsWatcher {
2197
col: actionNotificationsC,
2198
filter: makeIdFilter(st, actionMarker, receivers...),
2199
idconv: actionNotificationIdToActionId,
2200
})
2203
// WatchControllerStatusChanges starts and returns a StringsWatcher that
2204
// notifies when the status of a controller machine changes.
2205
// TODO(cherylj) Add unit tests for this, as per bug 1543408.
2206
func (st *State) WatchControllerStatusChanges() StringsWatcher {
2213
func makeControllerIdFilter(st *State) func(interface{}) bool {
2214
initialInfo, err := st.ControllerInfo()
2215
if err != nil {
2216
return nil
2218
machines := initialInfo.MachineIds
2219
return func(key interface{}) bool {
2220
switch key.(type) {
2221
case string:
2222
info, err := st.ControllerInfo()
2223
if err != nil {
2224
// Most likely, things will be killed and
2225
// restarted if we hit this error. Just use
2226
// the machine list we knew about last time.
2227
logger.Debugf("unable to get controller info: %v", err)
2228
} else {
2229
machines = info.MachineIds
2231
for _, machine := range machines {
2232
if strings.HasSuffix(key.(string), fmt.Sprintf("m#%s", machine)) {
2233
return true
2234
}
2244
// WatchActionResults starts and returns a StringsWatcher that
2245
// notifies on new ActionResults being added.
2246
func (st *State) WatchActionResults() StringsWatcher {
2247
return st.WatchActionResultsFilteredBy()
2248
}
2250
// WatchActionResultsFilteredBy starts and returns a StringsWatcher
2251
// that notifies on new ActionResults being added for the ActionRecevers
2252
// being watched.
2253
func (st *State) WatchActionResultsFilteredBy(receivers ...ActionReceiver) StringsWatcher {
2254
return newActionStatusWatcher(st, receivers, []ActionStatus{ActionCompleted, ActionCancelled, ActionFailed}...)
2256
2257
// openedPortsWatcher notifies of changes in the openedPorts
2258
// collection
2259
type openedPortsWatcher struct {
2260
commonWatcher
2267
// WatchOpenedPorts starts and returns a StringsWatcher notifying of changes to
2268
// the openedPorts collection. Reported changes have the following format:
2269
// "<machine-id>:[<subnet-CIDR>]", i.e. "0:10.20.0.0/16" or "1:" (empty subnet
2270
// ID is allowed for backwards-compatibility).
2271
func (st *State) WatchOpenedPorts() StringsWatcher {
2272
return newOpenedPortsWatcher(st)
2273
}
2274
2279
out: make(chan []string),
2280
}
2281
go func() {
2282
defer w.tomb.Done()
2283
defer close(w.out)
2284
w.tomb.Kill(w.loop())
2285
}()
2286
2287
return w
2288
}
2289
2290
// Changes returns the event channel for w
2291
func (w *openedPortsWatcher) Changes() <-chan []string {
2292
return w.out
2293
}
2294
2296
// "m#42#0.1.2.0/24") into a colon-separated string with the machine and subnet
2297
// IDs (e.g. "42:0.1.2.0/24"). Subnet ID (a.k.a. CIDR) can be empty for
2298
// backwards-compatibility.
2299
func (w *openedPortsWatcher) transformID(globalKey string) (string, error) {
2300
parts, err := extractPortsIDParts(globalKey)
2329
}
2330
2331
func (w *openedPortsWatcher) loop() error {
2332
in := make(chan watcher.Change)
2333
changes, err := w.initial()
2334
if err != nil {
2335
return errors.Trace(err)
2336
}
2361
func (w *openedPortsWatcher) merge(ids set.Strings, change watcher.Change) error {
2362
id, ok := change.Id.(string)
2363
if !ok {
2364
return errors.Errorf("id %v is not of type string, got %T", id, id)
2381
currentRevno, err := getTxnRevno(openedPorts, id)
2382
closer()
2383
if err != nil {
2384
return err
2385
}
2399
// WatchForRebootEvent returns a notify watcher that will trigger an event
2400
// when the reboot flag is set on our machine agent, our parent machine agent
2401
// or grandparent machine agent
2403
machineIds := m.machinesToCareAboutRebootsFor()
2404
machines := set.NewStrings(machineIds...)
2405
2406
filter := func(key interface{}) bool {
2407
if id, ok := key.(string); ok {
2408
if id, err := m.st.strictLocalID(id); err == nil {
2409
return machines.Contains(id)
2429
func newBlockDevicesWatcher(backend modelBackend, machineId string) NotifyWatcher {
2434
}
2435
go func() {
2436
defer w.tomb.Done()
2437
defer close(w.out)
2438
w.tomb.Kill(w.loop())
2439
}()
2440
return w
2441
}
2442
2443
// Changes returns the event channel for w.
2449
docID := w.backend.docID(w.machineId)
2450
coll, closer := w.db.GetCollection(blockDevicesC)
2457
w.watcher.Watch(blockDevicesC, docID, revno, changes)
2458
defer w.watcher.Unwatch(blockDevicesC, docID, changes)
2475
if !reflect.DeepEqual(newBlockDevices, blockDevices) {
2476
blockDevices = newBlockDevices
2485
// WatchForMigration returns a notify watcher which reports when
2486
// a migration is in progress for the model associated with the
2487
// State.
2488
func (st *State) WatchForMigration() NotifyWatcher {
2489
return newMigrationActiveWatcher(st)
2490
}
2491
2492
type migrationActiveWatcher struct {
2493
commonWatcher
2494
collName string
2495
id string
2496
sink chan struct{}
2497
}
2498
2499
func newMigrationActiveWatcher(st *State) NotifyWatcher {
2500
w := &migrationActiveWatcher{
2501
commonWatcher: newCommonWatcher(st),
2502
collName: migrationsActiveC,
2503
id: st.ModelUUID(),
2504
sink: make(chan struct{}),
2505
}
2506
go func() {
2507
defer w.tomb.Done()
2508
defer close(w.sink)
2509
w.tomb.Kill(w.loop())
2510
}()
2511
return w
2512
}
2513
2514
// Changes returns the event channel for this watcher.
2515
func (w *migrationActiveWatcher) Changes() <-chan struct{} {
2516
return w.sink
2517
}
2518
2519
func (w *migrationActiveWatcher) loop() error {
2521
revno, err := getTxnRevno(collection, w.id)
2522
closer()
2523
if err != nil {
2524
return errors.Trace(err)
2525
}
2526
2527
in := make(chan watcher.Change)
2528
w.watcher.Watch(w.collName, w.id, revno, in)
2529
defer w.watcher.Unwatch(w.collName, w.id, in)
2530
2531
out := w.sink
2532
for {
2533
select {
2534
case <-w.tomb.Dying():
2535
return tomb.ErrDying
2536
case <-w.watcher.Dead():
2537
return stateWatcherDeadError(w.watcher.Err())
2538
case change := <-in:
2539
if _, ok := collect(change, in, w.tomb.Dying()); !ok {
2540
return tomb.ErrDying
2541
}
2542
out = w.sink
2543
case out <- struct{}{}:
2544
out = nil
2545
}
2546
}
2547
}
2548
2549
// WatchMigrationStatus returns a NotifyWatcher which triggers
2550
// whenever the status of latest migration for the State's model
2551
// changes. One instance can be used across migrations. The watcher
2552
// will report changes when one migration finishes and another one
2553
// begins.
2554
//
2555
// Note that this watcher does not produce an initial event if there's
2556
// never been a migration attempt for the model.
2557
func (st *State) WatchMigrationStatus() NotifyWatcher {
2558
// Watch the entire migrationsStatusC collection for migration
2559
// status updates related to the State's model. This is more
2560
// efficient and simpler than tracking the current active
2561
// migration (and changing watchers when one migration finishes
2562
// and another starts.
2563
//
2564
// This approach is safe because there are strong guarantees that
2565
// there will only be one active migration per model. The watcher
2566
// will only see changes for one migration status document at a
2567
// time for the model.
2568
return newNotifyCollWatcher(st, migrationsStatusC, isLocalID(st))
2569
}
2570
2571
// WatchMachineRemovals returns a NotifyWatcher which triggers
2572
// whenever machine removal records are added or removed.
2573
func (st *State) WatchMachineRemovals() NotifyWatcher {
2574
return newNotifyCollWatcher(st, machineRemovalsC, isLocalID(st))
2575
}
2576
2577
// notifyCollWatcher implements NotifyWatcher, triggering when a
2578
// change is seen in a specific collection matching the provided
2579
// filter function.
2580
type notifyCollWatcher struct {
2581
commonWatcher
2582
collName string
2583
filter func(interface{}) bool
2584
sink chan struct{}
2585
}
2586
2587
func newNotifyCollWatcher(backend modelBackend, collName string, filter func(interface{}) bool) NotifyWatcher {
2590
collName: collName,
2591
filter: filter,
2592
sink: make(chan struct{}),
2593
}
2594
go func() {
2595
defer w.tomb.Done()
2596
defer close(w.sink)
2597
w.tomb.Kill(w.loop())
2598
}()
2599
return w
2600
}
2601
2602
// Changes returns the event channel for this watcher.
2603
func (w *notifyCollWatcher) Changes() <-chan struct{} {
2604
return w.sink
2605
}
2606
2607
func (w *notifyCollWatcher) loop() error {
2608
in := make(chan watcher.Change)
2609
2610
w.watcher.WatchCollectionWithFilter(w.collName, in, w.filter)
2611
defer w.watcher.UnwatchCollection(w.collName, in)
2612
2613
out := w.sink // out set so that initial event is sent.
2614
for {
2615
select {
2616
case <-w.tomb.Dying():
2617
return tomb.ErrDying
2618
case <-w.watcher.Dead():
2619
return stateWatcherDeadError(w.watcher.Err())
2620
case change := <-in:
2621
if _, ok := collect(change, in, w.tomb.Dying()); !ok {
2622
return tomb.ErrDying
2623
}
2624
out = w.sink
2625
case out <- struct{}{}:
2626
out = nil
2627
}
2628
}
2629
}
2630
2631
// WatchRemoteRelations returns a StringsWatcher that notifies of changes to
2632
// the lifecycles of the remote relations in the model.
2633
func (st *State) WatchRemoteRelations() StringsWatcher {
2634
// Use a no-op transform func to record the known ids.
2635
known := make(map[interface{}]bool)
2636
tr := func(id string) string {
2637
known[id] = true
2638
return id
2639
}
2640
2641
filter := func(id interface{}) bool {
2642
id, err := st.strictLocalID(id.(string))
2643
if err != nil {
2644
return false
2645
}
2646
2647
// Gather the remote app names.
2649
defer closer()
2650
2651
type remoteAppDoc struct {
2652
Name string
2653
}
2654
remoteAppNameField := bson.D{{"name", 1}}
2655
var apps []remoteAppDoc
2656
err = remoteApps.Find(nil).Select(remoteAppNameField).All(&apps)
2657
if err != nil {
2658
watchLogger.Errorf("could not lookup remote application names: %v", err)
2659
return false
2660
}
2661
remoteAppNames := set.NewStrings()
2662
for _, a := range apps {
2663
remoteAppNames.Add(a.Name)
2664
}
2665
2666
// Run a query to pickup any relations to those remote apps.
2668
defer closer()
2669
2670
query := bson.D{
2671
{"key", id},
2672
{"endpoints.applicationname", bson.D{{"$in", remoteAppNames.Values()}}},
2673
}
2674
num, err := relations.Find(query).Count()
2675
if err != nil {
2676
watchLogger.Errorf("could not lookup remote relations: %v", err)
2677
return false
2678
}
2679
// The relation (or remote app) may have been deleted, but if it has been
2680
// seen previously, return true.
2681
if num == 0 {
2682
_, seen := known[id]
2683
delete(known, id)
2684
return seen
2685
}
2686
return num > 0
2687
}
2688
return newLifecycleWatcher(st, relationsC, nil, filter, tr)
2689
}
2690
2691
// WatchSubnets returns a StringsWatcher that notifies of changes to
2692
// the lifecycles of the subnets in the model.
2693
func (st *State) WatchSubnets() StringsWatcher {
2694
return newLifecycleWatcher(st, subnetsC, nil, isLocalID(st), nil)
2695
}
2696
2697
// isLocalID returns a watcher filter func that rejects ids not specific
2698
// to the supplied modelBackend.
2699
func isLocalID(st modelBackend) func(interface{}) bool {
2700
return func(id interface{}) bool {
2701
key, ok := id.(string)
2702
if !ok {
2703
return false
2704
}
2705
_, err := st.strictLocalID(key)
2706
return err == nil
2707
}
2708
}