worker/peergrouper: publish API addresses

worker/peergrouper/worker.go

 // Copyright 2014 Canonical Ltd.
 // Licensed under the AGPLv3, see LICENCE file for details.
 
 package peergrouper
 
 import (
         "fmt"
         "sync"
         "time"
 
         "launchpad.net/tomb"
 
         "launchpad.net/juju-core/errors"
+        "launchpad.net/juju-core/instance"
         "launchpad.net/juju-core/replicaset"
         "launchpad.net/juju-core/state"
         "launchpad.net/juju-core/worker"
 )
 
 type stateInterface interface {
         Machine(id string) (stateMachine, error)
         WatchStateServerInfo() state.NotifyWatcher
         StateServerInfo() (*state.StateServerInfo, error)
         MongoSession() mongoSession
 }
 
 type stateMachine interface {
         Id() string
         Refresh() error
         Watch() state.NotifyWatcher
         WantsVote() bool
         HasVote() bool
         SetHasVote(hasVote bool) error
-        StateHostPort() string
+        APIHostPorts() []instance.HostPort
+        MongoHostPorts() []instance.HostPort
 }
 
 type mongoSession interface {
         CurrentStatus() (*replicaset.Status, error)
         CurrentMembers() ([]replicaset.Member, error)
         Set([]replicaset.Member) error
 }
 
+type publisherInterface interface {
+        // publish publishes information about the given state servers

[mfoord, 2014/03/19 17:36:41]

Should be"publishAPIServers publishes..."

+        // to whomsoever it may concern. When it is called there
+        // is no guarantee that any of the information has actually changed.
+        publishAPIServers(apiServers [][]instance.HostPort) error
+}
+
 // notifyFunc holds a function that is sent
 // to the main worker loop to fetch new information
 // when something changes. It reports whether
 // the information has actually changed (and by implication
 // whether the replica set may need to be changed).
-type notifyFunc func() (bool, error)
+type notifyFunc func() (changed bool, err error)
 
 var (
         // If we fail to set the mongo replica set members,
         // we retry at the following interval until we succeed.
         retryInterval = 2 * time.Second
 
         // pollInterval holds the interval at which the replica set
         // members will be updated even in the absence of changes
         // to State. This enables us to make changes to members
         // that are triggered by changes to member status.
         //
         // 10 seconds is the default time interval used by
         // mongo to keep its replicas up to date.
         pollInterval = 10 * time.Second
 )
 
 // pgWorker holds all the mutable state that we are watching.
 // The only goroutine that is allowed to modify this
 // is worker.loop - other watchers modify the
 // current state by calling worker.notify instead of
 // modifying it directly.
 type pgWorker struct {
         tomb tomb.Tomb
 
         // wg represents all the currently running goroutines.
         // The worker main loop waits for all of these to exit
         // before finishing.
         wg sync.WaitGroup
 
-        // st represents the State. It is an interface for testing
-        // purposes only.
+        // st represents the State. It is an interface so we can swap
+        // out the implementation during testing.
         st stateInterface
 
-        // When something changes that might might affect
+        // When something changes that might affect
         // the peer group membership, it sends a function
         // on notifyCh that is run inside the main worker
         // goroutine to mutate the state. It reports whether
         // the state has actually changed.
         notifyCh chan notifyFunc
 
         // machines holds the set of machines we are currently
         // watching (all the state server machines). Each one has an
         // associated goroutine that
         // watches attributes of that machine.
         machines map[string]*machine
+
+        // publisher holds the implementation of the API
+        // address publisher.
+        publisher publisherInterface
 }
 
 // New returns a new worker that maintains the mongo replica set
 // with respect to the given state.
 func New(st *state.State) (worker.Worker, error) {
         cfg, err := st.EnvironConfig()
         if err != nil {
                 return nil, err
         }
         return newWorker(&stateShim{
                 State:     st,
                 mongoPort: cfg.StatePort(),
-        }), nil
+                apiPort:   cfg.APIPort(),
+        }, newPublisher(st)), nil
 }
 
-func newWorker(st stateInterface) worker.Worker {
+func newWorker(st stateInterface, pub publisherInterface) worker.Worker {
         w := &pgWorker{
-                st:       st,
-                notifyCh: make(chan notifyFunc),
-                machines: make(map[string]*machine),
+                st:        st,
+                notifyCh:  make(chan notifyFunc),
+                machines:  make(map[string]*machine),
+                publisher: pub,
         }
+        logger.Infof("worker starting")
         go func() {
                 defer w.tomb.Done()
                 if err := w.loop(); err != nil {
                         logger.Errorf("peergrouper loop terminated: %v", err)
                         w.tomb.Kill(err)
                 }
                 // Wait for the various goroutines to be killed.
                 // N.B. we don't defer this call because
                 // if we do and a bug causes a panic, Wait will deadlock
                 // waiting for the unkilled goroutines to exit.
@@ skipping 23 matching lines @@
                         changed, err := f()
                         if err != nil {
                                 return err
                         }
                         if !changed {
                                 break
                         }
                         // Try to update the replica set immediately.
                         retry.Reset(0)
                 case <-retry.C:
+                        ok := true
+                        if err := w.publisher.publishAPIServers(w.apiHostPorts()); err != nil {
+                                logger.Errorf("cannot publish state server addresses: %v", err)
+                                ok = false
+                        }
                         if err := w.updateReplicaset(); err != nil {
                                 if _, isReplicaSetError := err.(*replicaSetError); !isReplicaSetError {
                                         return err
                                 }
                                 logger.Errorf("cannot set replicaset: %v", err)
+                                ok = false
+                        }
+                        if ok {
+                                logger.Infof("polling after %v", pollInterval)
+                                // Update the replica set members occasionally
+                                // to keep them up to date with the current
+                                // replica set member statuses.
+                                retry.Reset(pollInterval)
+                        } else {
+                                logger.Infof("retrying after %v", retryInterval)
                                 retry.Reset(retryInterval)
-                                break
                         }
 
-                        // Update the replica set members occasionally
-                        // to keep them up to date with the current
-                        // replica set member statuses.
-                        retry.Reset(pollInterval)
                 case <-w.tomb.Dying():
                         return tomb.ErrDying
                 }
         }
 }
 
+func (w *pgWorker) apiHostPorts() [][]instance.HostPort {
+        servers := make([][]instance.HostPort, 0, len(w.machines))
+        for _, m := range w.machines {
+                if len(m.apiHostPorts) > 0 {
+                        servers = append(servers, m.apiHostPorts)
+                }
+        }
+        return servers
+}
+
 // notify sends the given notification function to
 // the worker main loop to be executed.
 func (w *pgWorker) notify(f notifyFunc) bool {
         select {
         case w.notifyCh <- f:
                 return true
         case <-w.tomb.Dying():
                 return false
         }
 }
 
-// getPeerGroupInfo collates current session information about the
+// peerGroupInfo collates current session information about the
 // mongo peer group with information from state machines.
 func (w *pgWorker) peerGroupInfo() (*peerGroupInfo, error) {
         session := w.st.MongoSession()
         info := &peerGroupInfo{}
         var err error
         status, err := session.CurrentStatus()
         if err != nil {
                 return nil, fmt.Errorf("cannot get replica set status: %v", err)
         }
         info.statuses = status.Members
@@ skipping 84 matching lines @@
                 defer w.wg.Done()
                 if err := loop(); err != nil {
                         w.tomb.Kill(err)
                 }
         }()
 }
 
 // setHasVote sets the HasVote status of all the given
 // machines to hasVote.
 func setHasVote(ms []*machine, hasVote bool) error {
-
         for _, m := range ms {
                 if err := m.stm.SetHasVote(hasVote); err != nil {
                         return fmt.Errorf("cannot set voting status of %q to %v: %v", m.id, hasVote, err)
                 }
         }
         return nil
 }
 
 // serverInfoWatcher watches the state server info and
 // notifies the worker when it changes.
@@ skipping 65 matching lines @@
                         return false, fmt.Errorf("cannot get machine %q: %v", id, err)
                 }
                 infow.worker.machines[id] = infow.worker.newMachine(stm)
                 changed = true
         }
         return changed, nil
 }
 
 // machine represents a machine in State.
 type machine struct {
-        id        string
-        wantsVote bool
-        hostPort  string
+        id             string
+        wantsVote      bool
+        apiHostPorts   []instance.HostPort
+        mongoHostPorts []instance.HostPort
 
         worker         *pgWorker
         stm            stateMachine
         machineWatcher state.NotifyWatcher
 }
 
+func (m *machine) mongoHostPort() string {
+        return instance.SelectInternalHostPort(m.mongoHostPorts, false)
+}
+
 func (m *machine) String() string {
         return m.id
 }
 
 func (m *machine) GoString() string {
-        return fmt.Sprintf("&peergrouper.machine{id: %q, wantsVote: %v, hostPort: %q}", m.id, m.wantsVote, m.hostPort)
+        return fmt.Sprintf("&peergrouper.machine{id: %q, wantsVote: %v, hostPort: %q}", m.id, m.wantsVote, m.mongoHostPort())
 }
 
 func (w *pgWorker) newMachine(stm stateMachine) *machine {
         m := &machine{
                 worker:         w,
                 id:             stm.Id(),
                 stm:            stm,
-                hostPort:       stm.StateHostPort(),
+                apiHostPorts:   stm.APIHostPorts(),
+                mongoHostPorts: stm.MongoHostPorts(),
                 wantsVote:      stm.WantsVote(),
                 machineWatcher: stm.Watch(),
         }
         w.start(m.loop)
         return m
 }
 
 func (m *machine) loop() error {
         for {
                 select {
@@ skipping 23 matching lines @@
                         // loop will be stopped very soon anyway.
                         return false, nil
                 }
                 return false, err
         }
         changed := false
         if wantsVote := m.stm.WantsVote(); wantsVote != m.wantsVote {
                 m.wantsVote = wantsVote
                 changed = true
         }
-        if hostPort := m.stm.StateHostPort(); hostPort != m.hostPort {
-                m.hostPort = hostPort
+        if hps := m.stm.MongoHostPorts(); !hostPortsEqual(hps, m.mongoHostPorts) {
+                m.mongoHostPorts = hps
+                changed = true
+        }
+        if hps := m.stm.APIHostPorts(); !hostPortsEqual(hps, m.apiHostPorts) {
+                m.apiHostPorts = hps
                 changed = true
         }
         return changed, nil
 }
 
+func hostPortsEqual(hps1, hps2 []instance.HostPort) bool {
+        if len(hps1) != len(hps2) {
+                return false
+        }
+        for i := range hps1 {
+
+                if hps1[i] != hps2[i] {
+                        return false
+                }
+        }
+        return true
+}
+
 func inStrings(t string, ss []string) bool {
         for _, s := range ss {
                 if s == t {
                         return true
                 }
         }
         return false
 }