Merge lp:~niemeyer/juju-core/mstate-machine-watcher into lp:~juju/juju-core/trunk

[Gustavo Niemeyer]

mstate: add machine watcher

This is the first high-level watcher after the foundation for
watcher is in place, so it adds a few missing pieces onto the
state logic itself too.

https://codereview.appspot.com/6489104/

[Gustavo Niemeyer (niemeyer)]

Reviewers: mp+123614_code.launchpad.net,

Message:
Please take a look.

Description:
mstate: add machine watcher

This is the first high-level watcher after the foundation for
watcher is in place, so it adds a few missing pieces onto the
state logic itself too.

https://code.launchpad.net/~niemeyer/juju-core/mstate-machine-watcher/+merge/123614

Requires:
https://code.launchpad.net/~niemeyer/juju-core/mstate-watcher-helpers/+merge/123596

(do not edit description out of merge proposal)

Please review this at https://codereview.appspot.com/6489104/

Affected files:
   A [revision details]
   M mstate/export_test.go
   M mstate/machine.go
   M mstate/machine_test.go
   M mstate/open.go
   M mstate/state.go
   M mstate/state_test.go
   A mstate/watcher.go

[William Reade (fwereade)]

LGTM modulo locale worries.

https://codereview.appspot.com/6489104/diff/1/mstate/open.go
File mstate/open.go (right):

https://codereview.appspot.com/6489104/diff/1/mstate/open.go#newcode45
mstate/open.go:45: // Quite unfortunate that the error has no
appropriate code.
Very much so. Can we be screwed by locale settings here?

https://codereview.appspot.com/6489104/

[Gustavo Niemeyer (niemeyer)]

*** Submitted:

mstate: add machine watcher

This is the first high-level watcher after the foundation for
watcher is in place, so it adds a few missing pieces onto the
state logic itself too.

R=fwereade
CC=
https://codereview.appspot.com/6489104

https://codereview.appspot.com/6489104/diff/1/mstate/open.go
File mstate/open.go (right):

https://codereview.appspot.com/6489104/diff/1/mstate/open.go#newcode45
mstate/open.go:45: // Quite unfortunate that the error has no
appropriate code.
On 2012/09/11 07:28:07, fwereade wrote:
> Very much so. Can we be screwed by locale settings here?

Not at the moment at least. I've checked the upstream code and it
doesn't translate the message.

I've included in the comment the URL of the upstream bug reported on
this issue.

https://codereview.appspot.com/6489104/

[Aram Hăvărneanu (aramh)]

https://codereview.appspot.com/6489104/diff/5001/mstate/machine.go
File mstate/machine.go (right):

https://codereview.appspot.com/6489104/diff/5001/mstate/machine.go#newcode23
mstate/machine.go:23: TxnRevno int64 `bson:"txn-revno"`
This complicates tests and watchers because Insert can't set TxnRevno
and DeepEquals will fail. How about reverting this, and create a new
machineDocTxn like this:

type machineDocTxn struct {
 machineDoc `bson:",inline"`
 TxnRevno int64 `bson:"txn-revno"`
}

and use that where we need it?

https://codereview.appspot.com/6489104/diff/5001/mstate/watcher.go
File mstate/watcher.go (right):

https://codereview.appspot.com/6489104/diff/5001/mstate/watcher.go#newcode68
mstate/watcher.go:68: case w.changeChan <- m:
Shouldn't this be a default case, and w.changeChan <- m *after* the
select? That way we will always completely drain ch before sending m.
Now there's a chance we will send m before draining ch.

https://codereview.appspot.com/6489104/

[Gustavo Niemeyer (niemeyer)]

https://codereview.appspot.com/6489104/diff/5001/mstate/machine.go
File mstate/machine.go (right):

https://codereview.appspot.com/6489104/diff/5001/mstate/machine.go#newcode23
mstate/machine.go:23: TxnRevno int64 `bson:"txn-revno"`
On 2012/09/11 17:40:53, aram wrote:
> This complicates tests and watchers because Insert can't set TxnRevno
and
> DeepEquals will fail. How about reverting this, and create a new
machineDocTxn
> like this:

The reasoning for doing it doesn't sound great. Doing DeepEquals on such
a structure is most definitely wrong. Private data is private, and by
definition we should be able to cache information freely or change it
without breaking tests that work on the public APIs.

It sounds sensible to have a omitempty setting there, though, so that we
don't try to insert it. Can you please make the change on a follow up
CL?

https://codereview.appspot.com/6489104/diff/5001/mstate/watcher.go
File mstate/watcher.go (right):

https://codereview.appspot.com/6489104/diff/5001/mstate/watcher.go#newcode68
mstate/watcher.go:68: case w.changeChan <- m:
On 2012/09/11 17:40:53, aram wrote:
> Shouldn't this be a default case, and w.changeChan <- m *after* the
select? That
> way we will always completely drain ch before sending m. Now there's a
chance we
> will send m before draining ch.

No, there's some misunderstanding about how the watcher works. The
channel is never drained, and the construction here is purposefully
operating *while* trying to send.

https://codereview.appspot.com/6489104/

[Aram Hăvărneanu (aramh)]

> The reasoning for doing it doesn't sound great. Doing DeepEquals on
such
> a structure is most definitely wrong. Private data is private, and by
> definition we should be able to cache information freely or change it
> without breaking tests that work on the public APIs.

It may be wrong, but in that case we have to do lots of work to fix it
because all the documents will have this field.

  white:mstate$ cd state
  /home/aram/src/launchpad.net/juju-core/state
  white:state$ g DeepEquals | wc -l
  89
  white:state$ cd mstate
  /home/aram/src/launchpad.net/juju-core/mstate
  white:mstate$ g DeepEquals | wc -l
  58

To be honest, using DeepEquals in tests always felt wrong to be because
it was checking the implementation details instead of relying on the
interface, but I assumed it was ok since this is the way state tests
were written in the first place.

I will put omitempty, but this doesn't fix the tests. Entities returned
by Insert will have TxtRevno always zero but entities returned by other
querying functions or by Refresh will have a non-zero TxtRevno.

If you are against this change, perhaps an alternative is to call
Refresh() in tests where this might matter.

https://codereview.appspot.com/6489104/

[Gustavo Niemeyer (niemeyer)]

On Tue, Sep 11, 2012 at 3:13 PM, <email address hidden> wrote:
> white:mstate$ cd state
> /home/aram/src/launchpad.net/juju-core/state
> white:state$ g DeepEquals | wc -l
> 89
> white:state$ cd mstate
> /home/aram/src/launchpad.net/juju-core/mstate
> white:mstate$ g DeepEquals | wc -l
> 58

This is counting the number of times we've used DeepEquals, which
doesn't really mean much.

How many of those entries are comparing the private fields of machine?

> I will put omitempty, but this doesn't fix the tests. Entities returned
> by Insert will have TxtRevno always zero but entities returned by other
> querying functions or by Refresh will have a non-zero TxtRevno.

That's a good point. How can we fix it?

gustavo @ http://niemeyer.net

=== modified file 'mstate/export_test.go'
--- mstate/export_test.go	2012-08-16 13:44:09 +0000
+++ mstate/export_test.go	2012-09-10 18:17:20 +0000
@@ -12,3 +12,7 @@
 	m := &Machine{doc: machineDoc(*doc)}
 	return m.String()
 }
+
+func init() {
+	logSize = logSizeTests
+}

=== modified file 'mstate/machine.go'
--- mstate/machine.go	2012-09-06 17:04:54 +0000
+++ mstate/machine.go	2012-09-10 18:17:20 +0000
@@ -20,6 +20,7 @@
 	Id         int `bson:"_id"`
 	InstanceId string
 	Life       Life
+	TxnRevno   int64 `bson:"txn-revno"`
 }
 
 func newMachine(st *State, doc *machineDoc) *Machine {
@@ -73,6 +74,11 @@
 	return nil
 }
 
+// Watch returns a watcher that fires when the machine changes.
+func (m *Machine) Watch() *MachineWatcher {
+	return newMachineWatcher(m)
+}
+
 // AgentAlive returns whether the respective remote agent is alive.
 func (m *Machine) AgentAlive() bool {
 	return m.st.presencew.Alive(m.globalKey())

=== modified file 'mstate/machine_test.go'
--- mstate/machine_test.go	2012-09-06 17:04:54 +0000
+++ mstate/machine_test.go	2012-09-10 18:17:20 +0000
@@ -3,6 +3,7 @@
 import (
 	. "launchpad.net/gocheck"
 	state "launchpad.net/juju-core/mstate"
+	"launchpad.net/juju-core/version"
 	"sort"
 	"time"
 )
@@ -220,3 +221,103 @@
 	sort.Strings(names)
 	return names
 }
+
+type machineInfo struct {
+	tools      *state.Tools
+	instanceId string
+}
+
+func tools(tools int, url string) *state.Tools {
+	return &state.Tools{
+		URL: url,
+		Binary: version.Binary{
+			Number: version.Number{0, 0, tools},
+			Series: "series",
+			Arch:   "arch",
+		},
+	}
+}
+
+var watchMachineTests = []struct {
+	test func(m *state.Machine) error
+	want machineInfo
+}{
+	{
+		func(m *state.Machine) error {
+			return nil
+		},
+		machineInfo{
+			tools: &state.Tools{},
+		},
+	},
+	{
+		func(m *state.Machine) error {
+			return m.SetInstanceId("m-foo")
+		},
+		machineInfo{
+			tools:      &state.Tools{},
+			instanceId: "m-foo",
+		},
+	},
+	{
+		func(m *state.Machine) error {
+			return m.SetInstanceId("")
+		},
+		machineInfo{
+			tools:      &state.Tools{},
+			instanceId: "",
+		},
+	},
+	// TODO SetAgentTools is missing.
+	//{
+	//	func(m *state.Machine) error {
+	//		return m.SetAgentTools(tools(3, "baz"))
+	//	},
+	//	machineInfo{
+	//		tools: tools(3, "baz"),
+	//	},
+	//},
+	//{
+	//	func(m *state.Machine) error {
+	//		return m.SetAgentTools(tools(4, "khroomph"))
+	//	},
+	//	machineInfo{
+	//		tools: tools(4, "khroomph"),
+	//	},
+	//},
+}
+
+func (s *MachineSuite) TestWatchMachine(c *C) {
+	w := s.machine.Watch()
+	defer func() {
+		c.Assert(w.Stop(), IsNil)
+	}()
+	for i, test := range watchMachineTests {
+		c.Logf("test %d", i)
+		err := test.test(s.machine)
+		c.Assert(err, IsNil)
+		s.State.StartSync()
+		select {
+		case m, ok := <-w.Changes():
+			c.Assert(ok, Equals, true)
+			c.Assert(m.Id(), Equals, s.machine.Id())
+			var info machineInfo
+			// TODO AgentTools is missing.
+			info.tools = test.want.tools
+			//info.tools, err = m.AgentTools()
+			//c.Assert(err, IsNil)
+			info.instanceId, err = m.InstanceId()
+			if _, ok := err.(*state.NotFoundError); !ok {
+				c.Assert(err, IsNil)
+			}
+			c.Assert(info, DeepEquals, test.want)
+		case <-time.After(500 * time.Millisecond):
+			c.Fatalf("did not get change: %v", test.want)
+		}
+	}
+	select {
+	case got := <-w.Changes():
+		c.Fatalf("got unexpected change: %#v", got)
+	case <-time.After(100 * time.Millisecond):
+	}
+}

=== modified file 'mstate/open.go'
--- mstate/open.go	2012-09-06 22:12:07 +0000
+++ mstate/open.go	2012-09-10 18:17:20 +0000
@@ -1,10 +1,12 @@
 package mstate
 
 import (
+	"fmt"
 	"labix.org/v2/mgo"
 	"labix.org/v2/mgo/txn"
 	"launchpad.net/juju-core/log"
 	"launchpad.net/juju-core/mstate/presence"
+	"launchpad.net/juju-core/mstate/watcher"
 )
 
 var indexes = []mgo.Index{
@@ -12,6 +14,14 @@
 	{Key: []string{"endpoints.servicename"}},
 }
 
+// The capped collection used for transaction logs defaults to 200MB.
+// It's tweaked in export_test.go to 1MB to avoid the overhead of
+// creating and deleting the large file repeatedly.
+var (
+	logSize      = 200000000
+	logSizeTests = 1000000
+)
+
 func Dial(servers string) (*State, error) {
 	log.Printf("opening state with servers: %q", servers)
 	session, err := mgo.Dial(servers)
@@ -19,32 +29,44 @@
 		return nil, err
 	}
 	db := session.DB("juju")
-	presencedb := session.DB("presence")
-	txns := db.C("txns")
+	pdb := session.DB("presence")
 	st := &State{
-		db:         db,
-		presencedb: presencedb,
-		charms:     db.C("charms"),
-		machines:   db.C("machines"),
-		relations:  db.C("relations"),
-		services:   db.C("services"),
-		settings:   db.C("settings"),
-		units:      db.C("units"),
-		presence:   presencedb.C("presence"),
-		runner:     txn.NewRunner(txns),
-	}
-	st.presencew = presence.NewWatcher(st.presence)
+		db:        db,
+		charms:    db.C("charms"),
+		machines:  db.C("machines"),
+		relations: db.C("relations"),
+		services:  db.C("services"),
+		settings:  db.C("settings"),
+		units:     db.C("units"),
+		presence:  pdb.C("presence"),
+	}
+	log := db.C("txns.log")
+	info := mgo.CollectionInfo{Capped: true, MaxBytes: logSize}
+	// Quite unfortunate that the error has no appropriate code.
+	if err := log.Create(&info); err != nil && err.Error() != "collection already exists" {
+		return nil, fmt.Errorf("cannot create log collection: %v", err)
+	}
+	st.runner = txn.NewRunner(db.C("txns"))
+	st.runner.ChangeLog(db.C("txns.log"))
+	st.watcher = watcher.New(db.C("txns.log"))
+	st.presencew = presence.NewWatcher(pdb.C("presence"))
 	for _, index := range indexes {
 		err = st.relations.EnsureIndex(index)
 		if err != nil {
-			return nil, err
+			return nil, fmt.Errorf("cannot create database index: %v", err)
 		}
 	}
 	return st, nil
 }
 
 func (st *State) Close() error {
-	err := st.presencew.Stop()
+	err1 := st.presencew.Stop()
+	err2 := st.watcher.Stop()
 	st.db.Session.Close()
-	return err
+	for _, err := range []error{err1, err2} {
+		if err != nil {
+			return err
+		}
+	}
+	return nil
 }

=== modified file 'mstate/state.go'
--- mstate/state.go	2012-09-06 16:54:22 +0000
+++ mstate/state.go	2012-09-10 18:17:20 +0000
@@ -11,6 +11,7 @@
 	"launchpad.net/juju-core/charm"
 	"launchpad.net/juju-core/environs/config"
 	"launchpad.net/juju-core/mstate/presence"
+	"launchpad.net/juju-core/mstate/watcher"
 	"launchpad.net/juju-core/trivial"
 	"launchpad.net/juju-core/version"
 	"net/url"
@@ -28,7 +29,6 @@
 // managed by juju.
 type State struct {
 	db         *mgo.Database
-	presencedb *mgo.Database
 	charms     *mgo.Collection
 	machines   *mgo.Collection
 	relations  *mgo.Collection
@@ -36,8 +36,9 @@
 	settings   *mgo.Collection
 	units      *mgo.Collection
 	presence   *mgo.Collection
+	runner     *txn.Runner
+	watcher    *watcher.Watcher
 	presencew  *presence.Watcher
-	runner     *txn.Runner
 }
 
 func deadOnAbort(err error) error {
@@ -369,3 +370,11 @@
 func (s *State) ForcePresenceRefresh() {
 	s.presencew.ForceRefresh()
 }
+
+// StartSync forces watchers to resynchronize their state with the
+// database immediately. This will happen periodically automatically.
+func (s *State) StartSync() {
+	// TODO Make presence more like watcher, add it here, and
+	// remove ForcePresenceRefresh.
+	s.watcher.StartSync()
+}

=== modified file 'mstate/state_test.go'
--- mstate/state_test.go	2012-09-06 16:54:22 +0000
+++ mstate/state_test.go	2012-09-10 18:17:20 +0000
@@ -7,7 +7,7 @@
 	"launchpad.net/juju-core/charm"
 	"launchpad.net/juju-core/environs/config"
 	state "launchpad.net/juju-core/mstate"
-	coretesting "launchpad.net/juju-core/testing"
+	"launchpad.net/juju-core/testing"
 	"net/url"
 )
 
@@ -19,9 +19,18 @@
 
 var _ = Suite(&StateSuite{})
 
+func (s *StateSuite) TestDialAgain(c *C) {
+	// Ensure idempotent operations on Dial are working fine.
+	for i := 0; i < 2; i++ {
+		st, err := state.Dial(testing.MgoAddr)
+		c.Assert(err, IsNil)
+		c.Assert(st.Close(), IsNil)
+	}
+}
+
 func (s *StateSuite) TestAddCharm(c *C) {
 	// Check that adding charms from scratch works correctly.
-	ch := coretesting.Charms.Dir("dummy")
+	ch := testing.Charms.Dir("dummy")
 	curl := charm.MustParseURL(
 		fmt.Sprintf("local:series/%s-%d", ch.Meta().Name, ch.Revision()),
 	)

=== added file 'mstate/watcher.go'
--- mstate/watcher.go	1970-01-01 00:00:00 +0000
+++ mstate/watcher.go	2012-09-10 18:17:20 +0000
@@ -0,0 +1,74 @@
+package mstate
+
+import (
+	"launchpad.net/juju-core/mstate/watcher"
+	"launchpad.net/tomb"
+)
+
+// MachineWatcher observes changes to the settings of a machine.
+type MachineWatcher struct {
+	changeChan chan *Machine
+	tomb       tomb.Tomb
+}
+
+// newMachineWatcher creates and starts a watcher to watch information
+// about the machine.
+func newMachineWatcher(m *Machine) *MachineWatcher {
+	w := &MachineWatcher{changeChan: make(chan *Machine)}
+	go func() {
+		defer w.tomb.Done()
+		defer close(w.changeChan)
+		w.tomb.Kill(w.loop(m))
+	}()
+	return w
+}
+
+// Changes returns a channel that will receive the new
+// *Machine when a change is detected. Note that multiple
+// changes may be observed as a single event in the channel.
+// The first event on the channel holds the initial state
+// as returned by Machine.Info.
+func (w *MachineWatcher) Changes() <-chan *Machine {
+	return w.changeChan
+}
+
+func (w *MachineWatcher) Stop() error {
+	w.tomb.Kill(nil)
+	return w.tomb.Wait()
+}
+
+func (w *MachineWatcher) loop(m *Machine) (err error) {
+	ch := make(chan watcher.Change)
+	id := m.Id()
+	st := m.st
+	st.watcher.Watch(st.machines.Name, id, m.doc.TxnRevno, ch)
+	defer st.watcher.Unwatch(st.machines.Name, id, ch)
+	for {
+		select {
+		case <-st.watcher.Dying():
+			return watcher.MustErr(st.watcher)
+		case <-w.tomb.Dying():
+			return tomb.ErrDying
+		case <-ch:
+		}
+		if m, err = st.Machine(id); err != nil {
+			return err
+		}
+		for {
+			select {
+			case <-st.watcher.Dying():
+				return watcher.MustErr(st.watcher)
+			case <-w.tomb.Dying():
+				return tomb.ErrDying
+			case <-ch:
+				if err := m.Refresh(); err != nil {
+					return err
+				}
+				continue
+			case w.changeChan <- m:
+			}
+			break
+		}
+	}
+	return nil
+}