Merge lp:~axwalk/juju-core/instancedistributor-policy into lp:~go-bot/juju-core/trunk

[Andrew Wilkins]

Introduce InstanceDistributor policy

This is a policy that is invoked whenever
we attempt to assign a unit to an empty
machine.

The policy will be invoked with the
associated instances (i.e. provisioned,
empty machines) and the distribution group
instances (i.e. all other instances which
have assigned units of the same type.)

Azure will return nil, whereas ec2 will
want to check which availability zone each
machine is in and distribute for maximum
distance.

https://codereview.appspot.com/81340043/

=== modified file 'environs/statepolicy.go'
--- environs/statepolicy.go	2014-03-25 13:01:16 +0000
+++ environs/statepolicy.go	2014-03-27 07:16:15 +0000
@@ -33,7 +33,21 @@
 	}
 	return nil, errors.NewNotImplementedError("Prechecker")
 }
+<<<<<<< TREE
 
 func (environStatePolicy) ConfigValidator(providerType string) (state.ConfigValidator, error) {
 	return Provider(providerType)
 }
+=======
+
+func (environStatePolicy) InstanceDistributor(cfg *config.Config) (state.InstanceDistributor, error) {
+	env, err := New(cfg)
+	if err != nil {
+		return nil, err
+	}
+	if p, ok := env.(state.InstanceDistributor); ok {
+		return p, nil
+	}
+	return nil, errors.NewNotImplementedError("InstanceDistributor")
+}
+>>>>>>> MERGE-SOURCE
 
=== modified file 'state/apiserver/provisioner/provisioner.go'
--- state/apiserver/provisioner/provisioner.go	2014-03-27 07:16:13 +0000
+++ state/apiserver/provisioner/provisioner.go	2014-03-27 07:16:15 +0000
@@ -351,29 +351,12 @@
 		if err != nil {
 			return nil, err
 		}
-		allUnits, err := service.AllUnits()
+		instanceIds, err := service.ServiceInstances()
 		if err != nil {
 			return nil, err
 		}
-		for _, unit := range allUnits {
-			machineId, err := unit.AssignedMachineId()
-			if state.IsNotAssigned(err) {
-				continue
-			} else if err != nil {
-				return nil, err
-			}
-			machine, err := st.Machine(machineId)
-			if err != nil {
-				return nil, err
-			}
-			instanceId, err := machine.InstanceId()
-			if err == nil {
-				instanceIdSet.Add(string(instanceId))
-			} else if state.IsNotProvisionedError(err) {
-				continue
-			} else {
-				return nil, err
-			}
+		for _, instanceId := range instanceIds {
+			instanceIdSet.Add(string(instanceId))
 		}
 	}
 	instanceIds := make([]instance.Id, instanceIdSet.Size())
 
=== added file 'state/distribution.go'
--- state/distribution.go	1970-01-01 00:00:00 +0000
+++ state/distribution.go	2014-03-27 07:16:15 +0000
@@ -0,0 +1,77 @@
+// Copyright 2014 Canonical Ltd.
+// Licensed under the AGPLv3, see LICENCE file for details.
+
+package state
+
+import (
+	"fmt"
+
+	"launchpad.net/juju-core/errors"
+	"launchpad.net/juju-core/instance"
+)
+
+// distribute takes a unit and set of clean, empty instances and asks the
+// InstanceDistributor policy (if any) which ones are suitable for assigning
+// the unit to. If there is no InstanceDistributor, or the distribution group
+// is empty, then all of the candidates will be returned.
+func (u *Unit) distribute(candidates []instance.Id) ([]instance.Id, error) {
+	if u.st.policy == nil {
+		return candidates, nil
+	}
+	cfg, err := u.st.EnvironConfig()
+	if err != nil {
+		return nil, err
+	}
+	distributor, err := u.st.policy.InstanceDistributor(cfg)
+	if errors.IsNotImplementedError(err) {
+		return candidates, nil
+	} else if err != nil {
+		return nil, err
+	}
+	if distributor == nil {
+		return nil, fmt.Errorf("policy returned nil instance distributor without an error")
+	}
+	service, err := u.Service()
+	if err != nil {
+		return nil, err
+	}
+	distributionGroup, err := service.ServiceInstances()
+	if err != nil {
+		return nil, err
+	}
+	if len(distributionGroup) == 0 {
+		return candidates, nil
+	}
+	return distributor.DistributeInstances(candidates, distributionGroup)
+}
+
+// ServiceInstances returns the instance IDs of provisioned
+// machines that are assigned units of this service.
+func (service *Service) ServiceInstances() ([]instance.Id, error) {
+	units, err := service.AllUnits()
+	if err != nil {
+		return nil, err
+	}
+	instanceIds := make([]instance.Id, 0, len(units))
+	for _, unit := range units {
+		machineId, err := unit.AssignedMachineId()
+		if IsNotAssigned(err) {
+			continue
+		} else if err != nil {
+			return nil, err
+		}
+		machine, err := service.st.Machine(machineId)
+		if err != nil {
+			return nil, err
+		}
+		instanceId, err := machine.InstanceId()
+		if err == nil {
+			instanceIds = append(instanceIds, instanceId)
+		} else if IsNotProvisionedError(err) {
+			continue
+		} else {
+			return nil, err
+		}
+	}
+	return instanceIds, nil
+}
 
=== modified file 'state/policy.go'
--- state/policy.go	2014-03-03 01:09:00 +0000
+++ state/policy.go	2014-03-27 07:16:15 +0000
@@ -9,6 +9,7 @@
 	"launchpad.net/juju-core/constraints"
 	"launchpad.net/juju-core/environs/config"
 	"launchpad.net/juju-core/errors"
+	"launchpad.net/juju-core/instance"
 )
 
 // Policy is an interface provided to State that may
@@ -24,10 +25,18 @@
 	// Prechecker takes a *config.Config and returns
 	// a (possibly nil) Prechecker or an error.
 	Prechecker(*config.Config) (Prechecker, error)
+<<<<<<< TREE
 
 	// ConfigValidator takes a string (environ type) and returns
 	// a (possibly nil) ConfigValidator or an error.
 	ConfigValidator(string) (ConfigValidator, error)
+=======
+
+	// InstanceDistributor takes a *config.Config
+	// and returns a (possibly nil) UnitDistributor
+	// or an error.
+	InstanceDistributor(*config.Config) (InstanceDistributor, error)
+>>>>>>> MERGE-SOURCE
 }
 
 // Prechecker is a policy interface that is provided to State
@@ -71,6 +80,7 @@
 	}
 	return prechecker.PrecheckInstance(series, cons)
 }
+<<<<<<< TREE
 
 // validate calls the state's assigned policy, if non-nil, to obtain
 // a Validator, and calls validate if a non-nil Validator is returned.
@@ -89,3 +99,23 @@
 	}
 	return configValidator.Validate(cfg, old)
 }
+=======
+
+// InstanceDistributor is a policy interface that is provided
+// to State to perform distribution of units across instances
+// for high availability.
+type InstanceDistributor interface {
+	// DistributeInstance takes a set of clean, empty
+	// instances, and a distribution group, and returns
+	// the subset of candidates which the policy will
+	// allow to enter into the distribution group.
+	//
+	// TODO(axw) move this comment
+	// The unit assigner will attempt to assign a unit
+	// to each of the resulting instances until it
+	// succeeds. If no instances can be assigned
+	// (e.g. because of concurrent deployments), then
+	// a new machine will be allocated.
+	DistributeInstances(candidates, distributionGroup []instance.Id) ([]instance.Id, error)
+}
+>>>>>>> MERGE-SOURCE
 
=== modified file 'state/unit.go'
--- state/unit.go	2014-03-26 12:56:50 +0000
+++ state/unit.go	2014-03-27 07:16:15 +0000
@@ -1225,14 +1225,63 @@
 		return nil, err
 	}
 
-	// TODO(rog) Fix so this is more efficient when there are concurrent uses.
-	// Possible solution: pick the highest and the smallest id of all
-	// unused machines, and try to assign to the first one >= a random id in the
-	// middle.
-	iter := query.Batch(1).Prefetch(0).Iter()
-	var mdoc machineDoc
-	for iter.Next(&mdoc) {
-		m := newMachine(u.st, &mdoc)
+	// Find all of the candidate machines, and associated
+	// instances for those that are provisioned. Instances
+	// will be distributed across in preference to
+	// unprovisioned machines.
+	var mdocs []*machineDoc
+	if err := query.All(&mdocs); err != nil {
+		assignContextf(&err, u, context)
+		return nil, err
+	}
+	var unprovisioned []*Machine
+	var instances []instance.Id
+	instanceMachines := make(map[instance.Id]*Machine)
+	for _, mdoc := range mdocs {
+		m := newMachine(u.st, mdoc)
+		instance, err := m.InstanceId()
+		if IsNotProvisionedError(err) {
+			unprovisioned = append(unprovisioned, m)
+		} else if err != nil {
+			assignContextf(&err, u, context)
+			return nil, err
+		} else {
+			instances = append(instances, instance)
+			instanceMachines[instance] = m
+		}
+	}
+
+	// Filter the list of instances that are suitable for
+	// distribution, and then map them back to machines.
+	//
+	// TODO(axw) 2014-03-27 #XXXXXX
+	// Shuffle machines to reduce likelihood of collisions.
+	// The partition of provisioned/unprovisioned machines
+	// must be maintained.
+	if instances, err = u.distribute(instances); err != nil {
+		assignContextf(&err, u, context)
+		return nil, err
+	}
+	machines := make([]*Machine, len(instances), len(instances)+len(unprovisioned))
+	for i, instance := range instances {
+		m, ok := instanceMachines[instance]
+		if !ok {
+			err := fmt.Errorf("invalid instance returned: %v", instance)
+			assignContextf(&err, u, context)
+			return nil, err
+		}
+		machines[i] = m
+	}
+	machines = append(machines, unprovisioned...)
+
+	// TODO(axw) 2014-03-27 #XXXXXX
+	// We should not select a machine that is in the process
+	// of being provisioned. There's no point asserting that
+	// the machine hasn't been provisioned, as there'll still
+	// be a period of time during which the machine may be
+	// provisioned without the fact having yet been recorded
+	// in state.
+	for _, m := range machines {
 		err := u.assignToMachine(m, true)
 		if err == nil {
 			return m, nil
@@ -1242,10 +1291,6 @@
 			return nil, err
 		}
 	}
-	if err := iter.Err(); err != nil {
-		assignContextf(&err, u, context)
-		return nil, err
-	}
 	return nil, noCleanMachines
 }