Merge pull request #3850 from mjs/extract-upgrade-steps

Extract the upgradesteps worker to its own package

This change moves the upgradesteps worker from cmd/jujud/agent to worker/upgradesteps and makes the required changes to allow this to work. This move is required for the upcoming change to get this worker running under the dependency engine.

A number of cleanups were made along the way including:
- a state opening function is now passed in to the worker instead of the worker having knowledge of how to do this.
- the function to set machine status is now passed via an interface instead of pulling this off the agent

The upgradesteps worker code is now cleaner but there are a number of other improvements to make. These are documented in a new set of TODOs at the top of worker/upgradesteps/worker.go and will be addressed in the next PR.

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

cmd/jujud/agent/machine.go

@@ -103,6 +103,7 @@ import (
 	"github.com/juju/juju/worker/txnpruner"
 	"github.com/juju/juju/worker/unitassigner"
 	"github.com/juju/juju/worker/upgrader"
+	"github.com/juju/juju/worker/upgradesteps"
 )
 
 const bootstrapMachineId = "0"
@@ -279,7 +280,7 @@ func MachineAgentFactoryFn(
 			machineId,
 			agentConfWriter,
 			bufferedLogs,
-			NewUpgradeWorkerContext(),
+			upgradesteps.NewUpgradeWorkerContext(),
 			worker.NewRunner(cmdutil.IsFatal, cmdutil.MoreImportant, worker.RestartDelay),
 			loopDeviceManager,
 			rootDir,
@@ -292,7 +293,7 @@ func NewMachineAgent(
 	machineId string,
 	agentConfWriter AgentConfigWriter,
 	bufferedLogs logsender.LogRecordCh,
-	upgradeWorkerContext *upgradeWorkerContext,
+	upgradeWorkerContext *upgradesteps.UpgradeWorkerContext,
 	runner worker.Runner,
 	loopDeviceManager looputil.LoopDeviceManager,
 	rootDir string,
@@ -322,7 +323,7 @@ type MachineAgent struct {
 	rootDir              string
 	bufferedLogs         logsender.LogRecordCh
 	configChangedVal     voyeur.Value
-	upgradeWorkerContext *upgradeWorkerContext
+	upgradeWorkerContext *upgradesteps.UpgradeWorkerContext
 	workersStarted       chan struct{}
 
 	// XXX(fwereade): these smell strongly of goroutine-unsafeness.
@@ -701,10 +702,10 @@ func (a *MachineAgent) APIWorker() (_ worker.Worker, err error) {
 
 	runner := newConnRunner(st)
 
-	// Run the agent upgrader and the upgrade-steps worker without waiting for
+	// Run the agent upgrader and the upgradesteps worker without waiting for
 	// the upgrade steps to complete.
 	runner.StartWorker("upgrader", a.agentUpgraderWorkerStarter(st.Upgrader(), agentConfig))
-	runner.StartWorker("upgrade-steps", a.upgradeStepsWorkerStarter(st, machine.Jobs()))
+	runner.StartWorker("upgradesteps", a.upgradeStepsWorkerStarter(st, machine.Jobs()))
 
 	// All other workers must wait for the upgrade steps to complete before starting.
 	a.startWorkerAfterUpgrade(runner, "api-post-upgrade", func() (worker.Worker, error) {
@@ -917,12 +918,53 @@ func (a *MachineAgent) Restart() error {
 }
 
 func (a *MachineAgent) upgradeStepsWorkerStarter(
-	st api.Connection,
+	apiConn api.Connection,
 	jobs []multiwatcher.MachineJob,
 ) func() (worker.Worker, error) {
 	return func() (worker.Worker, error) {
-		return a.upgradeWorkerContext.Worker(a, st, jobs), nil
+		tag, ok := a.Tag().(names.MachineTag)
+		if !ok {
+			return nil, errors.New("agent's tag is not a machine tag")
+		}
+		machine, err := apiConn.Machiner().Machine(tag)
+		if err != nil {
+			return nil, errors.Trace(err)
+		}
+		return a.upgradeWorkerContext.Worker(a, apiConn, jobs, a.openStateForUpgrade, machine)
+	}
+}
+
+// openStateForUpgrade exists to be passed into the upgradesteps
+// worker. The upgradesteps worker opens state independently of the
+// state worker so that it isn't affected by the state worker's
+// lifetime. It ensures the MongoDB server is configured and started,
+// and then opens a state connection.
+//
+// TODO(mjs)- review the need for this once the dependency engine is
+// in use. Why can't upgradesteps depend on the main state connection?
+func (a *MachineAgent) openStateForUpgrade() (*state.State, func(), error) {
+	agentConfig := a.CurrentConfig()
+	if err := a.ensureMongoServer(agentConfig); err != nil {
+		return nil, nil, errors.Trace(err)
+	}
+	info, ok := agentConfig.MongoInfo()
+	if !ok {
+		return nil, nil, errors.New("no state info available")
 	}
+	st, err := state.Open(agentConfig.Environment(), info, mongo.DefaultDialOpts(), environs.NewStatePolicy())
+	if err != nil {
+		return nil, nil, errors.Trace(err)
+	}
+
+	// Ensure storage is available during upgrades.
+	stor := statestorage.NewStorage(st.EnvironUUID(), st.MongoSession())
+	registerSimplestreamsDataSource(stor)
+
+	closer := func() {
+		unregisterSimplestreamsDataSource()
+		st.Close()
+	}
+	return st, closer, nil
 }
 
 func (a *MachineAgent) agentUpgraderWorkerStarter(
@@ -1715,18 +1757,6 @@ func (a *MachineAgent) upgradeWaiterWorker(name string, start func() (worker.Wor
 	})
 }
 
-func (a *MachineAgent) setMachineStatus(apiState api.Connection, status params.Status, info string) error {
-	tag := a.Tag().(names.MachineTag)
-	machine, err := apiState.Machiner().Machine(tag)
-	if err != nil {
-		return errors.Trace(err)
-	}
-	if err := machine.SetStatus(status, info, nil); err != nil {
-		return errors.Trace(err)
-	}
-	return nil
-}
-
 // WorkersStarted returns a channel that's closed once all top level workers
 // have been started. This is provided for testing purposes.
 func (a *MachineAgent) WorkersStarted() <-chan struct{} {

cmd/jujud/agent/unit/manifolds.go

@@ -119,7 +119,7 @@ func Manifolds(config ManifoldsConfig) dependency.Manifolds {
 		// and the agent to be restarted running the new tools. We should only
 		// need one of these in a consolidated agent, but we'll need to be
 		// careful about behavioural differences, and interactions with the
-		// upgrade-steps worker.
+		// upgradesteps worker.
 		UpgraderName: upgrader.Manifold(upgrader.ManifoldConfig{
 			AgentName:     AgentName,
 			APICallerName: APICallerName,

featuretests/upgrade_test.go

@@ -33,6 +33,7 @@ import (
 	"github.com/juju/juju/upgrades"
 	"github.com/juju/juju/version"
 	"github.com/juju/juju/worker/upgrader"
+	"github.com/juju/juju/worker/upgradesteps"
 )
 
 type exposedAPI bool
@@ -59,8 +60,8 @@ func (s *upgradeSuite) SetUpTest(c *gc.C) {
 	s.oldVersion.Minor = 16
 
 	// Don't wait so long in tests.
-	s.PatchValue(&agentcmd.UpgradeStartTimeoutMaster, time.Duration(time.Millisecond*50))
-	s.PatchValue(&agentcmd.UpgradeStartTimeoutSecondary, time.Duration(time.Millisecond*60))
+	s.PatchValue(&upgradesteps.UpgradeStartTimeoutMaster, time.Duration(time.Millisecond*50))
+	s.PatchValue(&upgradesteps.UpgradeStartTimeoutSecondary, time.Duration(time.Millisecond*60))
 
 	// TODO(mjs) - the following should maybe be part of AgentSuite.SetUpTest()
 	s.PatchValue(&cmdutil.EnsureMongoServer, func(mongo.EnsureServerParams) error {
@@ -102,7 +103,7 @@ func (s *upgradeSuite) TestLoginsDuringUpgrade(c *gc.C) {
 		}
 		return nil
 	}
-	s.PatchValue(&agentcmd.PerformUpgrade, fakePerformUpgrade)
+	s.PatchValue(&upgradesteps.PerformUpgrade, fakePerformUpgrade)
 
 	a := s.newAgent(c, machine)
 	go func() { c.Check(a.Run(nil), jc.ErrorIsNil) }()
@@ -157,7 +158,7 @@ func (s *upgradeSuite) TestDowngradeOnMasterWhenOtherStateServerDoesntStartUpgra
 	fakeIsMachineMaster := func(*state.State, string) (bool, error) {
 		return true, nil
 	}
-	s.PatchValue(&agentcmd.IsMachineMaster, fakeIsMachineMaster)
+	s.PatchValue(&upgradesteps.IsMachineMaster, fakeIsMachineMaster)
 
 	// Start the agent
 	agent := s.newAgent(c, machineA)

worker/upgradesteps/package_test.go

@@ -0,0 +1,14 @@
+// Copyright 2015 Canonical Ltd.
+// Licensed under the AGPLv3, see LICENCE file for details.
+
+package upgradesteps
+
+import (
+	stdtesting "testing"
+
+	"github.com/juju/juju/testing"
+)
+
+func TestAll(t *stdtesting.T) {
+	testing.MgoTestPackage(t)
+}