Merge lp:~themue/pyjuju/go-state-more-unit-watchers into lp:pyjuju/go

[Frank Mueller]

More watchers for Unit.

Adding of NeedsUpgradeWatcher, ResolvedWatcher and PortsWatcher
for Units. They all are based on the new ExistenceWatcher for
the creation, content changing and deletion of nodes.

https://codereview.appspot.com/6011047/

[Roger Peppe (rogpeppe)]

looks good.

https://codereview.appspot.com/6011047/diff/1/state/unit.go
File state/unit.go (right):

https://codereview.appspot.com/6011047/diff/1/state/unit.go#newcode489
state/unit.go:489: // NeedsUpgradeWatcher observes changes to a upgrade
flag node.
  // NeedsUpgradeWatcher observes changes to a unit's upgrade flag.
?

https://codereview.appspot.com/6011047/diff/1/state/unit.go#newcode531
state/unit.go:531: defer close(w.changeChan)
is this how callers are supposed to know that the upgrade watcher is
shutting down? there's no Dying method, so i guess so. if so, it should
be documented (it's different from the ExistenceWatcher way); if not
then you should a a Dying method and ensure that w.tomb is in a dying
state before closing changeChan.

https://codereview.appspot.com/6011047/diff/1/state/unit.go#newcode614
state/unit.go:614: // PortsWatcher observes changes to a ports node.
// PortsWatcher observes changes to a unit's open ports.

https://codereview.appspot.com/6011047/diff/1/state/unit.go#newcode656
state/unit.go:656: defer close(w.changeChan)
same applies here as to NeedsUpgradeWatcher.

https://codereview.appspot.com/6011047/diff/1/state/unit.go#newcode663
state/unit.go:663: ports := &struct{ Open []Port }{}
var ports struct {
     Open []Port
}
would be clearer i think (and would avoid the unnecessary double
indirection).

https://codereview.appspot.com/6011047/diff/1/state/watch_test.go
File state/watch_test.go (right):

https://codereview.appspot.com/6011047/diff/1/state/watch_test.go#newcode13
state/watch_test.go:13: return change, ok, false
you could probably just Assert(ok, Equals, true) here
and thus simplify the interface to receiveChange.
That applies to most of the similar tests below - those that don't test
for the changes channel being closed can factor out the closed test.

another possibility is to pass expected values for timeout and closed
into the function:

receiveChange := func(w *state.ConfigWatcher, expectClosed,
expectTimeout bool) *state.ConfigNode {
     etc
}

https://codereview.appspot.com/6011047/diff/1/state/watch_test.go#newcode14
state/watch_test.go:14: case <-time.After(time.Second):
i think a second is too long here. 100 * time.Millsecond?
same applies below, let's not make testing take too long.

https://codereview.appspot.com/6011047/diff/1/state/watcher/watcher.go
File state/watcher/watcher.go (right):

https://codereview.appspot.com/6011047/diff/1/state/watcher/watcher.go#newcode10
state/watcher/watcher.go:10: // of a node and, if it's true, its
content.
// ExistenceChange holds information on the existence
// and contents of a node. Content will be empty when the
// node does not exist.
?

https://codereview.appspot.com/6011047/diff/1/state/watcher/watcher.go#newcode17
state/watcher/watcher.go:17: // notification (including the content)
when it is created, changed
s/ (including the content)//

it's implied by the rest of the sentence and the ExistenceChange type.

https://codereview.appspot.com/6011047/diff/1/state/watcher/watcher.go#newcode94
state/watcher/watcher.go:94: content, _, watch, err := w.zk.GetW(w.path)
it seems slightly odd that when a node gets deleted, we ignore what
we've just...

[lp:~themue/pyjuju/go-state-more-unit-watchers]

113. By Frank Mueller on 2012-04-16

Changed after review.

[Roger Peppe (rogpeppe)]

https://codereview.appspot.com/6011047/diff/1/state/watch_test.go
File state/watch_test.go (right):

https://codereview.appspot.com/6011047/diff/1/state/watch_test.go#newcode13
state/watch_test.go:13: return change, ok, false
On 2012/04/16 15:45:31, TheMue wrote:
> On 2012/04/16 11:43:01, rog wrote:
> > you could probably just Assert(ok, Equals, true) here
> > and thus simplify the interface to receiveChange.
> > That applies to most of the similar tests below - those that don't
test for
> the
> > changes channel being closed can factor out the closed test.
> >
> > another possibility is to pass expected values for timeout and
closed into the
> > function:
> >
> > receiveChange := func(w *state.ConfigWatcher, expectClosed,
expectTimeout
> bool)
> > *state.ConfigNode {
> > etc
> > }

> Thought about it too. But then I decided for a clean seperation of the
simple
> retrievement with a possible timeout (had some due to errors while
development)
> and the assertions in the test. To me this way is more readable, even
if the
> code is bit more bloated.

fair enough.

https://codereview.appspot.com/6011047/diff/1/state/watcher/watcher.go
File state/watcher/watcher.go (right):

https://codereview.appspot.com/6011047/diff/1/state/watcher/watcher.go#newcode94
state/watcher/watcher.go:94: content, _, watch, err := w.zk.GetW(w.path)
On 2012/04/16 15:45:31, TheMue wrote:
> On 2012/04/16 11:43:01, rog wrote:
> > it seems slightly odd that when a node gets deleted, we ignore what
we've just
> > been told and call GetW and then ExistsW on the node anyway.
> > i appreciate that passing the event type into update might
complicate the
> logic
> > a little, but i think it might be worth doing. something like this
perhaps?
> >
> > func (w *ExistenceWatcher) update(event int) (nextWatch <-chan
> zookeeper.Event,
> > err error
> > ) {
> > var content string
> > var watch <-chan zookeeper.Event
> > if event != zookeeper.EVENT_DELETED {
> > content, _, watch, err = w.zk.GetW(w.path)
> > }
> > switch {
> > case event == zookeeper.EVENT_DELETED || zookeeper.IsError(err,
> > zookeeper.ZNONODE):
> > call existsW
> > etc
> > }

> Hmm, ok, which event would you pass for the first call of update()? At
this
> moment you're unsure if it exists. And where does err in switch come
from?

you could pass any event that's not EVENT_DELETED - EVENT_CHANGED would
be fine.

err is defined in the same place it is currently - as a return
parameter.

https://codereview.appspot.com/6011047/

[Gustavo Niemeyer (niemeyer)]

As we discussed over IRC, can you please break that down into smaller
chunks?

https://codereview.appspot.com/6011047/diff/6001/state/state.go
File state/state.go (right):

https://codereview.appspot.com/6011047/diff/6001/state/state.go#newcode273
state/state.go:273: func (w *ConfigWatcher) Dying() <-chan struct{} {
Why is this necessary?

https://codereview.appspot.com/6011047/diff/6001/state/unit.go
File state/unit.go (right):

https://codereview.appspot.com/6011047/diff/6001/state/unit.go#newcode316
state/unit.go:316: // WatchResolved creates a watcher for the resolved
node of the unit.
There's no such thing as a "resolved node" documented in this API. We
need to document these methods in terms of what they mean semantically
for juju, rather than what they mean for the interaction of these
methods with zookeeper. For example:

// WatchResolved returns a watcher that fires when the unit
// is marked as having had its problems resolved. See
// SetResolved for details.

Please review the documentation for the other methods.

https://codereview.appspot.com/6011047/diff/6001/state/unit.go#newcode480
state/unit.go:480: if none && mode == ResolvedNone {
none? Does it mean acceptNone? Please fix the argument name and document
it.

https://codereview.appspot.com/6011047/

[Gustavo Niemeyer (niemeyer)]

More notes on the changes in watcher to get you going tomorrow.

https://codereview.appspot.com/6011047/diff/6001/state/watcher/watcher.go
File state/watcher/watcher.go (right):

https://codereview.appspot.com/6011047/diff/6001/state/watcher/watcher.go#newcode12
state/watcher/watcher.go:12: type ExistenceChange struct {
This isn't watching just existence.. it's watching both existence and
contents. How's that different from the ContentsWatcher, and why do we
need both? Any higher-level watcher that needs contents may be
implemented on top of a watcher that offers existence *and* contents. I
suggest dropping this type, and modifying ContentsWatcher to handle both
problems.

This should be done in an independent branch.

https://codereview.appspot.com/6011047/diff/6001/state/watcher/watcher.go#newcode96
state/watcher/watcher.go:96: case err != nil && !zookeeper.IsError(err,
zookeeper.ZNONODE):
This check should be inverted. We care about the case ZNONODE, so let's
be explicit about it:

case zookeeper.IsError(err, zookeeper.ZNONODE):
    ...

https://codereview.appspot.com/6011047/diff/6001/state/watcher/watcher.go#newcode100
state/watcher/watcher.go:100: _, watch, err = w.zk.ExistsW(w.path)
This is bogus. It is assuming that the remote state of the node is known
here, but it's not. The current state of the node is in that parameter
that is being ignored.

The overall logic must be fixed to take that into consideration.

https://codereview.appspot.com/6011047/

[Gustavo Niemeyer (niemeyer)]

For the record, this work is being split down into other smaller
branches. Please do keep these reviews in mind, though.

https://codereview.appspot.com/6011047/

Unmerged revisions

113. By Frank Mueller on 2012-04-16

Changed after review.

112. By Frank Mueller on 2012-04-12

Extended watcher package by ExistenceWatcher with looks for creation,
content changes and deletions. Used it in NeedsUpgradeWatcher, ResolvedWatcher
and PortsWatcher for Units.

=== modified file 'state/state.go'
--- state/state.go	2012-04-05 13:01:51 +0000
+++ state/state.go	2012-04-16 15:47:21 +0000
@@ -268,6 +268,12 @@
 	return w.changeChan
 }
 
+// Dying returns a channel that is closed when the
+// watcher has stopped or is about to stop.
+func (w *ConfigWatcher) Dying() <-chan struct{} {
+	return w.tomb.Dying()
+}
+
 // Stop stops the watch and returns any error encountered
 // while watching. This method should always be called
 // before discarding the watcher.
 
=== modified file 'state/state_test.go'
--- state/state_test.go	2012-03-26 12:47:37 +0000
+++ state/state_test.go	2012-04-16 15:47:21 +0000
@@ -778,7 +778,7 @@
 	c.Assert(err, ErrorMatches, "no unused machine found")
 }
 
-func (s *StateSuite) TestGetSetClearUnitUpgrate(c *C) {
+func (s *StateSuite) TestGetSetClearUnitUpgrade(c *C) {
 	// Check that setting and clearing an upgrade flag on a unit works.
 	dummy, _ := addDummyCharm(c, s.st)
 	wordpress, err := s.st.AddService("wordpress", dummy)
 
=== modified file 'state/unit.go'
--- state/unit.go	2012-03-20 20:28:54 +0000
+++ state/unit.go	2012-04-16 15:47:21 +0000
@@ -11,6 +11,8 @@
 	"launchpad.net/gozk/zookeeper"
 	"launchpad.net/juju/go/charm"
 	"launchpad.net/juju/go/state/presence"
+	"launchpad.net/juju/go/state/watcher"
+	"launchpad.net/tomb"
 	"strconv"
 	"strings"
 	"time"
@@ -261,6 +263,11 @@
 	return err
 }
 
+// WatchNeedsUpgrade creates a watcher for the upgrade flag of the unit.
+func (u *Unit) WatchNeedsUpgrade() *NeedsUpgradeWatcher {
+	return newNeedsUpgradeWatcher(u)
+}
+
 // Resolved returns the resolved mode for the unit.
 func (u *Unit) Resolved() (ResolvedMode, error) {
 	yaml, _, err := u.st.zk.Get(u.zkResolvedPath())
@@ -271,15 +278,7 @@
 	if err != nil {
 		return ResolvedNone, err
 	}
-	setting := &struct{ Retry ResolvedMode }{}
-	if err = goyaml.Unmarshal([]byte(yaml), setting); err != nil {
-		return ResolvedNone, err
-	}
-	mode := setting.Retry
-	if err := validResolvedMode(mode); err != nil {
-		return ResolvedNone, err
-	}
-	return mode, nil
+	return parseResolvedMode(yaml)
 }
 
 // SetResolved marks the unit as having had any previous state
@@ -289,7 +288,7 @@
 // reexecute previous failed hooks or to continue as if they had 
 // succeeded before.
 func (u *Unit) SetResolved(mode ResolvedMode) error {
-	if err := validResolvedMode(mode); err != nil {
+	if err := validResolvedMode(mode, false); err != nil {
 		return err
 	}
 	setting := &struct{ Retry ResolvedMode }{mode}
@@ -314,6 +313,11 @@
 	return err
 }
 
+// WatchResolved creates a watcher for the resolved node of the unit.
+func (u *Unit) WatchResolved() *ResolvedWatcher {
+	return newResolvedWatcher(u)
+}
+
 // OpenPort sets the policy of the port with protocol and number to be opened.
 func (u *Unit) OpenPort(protocol string, number int) error {
 	openPort := func(oldYaml string, stat *zookeeper.Stat) (string, error) {
@@ -386,6 +390,11 @@
 	return ports.Open, nil
 }
 
+// WatchPorts creates a watcher for the ports node of the unit.
+func (u *Unit) WatchPorts() *PortsWatcher {
+	return newPortsWatcher(u)
+}
+
 // AgentAlive returns whether the respective remote agent is alive.
 func (u *Unit) AgentAlive() (bool, error) {
 	return presence.Alive(u.st.zk, u.zkAgentPath())
@@ -417,11 +426,6 @@
 	return fmt.Sprintf("/units/%s", u.key)
 }
 
-// zkPortsPath returns the ZooKeeper path for the open ports.
-func (u *Unit) zkPortsPath() string {
-	return fmt.Sprintf("/units/%s/ports", u.key)
-}
-
 // zkAgentPath returns the ZooKeeper path for the unit agent.
 func (u *Unit) zkAgentPath() string {
 	return fmt.Sprintf("/units/%s/agent", u.key)
@@ -437,6 +441,11 @@
 	return fmt.Sprintf("/units/%s/resolved", u.key)
 }
 
+// zkPortsPath returns the ZooKeeper path for the open ports.
+func (u *Unit) zkPortsPath() string {
+	return fmt.Sprintf("/units/%s/ports", u.key)
+}
+
 // parseUnitName parses a unit name like "wordpress/0" into
 // its service name and sequence number parts.
 func parseUnitName(name string) (serviceName string, seqNo int, err error) {
@@ -451,11 +460,232 @@
 	return parts[0], int(sequenceNo), nil
 }
 
+// parseResolveMode parses a given YAML for the resolve mode
+// and checks if it's valid.
+func parseResolvedMode(yaml string) (ResolvedMode, error) {
+	setting := &struct{ Retry ResolvedMode }{}
+	if err := goyaml.Unmarshal([]byte(yaml), setting); err != nil {
+		return ResolvedNone, err
+	}
+	mode := setting.Retry
+	if err := validResolvedMode(mode, true); err != nil {
+		return ResolvedNone, err
+	}
+	return mode, nil
+}
+
 // validResolvedMode ensures that only valid values for the
 // resolved mode are used.
-func validResolvedMode(mode ResolvedMode) error {
+func validResolvedMode(mode ResolvedMode, none bool) error {
+	if none && mode == ResolvedNone {
+		return nil
+	}
 	if mode != ResolvedRetryHooks && mode != ResolvedNoHooks {
 		return fmt.Errorf("invalid error resolution mode: %d", mode)
 	}
 	return nil
 }
+
+// NeedsUpgradeWatcher observes changes to a unit's upgrade flag.
+type NeedsUpgradeWatcher struct {
+	unit       *Unit
+	tomb       tomb.Tomb
+	watcher    *watcher.ExistenceWatcher
+	changeChan chan bool
+}
+
+// newNeedsUpgradeWatcher creates and starts a new resolved flag node 
+// watcher for the given path.
+func newNeedsUpgradeWatcher(u *Unit) *NeedsUpgradeWatcher {
+	w := &NeedsUpgradeWatcher{
+		unit:       u,
+		changeChan: make(chan bool),
+		watcher:    watcher.NewExistenceWatcher(u.st.zk, u.zkNeedsUpgradePath()),
+	}
+	go w.loop()
+	return w
+}
+
+// Changes returns a channel that will receive the new
+// resolved mode when a change is detected. Note that multiple
+// changes may be observed as a single event in the channel.
+func (w *NeedsUpgradeWatcher) Changes() <-chan bool {
+	return w.changeChan
+}
+
+// Dying returns a channel that is closed when the
+// watcher has stopped or is about to stop.
+func (w *NeedsUpgradeWatcher) Dying() <-chan struct{} {
+	return w.tomb.Dying()
+}
+
+// Stop stops the watch and returns any error encountered
+// while watching. This method should always be called
+// before discarding the watcher.
+func (w *NeedsUpgradeWatcher) Stop() error {
+	w.tomb.Kill(nil)
+	if err := w.watcher.Stop(); err != nil {
+		w.tomb.Wait()
+		return err
+	}
+	return w.tomb.Wait()
+}
+
+// loop is the backend for watching the resolved flag node.
+func (w *NeedsUpgradeWatcher) loop() {
+	defer w.tomb.Done()
+	defer close(w.changeChan)
+
+	for {
+		select {
+		case <-w.tomb.Dying():
+			return
+		case change := <-w.watcher.Changes():
+			select {
+			case <-w.watcher.Dying():
+				return
+			case <-w.tomb.Dying():
+				return
+			case w.changeChan <- change.Exists:
+			}
+		}
+	}
+}
+
+// ResolvedWatcher observes changes to a resolved flag node.
+type ResolvedWatcher struct {
+	unit       *Unit
+	tomb       tomb.Tomb
+	watcher    *watcher.ExistenceWatcher
+	changeChan chan ResolvedMode
+}
+
+// newResolvedWatcher creates and starts a new resolved flag node 
+// watcher for the given path.
+func newResolvedWatcher(u *Unit) *ResolvedWatcher {
+	w := &ResolvedWatcher{
+		unit:       u,
+		changeChan: make(chan ResolvedMode),
+		watcher:    watcher.NewExistenceWatcher(u.st.zk, u.zkResolvedPath()),
+	}
+	go w.loop()
+	return w
+}
+
+// Changes returns a channel that will receive the new
+// resolved mode when a change is detected. Note that multiple
+// changes may be observed as a single event in the channel.
+func (w *ResolvedWatcher) Changes() <-chan ResolvedMode {
+	return w.changeChan
+}
+
+// Dying returns a channel that is closed when the
+// watcher has stopped or is about to stop.
+func (w *ResolvedWatcher) Dying() <-chan struct{} {
+	return w.tomb.Dying()
+}
+
+// Stop stops the watch and returns any error encountered
+// while watching. This method should always be called
+// before discarding the watcher.
+func (w *ResolvedWatcher) Stop() error {
+	w.tomb.Kill(nil)
+	if err := w.watcher.Stop(); err != nil {
+		w.tomb.Wait()
+		return err
+	}
+	return w.tomb.Wait()
+}
+
+// loop is the backend for watching the resolved flag node.
+func (w *ResolvedWatcher) loop() {
+	defer w.tomb.Done()
+	defer close(w.changeChan)
+
+	for {
+		select {
+		case <-w.tomb.Dying():
+			return
+		case change := <-w.watcher.Changes():
+			mode, err := parseResolvedMode(change.Content)
+			if err != nil {
+				w.tomb.Kill(err)
+				return
+			}
+			select {
+			case <-w.watcher.Dying():
+				return
+			case <-w.tomb.Dying():
+				return
+			case w.changeChan <- mode:
+			}
+		}
+	}
+}
+
+// PortsWatcher observes changes to a unit's open ports.
+type PortsWatcher struct {
+	unit       *Unit
+	tomb       tomb.Tomb
+	watcher    *watcher.ExistenceWatcher
+	changeChan chan []Port
+}
+
+// newPortsWatcher creates and starts a new resolved flag node 
+// watcher for the given path.
+func newPortsWatcher(u *Unit) *PortsWatcher {
+	w := &PortsWatcher{
+		unit:       u,
+		changeChan: make(chan []Port),
+		watcher:    watcher.NewExistenceWatcher(u.st.zk, u.zkPortsPath()),
+	}
+	go w.loop()
+	return w
+}
+
+// Changes returns a channel that will receive the new
+// ports when a change is detected. Note that multiple
+// changes may be observed as a single event in the channel.
+func (w *PortsWatcher) Changes() <-chan []Port {
+	return w.changeChan
+}
+
+// Stop stops the watch and returns any error encountered
+// while watching. This method should always be called
+// before discarding the watcher.
+func (w *PortsWatcher) Stop() error {
+	w.tomb.Kill(nil)
+	if err := w.watcher.Stop(); err != nil {
+		w.tomb.Wait()
+		return err
+	}
+	return w.tomb.Wait()
+}
+
+// loop is the backend for watching the ports node.
+func (w *PortsWatcher) loop() {
+	defer w.tomb.Done()
+	defer close(w.changeChan)
+
+	for {
+		select {
+		case <-w.tomb.Dying():
+			return
+		case change := <-w.watcher.Changes():
+			var ports struct {
+				Open []Port
+			}
+			if err := goyaml.Unmarshal([]byte(change.Content), &ports); err != nil {
+				w.tomb.Kill(err)
+				return
+			}
+			select {
+			case <-w.watcher.Dying():
+				return
+			case <-w.tomb.Dying():
+				return
+			case w.changeChan <- ports.Open:
+			}
+		}
+	}
+}
 
=== modified file 'state/watch_test.go'
--- state/watch_test.go	2012-04-05 13:01:51 +0000
+++ state/watch_test.go	2012-04-16 15:47:21 +0000
@@ -2,10 +2,20 @@
 
 import (
 	. "launchpad.net/gocheck"
+	"launchpad.net/juju/go/state"
 	"time"
 )
 
 func (s *StateSuite) TestServiceWatchConfig(c *C) {
+	receiveChange := func(w *state.ConfigWatcher) (*state.ConfigNode, bool, bool) {
+		select {
+		case change, ok := <-w.Changes():
+			return change, ok, false
+		case <-time.After(100 * time.Millisecond):
+			return nil, false, true
+		}
+		return nil, false, false
+	}
 	dummy, _ := addDummyCharm(c, s.st)
 	wordpress, err := s.st.AddService("wordpress", dummy)
 	c.Assert(err, IsNil)
@@ -17,7 +27,9 @@
 	watcher := wordpress.WatchConfig()
 
 	// Recieve initial event after creation.
-	changedConfig := <-watcher.Changes()
+	changedConfig, ok, timeout := receiveChange(watcher)
+	c.Assert(timeout, Equals, false)
+	c.Assert(ok, Equals, true)
 	c.Assert(changedConfig.Keys(), HasLen, 0)
 
 	// Two more change events.
@@ -25,20 +37,23 @@
 	config.Set("baz", "yadda")
 	_, err = config.Write()
 	c.Assert(err, IsNil)
-
 	time.Sleep(100 * time.Millisecond)
 	config.Delete("foo")
 	_, err = config.Write()
 	c.Assert(err, IsNil)
 
 	// Receive the two changes.
-	changedConfig = <-watcher.Changes()
+	changedConfig, ok, timeout = receiveChange(watcher)
+	c.Assert(timeout, Equals, false)
+	c.Assert(ok, Equals, true)
 	c.Assert(changedConfig.Map(), DeepEquals, map[string]interface{}{"foo": "bar", "baz": "yadda"})
 	foo, found := changedConfig.Get("foo")
 	c.Assert(found, Equals, true)
 	c.Assert(foo, Equals, "bar")
 
-	changedConfig = <-watcher.Changes()
+	changedConfig, ok, timeout = receiveChange(watcher)
+	c.Assert(timeout, Equals, false)
+	c.Assert(ok, Equals, true)
 	c.Assert(changedConfig.Map(), DeepEquals, map[string]interface{}{"baz": "yadda"})
 	foo, found = changedConfig.Get("foo")
 	c.Assert(found, Equals, false)
@@ -47,18 +62,23 @@
 	c.Assert(baz, Equals, "yadda")
 
 	// No more changes.
-	select {
-	case <-watcher.Changes():
-		c.Fatalf("no more config changes expected")
-	case <-time.After(200 * time.Millisecond):
-		// The timeout is expected.
-	}
+	_, _, timeout = receiveChange(watcher)
+	c.Assert(timeout, Equals, true)
 
 	err = watcher.Stop()
 	c.Assert(err, IsNil)
 }
 
-func (s *StateSuite) TestServiceWatchConfigIllegalData(c *C) {
+func (s *StateSuite) TestServiceWatchConfigIllegalDatal(c *C) {
+	receiveChange := func(w *state.ConfigWatcher) (*state.ConfigNode, bool, bool) {
+		select {
+		case change, ok := <-w.Changes():
+			return change, ok, false
+		case <-time.After(100 * time.Millisecond):
+			return nil, false, true
+		}
+		return nil, false, false
+	}
 	dummy, _ := addDummyCharm(c, s.st)
 	wordpress, err := s.st.AddService("wordpress", dummy)
 	c.Assert(err, IsNil)
@@ -69,15 +89,167 @@
 	_, err = s.zkConn.Set("/services/service-0000000000/config", "---", -1)
 	c.Assert(err, IsNil)
 
-	// Changes() has to be closed
-	select {
-	case _, ok := <-watcher.Changes():
-		c.Assert(ok, Equals, false)
-	case <-time.After(200 * time.Millisecond):
-		// Timeout should not be needed.
-		c.Fatalf("config change channel should have been closed due to the illegal data")
-	}
+	// Changes() has to be closed.
+	_, ok, timeout := receiveChange(watcher)
+	c.Assert(ok, Equals, false)
+	c.Assert(timeout, Equals, false)
 
 	err = watcher.Stop()
 	c.Assert(err, ErrorMatches, "YAML error: .*")
 }
+
+func (s *StateSuite) TestUnitWatchNeedsUpgrade(c *C) {
+	receiveChange := func(w *state.NeedsUpgradeWatcher) (bool, bool, bool) {
+		select {
+		case change, ok := <-w.Changes():
+			return change, ok, false
+		case <-time.After(100 * time.Millisecond):
+			return false, false, true
+		}
+		return false, false, false
+	}
+	dummy, _ := addDummyCharm(c, s.st)
+	wordpress, err := s.st.AddService("wordpress", dummy)
+	c.Assert(err, IsNil)
+	c.Assert(wordpress.Name(), Equals, "wordpress")
+	unit, err := wordpress.AddUnit()
+	c.Assert(err, IsNil)
+	watcher := unit.WatchNeedsUpgrade()
+
+	go func() {
+		time.Sleep(50 * time.Millisecond)
+		err = unit.SetNeedsUpgrade()
+		c.Assert(err, IsNil)
+		time.Sleep(50 * time.Millisecond)
+		err = unit.ClearNeedsUpgrade()
+		c.Assert(err, IsNil)
+	}()
+
+	// Receive the changes.
+	upgrade, ok, timeout := receiveChange(watcher)
+	c.Assert(timeout, Equals, false)
+	c.Assert(ok, Equals, true)
+	c.Assert(upgrade, Equals, true)
+	upgrade, ok, timeout = receiveChange(watcher)
+	c.Assert(ok, Equals, true)
+	c.Assert(timeout, Equals, false)
+	c.Assert(upgrade, Equals, false)
+
+	// No more changes.
+	_, _, timeout = receiveChange(watcher)
+	c.Assert(timeout, Equals, true)
+
+	err = watcher.Stop()
+	c.Assert(err, IsNil)
+}
+
+func (s *StateSuite) TestUnitWatchResolved(c *C) {
+	receiveChange := func(w *state.ResolvedWatcher) (state.ResolvedMode, bool, bool) {
+		select {
+		case change, ok := <-w.Changes():
+			return change, ok, false
+		case <-time.After(100 * time.Millisecond):
+			return state.ResolvedNoHooks, false, true
+		}
+		return state.ResolvedNoHooks, false, false
+	}
+	dummy, _ := addDummyCharm(c, s.st)
+	wordpress, err := s.st.AddService("wordpress", dummy)
+	c.Assert(err, IsNil)
+	c.Assert(wordpress.Name(), Equals, "wordpress")
+	unit, err := wordpress.AddUnit()
+	c.Assert(err, IsNil)
+	watcher := unit.WatchResolved()
+
+	go func() {
+		time.Sleep(50 * time.Millisecond)
+		err = unit.SetResolved(state.ResolvedRetryHooks)
+		c.Assert(err, IsNil)
+		time.Sleep(50 * time.Millisecond)
+		err = unit.ClearResolved()
+		c.Assert(err, IsNil)
+		time.Sleep(50 * time.Millisecond)
+		err = unit.SetResolved(state.ResolvedNoHooks)
+		c.Assert(err, IsNil)
+	}()
+
+	// Receive the changes.
+	resolved, ok, timeout := receiveChange(watcher)
+	c.Assert(timeout, Equals, false)
+	c.Assert(ok, Equals, true)
+	c.Assert(resolved, Equals, state.ResolvedRetryHooks)
+	resolved, ok, timeout = receiveChange(watcher)
+	c.Assert(timeout, Equals, false)
+	c.Assert(ok, Equals, true)
+	c.Assert(resolved, Equals, state.ResolvedNone)
+	resolved, ok, timeout = receiveChange(watcher)
+	c.Assert(timeout, Equals, false)
+	c.Assert(ok, Equals, true)
+	c.Assert(resolved, Equals, state.ResolvedNoHooks)
+
+	// No more changes.
+	_, _, timeout = receiveChange(watcher)
+	c.Assert(timeout, Equals, true)
+
+	err = watcher.Stop()
+	c.Assert(err, IsNil)
+}
+
+func (s *StateSuite) TestUnitWatchPorts(c *C) {
+	receiveChange := func(w *state.PortsWatcher) ([]state.Port, bool, bool) {
+		select {
+		case change, ok := <-w.Changes():
+			return change, ok, false
+		case <-time.After(100 * time.Millisecond):
+			return nil, false, true
+		}
+		return nil, false, false
+	}
+	dummy, _ := addDummyCharm(c, s.st)
+	wordpress, err := s.st.AddService("wordpress", dummy)
+	c.Assert(err, IsNil)
+	c.Assert(wordpress.Name(), Equals, "wordpress")
+	unit, err := wordpress.AddUnit()
+	c.Assert(err, IsNil)
+	watcher := unit.WatchPorts()
+
+	go func() {
+		time.Sleep(50 * time.Millisecond)
+		err = unit.OpenPort("tcp", 80)
+		c.Assert(err, IsNil)
+		time.Sleep(50 * time.Millisecond)
+		err = unit.OpenPort("udp", 53)
+		c.Assert(err, IsNil)
+		time.Sleep(50 * time.Millisecond)
+		err = unit.ClosePort("tcp", 80)
+		c.Assert(err, IsNil)
+	}()
+
+	// Receive the changes.
+	open, ok, timeout := receiveChange(watcher)
+	c.Assert(timeout, Equals, false)
+	c.Assert(ok, Equals, true)
+	c.Assert(open, DeepEquals, []state.Port{
+		{"tcp", 80},
+	})
+	open, ok, timeout = receiveChange(watcher)
+	c.Assert(timeout, Equals, false)
+	c.Assert(ok, Equals, true)
+	c.Assert(open, DeepEquals, []state.Port{
+		{"tcp", 80},
+		{"udp", 53},
+	})
+	open, ok, timeout = receiveChange(watcher)
+	c.Assert(timeout, Equals, false)
+	c.Assert(ok, Equals, true)
+	c.Assert(open, DeepEquals, []state.Port{
+		{"udp", 53},
+	})
+
+	// No more changes.
+	_, _, timeout = receiveChange(watcher)
+	c.Assert(timeout, Equals, true)
+
+	err = watcher.Stop()
+	c.Assert(err, IsNil)
+}
 
=== modified file 'state/watcher/watcher.go'
--- state/watcher/watcher.go	2012-04-05 13:01:51 +0000
+++ state/watcher/watcher.go	2012-04-16 15:47:21 +0000
@@ -6,6 +6,121 @@
 	"launchpad.net/tomb"
 )
 
+// ExistenceChange holds information on the existence
+// and contents of a node. Content will be empty when the
+// node does not exist.
+type ExistenceChange struct {
+	Exists  bool
+	Content string
+}
+
+// ExistenceWatcher observes a ZooKeeper node and delivers a
+// notification when it is created, changed or deleted.
+type ExistenceWatcher struct {
+	zk         *zookeeper.Conn
+	path       string
+	tomb       tomb.Tomb
+	changeChan chan ExistenceChange
+	existence  ExistenceChange
+}
+
+// NewExistenceWatcher creates a ExistenceWatcher observing
+// the ZooKeeper node at watchedPath.
+func NewExistenceWatcher(zk *zookeeper.Conn, watchedPath string) *ExistenceWatcher {
+	w := &ExistenceWatcher{
+		zk:         zk,
+		path:       watchedPath,
+		changeChan: make(chan ExistenceChange),
+		existence:  ExistenceChange{},
+	}
+	go w.loop()
+	return w
+}
+
+// Changes returns a channel that will receive the change if a node
+// is created or deleted. In case or a created node it also contains
+// the content. Note that multiple changes may be observed as a single 
+// event in the channel.
+func (w *ExistenceWatcher) Changes() <-chan ExistenceChange {
+	return w.changeChan
+}
+
+// Dying returns a channel that is closed when the
+// watcher has stopped or is about to stop.
+func (w *ExistenceWatcher) Dying() <-chan struct{} {
+	return w.tomb.Dying()
+}
+
+// Stop stops the watch and returns any error encountered
+// while watching. This method should always be called before
+// discarding the watcher.
+func (w *ExistenceWatcher) Stop() error {
+	w.tomb.Kill(nil)
+	return w.tomb.Wait()
+}
+
+// loop is the backend for watching.
+func (w *ExistenceWatcher) loop() {
+	defer w.tomb.Done()
+	defer close(w.changeChan)
+
+	watch, err := w.update()
+	if err != nil {
+		w.tomb.Kill(err)
+		return
+	}
+
+	for {
+		select {
+		case <-w.tomb.Dying():
+			return
+		case evt := <-watch:
+			if !evt.Ok() {
+				w.tomb.Killf("watcher: critical session event: %v", evt)
+				return
+			}
+			watch, err = w.update()
+			if err != nil {
+				w.tomb.Kill(err)
+				return
+			}
+		}
+	}
+}
+
+// update checks the node existence and emits changes of that state to the 
+// change channel if it has changed. It returns the next watch.
+func (w *ExistenceWatcher) update() (nextWatch <-chan zookeeper.Event, err error) {
+	content, _, watch, err := w.zk.GetW(w.path)
+	switch {
+	case err != nil && !zookeeper.IsError(err, zookeeper.ZNONODE):
+		return nil, fmt.Errorf("watcher: can't get the content of node %q: %v", w.path, err)
+	case err != nil:
+		// Need a new watch to signal creation of node.
+		_, watch, err = w.zk.ExistsW(w.path)
+		if err != nil {
+			return nil, fmt.Errorf("watcher: can't check the existence of node %q: %v", w.path, err)
+		}
+		if !w.existence.Exists {
+			return watch, nil
+		}
+		w.existence.Exists = false
+		w.existence.Content = ""
+	default:
+		if w.existence.Exists && content == w.existence.Content {
+			return watch, nil
+		}
+		w.existence.Exists = true
+		w.existence.Content = content
+	}
+	select {
+	case <-w.tomb.Dying():
+		return nil, tomb.ErrDying
+	case w.changeChan <- w.existence:
+	}
+	return watch, nil
+}
+
 // ContentWatcher observes a ZooKeeper node and delivers a
 // notification when a content change is detected.
 type ContentWatcher struct {
 
=== modified file 'state/watcher/watcher_test.go'
--- state/watcher/watcher_test.go	2012-04-05 13:01:51 +0000
+++ state/watcher/watcher_test.go	2012-04-16 15:47:21 +0000
@@ -39,122 +39,215 @@
 
 	s.zkConn = zk
 	s.path = "/watcher"
-
-	_, err = s.zkConn.Create(s.path, "", 0, zookeeper.WorldACL(zookeeper.PERM_ALL))
-	c.Assert(err, IsNil)
 }
 
 func (s *WatcherSuite) TearDownTest(c *C) {
-	testing.ZkRemoveTree(s.zkConn, s.path)
+	s.removePath(c)
 	s.zkConn.Close()
 }
 
+func (s *WatcherSuite) TestExistenceWatcher(c *C) {
+	receiveChange := func(w *watcher.ExistenceWatcher) (*watcher.ExistenceChange, bool, bool) {
+		select {
+		case change, ok := <-w.Changes():
+			return &change, ok, false
+		case <-time.After(200 * time.Millisecond):
+			return nil, false, true
+		}
+		return nil, false, false
+	}
+	watcher := watcher.NewExistenceWatcher(s.zkConn, s.path)
+
+	go func() {
+		time.Sleep(50 * time.Millisecond)
+		s.createPath(c, "init")
+		time.Sleep(50 * time.Millisecond)
+		s.changeContent(c, "foo")
+		time.Sleep(50 * time.Millisecond)
+		s.removePath(c)
+		time.Sleep(50 * time.Millisecond)
+		s.createPath(c, "done")
+	}()
+
+	// Receive the four changes.
+	change, ok, timeout := receiveChange(watcher)
+	c.Assert(timeout, Equals, false)
+	c.Assert(ok, Equals, true)
+	c.Assert(change.Exists, Equals, true)
+	c.Assert(change.Content, Equals, "init")
+
+	change, ok, timeout = receiveChange(watcher)
+	c.Assert(timeout, Equals, false)
+	c.Assert(ok, Equals, true)
+	c.Assert(change.Exists, Equals, true)
+	c.Assert(change.Content, Equals, "foo")
+
+	change, ok, timeout = receiveChange(watcher)
+	c.Assert(timeout, Equals, false)
+	c.Assert(ok, Equals, true)
+	c.Assert(change.Exists, Equals, false)
+	c.Assert(change.Content, Equals, "")
+
+	change, ok, timeout = receiveChange(watcher)
+	c.Assert(timeout, Equals, false)
+	c.Assert(ok, Equals, true)
+	c.Assert(change.Exists, Equals, true)
+	c.Assert(change.Content, Equals, "done")
+
+	// No more changes.
+	_, _, timeout = receiveChange(watcher)
+	c.Assert(timeout, Equals, true)
+
+	err := watcher.Stop()
+	c.Assert(err, IsNil)
+
+	// Changes() has to be closed.
+	_, ok, timeout = receiveChange(watcher)
+	c.Assert(ok, Equals, false)
+	c.Assert(timeout, Equals, false)
+}
+
 func (s *WatcherSuite) TestContentWatcher(c *C) {
+	receiveChange := func(w *watcher.ContentWatcher) (string, bool, bool) {
+		select {
+		case change, ok := <-w.Changes():
+			return change, ok, false
+		case <-time.After(200 * time.Millisecond):
+			return "", false, true
+		}
+		return "", false, false
+	}
+	s.createPath(c, "init")
 	watcher := watcher.NewContentWatcher(s.zkConn, s.path)
 
 	go func() {
 		time.Sleep(50 * time.Millisecond)
 		s.changeContent(c, "foo")
-
 		time.Sleep(50 * time.Millisecond)
 		s.changeContent(c, "foo")
-
 		time.Sleep(50 * time.Millisecond)
 		s.changeContent(c, "bar")
 	}()
 
-	// Receive the two changes.
-	change := <-watcher.Changes()
+	// Receive the three changes. First one is from creation, 
+	// the content watcher needs an existing node.
+	change, ok, timeout := receiveChange(watcher)
+	c.Assert(timeout, Equals, false)
+	c.Assert(ok, Equals, true)
+	c.Assert(change, Equals, "init")
+
+	change, ok, timeout = receiveChange(watcher)
+	c.Assert(timeout, Equals, false)
+	c.Assert(ok, Equals, true)
 	c.Assert(change, Equals, "foo")
 
-	change = <-watcher.Changes()
+	change, ok, timeout = receiveChange(watcher)
+	c.Assert(timeout, Equals, false)
+	c.Assert(ok, Equals, true)
 	c.Assert(change, Equals, "bar")
 
 	// No more changes.
-	select {
-	case <-watcher.Changes():
-		c.Fatalf("no more changes expected")
-	case <-time.After(200 * time.Millisecond):
-		// The timeout is expected.
-	}
+	_, _, timeout = receiveChange(watcher)
+	c.Assert(timeout, Equals, true)
 
 	err := watcher.Stop()
 	c.Assert(err, IsNil)
 
-	// Changes() has to be closed
-	select {
-	case _, ok := <-watcher.Changes():
-		c.Assert(ok, Equals, false)
-	case <-time.After(200 * time.Millisecond):
-		// Timeout should not be needed.
-		c.Fatalf("timeout should not happen")
+	// Changes() has to be closed.
+	_, ok, timeout = receiveChange(watcher)
+	c.Assert(ok, Equals, false)
+	c.Assert(timeout, Equals, false)
+}
+
+func (s *WatcherSuite) TestContentWatcherDeletedNode(c *C) {
+	receiveChange := func(w *watcher.ContentWatcher) (string, bool, bool) {
+		select {
+		case change, ok := <-w.Changes():
+			return change, ok, false
+		case <-time.After(200 * time.Millisecond):
+			return "", false, true
+		}
+		return "", false, false
 	}
+	s.createPath(c, "init")
+	watcher := watcher.NewContentWatcher(s.zkConn, s.path)
+
+	// Receive initiol creation event.
+	change := <-watcher.Changes()
+	c.Assert(change, Equals, "init")
+
+	go func() {
+		time.Sleep(50 * time.Millisecond)
+		s.removePath(c)
+	}()
+
+	// Changes() has to be closed.
+	_, ok, timeout := receiveChange(watcher)
+	c.Assert(ok, Equals, false)
+	c.Assert(timeout, Equals, false)
+
+	err := watcher.Stop()
+	c.Assert(err, ErrorMatches, `watcher: node "/watcher" has been deleted`)
 }
 
 func (s *WatcherSuite) TestChildrenWatcher(c *C) {
+	receiveChange := func(w *watcher.ChildrenWatcher) (*watcher.ChildrenChange, bool, bool) {
+		select {
+		case change, ok := <-w.Changes():
+			return &change, ok, false
+		case <-time.After(200 * time.Millisecond):
+			return nil, false, true
+		}
+		return nil, false, false
+	}
+	s.createPath(c, "init")
 	watcher := watcher.NewChildrenWatcher(s.zkConn, s.path)
 
 	go func() {
 		time.Sleep(50 * time.Millisecond)
 		s.changeChildren(c, true, "foo")
-
 		time.Sleep(50 * time.Millisecond)
 		s.changeChildren(c, true, "bar")
-
 		time.Sleep(50 * time.Millisecond)
 		s.changeChildren(c, false, "foo")
 	}()
 
 	// Receive the three changes.
-	change := <-watcher.Changes()
+	change, ok, timeout := receiveChange(watcher)
+	c.Assert(timeout, Equals, false)
+	c.Assert(ok, Equals, true)
 	c.Assert(change.Added, DeepEquals, []string{"foo"})
 
-	change = <-watcher.Changes()
+	change, ok, timeout = receiveChange(watcher)
+	c.Assert(timeout, Equals, false)
+	c.Assert(ok, Equals, true)
 	c.Assert(change.Added, DeepEquals, []string{"bar"})
 
-	change = <-watcher.Changes()
+	change, ok, timeout = receiveChange(watcher)
+	c.Assert(timeout, Equals, false)
+	c.Assert(ok, Equals, true)
 	c.Assert(change.Deleted, DeepEquals, []string{"foo"})
 
 	// No more changes.
-	select {
-	case <-watcher.Changes():
-		c.Fatalf("no more changes expected")
-	case <-time.After(time.Second):
-		// The timeout is expected.
-	}
-
-	err := watcher.Stop()
-	c.Assert(err, IsNil)
-
-	// Changes() has to be closed
-	select {
-	case _, ok := <-watcher.Changes():
-		c.Assert(ok, Equals, false)
-	case <-time.After(200 * time.Millisecond):
-		// Timeout should not be needed.
-		c.Fatalf("timeout should not happen")
-	}
-}
-
-func (s *WatcherSuite) TestDeletedNode(c *C) {
-	watcher := watcher.NewContentWatcher(s.zkConn, s.path)
-
-	go func() {
-		time.Sleep(50 * time.Millisecond)
-		testing.ZkRemoveTree(s.zkConn, s.path)
-	}()
-
-	// Changes() has to be closed
-	select {
-	case _, ok := <-watcher.Changes():
-		c.Assert(ok, Equals, false)
-	case <-time.After(200 * time.Millisecond):
-		// Timeout should not be needed.
-		c.Fatalf("timeout should not happen")
-	}
-
-	err := watcher.Stop()
-	c.Assert(err, ErrorMatches, `watcher: node "/watcher" has been deleted`)
+	_, _, timeout = receiveChange(watcher)
+	c.Assert(timeout, Equals, true)
+
+	err := watcher.Stop()
+	c.Assert(err, IsNil)
+
+	// Changes() has to be closed.
+	_, ok, timeout = receiveChange(watcher)
+	c.Assert(ok, Equals, false)
+	c.Assert(timeout, Equals, false)
+}
+
+func (s *WatcherSuite) createPath(c *C, content string) {
+	_, err := s.zkConn.Create(s.path, content, 0, zookeeper.WorldACL(zookeeper.PERM_ALL))
+	c.Assert(err, IsNil)
+}
+
+func (s *WatcherSuite) removePath(c *C) {
+	testing.ZkRemoveTree(s.zkConn, s.path)
 }
 
 func (s *WatcherSuite) changeContent(c *C, content string) {