Merge lp:~fwereade/juju-core/cleaner-resumer-integrate into lp:~go-bot/juju-core/trunk

[William Reade]

jujud: integrate cleaner, resumer

the stateReporter chan may be useful elsewhere, but I haven't trawled for
further simplifications. Main point is integrating these tasks and testing wat
we can.

https://codereview.appspot.com/10825044/

jujud: integrate cleaner, resumer

the stateReporter chan may be useful elsewhere, but I haven't trawled for
further simplifications. Main point is integrating these tasks and testing wat
we can.

https://codereview.appspot.com/10825044/

[William Reade (fwereade)]

Reviewers: mp+174013_code.launchpad.net,

Message:
Please take a look.

Description:
jujud: integrate cleaner, resumer

the stateReporter chan may be useful elsewhere, but I haven't trawled
for
further simplifications. Main point is integrating these tasks and
testing wat
we can.

https://code.launchpad.net/~fwereade/juju-core/cleaner-resumer-integrate/+merge/174013

(do not edit description out of merge proposal)

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

Affected files:
   A [revision details]
   M cmd/jujud/machine.go
   M cmd/jujud/machine_test.go

[Dimiter Naydenov (dimitern)]

LGTM, only a couple of clarification requests.

https://codereview.appspot.com/10825044/diff/1/cmd/jujud/machine_test.go
File cmd/jujud/machine_test.go (right):

https://codereview.appspot.com/10825044/diff/1/cmd/jujud/machine_test.go#newcode318
cmd/jujud/machine_test.go:318: agentStates := make(chan *state.State,
1000)
why 1000? why not unbuffered?

https://codereview.appspot.com/10825044/diff/1/cmd/jujud/machine_test.go#newcode319
cmd/jujud/machine_test.go:319: defer
sendOpenedStates(sendOpenedStates(agentStates))
this a but confusing - perhaps a comment before this whole block
explaining what's being done will help

https://codereview.appspot.com/10825044/

[John A Meinel (jameinel)]

LGTM

I have some comments, but I can live with the code as it is written. So
if having a helper is too much distraction, just land it.

https://codereview.appspot.com/10825044/diff/1/cmd/jujud/machine_test.go
File cmd/jujud/machine_test.go (right):

https://codereview.appspot.com/10825044/diff/1/cmd/jujud/machine_test.go#newcode318
cmd/jujud/machine_test.go:318: agentStates := make(chan *state.State,
1000)
On 2013/07/10 18:28:46, dimitern wrote:
> why 1000? why not unbuffered?

I don't think we want unbuffered, because then when the other loop
starts up it will block until we do our select (I think).
The trick is that once you call 'sendOpenedStates' *all* things that
start up are going to be trying to send their state to this channel.

The general idea is: "I'm going to start something up, and I want a
backdoor into the State object it is using, so send it onto this channel
so I can pull it out in this code."

However, once you set the channel, more things are going to be sent on
it than you really need. We could just create a local goroutine that
consumes them all and puts it somewhere we can use it.

https://codereview.appspot.com/10825044/diff/1/cmd/jujud/machine_test.go#newcode319
cmd/jujud/machine_test.go:319: defer
sendOpenedStates(sendOpenedStates(agentStates))
On 2013/07/10 18:28:46, dimitern wrote:
> this a but confusing - perhaps a comment before this whole block
explaining
> what's being done will help

A slightly better pattern is probably:

oldChan := sendOpenedStates(agentStates)
defer sendOpenedStates(oldChan)

For things like this, I usually prefer to have the sendOpenedStates type
function actually return a func() that does the cleanup. So the code
becomes:

func sendOpenedStates(ch <-chan *state.State) {
   old, agentStates := agentStates, ch
   return func() { agentStates := old) }
}

As then the code in the test case is:

cleanup := sendOpenedStates(agentStates)
defer cleanup()

And that is a common pattern regardless of what internal poking a given
test helper has to do.

Nested function calls and defer are hard to parse for mere mortals,
which is why I don't spell it as:

defer sendOpenedStates(agentStates)()
     ^------------------------------^
The indicator of what you are deferring is very far from 'defer' vs

cleanup := ...
defer cleanup()

It is also why I'm not 100% sold on the:

go func() {
...
...
...
...
...
..
...
.
.
.
.
..
}()

Syntax. At least it is "go func() {" which makes it clear you are
thinking about calling func at the end of this. But when you leave off
those last two parenthesis it gets pretty confusing. :)

https://codereview.appspot.com/10825044/diff/1/cmd/jujud/machine_test.go#newcode329
cmd/jujud/machine_test.go:329: case <-time.After(500 *
time.Millisecond):
Please use the testing.LongWait constants. It will help signal *why* you
are picking a given timeout, and lets us tweak them across the codebase
if we find they are too long/short.

https://codereview.appspot.com/10825044/diff/1/cmd/jujud/machine_test.go#newcode386
cmd/jujud/machine_test.go:386: timeout := time.After(500 *
time.Millisecond)
Same here on the timeout.

This loop has been used elsewhere, but I'm not quite sure why we hav...

[William Reade (fwereade)]

https://codereview.appspot.com/10825044/diff/1/cmd/jujud/machine_test.go
File cmd/jujud/machine_test.go (right):

https://codereview.appspot.com/10825044/diff/1/cmd/jujud/machine_test.go#newcode318
cmd/jujud/machine_test.go:318: agentStates := make(chan *state.State,
1000)
On 2013/07/11 05:50:16, jameinel wrote:
> On 2013/07/10 18:28:46, dimitern wrote:
> > why 1000? why not unbuffered?

> I don't think we want unbuffered, because then when the other loop
starts up it
> will block until we do our select (I think).
> The trick is that once you call 'sendOpenedStates' *all* things that
start up
> are going to be trying to send their state to this channel.

> The general idea is: "I'm going to start something up, and I want a
backdoor
> into the State object it is using, so send it onto this channel so I
can pull it
> out in this code."

> However, once you set the channel, more things are going to be sent on
it than
> you really need. We could just create a local goroutine that consumes
them all
> and puts it somewhere we can use it.

In practice, only one state should be sent, because we're not expecting
errors and restarts. But if we get them, that buffer should be big
enough to ensure nothing blocks. That's all :). I don't *think* it's
worth the extra infrastructure to check that no more come in; if things
are broken I think we'll detect it in other ways.

https://codereview.appspot.com/10825044/diff/1/cmd/jujud/machine_test.go#newcode319
cmd/jujud/machine_test.go:319: defer
sendOpenedStates(sendOpenedStates(agentStates))
On 2013/07/11 05:50:16, jameinel wrote:
> On 2013/07/10 18:28:46, dimitern wrote:
> > this a but confusing - perhaps a comment before this whole block
explaining
> > what's being done will help

> A slightly better pattern is probably:

> oldChan := sendOpenedStates(agentStates)
> defer sendOpenedStates(oldChan)

> For things like this, I usually prefer to have the sendOpenedStates
type
> function actually return a func() that does the cleanup. So the code
becomes:

> func sendOpenedStates(ch <-chan *state.State) {
> old, agentStates := agentStates, ch
> return func() { agentStates := old) }
> }

> As then the code in the test case is:

> cleanup := sendOpenedStates(agentStates)
> defer cleanup()

> And that is a common pattern regardless of what internal poking a
given test
> helper has to do.

> Nested function calls and defer are hard to parse for mere mortals,
which is why
> I don't spell it as:

> defer sendOpenedStates(agentStates)()
> ^------------------------------^
> The indicator of what you are deferring is very far from 'defer' vs

> cleanup := ...
> defer cleanup()

> It is also why I'm not 100% sold on the:

> go func() {
> ...
> ...
> ...
> ...
> ...
> ..
> ...
> .
> .
> .
> .
> ..
> }()

> Syntax. At least it is "go func() {" which makes it clear you are
thinking about
> calling func at the end of this. But when you leave off those last two
> parenthesis it gets pretty confusing. :)

Done as undo func.

https://codereview.appspot.com/10825044/diff/1/cmd/jujud/machine_test.go#newcode329
cmd/jujud/machine_test.go:329: case <-time.After(500 *
time.Millisecond):
On 2013/07/11 05:50:16, jameinel wrot...

=== modified file 'cmd/jujud/machine.go'
--- cmd/jujud/machine.go	2013-07-08 18:40:36 +0000
+++ cmd/jujud/machine.go	2013-07-11 12:18:26 +0000
@@ -16,9 +16,11 @@
 	"launchpad.net/juju-core/state/api/params"
 	"launchpad.net/juju-core/state/apiserver"
 	"launchpad.net/juju-core/worker"
+	"launchpad.net/juju-core/worker/cleaner"
 	"launchpad.net/juju-core/worker/firewaller"
 	"launchpad.net/juju-core/worker/machiner"
 	"launchpad.net/juju-core/worker/provisioner"
+	"launchpad.net/juju-core/worker/resumer"
 	"launchpad.net/tomb"
 	"path/filepath"
 	"time"
@@ -154,6 +156,7 @@
 	if err != nil {
 		return nil, err
 	}
+	reportOpenedState(st)
 	m := entity.(*state.Machine)
 	// TODO(rog) use more discriminating test for errors
 	// rather than taking everything down indiscriminately.
@@ -206,6 +209,15 @@
 				}
 				return apiserver.NewServer(st, fmt.Sprintf(":%d", a.Conf.APIPort), a.Conf.StateServerCert, a.Conf.StateServerKey)
 			})
+			runner.StartWorker("cleaner", func() (worker.Worker, error) {
+				return cleaner.NewCleaner(st), nil
+			})
+			runner.StartWorker("resumer", func() (worker.Worker, error) {
+				// The action of resumer is so subtle that it is not tested,
+				// because we can't figure out how to do so without brutalising
+				// the transaction log.
+				return resumer.NewResumer(st), nil
+			})
 		default:
 			log.Warningf("ignoring unknown job %q", job)
 		}
@@ -241,3 +253,22 @@
 func (a *MachineAgent) Tag() string {
 	return state.MachineTag(a.MachineId)
 }
+
+// Below pieces are used for testing,to give us access to the *State opened
+// by the agent, and allow us to trigger syncs without waiting 5s for them
+// to happen automatically.
+
+var stateReporter chan<- *state.State
+
+func reportOpenedState(st *state.State) {
+	select {
+	case stateReporter <- st:
+	default:
+	}
+}
+
+func sendOpenedStates(dst chan<- *state.State) (undo func()) {
+	var original chan<- *state.State
+	original, stateReporter = stateReporter, dst
+	return func() { stateReporter = original }
+}
 
=== modified file 'cmd/jujud/machine_test.go'
--- cmd/jujud/machine_test.go	2013-07-09 11:31:00 +0000
+++ cmd/jujud/machine_test.go	2013-07-11 12:18:26 +0000
@@ -307,35 +307,44 @@
 	RetryDelay: 10 * time.Millisecond,
 }
 
-func (s *MachineSuite) TestServeAPI(c *C) {
-	stm, conf, _ := s.primeAgent(c, state.JobManageState)
+func (s *MachineSuite) assertJobWithState(c *C, job state.MachineJob, test func(*agent.Conf, *state.State)) {
+	stm, conf, _ := s.primeAgent(c, job)
 	addAPIInfo(conf, stm)
 	err := conf.Write()
 	c.Assert(err, IsNil)
 	a := s.newAgent(c, stm)
+	defer a.Stop()
+
+	agentStates := make(chan *state.State, 1000)
+	undo := sendOpenedStates(agentStates)
+	defer undo()
+
 	done := make(chan error)
 	go func() {
 		done <- a.Run(nil)
 	}()
 
-	st, err := api.Open(conf.APIInfo, fastDialOpts)
-	c.Assert(err, IsNil)
-	defer st.Close()
-
-	// This just verifies we can log in successfully.
-	m, err := st.Machiner().Machine(stm.Tag())
-	c.Assert(err, IsNil)
-	c.Assert(m.Life(), Equals, params.Alive)
+	select {
+	case agentState := <-agentStates:
+		c.Assert(agentState, NotNil)
+		test(conf, agentState)
+	case <-time.After(testing.LongWait):
+		c.Fatalf("state not opened")
+	}
 
 	err = a.Stop()
-	// When shutting down, the API server can be shut down before
-	// the other workers that connect to it, so they get an error so
-	// they then die, causing Stop to return an error.  It's not
-	// easy to control the actual error that's received in this
-	// circumstance so we just log it rather than asserting that it
-	// is not nil.
-	if err != nil {
-		c.Logf("error shutting down: %v", err)
+	if job == state.JobManageState {
+		// When shutting down, the API server can be shut down before
+		// the other workers that connect to it, so they get an error so
+		// they then die, causing Stop to return an error.  It's not
+		// easy to control the actual error that's received in this
+		// circumstance so we just log it rather than asserting that it
+		// is not nil.
+		if err != nil {
+			c.Logf("error shutting down state manager: %v", err)
+		}
+	} else {
+		c.Assert(err, IsNil)
 	}
 
 	select {
@@ -346,6 +355,55 @@
 	}
 }
 
+func (s *MachineSuite) TestManageStateServesAPI(c *C) {
+	s.assertJobWithState(c, state.JobManageState, func(conf *agent.Conf, agentState *state.State) {
+		st, err := api.Open(conf.APIInfo, fastDialOpts)
+		c.Assert(err, IsNil)
+		defer st.Close()
+		m, err := st.Machiner().Machine(conf.APIInfo.Tag)
+		c.Assert(err, IsNil)
+		c.Assert(m.Life(), Equals, params.Alive)
+	})
+}
+
+func (s *MachineSuite) TestManageStateRunsCleaner(c *C) {
+	s.assertJobWithState(c, state.JobManageState, func(conf *agent.Conf, agentState *state.State) {
+		// Create a service and unit, and destroy the service.
+		service, err := s.State.AddService("wordpress", s.AddTestingCharm(c, "wordpress"))
+		c.Assert(err, IsNil)
+		unit, err := service.AddUnit()
+		c.Assert(err, IsNil)
+		err = service.Destroy()
+		c.Assert(err, IsNil)
+
+		// Check the unit was not yet removed.
+		err = unit.Refresh()
+		c.Assert(err, IsNil)
+		w := unit.Watch()
+		defer w.Stop()
+
+		// Trigger a sync on the state used by the agent, and wait
+		// for the unit to be removed.
+		agentState.Sync()
+		timeout := time.After(testing.LongWait)
+		for done := false; !done; {
+			select {
+			case <-timeout:
+				c.Fatalf("unit not cleaned up")
+			case <-time.After(testing.ShortWait):
+				s.State.StartSync()
+			case <-w.Changes():
+				err := unit.Refresh()
+				if errors.IsNotFoundError(err) {
+					done = true
+				} else {
+					c.Assert(err, IsNil)
+				}
+			}
+		}
+	})
+}
+
 var serveAPIWithBadConfTests = []struct {
 	change func(c *agent.Conf)
 	err    string