Merge pull request #4266 from wallyworld/lease-manager-restart

Pass new state to apiserver worker each time it is started

Fixes: https://bugs.launchpad.net/juju-core/+bug/1539656

Each time apiserverworker is started, we need a fresh copy of state.
I experimented with aborting/resetting the lease manager to avoid having to create a new copy of state but it got messy. This is the simplest approach. We can revisit once William's todo with setting up the lease manager (in the HackLeadership code) is done.

(Review request: http://reviews.vapour.ws/r/3704/)

apiserver/apiserver.go

@@ -319,6 +319,7 @@ func (srv *Server) run(lis net.Listener) {
 		srv.wg.Wait()              // wait for any outstanding requests to complete.
 		srv.tomb.Done()
 		srv.statePool.Close()
+		srv.state.Close()
 	}()
 
 	srv.wg.Add(1)

cmd/jujud/agent/machine.go

@@ -1096,7 +1096,14 @@ func (a *MachineAgent) StateWorker() (worker.Worker, error) {
 			//
 			// TODO(ericsnow) For now we simply do not close the channel.
 			certChangedChan := make(chan params.StateServingInfo, 1)
-			runner.StartWorker("apiserver", a.apiserverWorkerStarter(st, certChangedChan))
+			// Each time aipserver worker is restarted, we need a fresh copy of state due
+			// to the fact that state holds lease managers which are killed and need to be reset.
+			stateOpener := func() (*state.State, error) {
+				logger.Debugf("opening state for apistate worker")
+				st, _, err := openState(agentConfig, stateWorkerDialOpts)
+				return st, err
+			}
+			runner.StartWorker("apiserver", a.apiserverWorkerStarter(stateOpener, certChangedChan))
 			var stateServingSetter certupdater.StateServingInfoSetter = func(info params.StateServingInfo, done <-chan struct{}) error {
 				return a.ChangeConfig(func(config agent.ConfigSetter) error {
 					config.SetStateServingInfo(info)
@@ -1419,8 +1426,16 @@ func _getFirewallMode(apiSt api.Connection) (string, error) {
 // journaling is enabled.
 var stateWorkerDialOpts mongo.DialOpts
 
-func (a *MachineAgent) apiserverWorkerStarter(st *state.State, certChanged chan params.StateServingInfo) func() (worker.Worker, error) {
-	return func() (worker.Worker, error) { return a.newApiserverWorker(st, certChanged) }
+func (a *MachineAgent) apiserverWorkerStarter(
+	stateOpener func() (*state.State, error), certChanged chan params.StateServingInfo,
+) func() (worker.Worker, error) {
+	return func() (worker.Worker, error) {
+		st, err := stateOpener()
+		if err != nil {
+			return nil, errors.Trace(err)
+		}
+		return a.newApiserverWorker(st, certChanged)
+	}
 }
 
 func (a *MachineAgent) newApiserverWorker(st *state.State, certChanged chan params.StateServingInfo) (worker.Worker, error) {