worker/instancepoller: batch instancepoller

worker/instancepoller: batch instancepoller

use ratelimit Token Bucket to batch 
requests in the instance poller.

Mitigates launchpad issue 1277397

https://code.launchpad.net/~mfoord/juju-core/instancepoller-aggregate/+merge/209966

(do not edit description out of merge proposal)

[fuzzyman, 2014-03-12 18:17:11]

Please take a look.

[natefinch, 2014-03-12 19:48:50]

Generally LGTM... just a few minor things.

https://codereview.appspot.com/74900044/diff/1/worker/instancepoller/aggregat...
File worker/instancepoller/aggregate.go (right):

https://codereview.appspot.com/74900044/diff/1/worker/instancepoller/aggregat...
worker/instancepoller/aggregate.go:22: type aggregator struct {
Commenting this would make it easier for people to understand what it does. 
It's not exported, so it's not strictly necessary, but it's nice :)

https://codereview.appspot.com/74900044/diff/1/worker/instancepoller/aggregat...
worker/instancepoller/aggregate.go:60: var GatherTime = 3 * time.Second
Always comment all public fields :)

Actually... why are these public? They seem to be an implementation detail.

https://codereview.appspot.com/74900044/diff/1/worker/instancepoller/aggregat...
worker/instancepoller/aggregate.go:67: bucket := ratelimit.New(GatherTime,
Capacity)
I'm not sure we really need a rate limiter here.  A simple timer channel would
work just as well.  I don't know that it's worth changing at this point, but
take a look at time.After(), it's a super useful construct for making a channel
that fires after a certain period of time.

https://codereview.appspot.com/74900044/diff/1/worker/instancepoller/aggregat...
worker/instancepoller/aggregate.go:101: // instanceInfo returns the instance
info for the given id
name in comment doesn't match the function

https://codereview.appspot.com/74900044/diff/1/worker/instancepoller/aggregat...
worker/instancepoller/aggregate.go:103: func (*aggregator) instInfo(id
instance.Id, inst instance.Instance) (instanceInfo, error) {
Since you're not using the receiver (*aggregator), you can just make this a bare
function and not a method on aggregator.

https://codereview.appspot.com/74900044/diff/1/worker/instancepoller/aggregat...
worker/instancepoller/aggregate.go:109: return instanceInfo{}, err
probably a good idea to prepend this error with some identifying information,
like fmt.Errorf("error getting addresses for instance %v: %v", id, err)

https://codereview.appspot.com/74900044/diff/1/worker/instancepoller/aggregat...
File worker/instancepoller/aggregate_test.go (right):

https://codereview.appspot.com/74900044/diff/1/worker/instancepoller/aggregat...
worker/instancepoller/aggregate_test.go:60: return
bare returns are generally discouraged except where required for doing tricks
with defer.  you can just make this return result, err... it seem redundant, but
it makes it more clear what's actually being returned.

https://codereview.appspot.com/74900044/diff/1/worker/instancepoller/aggregat...
worker/instancepoller/aggregate_test.go:177: c.Assert(reply.err, gc.DeepEquals,
fmt.Errorf("gotcha"))
minor, but better if you can factor out the gotcha error to a global variable in
this file.

https://codereview.appspot.com/74900044/diff/1/worker/instancepoller/aggregat...
worker/instancepoller/aggregate_test.go:193: c.Assert(err, gc.NotNil)
we have a policy in this case to check for the specific error, even though it's
not exposed by the package at all.

[rog, 2014-03-12 20:50:06]

The code and the test coverage looks generally great, thanks. Quite a few
comments here but nothing particularly substantial.

https://codereview.appspot.com/74900044/diff/1/worker/instancepoller/aggregat...
File worker/instancepoller/aggregate.go (right):

https://codereview.appspot.com/74900044/diff/1/worker/instancepoller/aggregat...
worker/instancepoller/aggregate.go:60: var GatherTime = 3 * time.Second
On 2014/03/12 19:48:51, nate.finch wrote:
> Always comment all public fields :)
> 
> Actually... why are these public? They seem to be an implementation detail.

agreed - there's no need to export this, or Capcity.

https://codereview.appspot.com/74900044/diff/1/worker/instancepoller/aggregat...
worker/instancepoller/aggregate.go:67: bucket := ratelimit.New(GatherTime,
Capacity)
On 2014/03/12 19:48:51, nate.finch wrote:
> I'm not sure we really need a rate limiter here.  A simple timer channel would
> work just as well.  I don't know that it's worth changing at this point, but
> take a look at time.After(), it's a super useful construct for making a
channel
> that fires after a certain period of time.  

We're already using time.After here (well, timer.Reset, which saves us making a
new timer every time). We're using the rate limiter to inform the time value
that gets passed to the timer. We could probably inline the logic here, but it's
not quite as simple as it might seem at first glance (the original version of
this logic just waited for three seconds from the first request received, but
that's not ideal because we don't really always want to delay sporadic requests
by 3 seconds)

https://codereview.appspot.com/74900044/diff/1/worker/instancepoller/aggregat...
worker/instancepoller/aggregate.go:72: case req, ok := <-a.reqc:
We never close reqc, so there's no need to check whether it's closed.

case req := <-a.reqc:

should be sufficient.

https://codereview.appspot.com/74900044/diff/1/worker/instancepoller/aggregat...
worker/instancepoller/aggregate.go:103: func (*aggregator) instInfo(id
instance.Id, inst instance.Instance) (instanceInfo, error) {
On 2014/03/12 19:48:51, nate.finch wrote:
> Since you're not using the receiver (*aggregator), you can just make this a
bare
> function and not a method on aggregator.  

Given that it's only used as part of aggregator, it seems reasonable to make it
a method, so it's obviously distinct from the rest of the package.

https://codereview.appspot.com/74900044/diff/1/worker/instancepoller/aggregat...
worker/instancepoller/aggregate.go:109: return instanceInfo{}, err
On 2014/03/12 19:48:51, nate.finch wrote:
> probably a good idea to prepend this error with some identifying information,
> like fmt.Errorf("error getting addresses for instance %v: %v", id, err)

That would break logic elsewhere in the package, which checks
errors.NotImplementedError. When we start using errgo, that
information flow will become more evident.

https://codereview.appspot.com/74900044/diff/1/worker/instancepoller/aggregat...
File worker/instancepoller/aggregate_test.go (right):

https://codereview.appspot.com/74900044/diff/1/worker/instancepoller/aggregat...
worker/instancepoller/aggregate_test.go:28: err       bool
perhaps make this of type error instead of bool - that way you
don't need a special "gotcha" value, so the specific error
can easily be defined in the test, closer to where it's tested for.

https://codereview.appspot.com/74900044/diff/1/worker/instancepoller/aggregat...
worker/instancepoller/aggregate_test.go:45: ids     []instance.Id
perhaps a comment on this field:

// ids is set when the Instances method is called.

I should say that, strictly speaking, this is playing a little
fast and loose with concurrency. Given that the worker
is in its own goroutine and has theoretical access to the
testInstanceGetter at any point, we can have unguarded
concurrent access (if you haven't read http://golang.org/ref/mem, you
should, BTW).

I don't mind too much in this particular instance, as
the worker *should* be dormant when it's not processing
any requests, but to do things right, you'd add
a sync.Mutex to the testInstanceGetter, and lock it
inside Instances, and in the test when you retrieve the
value of ids.

https://codereview.appspot.com/74900044/diff/1/worker/instancepoller/aggregat...
worker/instancepoller/aggregate_test.go:53: for _, inst := range i.results {
If you declared the results field as []instance.Instance, this could just be:

return i.results, i.err

or, for peace of mind:

return append([]instance.Instance(nil), i.results...), i.err

https://codereview.appspot.com/74900044/diff/1/worker/instancepoller/aggregat...
worker/instancepoller/aggregate_test.go:65: for _, address := range addresses {
I've seen this a few times now. We should perhaps consider adding an
instance.NewAddresses function.

https://codereview.appspot.com/74900044/diff/1/worker/instancepoller/aggregat...
worker/instancepoller/aggregate_test.go:82: aggregator.reqc <- req
reqc is really an implementation detail, and you don't need
to use it.

you can just do:

info, err := aggregator.instanceInfo("foo")
c.Assert(err, gc.IsNil)
c.Assert(info, gc.DeepEquals, ...etc)

That way you're actually testing more of the actual machinery too.

As above, no need to use the channels directly. Same applies to other tests
below.

https://codereview.appspot.com/74900044/diff/1/worker/instancepoller/aggregat...
worker/instancepoller/aggregate_test.go:89: c.Assert(testGetter.ids,
gc.DeepEquals, []instance.Id{instance.Id("foo")})
The instance.Id type conversion is unnecessary here ("foo" is an untyped string
and is assignable to any string type).

https://codereview.appspot.com/74900044/diff/1/worker/instancepoller/aggregat...
worker/instancepoller/aggregate_test.go:105: aggregator.reqc <- req
Perhaps a comment here might be appropriate.
Something like:
// The first request we send will be replied to immediately.
// Subsequent requests within gatherTime should be
// aggregated.

?

https://codereview.appspot.com/74900044/diff/1/worker/instancepoller/aggregat...
worker/instancepoller/aggregate_test.go:106: reply := <-replyChan
As above, no need to use the channels directly.

https://codereview.appspot.com/74900044/diff/1/worker/instancepoller/aggregat...
worker/instancepoller/aggregate_test.go:115: aggregator.reqc <-
instanceInfoReq{reply: replyChan2, instId: instance.Id("foo2")}
Slightly more awkward to avoid using reqc here, but I'd do it anyway.

Something like this, perhaps?

var wg sync.WaitGroup
checkInfo := func(id instance.Id, expectStatus string) {
    info, err := aggregator.instanceInfo(id)
    c.Check(err, gc.IsNil)
    c.Check(info.status, gc.Equals, expectStatus)
    wg.Done()
}

wg.Add(2)
go checkInfo("foo2", "not foobar")
go checkInfo("foo3", "ok-ish")
wg.Wait()
c.Assert(testGetter.ids, gc.DeepEquals, []instance.Id{"foo2", "foo3"})

https://codereview.appspot.com/74900044/diff/1/worker/instancepoller/aggregat...
worker/instancepoller/aggregate_test.go:177: c.Assert(reply.err, gc.DeepEquals,
fmt.Errorf("gotcha"))
On 2014/03/12 19:48:51, nate.finch wrote:
> minor, but better if you can factor out the gotcha error to a global variable
in
> this file.

+1 - then you can use gc.Equals, rather than DeepEquals.

https://codereview.appspot.com/74900044/diff/1/worker/instancepoller/aggregat...
worker/instancepoller/aggregate_test.go:188: func (s *aggregateSuite)
TestLoopDying(c *gc.C) {
I don't think we need to test this case - nothing closes the reqc channel
(easily verified), so there's actually no way for the loop to die without the
aggregator being killed. I'd lose this test.

[mfoord, 2014-03-13 14:37:40]

There's new code (tests especially) that needs reviewing. I may have one or two
more points from these reviews to address. "lbox propose" is currently failing
so it maybe a case of reverting to the diff in the mp I'm afraid.

[mfoord, 2014-03-13 14:41:23]

On 2014/03/13 14:37:40, mfoord wrote:
> There's new code (tests especially) that needs reviewing. I may have one or
two
> more points from these reviews to address. "lbox propose" is currently failing
> so it maybe a case of reverting to the diff in the mp I'm afraid.

I still need to add the NewAddresses utility function and add a comment about
the batching. Possibly also remove some unneeded instance.Id(...) type
conversions. The type I'm using for testInstanceGetter.counter is wrong (should
really be a thread safe counter even though it's strictly unnecessary). 

I have removed the loop return on channel dying code, so there are code changes
as well as test changes.

[fuzzyman, 2014-03-13 16:13:10]

Please take a look.

[rog, 2014-03-13 16:32:59]

LGTM with the timing-based test fixed to make it less potentially flaky.

https://codereview.appspot.com/74900044/diff/20001/worker/instancepoller/aggr...
File worker/instancepoller/aggregate.go (right):

https://codereview.appspot.com/74900044/diff/20001/worker/instancepoller/aggr...
worker/instancepoller/aggregate.go:60: var capacity int64 = 1
const capacity = 1

or just inline it, possibly with a comment saying
why we choose a capacity of 1 for the bucket.

https://codereview.appspot.com/74900044/diff/20001/worker/instancepoller/aggr...
File worker/instancepoller/aggregate_test.go (right):

https://codereview.appspot.com/74900044/diff/20001/worker/instancepoller/aggr...
worker/instancepoller/aggregate_test.go:47: ids     []instance.Id
It would still be nice to see a comment here.

https://codereview.appspot.com/74900044/diff/20001/worker/instancepoller/aggr...
worker/instancepoller/aggregate_test.go:48: results []*testInstance
I'd still suggest that this is []instance.Instance, unless there's a particular
reason not to.

Then Instances is just:

func (i *testInstanceGetter) Instances(ids []instance.Id) (result
[]instance.Instance, err error) {
     i.id = ids
     atomic.AddInt32(i.counter, 1)
     return i.result, i.err
}

https://codereview.appspot.com/74900044/diff/20001/worker/instancepoller/aggr...
worker/instancepoller/aggregate_test.go:148: c.Assert(testGetter.counter,
jc.GreaterThan, 10)
I think this is potentially flaky. If this test runs on a heavily loaded
machine, the millisecond sleeps may expand indefinitely.

I think a better approach might be to measure the total time taken between
starting the requests and receiving all the results, then checking that the
number of requests is no greater than that time divided by 10ms + 1.

[fuzzyman, 2014-03-13 17:58:38]

Please take a look.

[rog, 2014-03-13 18:26:57]

LGTM with the fix below.

https://codereview.appspot.com/74900044/diff/40001/worker/instancepoller/aggr...
File worker/instancepoller/aggregate_test.go (right):

https://codereview.appspot.com/74900044/diff/40001/worker/instancepoller/aggr...
worker/instancepoller/aggregate_test.go:57: i.counter =
atomic.AddInt32(&i.counter, 1)
ha ha, this is wrong, although it will work in the single-threaded case. The
reason AddInt32 takes a pointer is because it does the atomic increment itself.
By assigning to i.counter outside of that, we raise the possibility that we
might undo an add made by another concurrent call to AddInt32.

just lose the "i.counter = " bit.