state: add machine units watcher

state: add machine units watcher



https://code.launchpad.net/~aramh/juju-core/100-state-watchers-machine-units5/+merge/126790

(do not edit description out of merge proposal)

[aram, 2012-09-27 20:14:20]

Please take a look.

[aram, 2012-09-28 14:02:33]

WIP, for now. There are some obvious deficiencies which I am correcting.

[aram, 2012-10-01 16:54:46]

Please take a look.

[aram, 2012-10-01 16:55:19]

Rearchitectured to use a single goroutine per watch.

[niemeyer, 2012-10-02 00:08:44]

This is a good code spike to get a feeling for the problem, but really it needs
some careful consideration and polishing. As it is, it's hard to tell whether
the logic is right or not, and in fact it is not right.

I'd try to cut code down significantly by unifying the logic of subordinates and
principals. In the current implementation, they have entirely different code
paths despite the fact that their handling is exactly the same, if you leave
aside the fact they are discovered differently.

The implementation only needs to watch:

  1) the machine

  2) the specific units discovered (principal or not)

That's two channels. When it finds that a changed unit is a principal, it may
need to pick further subordinates, but that's about it, and that's in addition
to the common logic, not besides it.

I'd try to get an interface along the lines of this pseudo-sketch:

func (w) initial() (unitIds []string, err error) {
    for id in w.machine.Principals {
        unitIds, err = w.merge(unitIds, id)
        ...
    }
    ...
}

func (w) merge(unitIds []string, unitId string) ([]string, error) {
    doc = FindId(unitId)
    if !found {
        if known {
            add to changed result
        }
        unwatch
    } else {
        if !known {
            watch
            add to changed result
            for subId in doc.Subordinates {
                unitIds, err = w.merge(unitIds, subId)
            }
        } else {
            if life changed {
                add to changed result
            }
            for subId in doc.Subordinates {
                if !known[subId] {
                    unitIds, err = w.merge(unitIds, subId)
                }
            }
        }
    }
}

func (w) loop() error {
    w.initial()
    for {
        wait for unit and machine channels and merge
    }
}

I'm sure there are more details, but I'm hoping this will simplify it quite a
bit.

Please talk to me if you want to brainstorm on this.

https://codereview.appspot.com/6573064/diff/5001/state/watcher.go
File state/watcher.go (right):

https://codereview.appspot.com/6573064/diff/5001/state/watcher.go#newcode1290
state/watcher.go:1290: // to and from a machine.
// MachineUnitsWatcher notifies about assignments and lifecycle changes
// for all units of a machine.
// 
// The first event emitted will contain the unit names of all units currently
// assigned to the machine, irrespective of their life state. From then on a
// new event is emitted whenever a unit is assigned to or unassigned from the
// machine, or the lifecycle of a unit that is currently assigned to the
// machine changes.
//
// After a unit is found to be Dead, no further event will include it.

https://codereview.appspot.com/6573064/diff/5001/state/watcher.go#newcode1299
state/watcher.go:1299: // or removed from a machine.
// WatchUnits returns a new MachineUnitsWatcher for m.

https://codereview.appspot.com/6573064/diff/5001/state/watcher.go#newcode1326
state/watcher.go:1326: // initial state as returned by Machine.Units.
// Changes returns the event channel for w.

https://codereview.appspot.com/6573064/diff/5001/state/watcher.go#newcode1333
state/watcher.go:1333: err = w.st.units.Find(append(notDead, D{{"machineid",
w.machine.doc.Id}}...)).Select(lifeFields).All(&pudocs)
notDead? As discussed, we need to know about Dead resources.

I'm also slightly surprised to see this find here, rather  than logic that
observes the machine's principals field.

https://codereview.appspot.com/6573064/diff/5001/state/watcher.go#newcode1341
state/watcher.go:1341: err = w.st.units.Find(append(notDead, D{{"principal",
pudoc.Name}}...)).Select(lifeFields).All(&sudocs)
Ditto.

https://codereview.appspot.com/6573064/diff/5001/state/watcher.go#newcode1408
state/watcher.go:1408: if err != nil && err != mgo.ErrNotFound {
It's easy to build this logic without mentioning ErrNotFound twice.

https://codereview.appspot.com/6573064/diff/5001/state/watcher.go#newcode1436
state/watcher.go:1436: w.st.watcher.WatchCollection(w.st.units.Name, allCh)
We don't want every machine watching all units. We know exactly which units we
care about, and in general this will be a very small set for this watcher.

https://codereview.appspot.com/6573064/diff/5001/state/watcher.go#newcode1444
state/watcher.go:1444: for _, unit := range w.machine.doc.Principals {
w.initial() above doesn't look at Principals, but then we subscribe to a
different selection of units that doesn't necessarily match what we've seen.
There is surely a race condition here.

https://codereview.appspot.com/6573064/diff/5001/state/watcher.go#newcode1477
state/watcher.go:1477: if err != nil && err != mgo.ErrNotFound {
It's easy to build this logic without mentioning ErrNotFound twice.

https://codereview.appspot.com/6573064/diff/5001/state/watcher.go#newcode1485
state/watcher.go:1485: w.st.watcher.Watch(w.st.units.Name, unit, doc.TxnRevno,
principalCh)
The difference between a principal unit and a subordinate unit is simply in how
we get to know about their existence. Ideally we'd have a single place that
reckons we've found a new unit or not and adds it to the change selection or
not.

This will make it easier to handle details of that algorithm, such as avoiding
duplications in the changes slice and whatnot, which the above isn't doing.

https://codereview.appspot.com/6573064/diff/5001/state/watcher.go#newcode1492
state/watcher.go:1492: delete(w.known, unit)
And we never warn the observer that the unit went away? What if we unsubscribed
from the channel before the unit was seen on principalCh?

https://codereview.appspot.com/6573064/diff/5001/state/watcher.go#newcode1494
state/watcher.go:1494: knownPrincipals = newPrincipals
The block of code above may be updateMachine or something.

[aram, 2012-10-02 19:28:35]

> The implementation only needs to watch:
>   1) the machine
>   2) the specific units discovered (principal or not)

I changed it to do that. The reason I didn't do it in the first place
this way was to minimize the number of watchers we use. I thought using
2+nprincipals watchers was better than using nunits watchers.

https://codereview.appspot.com/6573064/diff/5001/state/watcher.go
File state/watcher.go (right):

https://codereview.appspot.com/6573064/diff/5001/state/watcher.go#newcode1290
state/watcher.go:1290: // to and from a machine.
On 2012/10/02 00:08:44, niemeyer wrote:
> // MachineUnitsWatcher notifies about assignments and lifecycle changes
> // for all units of a machine.
> // 
> // The first event emitted will contain the unit names of all units currently
> // assigned to the machine, irrespective of their life state. From then on a
> // new event is emitted whenever a unit is assigned to or unassigned from the
> // machine, or the lifecycle of a unit that is currently assigned to the
> // machine changes.
> //
> // After a unit is found to be Dead, no further event will include it.

Done.

https://codereview.appspot.com/6573064/diff/5001/state/watcher.go#newcode1299
state/watcher.go:1299: // or removed from a machine.
On 2012/10/02 00:08:44, niemeyer wrote:
> // WatchUnits returns a new MachineUnitsWatcher for m.

Done.

https://codereview.appspot.com/6573064/diff/5001/state/watcher.go#newcode1326
state/watcher.go:1326: // initial state as returned by Machine.Units.
On 2012/10/02 00:08:44, niemeyer wrote:
> // Changes returns the event channel for w.

Done.

https://codereview.appspot.com/6573064/diff/5001/state/watcher.go#newcode1333
state/watcher.go:1333: err = w.st.units.Find(append(notDead, D{{"machineid",
w.machine.doc.Id}}...)).Select(lifeFields).All(&pudocs)
> notDead? As discussed, we need to know about Dead resources.

I though we couldn't have dead units assigned to a machine because the
guy that sets units to dead is the same guy that does the unassignment,
and will set them to dead only so after the unassignment has completed.

Anyway, changed.

https://codereview.appspot.com/6573064/diff/5001/state/watcher.go#newcode1444
state/watcher.go:1444: for _, unit := range w.machine.doc.Principals {
On 2012/10/02 00:08:44, niemeyer wrote:
> w.initial() above doesn't look at Principals, but then we subscribe to a
> different selection of units that doesn't necessarily match what we've seen.
> There is surely a race condition here.

Done.

https://codereview.appspot.com/6573064/diff/5001/state/watcher.go#newcode1477
state/watcher.go:1477: if err != nil && err != mgo.ErrNotFound {
On 2012/10/02 00:08:44, niemeyer wrote:
> It's easy to build this logic without mentioning ErrNotFound twice.

Done.

[aram, 2012-10-02 19:28:57]

Please take a look.

[niemeyer, 2012-10-03 17:48:34]

So much better. Thank you!

Just a few details left:

https://codereview.appspot.com/6573064/diff/10002/state/watcher.go
File state/watcher.go (right):

https://codereview.appspot.com/6573064/diff/10002/state/watcher.go#newcode1318
state/watcher.go:1318: machine:       m,
machine: m.st.Machine(m.Id()), // Copy for free refreshing

https://codereview.appspot.com/6573064/diff/10002/state/watcher.go#newcode1339
state/watcher.go:1339: return w.updateMachine(changes)
Let's just inline this in the one place below.

https://codereview.appspot.com/6573064/diff/10002/state/watcher.go#newcode1365
state/watcher.go:1365: w.st.watcher.Unwatch(w.st.units.Name, unit, w.in)
Can't unwatch something we didn't watch before, and we may get here when coming
through update machine.

We need something along the lines of:

life, known := w.known[unit] (do this just once so we can reuse)

if err == mgo.ErrNotFound {
    if known {
        w.st.watcher.Unwatch(w.st.units.Name, unit, w.in)
        if life != Dead {
            pending = append(pending, unit)
        }
    }
    return pending, nil
}

https://codereview.appspot.com/6573064/diff/10002/state/watcher.go#newcode1371
state/watcher.go:1371: if _, ok := w.known[unit]; !ok {
if !known {

https://codereview.appspot.com/6573064/diff/10002/state/watcher.go#newcode1375
state/watcher.go:1375: for _, unit := range doc.Subordinates {
We only need a single one of these at the end. It'll work fine if we drop this
and the return below.

https://codereview.appspot.com/6573064/diff/10002/state/watcher.go#newcode1383
state/watcher.go:1383: if w.known[unit] != doc.Life {
} else if life != doc.Life {

https://codereview.appspot.com/6573064/diff/10002/state/watcher.go#newcode1387
state/watcher.go:1387: return pending, nil
Can't return here, or below. It's eating the verification for subordinates at
the end. Before fixing, please add a test that adds a subordinate + changes life
before the next sync.

https://codereview.appspot.com/6573064/diff/10002/state/watcher.go#newcode1392
state/watcher.go:1392: for _, unit := range doc.Subordinates {
s/unit/subunit/ to avoid shadowing. The w.known[unit] below means something else
than it does above.

[niemeyer, 2012-10-03 17:51:08]

https://codereview.appspot.com/6573064/diff/10002/state/watcher.go
File state/watcher.go (right):

https://codereview.appspot.com/6573064/diff/10002/state/watcher.go#newcode1318
state/watcher.go:1318: machine:       m,
On 2012/10/03 17:48:34, niemeyer wrote:
> machine: m.st.Machine(m.Id()), // Copy for free refreshing

Actually, we don't need to touch the database for that. I suggest doing this
instead:

m = &Machine{m.st, m.doc} // Copy so it may be freely refreshed

[aram, 2012-10-03 18:37:26]

https://codereview.appspot.com/6573064/diff/10002/state/watcher.go
File state/watcher.go (right):

https://codereview.appspot.com/6573064/diff/10002/state/watcher.go#newcode1318
state/watcher.go:1318: machine:       m,
On 2012/10/03 17:51:08, niemeyer wrote:
> On 2012/10/03 17:48:34, niemeyer wrote:
> > machine: m.st.Machine(m.Id()), // Copy for free refreshing
> 
> Actually, we don't need to touch the database for that. I suggest doing this
> instead:
> 
> m = &Machine{m.st, m.doc} // Copy so it may be freely refreshed

Done.

https://codereview.appspot.com/6573064/diff/10002/state/watcher.go#newcode1339
state/watcher.go:1339: return w.updateMachine(changes)
On 2012/10/03 17:48:34, niemeyer wrote:
> Let's just inline this in the one place below.

Done.

https://codereview.appspot.com/6573064/diff/10002/state/watcher.go#newcode1365
state/watcher.go:1365: w.st.watcher.Unwatch(w.st.units.Name, unit, w.in)
On 2012/10/03 17:48:34, niemeyer wrote:
> Can't unwatch something we didn't watch before, and we may get here when
coming
> through update machine.
> 
> We need something along the lines of:
> 
> life, known := w.known[unit] (do this just once so we can reuse)
> 
> if err == mgo.ErrNotFound {
>     if known {
>         w.st.watcher.Unwatch(w.st.units.Name, unit, w.in)
>         if life != Dead {
>             pending = append(pending, unit)
>         }
>     }
>     return pending, nil
> }

Done.

https://codereview.appspot.com/6573064/diff/10002/state/watcher.go#newcode1371
state/watcher.go:1371: if _, ok := w.known[unit]; !ok {
On 2012/10/03 17:48:34, niemeyer wrote:
> if !known {

Done.

https://codereview.appspot.com/6573064/diff/10002/state/watcher.go#newcode1383
state/watcher.go:1383: if w.known[unit] != doc.Life {
On 2012/10/03 17:48:34, niemeyer wrote:
> } else if life != doc.Life {

Done.

https://codereview.appspot.com/6573064/diff/10002/state/watcher.go#newcode1387
state/watcher.go:1387: return pending, nil
On 2012/10/03 17:48:34, niemeyer wrote:
> Can't return here, or below. It's eating the verification for subordinates at
> the end. Before fixing, please add a test that adds a subordinate + changes
life
> before the next sync.

Done.

https://codereview.appspot.com/6573064/diff/10002/state/watcher.go#newcode1392
state/watcher.go:1392: for _, unit := range doc.Subordinates {
On 2012/10/03 17:48:34, niemeyer wrote:
> s/unit/subunit/ to avoid shadowing. The w.known[unit] below means something
else
> than it does above.

Done.

[aram, 2012-10-03 18:37:47]

Please take a look.

[niemeyer, 2012-10-03 18:44:11]

LGTM with the revert of the Alive change, assuming all tests pass. Thank you!

[aram, 2012-10-03 18:44:29]

*** Submitted:

state: add machine units watcher

R=niemeyer
CC=
https://codereview.appspot.com/6573064