Merge lp:~axwalk/juju-core/apiclient-open-parallel into lp:~go-bot/juju-core/trunk

[Andrew Wilkins]

state/api: dial info.Addrs in parallel

This change is to dial all addresses in
parallel. The first one that successfully
connects is used.

https://codereview.appspot.com/82450043/

state/api: dial info.Addrs in parallel

This change is to dial all addresses in
parallel. The first one that successfully
connects is used.

https://codereview.appspot.com/82450043/

[Andrew Wilkins (axwalk)]

Reviewers: mp+213401_code.launchpad.net,

Message:
Please take a look.

Description:
state/api: dial info.Addrs in parallel

This change is to dial all addresses in
parallel. The first one that successfully
connects is used.

https://code.launchpad.net/~axwalk/juju-core/apiclient-open-parallel/+merge/213401

(do not edit description out of merge proposal)

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

Affected files (+135, -25 lines):
   A [revision details]
   M state/api/apiclient.go
   A state/api/apiclient_test.go

[Dimiter Naydenov (dimitern)]

LGTM

https://codereview.appspot.com/82450043/diff/1/state/api/apiclient_test.go
File state/api/apiclient_test.go (right):

https://codereview.appspot.com/82450043/diff/1/state/api/apiclient_test.go#newcode10
state/api/apiclient_test.go:10: //jc "github.com/juju/testing/checkers"
d

https://codereview.appspot.com/82450043/

[John A Meinel (jameinel)]

A couple of comments, but nothing that would block landing this.

LGTM

https://codereview.appspot.com/82450043/diff/1/state/api/apiclient.go
File state/api/apiclient.go (right):

https://codereview.appspot.com/82450043/diff/1/state/api/apiclient.go#newcode142
state/api/apiclient.go:142: // TODO what does "origin" really mean, and
is localhost always ok?
origin would have sense if you were doing actual web requests, I think
it is the location that redirected you here.
I believe the websocket requires something, but it accepts localhost, so
we just do it.

It isn't actually used in our implementation, so it is ok.

See something like:
http://learnitcorrect.com/blog/websocket-is-great-but-not-the-origin-policy.html

It is intended to avoid cross-site scripting, etc. But we just let
people connect to our websockets, and do the auth inside the websocket
(as part of Login) rather than assuming that connections to us are
already valid.

https://codereview.appspot.com/82450043/diff/1/state/api/apiclient.go#newcode167
state/api/apiclient.go:167: return conn, nil
Would we want to check <-stop at this point, in case we got a stop while
the dial was processing?

I guess it has to be handled up a layer anyway. We just know that
DialConfig is blocking, so we sort of expect to be in the middle of
dialing while we get a stop request. (because someone else succeeded.)

https://codereview.appspot.com/82450043/diff/1/state/api/apiclient.go#newcode169
state/api/apiclient.go:169: log.Debugf("state/api: %v", err)
maybe a little more context about "error trying to Dial API server, will
retry: %v" ?

It depends what errors we generally get, but it seems helpful to give a
bit more context than just an error.

https://codereview.appspot.com/82450043/

[Andrew Wilkins (axwalk)]

Please take a look.

https://codereview.appspot.com/82450043/diff/1/state/api/apiclient.go
File state/api/apiclient.go (right):

https://codereview.appspot.com/82450043/diff/1/state/api/apiclient.go#newcode142
state/api/apiclient.go:142: // TODO what does "origin" really mean, and
is localhost always ok?
On 2014/03/31 12:22:11, jameinel wrote:
> origin would have sense if you were doing actual web requests, I think
it is the
> location that redirected you here.
> I believe the websocket requires something, but it accepts localhost,
so we just
> do it.

> It isn't actually used in our implementation, so it is ok.

> See something like:

http://learnitcorrect.com/blog/websocket-is-great-but-not-the-origin-policy.html

> It is intended to avoid cross-site scripting, etc. But we just let
people
> connect to our websockets, and do the auth inside the websocket (as
part of
> Login) rather than assuming that connections to us are already valid.

I just moved that TODO, but thanks for the info :)
I'll remove the TODO, and update the comment.

https://codereview.appspot.com/82450043/diff/1/state/api/apiclient.go#newcode167
state/api/apiclient.go:167: return conn, nil
On 2014/03/31 12:22:11, jameinel wrote:
> Would we want to check <-stop at this point, in case we got a stop
while the
> dial was processing?

> I guess it has to be handled up a layer anyway. We just know that
DialConfig is
> blocking, so we sort of expect to be in the middle of dialing while we
get a
> stop request. (because someone else succeeded.)

It is indeed handled higher up; utils/parallel.Try.Start creates a
closure that will dispose of the result by Close()ing it once "stop"
("dying" inside Try) is closed.

https://codereview.appspot.com/82450043/diff/1/state/api/apiclient.go#newcode169
state/api/apiclient.go:169: log.Debugf("state/api: %v", err)
On 2014/03/31 12:22:11, jameinel wrote:
> maybe a little more context about "error trying to Dial API server,
will retry:
> %v" ?

> It depends what errors we generally get, but it seems helpful to give
a bit more
> context than just an error.

Done.

https://codereview.appspot.com/82450043/diff/1/state/api/apiclient_test.go
File state/api/apiclient_test.go (right):

https://codereview.appspot.com/82450043/diff/1/state/api/apiclient_test.go#newcode10
state/api/apiclient_test.go:10: //jc "github.com/juju/testing/checkers"
On 2014/03/31 11:55:23, dimitern wrote:
> d

Done.

https://codereview.appspot.com/82450043/

[Go Bot (go-bot)]

Attempt to merge into lp:juju-core failed due to conflicts:

text conflict in state/api/apiclient.go

[Andrew Wilkins (axwalk)]

Please take a look.

https://codereview.appspot.com/82450043/

[Roger Peppe (rogpeppe)]

A few fixes that would be nice to land when possible.

https://codereview.appspot.com/82450043/diff/40001/state/api/apiclient.go
File state/api/apiclient.go (right):

https://codereview.appspot.com/82450043/diff/40001/state/api/apiclient.go#newcode117
state/api/apiclient.go:117: if err := dialWebsocket(addr, opts, pool,
try); err != nil {
This doesn't seem quite right. Try.Start returns ErrStopped if an
earlier attempt has succeeded. If that happens, we want to stop calling
dialWebsocket.

I think something like this might be better:

for _, addr := range info.Addrs {
     err := dialWebsocket(addr, opts, pool, try)
     if err == parallel.ErrStopped {
        break
     }
     if err != nil {
         return nil, err
     }
     select{
     case <-time.After(dialAddressInterval):
     case <-try.Dead():
     }
}

with dialAddressInterval being something like 50ms,
so if the connect succeeds fast, we can avoid dialing
most of the addresses.

https://codereview.appspot.com/82450043/diff/40001/state/api/apiclient.go#newcode168
state/api/apiclient.go:168: err := parallel.ErrStopped
This isn't quite right - if we used up all our attempts, we probably
want to return a timeout error rather than ErrStopped.

I quite like using HasNext for this kind of thing:

for a := openAttempt.Start(); a.Next(); {
    select{
    case <-stop:
        return nil, parallel.ErrStopped
    default:
    }
    conn, err := websocket.DialConfig(...)
    if err == nil {
        return conn, nil
    }
    if !a.HasNext() {
        return nil, fmt.Errorf("timed out connecting to %q",
cfg.Location)
    }
}
panic("unreachable")

https://codereview.appspot.com/82450043/diff/40001/state/api/apiclient.go#newcode169
state/api/apiclient.go:169: for a := openAttempt.Start(); a.Next(); {
It's a pity we don't have a method on Attempt that lets us use
it in selects, so that it could be interrupted immediately.

At some point, something like:

// NextDuration reports the interval until the
// next attempt should be made.
func (a *Attempt) NextDuration() time.Duration

might be good. Then you could write:

for a := openAttempt.Start(); a.HasNext(); {
     select {
     case <-a.stop:
         return nil, parallel.ErrStopped
     case <-time.After(a.NextDuration()):
     }
     ....
}

https://codereview.appspot.com/82450043/diff/40001/state/api/apiclient.go#newcode172
state/api/apiclient.go:172: break
this is a no-op.
I think you probably want:

    return nil, parallel.ErrStopped

It could probably do with a test.

https://codereview.appspot.com/82450043/

[Andrew Wilkins (axwalk)]

Updates over here: https://codereview.appspot.com/82900045/

https://codereview.appspot.com/82450043/diff/40001/state/api/apiclient.go
File state/api/apiclient.go (right):

https://codereview.appspot.com/82450043/diff/40001/state/api/apiclient.go#newcode117
state/api/apiclient.go:117: if err := dialWebsocket(addr, opts, pool,
try); err != nil {
On 2014/03/31 15:39:01, rog wrote:
> This doesn't seem quite right. Try.Start returns ErrStopped if an
earlier
> attempt has succeeded. If that happens, we want to stop calling
dialWebsocket.

> I think something like this might be better:

> for _, addr := range info.Addrs {
> err := dialWebsocket(addr, opts, pool, try)
> if err == parallel.ErrStopped {
> break
> }
> if err != nil {
> return nil, err
> }
> select{
> case <-time.After(dialAddressInterval):
> case <-try.Dead():
> }
> }

> with dialAddressInterval being something like 50ms,
> so if the connect succeeds fast, we can avoid dialing
> most of the addresses.

Done.

https://codereview.appspot.com/82450043/diff/40001/state/api/apiclient.go#newcode168
state/api/apiclient.go:168: err := parallel.ErrStopped
On 2014/03/31 15:39:01, rog wrote:
> This isn't quite right - if we used up all our attempts, we probably
want to
> return a timeout error rather than ErrStopped.

> I quite like using HasNext for this kind of thing:

> for a := openAttempt.Start(); a.Next(); {
> select{
> case <-stop:
> return nil, parallel.ErrStopped
> default:
> }
> conn, err := websocket.DialConfig(...)
> if err == nil {
> return conn, nil
> }
> if !a.HasNext() {
> return nil, fmt.Errorf("timed out connecting to %q",
cfg.Location)
> }
> }
> panic("unreachable")

Done.

https://codereview.appspot.com/82450043/diff/40001/state/api/apiclient.go#newcode169
state/api/apiclient.go:169: for a := openAttempt.Start(); a.Next(); {
On 2014/03/31 15:39:01, rog wrote:
> It's a pity we don't have a method on Attempt that lets us use
> it in selects, so that it could be interrupted immediately.

> At some point, something like:

> // NextDuration reports the interval until the
> // next attempt should be made.
> func (a *Attempt) NextDuration() time.Duration

> might be good. Then you could write:

> for a := openAttempt.Start(); a.HasNext(); {
> select {
> case <-a.stop:
> return nil, parallel.ErrStopped
> case <-time.After(a.NextDuration()):
> }
> ....
> }

I looked briefly, but it looks a little complicated. I'll not attempt
(ha ha) this for the moment.

https://codereview.appspot.com/82450043/diff/40001/state/api/apiclient.go#newcode172
state/api/apiclient.go:172: break
On 2014/03/31 15:39:01, rog wrote:
> this is a no-op.
> I think you probably want:

> return nil, parallel.ErrStopped

> It could probably do with a test.

/facepalm
thanks, fixed and added a test.

https://codereview.appspot.com/82450043/

=== modified file 'state/api/apiclient.go'
--- state/api/apiclient.go	2014-03-31 12:24:52 +0000
+++ state/api/apiclient.go	2014-03-31 14:44:35 +0000
@@ -6,23 +6,31 @@
 import (
 	"crypto/tls"
 	"crypto/x509"
+	"io"
 	"time"
 
 	"code.google.com/p/go.net/websocket"
+	"github.com/juju/loggo"
 
 	"launchpad.net/juju-core/cert"
 	"launchpad.net/juju-core/instance"
-	"launchpad.net/juju-core/log"
 	"launchpad.net/juju-core/rpc"
 	"launchpad.net/juju-core/rpc/jsoncodec"
 	"launchpad.net/juju-core/state/api/params"
 	"launchpad.net/juju-core/utils"
+	"launchpad.net/juju-core/utils/parallel"
 )
 
+var logger = loggo.GetLogger("juju.state.api")
+
 // PingPeriod defines how often the internal connection health check
 // will run. It's a variable so it can be changed in tests.
 var PingPeriod = 1 * time.Minute
 
+// maxParallelDial defines the maximum number addresses to dial in
+// parallel.
+const maxParallelDial = 7
+
 type State struct {
 	client *rpc.Conn
 	conn   *websocket.Conn
@@ -95,48 +103,36 @@
 }
 
 func Open(info *Info, opts DialOpts) (*State, error) {
-	// TODO Select a random address from info.Addrs
-	// and only fail when we've tried all the addresses.
-	// TODO what does "origin" really mean, and is localhost always ok?
-	cfg, err := websocket.NewConfig("wss://"+info.Addrs[0]+"/", "http://localhost/")
-	if err != nil {
-		return nil, err
-	}
 	pool := x509.NewCertPool()
 	xcert, err := cert.ParseCert(info.CACert)
 	if err != nil {
 		return nil, err
 	}
 	pool.AddCert(xcert)
-	cfg.TlsConfig = &tls.Config{
-		RootCAs:    pool,
-		ServerName: "anything",
-	}
-	var conn *websocket.Conn
-	openAttempt := utils.AttemptStrategy{
-		Total: opts.Timeout,
-		Delay: opts.RetryDelay,
-	}
-	for a := openAttempt.Start(); a.Next(); {
-		log.Infof("state/api: dialing %q", cfg.Location)
-		conn, err = websocket.DialConfig(cfg)
-		if err == nil {
-			break
+
+	// Dial all addresses, with up to maxParallelDial in parallel.
+	try := parallel.NewTry(maxParallelDial, nil)
+	defer try.Kill()
+	for _, addr := range info.Addrs {
+		if err := dialWebsocket(addr, opts, pool, try); err != nil {
+			return nil, err
 		}
-		log.Errorf("state/api: %v", err)
 	}
+	try.Close()
+	result, err := try.Result()
 	if err != nil {
 		return nil, err
 	}
-	log.Infof("state/api: connection established")
+	conn := result.(*websocket.Conn)
+	logger.Infof("connection established to %q", conn.RemoteAddr())
 
 	client := rpc.NewConn(jsoncodec.NewWebsocket(conn), nil)
 	client.Start()
 	st := &State{
 		client:     client,
 		conn:       conn,
-		addr:       cfg.Location.Host,
-		serverRoot: "https://" + cfg.Location.Host,
+		addr:       conn.Config().Location.Host,
+		serverRoot: "https://" + conn.Config().Location.Host,
 		tag:        info.Tag,
 		password:   info.Password,
 	}
@@ -151,6 +147,43 @@
 	return st, nil
 }
 
+func dialWebsocket(addr string, opts DialOpts, rootCAs *x509.CertPool, try *parallel.Try) error {
+	// origin is required by the WebSocket API, used for "origin policy"
+	// in websockets. We pass localhost to satisfy the API; it is
+	// inconsequential to us.
+	const origin = "http://localhost/"
+	cfg, err := websocket.NewConfig("wss://"+addr+"/", origin)
+	if err != nil {
+		return err
+	}
+	cfg.TlsConfig = &tls.Config{
+		RootCAs:    rootCAs,
+		ServerName: "anything",
+	}
+	openAttempt := utils.AttemptStrategy{
+		Total: opts.Timeout,
+		Delay: opts.RetryDelay,
+	}
+	return try.Start(func(stop <-chan struct{}) (io.Closer, error) {
+		err := parallel.ErrStopped
+		for a := openAttempt.Start(); a.Next(); {
+			select {
+			case <-stop:
+				break
+			default:
+			}
+			logger.Infof("dialing %q", cfg.Location)
+			var conn *websocket.Conn
+			conn, err = websocket.DialConfig(cfg)
+			if err == nil {
+				return conn, nil
+			}
+			logger.Debugf("error dialing API server, will retry: %v", err)
+		}
+		return nil, err
+	})
+}
+
 func (s *State) heartbeatMonitor() {
 	for {
 		if err := s.Ping(); err != nil {

=== added file 'state/api/apiclient_test.go'
--- state/api/apiclient_test.go	1970-01-01 00:00:00 +0000
+++ state/api/apiclient_test.go	2014-03-31 14:44:35 +0000
@@ -0,0 +1,79 @@
+// Copyright 2014 Canonical Ltd.
+// Licensed under the AGPLv3, see LICENCE file for details.
+
+package api_test
+
+import (
+	"io"
+	"net"
+
+	gc "launchpad.net/gocheck"
+
+	jujutesting "launchpad.net/juju-core/juju/testing"
+	"launchpad.net/juju-core/state/api"
+)
+
+type apiclientSuite struct {
+	jujutesting.JujuConnSuite
+}
+
+var _ = gc.Suite(&apiclientSuite{})
+
+func (s *apiclientSuite) TestOpenMultiple(c *gc.C) {
+	// Create a socket that proxies to the API server.
+	info := s.APIInfo(c)
+	serverAddr := info.Addrs[0]
+	server, err := net.Dial("tcp", serverAddr)
+	c.Assert(err, gc.IsNil)
+	defer server.Close()
+	listener, err := net.Listen("tcp", ":0")
+	c.Assert(err, gc.IsNil)
+	defer listener.Close()
+	go func() {
+		for {
+			client, err := listener.Accept()
+			if err != nil {
+				return
+			}
+			go io.Copy(client, server)
+			go io.Copy(server, client)
+		}
+	}()
+
+	// Check that we can use the proxy to connect.
+	proxyAddr := listener.Addr().String()
+	info.Addrs = []string{proxyAddr}
+	st, err := api.Open(info, api.DialOpts{})
+	c.Assert(err, gc.IsNil)
+	defer st.Close()
+	c.Assert(st.Addr(), gc.Equals, proxyAddr)
+
+	// Now break Addrs[0], and ensure that Addrs[1]
+	// is successfully connected to.
+	info.Addrs = []string{proxyAddr, serverAddr}
+	listener.Close()
+	st, err = api.Open(info, api.DialOpts{})
+	c.Assert(err, gc.IsNil)
+	defer st.Close()
+	c.Assert(st.Addr(), gc.Equals, serverAddr)
+}
+
+func (s *apiclientSuite) TestOpenMultipleError(c *gc.C) {
+	listener, err := net.Listen("tcp", ":0")
+	c.Assert(err, gc.IsNil)
+	defer listener.Close()
+	go func() {
+		for {
+			client, err := listener.Accept()
+			if err != nil {
+				return
+			}
+			client.Close()
+		}
+	}()
+	info := s.APIInfo(c)
+	addr := listener.Addr().String()
+	info.Addrs = []string{addr, addr, addr}
+	_, err = api.Open(info, api.DialOpts{})
+	c.Assert(err, gc.ErrorMatches, "websocket.Dial .*: read tcp .*: connection reset by peer")
+}

=== modified file 'state/api/state.go'
--- state/api/state.go	2014-03-31 12:24:52 +0000
+++ state/api/state.go	2014-03-31 14:44:35 +0000
@@ -14,7 +14,7 @@
 	"launchpad.net/juju-core/state/api/environment"
 	"launchpad.net/juju-core/state/api/firewaller"
 	"launchpad.net/juju-core/state/api/keyupdater"
-	"launchpad.net/juju-core/state/api/logger"
+	apilogger "launchpad.net/juju-core/state/api/logger"
 	"launchpad.net/juju-core/state/api/machiner"
 	"launchpad.net/juju-core/state/api/params"
 	"launchpad.net/juju-core/state/api/provisioner"
@@ -128,8 +128,8 @@
 }
 
 // Logger returns access to the Logger API
-func (st *State) Logger() *logger.State {
-	return logger.NewState(st)
+func (st *State) Logger() *apilogger.State {
+	return apilogger.NewState(st)
 }
 
 // KeyUpdater returns access to the KeyUpdater API