state/api: fixes to parallel api.Open

state/api/apiclient.go

 // Copyright 2012, 2013 Canonical Ltd.
 // Licensed under the AGPLv3, see LICENCE file for details.
 
 package api
 
 import (
         "crypto/tls"
         "crypto/x509"
+        "fmt"
         "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/rpc"
         "launchpad.net/juju-core/rpc/jsoncodec"
@@ skipping 58 matching lines @@
         Password string
 
         // Nonce holds the nonce used when provisioning the machine. Used
         // only by the machine agent.
         Nonce string `yaml:",omitempty"`
 }
 
 // DialOpts holds configuration parameters that control the
 // Dialing behavior when connecting to a state server.
 type DialOpts struct {
+        // DialAddressInterval is the amount of time to wait
+        // before starting to dial another address.
+        DialAddressInterval time.Duration
+
         // Timeout is the amount of time to wait contacting
         // a state server.
         Timeout time.Duration
 
         // RetryDelay is the amount of time to wait between
         // unsucssful connection attempts.
         RetryDelay time.Duration
 }
 
 // DefaultDialOpts returns a DialOpts representing the default
 // parameters for contacting a state server.
 func DefaultDialOpts() DialOpts {
         return DialOpts{
-                Timeout:    10 * time.Minute,
-                RetryDelay: 2 * time.Second,
+                DialAddressInterval: 50 * time.Millisecond,
+                Timeout:             10 * time.Minute,
+                RetryDelay:          2 * time.Second,
         }
 }
 
 func Open(info *Info, opts DialOpts) (*State, error) {
         pool := x509.NewCertPool()
         xcert, err := cert.ParseCert(info.CACert)
         if err != nil {
                 return nil, err
         }
         pool.AddCert(xcert)
 
         // 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 {
+                err := dialWebsocket(addr, opts, pool, try)
+                if err == parallel.ErrStopped {
+                        break
+                } else if err != nil {

[rog, 2014/04/01 07:42:07]

s/else//

[axw, 2014/04/01 07:54:08]

Done.

                         return nil, err
                 }
+                select {
+                case <-time.After(opts.DialAddressInterval):
+                case <-try.Dead():
+                }
         }
         try.Close()
         result, err := try.Result()
         if err != nil {
                 return nil, err
         }
         conn := result.(*websocket.Conn)
         logger.Infof("connection established to %q", conn.RemoteAddr())
 
         client := rpc.NewConn(jsoncodec.NewWebsocket(conn), nil)
@@ skipping 23 matching lines @@
         // 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",
         }
+        return try.Start(newWebsocketDialer(cfg, opts))
+}
+
+// new WebsocketDialler returns a function that
+// can be passed to utils/parallel.Try.Start.
+func newWebsocketDialer(cfg *websocket.Config, opts DialOpts) func(<-chan struct{}) (io.Closer, error) {
         openAttempt := utils.AttemptStrategy{
                 Total: opts.Timeout,
                 Delay: opts.RetryDelay,
         }
-        return try.Start(func(stop <-chan struct{}) (io.Closer, error) {
-                err := parallel.ErrStopped
+        return func(stop <-chan struct{}) (io.Closer, error) {
                 for a := openAttempt.Start(); a.Next(); {
                         select {
                         case <-stop:
-                                break
+                                return nil, parallel.ErrStopped
                         default:
                         }
                         logger.Infof("dialing %q", cfg.Location)
-                        var conn *websocket.Conn
-                        conn, err = websocket.DialConfig(cfg)
+                        conn, err := websocket.DialConfig(cfg)
                         if err == nil {
                                 return conn, nil
                         }
-                        logger.Debugf("error dialing API server, will retry: %v", err)
+                        if a.HasNext() {
+                                logger.Debugf("error dialing API server, will retry: %v", err)

[rog, 2014/04/01 07:42:07]

perhaps include the cfg.Location here, otherwise when we've got a few of these
happening concurrently, we won't be able to tell which address is giving which
error.

[axw, 2014/04/01 07:54:08]

Done.

+                        } else {
+                                return nil, fmt.Errorf("timed out connecting to %q", cfg.Location)
+                        }
                 }
-                return nil, err
-        })
+                panic("unreachable")
+        }
 }
 
 func (s *State) heartbeatMonitor() {
         for {
                 if err := s.Ping(); err != nil {
                         close(s.broken)
                         return
                 }
                 time.Sleep(PingPeriod)
         }
@@ skipping 47 matching lines @@
 // Juju CLI, all addresses must be attempted, as the CLI may
 // be invoked both within and outside the environment (think
 // private clouds).
 func (s *State) APIHostPorts() [][]instance.HostPort {
         hostPorts := make([][]instance.HostPort, len(s.hostPorts))
         for i, server := range s.hostPorts {
                 hostPorts[i] = append([]instance.HostPort{}, server...)
         }
         return hostPorts
 }