openstack/deb-heat
Code Issues Proposed changes

Further split ASG code.

Browse Source 
This patch splits AWS and Heat version of AutoScalingGroups into
separate files and renames the autoscaling.py to instance_group.py.

The ugly class hierarchy will be fixed in following patches.

Change-Id: I4b776e3f4b74df3e083fcc6f8ba583822ca16580
Implements: partial-blueprint reorg-asg-code
This commit is contained in:
tengqm 2014-10-20 21:45:03 +08:00
parent 250035317a
commit ca6f897498
8 changed files with 956 additions and 901 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

893
heat/engine/resources/autoscaling.py
View File

@ -1,893 +0,0 @@
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import copy
import math
from oslo.utils import excutils
import six
from heat.common import environment_format
from heat.common import exception
from heat.common.i18n import _
from heat.common.i18n import _LI
from heat.common import timeutils as iso8601utils
from heat.engine import attributes
from heat.engine import constraints
from heat.engine import function
from heat.engine.notification import autoscaling as notification
from heat.engine import properties
from heat.engine import rsrc_defn
from heat.engine import scheduler
from heat.engine import stack_resource
from heat.engine import support
from heat.openstack.common import log as logging
from heat.scaling import cooldown
from heat.scaling import template
LOG = logging.getLogger(__name__)
(SCALED_RESOURCE_TYPE,) = ('OS::Heat::ScaledResource',)
(EXACT_CAPACITY, CHANGE_IN_CAPACITY, PERCENT_CHANGE_IN_CAPACITY) = (
'ExactCapacity', 'ChangeInCapacity', 'PercentChangeInCapacity')
def _calculate_new_capacity(current, adjustment, adjustment_type,
minimum, maximum):
"""
Given the current capacity, calculates the new capacity which results
from applying the given adjustment of the given adjustment-type. The
new capacity will be kept within the maximum and minimum bounds.
"""
if adjustment_type == CHANGE_IN_CAPACITY:
new_capacity = current + adjustment
elif adjustment_type == EXACT_CAPACITY:
new_capacity = adjustment
else:
# PercentChangeInCapacity
delta = current * adjustment / 100.0
if math.fabs(delta) < 1.0:
rounded = int(math.ceil(delta) if delta > 0.0
else math.floor(delta))
else:
rounded = int(math.floor(delta) if delta > 0.0
else math.ceil(delta))
new_capacity = current + rounded
if new_capacity > maximum:
LOG.debug(_('truncating growth to %s') % maximum)
return maximum
if new_capacity < minimum:
LOG.debug(_('truncating shrinkage to %s') % minimum)
return minimum
return new_capacity
class InstanceGroup(stack_resource.StackResource):
PROPERTIES = (
AVAILABILITY_ZONES, LAUNCH_CONFIGURATION_NAME, SIZE,
LOAD_BALANCER_NAMES, TAGS,
) = (
'AvailabilityZones', 'LaunchConfigurationName', 'Size',
'LoadBalancerNames', 'Tags',
)
_TAG_KEYS = (
TAG_KEY, TAG_VALUE,
) = (
'Key', 'Value',
)
_ROLLING_UPDATE_SCHEMA_KEYS = (
MIN_INSTANCES_IN_SERVICE, MAX_BATCH_SIZE, PAUSE_TIME
) = (
'MinInstancesInService', 'MaxBatchSize', 'PauseTime'
)
_UPDATE_POLICY_SCHEMA_KEYS = (ROLLING_UPDATE,) = ('RollingUpdate',)
ATTRIBUTES = (
INSTANCE_LIST,
) = (
'InstanceList',
)
properties_schema = {
AVAILABILITY_ZONES: properties.Schema(
properties.Schema.LIST,
_('Not Implemented.'),
required=True
),
LAUNCH_CONFIGURATION_NAME: properties.Schema(
properties.Schema.STRING,
_('The reference to a LaunchConfiguration resource.'),
required=True,
update_allowed=True
),
SIZE: properties.Schema(
properties.Schema.INTEGER,
_('Desired number of instances.'),
required=True,
update_allowed=True
),
LOAD_BALANCER_NAMES: properties.Schema(
properties.Schema.LIST,
_('List of LoadBalancer resources.')
),
TAGS: properties.Schema(
properties.Schema.LIST,
_('Tags to attach to this group.'),
schema=properties.Schema(
properties.Schema.MAP,
schema={
TAG_KEY: properties.Schema(
properties.Schema.STRING,
required=True
),
TAG_VALUE: properties.Schema(
properties.Schema.STRING,
required=True
),
},
)
),
}
attributes_schema = {
INSTANCE_LIST: attributes.Schema(
_("A comma-delimited list of server ip addresses. "
"(Heat extension).")
),
}
rolling_update_schema = {
MIN_INSTANCES_IN_SERVICE: properties.Schema(properties.Schema.NUMBER,
default=0),
MAX_BATCH_SIZE: properties.Schema(properties.Schema.NUMBER,
default=1),
PAUSE_TIME: properties.Schema(properties.Schema.STRING,
default='PT0S')
}
update_policy_schema = {
ROLLING_UPDATE: properties.Schema(properties.Schema.MAP,
schema=rolling_update_schema)
}
def __init__(self, name, json_snippet, stack):
"""
UpdatePolicy is currently only specific to InstanceGroup and
AutoScalingGroup. Therefore, init is overridden to parse for the
UpdatePolicy.
"""
super(InstanceGroup, self).__init__(name, json_snippet, stack)
self.update_policy = self.t.update_policy(self.update_policy_schema,
self.context)
def validate(self):
"""
Add validation for update_policy
"""
super(InstanceGroup, self).validate()
if self.update_policy:
self.update_policy.validate()
policy_name = self.update_policy_schema.keys()[0]
if self.update_policy[policy_name]:
pause_time = self.update_policy[policy_name][self.PAUSE_TIME]
if iso8601utils.parse_isoduration(pause_time) > 3600:
raise ValueError('Maximum PauseTime is 1 hour.')
def validate_launchconfig(self):
# It seems to be a common error to not have a dependancy on the
# launchconfiguration. This can happen if the the actual resource
# name is used instead of {get_resource: launch_conf} and no
# depends_on is used.
conf_refid = self.properties.get(self.LAUNCH_CONFIGURATION_NAME)
if conf_refid:
conf = self.stack.resource_by_refid(conf_refid)
if conf is None:
raise ValueError(_('%(lc)s (%(ref)s)'
' reference can not be found.')
% dict(lc=self.LAUNCH_CONFIGURATION_NAME,
ref=conf_refid))
if self.name not in conf.required_by():
raise ValueError(_('%(lc)s (%(ref)s)'
' requires a reference to the'
' configuration not just the name of the'
' resource.') % dict(
lc=self.LAUNCH_CONFIGURATION_NAME,
ref=conf_refid))
def get_instance_names(self):
"""Get a list of resource names of the instances in this InstanceGroup.
Failed resources will be ignored.
"""
return [r.name for r in self.get_instances()]
def get_instances(self):
"""Get a list of all the instance resources managed by this group.
Sort the list of instances first by created_time then by name.
"""
resources = []
if self.nested():
resources = [resource for resource in self.nested().itervalues()
if resource.status != resource.FAILED]
return sorted(resources, key=lambda r: (r.created_time, r.name))
def _environment(self):
"""Return the environment for the nested stack."""
return {
environment_format.PARAMETERS: {},
environment_format.RESOURCE_REGISTRY: {
SCALED_RESOURCE_TYPE: 'AWS::EC2::Instance',
},
}
def handle_create(self):
"""Create a nested stack and add the initial resources to it."""
self.validate_launchconfig()
num_instances = self.properties[self.SIZE]
initial_template = self._create_template(num_instances)
return self.create_with_template(initial_template, self._environment())
def check_create_complete(self, task):
"""
When stack creation is done, update the load balancer.
If any instances failed to be created, delete them.
"""
done = super(InstanceGroup, self).check_create_complete(task)
if done:
self._lb_reload()
return done
def handle_update(self, json_snippet, tmpl_diff, prop_diff):
"""
If Properties has changed, update self.properties, so we
get the new values during any subsequent adjustment.
"""
if tmpl_diff:
# parse update policy
if 'UpdatePolicy' in tmpl_diff:
up = json_snippet.update_policy(self.update_policy_schema,
self.context)
self.update_policy = up
if prop_diff:
self.properties = json_snippet.properties(self.properties_schema,
self.context)
# Replace instances first if launch configuration has changed
self._try_rolling_update(prop_diff)
# Get the current capacity, we may need to adjust if
# Size has changed
if self.SIZE in prop_diff:
inst_list = self.get_instances()
if len(inst_list) != self.properties[self.SIZE]:
self.resize(self.properties[self.SIZE])
def _tags(self):
"""
Make sure that we add a tag that Ceilometer can pick up.
These need to be prepended with 'metering.'.
"""
tags = self.properties.get(self.TAGS) or []
for t in tags:
if t[self.TAG_KEY].startswith('metering.'):
# the user has added one, don't add another.
return tags
return tags + [{self.TAG_KEY: 'metering.groupname',
self.TAG_VALUE: self.FnGetRefId()}]
def handle_delete(self):
return self.delete_nested()
def _get_instance_definition(self):
conf_refid = self.properties[self.LAUNCH_CONFIGURATION_NAME]
conf = self.stack.resource_by_refid(conf_refid)
props = function.resolve(conf.properties.data)
props['Tags'] = self._tags()
vpc_zone_ids = self.properties.get(AutoScalingGroup.VPCZONE_IDENTIFIER)
if vpc_zone_ids:
props['SubnetId'] = vpc_zone_ids[0]
return rsrc_defn.ResourceDefinition(None,
SCALED_RESOURCE_TYPE,
props,
conf.t.metadata())
def _get_instance_templates(self):
"""Get templates for resource instances."""
return [(instance.name, instance.t)
for instance in self.get_instances()]
def _create_template(self, num_instances, num_replace=0,
template_version=('HeatTemplateFormatVersion',
'2012-12-12')):
"""
Create a template to represent autoscaled instances.
Also see heat.scaling.template.resource_templates.
"""
instance_definition = self._get_instance_definition()
old_resources = self._get_instance_templates()
definitions = template.resource_templates(
old_resources, instance_definition, num_instances, num_replace)
return template.make_template(definitions, version=template_version)
def _try_rolling_update(self, prop_diff):
if (self.update_policy[self.ROLLING_UPDATE] and
self.LAUNCH_CONFIGURATION_NAME in prop_diff):
policy = self.update_policy[self.ROLLING_UPDATE]
pause_sec = iso8601utils.parse_isoduration(policy[self.PAUSE_TIME])
self._replace(policy[self.MIN_INSTANCES_IN_SERVICE],
policy[self.MAX_BATCH_SIZE],
pause_sec)
def _replace(self, min_in_service, batch_size, pause_sec):
"""
Replace the instances in the group using updated launch configuration
"""
def changing_instances(tmpl):
instances = self.get_instances()
current = set((i.name, i.t) for i in instances)
updated = set(tmpl.resource_definitions(self.nested()).items())
# includes instances to be updated and deleted
affected = set(k for k, v in current ^ updated)
return set(i.FnGetRefId() for i in instances if i.name in affected)
def pause_between_batch():
while True:
try:
yield
except scheduler.Timeout:
return
capacity = len(self.nested()) if self.nested() else 0
efft_bat_sz = min(batch_size, capacity)
efft_min_sz = min(min_in_service, capacity)
# effective capacity includes temporary capacity added to accommodate
# the minimum number of instances in service during update
efft_capacity = max(capacity - efft_bat_sz, efft_min_sz) + efft_bat_sz
batch_cnt = (efft_capacity + efft_bat_sz - 1) // efft_bat_sz
if pause_sec * (batch_cnt - 1) >= self.stack.timeout_secs():
raise ValueError('The current UpdatePolicy will result '
'in stack update timeout.')
try:
remainder = capacity
while remainder > 0 or efft_capacity > capacity:
if capacity - remainder >= efft_min_sz:
efft_capacity = capacity
template = self._create_template(efft_capacity, efft_bat_sz)
self._lb_reload(exclude=changing_instances(template))
updater = self.update_with_template(template,
self._environment())
updater.run_to_completion()
self.check_update_complete(updater)
remainder -= efft_bat_sz
if ((remainder > 0 or efft_capacity > capacity) and
pause_sec > 0):
self._lb_reload()
waiter = scheduler.TaskRunner(pause_between_batch)
waiter(timeout=pause_sec)
finally:
self._lb_reload()
def resize(self, new_capacity):
"""
Resize the instance group to the new capacity.
When shrinking, the oldest instances will be removed.
"""
new_template = self._create_template(new_capacity)
try:
updater = self.update_with_template(new_template,
self._environment())
updater.run_to_completion()
self.check_update_complete(updater)
finally:
# Reload the LB in any case, so it's only pointing at healthy
# nodes.
self._lb_reload()
def _lb_reload(self, exclude=None):
'''
Notify the LoadBalancer to reload its config to include
the changes in instances we have just made.
This must be done after activation (instance in ACTIVE state),
otherwise the instances' IP addresses may not be available.
'''
exclude = exclude or []
if self.properties[self.LOAD_BALANCER_NAMES]:
id_list = [inst.FnGetRefId() for inst in self.get_instances()
if inst.FnGetRefId() not in exclude]
for lb in self.properties[self.LOAD_BALANCER_NAMES]:
lb_resource = self.stack[lb]
props = copy.copy(lb_resource.properties.data)
if 'Instances' in lb_resource.properties_schema:
props['Instances'] = id_list
elif 'members' in lb_resource.properties_schema:
props['members'] = id_list
else:
raise exception.Error(
_("Unsupported resource '%s' in LoadBalancerNames") %
(lb,))
lb_defn = rsrc_defn.ResourceDefinition(
lb_resource.name,
lb_resource.type(),
properties=props,
metadata=lb_resource.t.get('Metadata'),
deletion_policy=lb_resource.t.get('DeletionPolicy'))
scheduler.TaskRunner(lb_resource.update, lb_defn)()
def FnGetRefId(self):
return self.physical_resource_name_or_FnGetRefId()
def _resolve_attribute(self, name):
'''
heat extension: "InstanceList" returns comma delimited list of server
ip addresses.
'''
if name == self.INSTANCE_LIST:
return u','.join(inst.FnGetAtt('PublicIp')
for inst in self.get_instances()) or None
def child_template(self):
num_instances = int(self.properties[self.SIZE])
return self._create_template(num_instances)
def child_params(self):
return self._environment()
class AutoScalingGroup(InstanceGroup, cooldown.CooldownMixin):
support_status = support.SupportStatus(version='2014.1')
PROPERTIES = (
AVAILABILITY_ZONES, LAUNCH_CONFIGURATION_NAME, MAX_SIZE, MIN_SIZE,
COOLDOWN, DESIRED_CAPACITY, HEALTH_CHECK_GRACE_PERIOD,
HEALTH_CHECK_TYPE, LOAD_BALANCER_NAMES, VPCZONE_IDENTIFIER, TAGS,
) = (
'AvailabilityZones', 'LaunchConfigurationName', 'MaxSize', 'MinSize',
'Cooldown', 'DesiredCapacity', 'HealthCheckGracePeriod',
'HealthCheckType', 'LoadBalancerNames', 'VPCZoneIdentifier', 'Tags',
)
_TAG_KEYS = (
TAG_KEY, TAG_VALUE,
) = (
'Key', 'Value',
)
_UPDATE_POLICY_SCHEMA_KEYS = (
ROLLING_UPDATE
) = (
'AutoScalingRollingUpdate'
)
_ROLLING_UPDATE_SCHEMA_KEYS = (
MIN_INSTANCES_IN_SERVICE, MAX_BATCH_SIZE, PAUSE_TIME
) = (
'MinInstancesInService', 'MaxBatchSize', 'PauseTime'
)
properties_schema = {
AVAILABILITY_ZONES: properties.Schema(
properties.Schema.LIST,
_('Not Implemented.'),
required=True
),
LAUNCH_CONFIGURATION_NAME: properties.Schema(
properties.Schema.STRING,
_('The reference to a LaunchConfiguration resource.'),
required=True,
update_allowed=True
),
MAX_SIZE: properties.Schema(
properties.Schema.INTEGER,
_('Maximum number of instances in the group.'),
required=True,
update_allowed=True
),
MIN_SIZE: properties.Schema(
properties.Schema.INTEGER,
_('Minimum number of instances in the group.'),
required=True,
update_allowed=True
),
COOLDOWN: properties.Schema(
properties.Schema.NUMBER,
_('Cooldown period, in seconds.'),
update_allowed=True
),
DESIRED_CAPACITY: properties.Schema(
properties.Schema.INTEGER,
_('Desired initial number of instances.'),
update_allowed=True
),
HEALTH_CHECK_GRACE_PERIOD: properties.Schema(
properties.Schema.INTEGER,
_('Not Implemented.'),
implemented=False
),
HEALTH_CHECK_TYPE: properties.Schema(
properties.Schema.STRING,
_('Not Implemented.'),
constraints=[
constraints.AllowedValues(['EC2', 'ELB']),
],
implemented=False
),
LOAD_BALANCER_NAMES: properties.Schema(
properties.Schema.LIST,
_('List of LoadBalancer resources.')
),
VPCZONE_IDENTIFIER: properties.Schema(
properties.Schema.LIST,
_('Use only with Neutron, to list the internal subnet to '
'which the instance will be attached; '
'needed only if multiple exist; '
'list length must be exactly 1.'),
schema=properties.Schema(
properties.Schema.STRING,
_('UUID of the internal subnet to which the instance '
'will be attached.')
)
),
TAGS: properties.Schema(
properties.Schema.LIST,
_('Tags to attach to this group.'),
schema=properties.Schema(
properties.Schema.MAP,
schema={
TAG_KEY: properties.Schema(
properties.Schema.STRING,
required=True
),
TAG_VALUE: properties.Schema(
properties.Schema.STRING,
required=True
),
},
)
),
}
rolling_update_schema = {
MIN_INSTANCES_IN_SERVICE: properties.Schema(properties.Schema.INTEGER,
default=0),
MAX_BATCH_SIZE: properties.Schema(properties.Schema.INTEGER,
default=1),
PAUSE_TIME: properties.Schema(properties.Schema.STRING,
default='PT0S')
}
update_policy_schema = {
ROLLING_UPDATE: properties.Schema(
properties.Schema.MAP,
schema=rolling_update_schema)
}
def handle_create(self):
self.validate_launchconfig()
return self.create_with_template(self.child_template(),
self._environment())
def check_create_complete(self, task):
"""Invoke the cooldown after creation succeeds."""
done = super(AutoScalingGroup, self).check_create_complete(task)
if done:
self._cooldown_timestamp(
"%s : %s" % (EXACT_CAPACITY, len(self.get_instances())))
return done
def handle_update(self, json_snippet, tmpl_diff, prop_diff):
"""
If Properties has changed, update self.properties, so we get the new
values during any subsequent adjustment.
"""
if tmpl_diff:
# parse update policy
if 'UpdatePolicy' in tmpl_diff:
up = json_snippet.update_policy(self.update_policy_schema,
self.context)
self.update_policy = up
if prop_diff:
self.properties = json_snippet.properties(self.properties_schema,
self.context)
# Replace instances first if launch configuration has changed
self._try_rolling_update(prop_diff)
if (self.DESIRED_CAPACITY in prop_diff and
self.properties[self.DESIRED_CAPACITY] is not None):
self.adjust(self.properties[self.DESIRED_CAPACITY],
adjustment_type=EXACT_CAPACITY)
else:
current_capacity = len(self.get_instances())
self.adjust(current_capacity, adjustment_type=EXACT_CAPACITY)
def adjust(self, adjustment, adjustment_type=CHANGE_IN_CAPACITY):
"""
Adjust the size of the scaling group if the cooldown permits.
"""
if self._cooldown_inprogress():
LOG.info(_LI("%(name)s NOT performing scaling adjustment, "
"cooldown %(cooldown)s"),
{'name': self.name,
'cooldown': self.properties[self.COOLDOWN]})
return
capacity = len(self.get_instances())
lower = self.properties[self.MIN_SIZE]
upper = self.properties[self.MAX_SIZE]
new_capacity = _calculate_new_capacity(capacity, adjustment,
adjustment_type, lower, upper)
if new_capacity == capacity:
LOG.debug('no change in capacity %d' % capacity)
return
# send a notification before, on-error and on-success.
notif = {
'stack': self.stack,
'adjustment': adjustment,
'adjustment_type': adjustment_type,
'capacity': capacity,
'groupname': self.FnGetRefId(),
'message': _("Start resizing the group %(group)s") % {
'group': self.FnGetRefId()},
'suffix': 'start',
}
notification.send(**notif)
try:
self.resize(new_capacity)
except Exception as resize_ex:
with excutils.save_and_reraise_exception():
try:
notif.update({'suffix': 'error',
'message': six.text_type(resize_ex),
})
notification.send(**notif)
except Exception:
LOG.exception(_('Failed sending error notification'))
else:
notif.update({
'suffix': 'end',
'capacity': new_capacity,
'message': _("End resizing the group %(group)s") % {
'group': notif['groupname']},
})
notification.send(**notif)
self._cooldown_timestamp("%s : %s" % (adjustment_type, adjustment))
def _tags(self):
"""Add Identifing Tags to all servers in the group.
This is so the Dimensions received from cfn-push-stats all include
the groupname and stack id.
Note: the group name must match what is returned from FnGetRefId
"""
autoscaling_tag = [{self.TAG_KEY: 'metering.AutoScalingGroupName',
self.TAG_VALUE: self.FnGetRefId()}]
return super(AutoScalingGroup, self)._tags() + autoscaling_tag
def validate(self):
res = super(AutoScalingGroup, self).validate()
if res:
return res
# check validity of group size
min_size = self.properties[self.MIN_SIZE]
max_size = self.properties[self.MAX_SIZE]
if max_size < min_size:
msg = _("MinSize can not be greater than MaxSize")
raise exception.StackValidationFailed(message=msg)
if min_size < 0:
msg = _("The size of AutoScalingGroup can not be less than zero")
raise exception.StackValidationFailed(message=msg)
if self.properties[self.DESIRED_CAPACITY] is not None:
desired_capacity = self.properties[self.DESIRED_CAPACITY]
if desired_capacity < min_size or desired_capacity > max_size:
msg = _("DesiredCapacity must be between MinSize and MaxSize")
raise exception.StackValidationFailed(message=msg)
# TODO(pasquier-s): once Neutron is able to assign subnets to
# availability zones, it will be possible to specify multiple subnets.
# For now, only one subnet can be specified. The bug #1096017 tracks
# this issue.
if self.properties.get(self.VPCZONE_IDENTIFIER) and \
len(self.properties[self.VPCZONE_IDENTIFIER]) != 1:
raise exception.NotSupported(feature=_("Anything other than one "
"VPCZoneIdentifier"))
def child_template(self):
if self.properties[self.DESIRED_CAPACITY]:
num_instances = self.properties[self.DESIRED_CAPACITY]
else:
num_instances = self.properties[self.MIN_SIZE]
return self._create_template(num_instances)
class AutoScalingResourceGroup(AutoScalingGroup):
"""An autoscaling group that can scale arbitrary resources."""
PROPERTIES = (
RESOURCE, MAX_SIZE, MIN_SIZE, COOLDOWN, DESIRED_CAPACITY,
ROLLING_UPDATES,
) = (
'resource', 'max_size', 'min_size', 'cooldown', 'desired_capacity',
'rolling_updates',
)
_ROLLING_UPDATES_SCHEMA = (
MIN_IN_SERVICE, MAX_BATCH_SIZE, PAUSE_TIME,
) = (
'min_in_service', 'max_batch_size', 'pause_time',
)
ATTRIBUTES = (
OUTPUTS, OUTPUTS_LIST, CURRENT_SIZE,
) = (
'outputs', 'outputs_list', 'current_size',
)
properties_schema = {
RESOURCE: properties.Schema(
properties.Schema.MAP,
_('Resource definition for the resources in the group, in HOT '
'format. The value of this property is the definition of a '
'resource just as if it had been declared in the template '
'itself.'),
required=True,
update_allowed=True,
),
MAX_SIZE: properties.Schema(
properties.Schema.INTEGER,
_('Maximum number of resources in the group.'),
required=True,
update_allowed=True,
constraints=[constraints.Range(min=0)],
),
MIN_SIZE: properties.Schema(
properties.Schema.INTEGER,
_('Minimum number of resources in the group.'),
required=True,
update_allowed=True,
constraints=[constraints.Range(min=0)]
),
COOLDOWN: properties.Schema(
properties.Schema.INTEGER,
_('Cooldown period, in seconds.'),
update_allowed=True
),
DESIRED_CAPACITY: properties.Schema(
properties.Schema.INTEGER,
_('Desired initial number of resources.'),
update_allowed=True
),
ROLLING_UPDATES: properties.Schema(
properties.Schema.MAP,
_('Policy for rolling updates for this scaling group.'),
required=False,
update_allowed=True,
schema={
MIN_IN_SERVICE: properties.Schema(
properties.Schema.NUMBER,
_('The minimum number of resources in service while '
'rolling updates are being executed.'),
constraints=[constraints.Range(min=0)],
default=0),
MAX_BATCH_SIZE: properties.Schema(
properties.Schema.NUMBER,
_('The maximum number of resources to replace at once.'),
constraints=[constraints.Range(min=0)],
default=1),
PAUSE_TIME: properties.Schema(
properties.Schema.NUMBER,
_('The number of seconds to wait between batches of '
'updates.'),
constraints=[constraints.Range(min=0)],
default=0),
},
),
}
attributes_schema = {
OUTPUTS: attributes.Schema(
_("A map of resource names to the specified attribute of each "
"individual resource.")
),
OUTPUTS_LIST: attributes.Schema(
_("A list of the specified attribute of each individual resource.")
),
CURRENT_SIZE: attributes.Schema(
_("The current size of AutoscalingResourceGroup.")
),
}
def _get_instance_definition(self):
rsrc = self.properties[self.RESOURCE]
return rsrc_defn.ResourceDefinition(
None, rsrc['type'],
properties=rsrc.get('properties'),
metadata=rsrc.get('metadata'))
def _lb_reload(self, exclude=None):
"""AutoScalingResourceGroup does not maintain load balancer
connections, so we just ignore calls to update the LB.
"""
pass
def _try_rolling_update(self, prop_diff):
if (self.properties[self.ROLLING_UPDATES] and
self.RESOURCE in prop_diff):
policy = self.properties[self.ROLLING_UPDATES]
self._replace(policy[self.MIN_IN_SERVICE],
policy[self.MAX_BATCH_SIZE],
policy[self.PAUSE_TIME])
def _create_template(self, num_instances, num_replace=0,
template_version=('heat_template_version',
'2013-05-23')):
"""Create a template in the HOT format for the nested stack."""
return super(AutoScalingResourceGroup,
self)._create_template(num_instances, num_replace,
template_version=template_version)
def FnGetAtt(self, key, *path):
if key == self.CURRENT_SIZE:
return len(self.get_instances())
if path:
attrs = ((rsrc.name,
rsrc.FnGetAtt(*path)) for rsrc in self.get_instances())
if key == self.OUTPUTS:
return dict(attrs)
if key == self.OUTPUTS_LIST:
return [value for name, value in attrs]
raise exception.InvalidTemplateAttribute(resource=self.name,
key=key)
def resource_mapping():
return {
'AWS::AutoScaling::AutoScalingGroup': AutoScalingGroup,
'OS::Heat::InstanceGroup': InstanceGroup,
'OS::Heat::AutoScalingGroup': AutoScalingResourceGroup,
}

357
heat/engine/resources/aws/autoscaling_group.py Normal file
View File

@ -0,0 +1,357 @@
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import math
from oslo.utils import excutils
import six
from heat.common import exception
from heat.common.i18n import _
from heat.common.i18n import _LI
from heat.engine import constraints
from heat.engine.notification import autoscaling as notification
from heat.engine import properties
from heat.engine.resources import instance_group as instgrp
from heat.engine import support
from heat.openstack.common import log as logging
from heat.scaling import cooldown
LOG = logging.getLogger(__name__)
(EXACT_CAPACITY, CHANGE_IN_CAPACITY, PERCENT_CHANGE_IN_CAPACITY) = (
'ExactCapacity', 'ChangeInCapacity', 'PercentChangeInCapacity')
def _calculate_new_capacity(current, adjustment, adjustment_type,
minimum, maximum):
"""
Given the current capacity, calculates the new capacity which results
from applying the given adjustment of the given adjustment-type. The
new capacity will be kept within the maximum and minimum bounds.
"""
if adjustment_type == CHANGE_IN_CAPACITY:
new_capacity = current + adjustment
elif adjustment_type == EXACT_CAPACITY:
new_capacity = adjustment
else:
# PercentChangeInCapacity
delta = current * adjustment / 100.0
if math.fabs(delta) < 1.0:
rounded = int(math.ceil(delta) if delta > 0.0
else math.floor(delta))
else:
rounded = int(math.floor(delta) if delta > 0.0
else math.ceil(delta))
new_capacity = current + rounded
if new_capacity > maximum:
LOG.debug(_('truncating growth to %s') % maximum)
return maximum
if new_capacity < minimum:
LOG.debug(_('truncating shrinkage to %s') % minimum)
return minimum
return new_capacity
class AutoScalingGroup(instgrp.InstanceGroup, cooldown.CooldownMixin):
support_status = support.SupportStatus(version='2014.1')
PROPERTIES = (
AVAILABILITY_ZONES, LAUNCH_CONFIGURATION_NAME, MAX_SIZE, MIN_SIZE,
COOLDOWN, DESIRED_CAPACITY, HEALTH_CHECK_GRACE_PERIOD,
HEALTH_CHECK_TYPE, LOAD_BALANCER_NAMES, VPCZONE_IDENTIFIER, TAGS,
) = (
'AvailabilityZones', 'LaunchConfigurationName', 'MaxSize', 'MinSize',
'Cooldown', 'DesiredCapacity', 'HealthCheckGracePeriod',
'HealthCheckType', 'LoadBalancerNames', 'VPCZoneIdentifier', 'Tags',
)
_TAG_KEYS = (
TAG_KEY, TAG_VALUE,
) = (
'Key', 'Value',
)
_UPDATE_POLICY_SCHEMA_KEYS = (
ROLLING_UPDATE
) = (
'AutoScalingRollingUpdate'
)
_ROLLING_UPDATE_SCHEMA_KEYS = (
MIN_INSTANCES_IN_SERVICE, MAX_BATCH_SIZE, PAUSE_TIME
) = (
'MinInstancesInService', 'MaxBatchSize', 'PauseTime'
)
properties_schema = {
AVAILABILITY_ZONES: properties.Schema(
properties.Schema.LIST,
_('Not Implemented.'),
required=True
),
LAUNCH_CONFIGURATION_NAME: properties.Schema(
properties.Schema.STRING,
_('The reference to a LaunchConfiguration resource.'),
required=True,
update_allowed=True
),
MAX_SIZE: properties.Schema(
properties.Schema.INTEGER,
_('Maximum number of instances in the group.'),
required=True,
update_allowed=True
),
MIN_SIZE: properties.Schema(
properties.Schema.INTEGER,
_('Minimum number of instances in the group.'),
required=True,
update_allowed=True
),
COOLDOWN: properties.Schema(
properties.Schema.NUMBER,
_('Cooldown period, in seconds.'),
update_allowed=True
),
DESIRED_CAPACITY: properties.Schema(
properties.Schema.INTEGER,
_('Desired initial number of instances.'),
update_allowed=True
),
HEALTH_CHECK_GRACE_PERIOD: properties.Schema(
properties.Schema.INTEGER,
_('Not Implemented.'),
implemented=False
),
HEALTH_CHECK_TYPE: properties.Schema(
properties.Schema.STRING,
_('Not Implemented.'),
constraints=[
constraints.AllowedValues(['EC2', 'ELB']),
],
implemented=False
),
LOAD_BALANCER_NAMES: properties.Schema(
properties.Schema.LIST,
_('List of LoadBalancer resources.')
),
VPCZONE_IDENTIFIER: properties.Schema(
properties.Schema.LIST,
_('Use only with Neutron, to list the internal subnet to '
'which the instance will be attached; '
'needed only if multiple exist; '
'list length must be exactly 1.'),
schema=properties.Schema(
properties.Schema.STRING,
_('UUID of the internal subnet to which the instance '
'will be attached.')
)
),
TAGS: properties.Schema(
properties.Schema.LIST,
_('Tags to attach to this group.'),
schema=properties.Schema(
properties.Schema.MAP,
schema={
TAG_KEY: properties.Schema(
properties.Schema.STRING,
required=True
),
TAG_VALUE: properties.Schema(
properties.Schema.STRING,
required=True
),
},
)
),
}
rolling_update_schema = {
MIN_INSTANCES_IN_SERVICE: properties.Schema(properties.Schema.INTEGER,
default=0),
MAX_BATCH_SIZE: properties.Schema(properties.Schema.INTEGER,
default=1),
PAUSE_TIME: properties.Schema(properties.Schema.STRING,
default='PT0S')
}
update_policy_schema = {
ROLLING_UPDATE: properties.Schema(
properties.Schema.MAP,
schema=rolling_update_schema)
}
def handle_create(self):
self.validate_launchconfig()
return self.create_with_template(self.child_template(),
self._environment())
def _get_conf_properties(self):
conf, props = super(AutoScalingGroup, self)._get_conf_properties()
vpc_zone_ids = self.properties.get(AutoScalingGroup.VPCZONE_IDENTIFIER)
if vpc_zone_ids:
props['SubnetId'] = vpc_zone_ids[0]
return conf, props
def check_create_complete(self, task):
"""Invoke the cooldown after creation succeeds."""
done = super(AutoScalingGroup, self).check_create_complete(task)
if done:
self._cooldown_timestamp(
"%s : %s" % (EXACT_CAPACITY, len(self.get_instances())))
return done
def handle_update(self, json_snippet, tmpl_diff, prop_diff):
"""
If Properties has changed, update self.properties, so we get the new
values during any subsequent adjustment.
"""
if tmpl_diff:
# parse update policy
if 'UpdatePolicy' in tmpl_diff:
up = json_snippet.update_policy(self.update_policy_schema,
self.context)
self.update_policy = up
if prop_diff:
self.properties = json_snippet.properties(self.properties_schema,
self.context)
# Replace instances first if launch configuration has changed
self._try_rolling_update(prop_diff)
if (self.DESIRED_CAPACITY in prop_diff and
self.properties[self.DESIRED_CAPACITY] is not None):
self.adjust(self.properties[self.DESIRED_CAPACITY],
adjustment_type=EXACT_CAPACITY)
else:
current_capacity = len(self.get_instances())
self.adjust(current_capacity, adjustment_type=EXACT_CAPACITY)
def adjust(self, adjustment, adjustment_type=CHANGE_IN_CAPACITY):
"""
Adjust the size of the scaling group if the cooldown permits.
"""
if self._cooldown_inprogress():
LOG.info(_LI("%(name)s NOT performing scaling adjustment, "
"cooldown %(cooldown)s"),
{'name': self.name,
'cooldown': self.properties[self.COOLDOWN]})
return
capacity = len(self.get_instances())
lower = self.properties[self.MIN_SIZE]
upper = self.properties[self.MAX_SIZE]
new_capacity = _calculate_new_capacity(capacity, adjustment,
adjustment_type, lower, upper)
if new_capacity == capacity:
LOG.debug('no change in capacity %d' % capacity)
return
# send a notification before, on-error and on-success.
notif = {
'stack': self.stack,
'adjustment': adjustment,
'adjustment_type': adjustment_type,
'capacity': capacity,
'groupname': self.FnGetRefId(),
'message': _("Start resizing the group %(group)s") % {
'group': self.FnGetRefId()},
'suffix': 'start',
}
notification.send(**notif)
try:
self.resize(new_capacity)
except Exception as resize_ex:
with excutils.save_and_reraise_exception():
try:
notif.update({'suffix': 'error',
'message': six.text_type(resize_ex),
})
notification.send(**notif)
except Exception:
LOG.exception(_('Failed sending error notification'))
else:
notif.update({
'suffix': 'end',
'capacity': new_capacity,
'message': _("End resizing the group %(group)s") % {
'group': notif['groupname']},
})
notification.send(**notif)
self._cooldown_timestamp("%s : %s" % (adjustment_type, adjustment))
def _tags(self):
"""Add Identifing Tags to all servers in the group.
This is so the Dimensions received from cfn-push-stats all include
the groupname and stack id.
Note: the group name must match what is returned from FnGetRefId
"""
autoscaling_tag = [{self.TAG_KEY: 'metering.AutoScalingGroupName',
self.TAG_VALUE: self.FnGetRefId()}]
return super(AutoScalingGroup, self)._tags() + autoscaling_tag
def validate(self):
res = super(AutoScalingGroup, self).validate()
if res:
return res
# check validity of group size
min_size = self.properties[self.MIN_SIZE]
max_size = self.properties[self.MAX_SIZE]
if max_size < min_size:
msg = _("MinSize can not be greater than MaxSize")
raise exception.StackValidationFailed(message=msg)
if min_size < 0:
msg = _("The size of AutoScalingGroup can not be less than zero")
raise exception.StackValidationFailed(message=msg)
if self.properties[self.DESIRED_CAPACITY] is not None:
desired_capacity = self.properties[self.DESIRED_CAPACITY]
if desired_capacity < min_size or desired_capacity > max_size:
msg = _("DesiredCapacity must be between MinSize and MaxSize")
raise exception.StackValidationFailed(message=msg)
# TODO(pasquier-s): once Neutron is able to assign subnets to
# availability zones, it will be possible to specify multiple subnets.
# For now, only one subnet can be specified. The bug #1096017 tracks
# this issue.
if self.properties.get(self.VPCZONE_IDENTIFIER) and \
len(self.properties[self.VPCZONE_IDENTIFIER]) != 1:
raise exception.NotSupported(feature=_("Anything other than one "
"VPCZoneIdentifier"))
def child_template(self):
if self.properties[self.DESIRED_CAPACITY]:
num_instances = self.properties[self.DESIRED_CAPACITY]
else:
num_instances = self.properties[self.MIN_SIZE]
return self._create_template(num_instances)
def resource_mapping():
return {
'AWS::AutoScaling::AutoScalingGroup': AutoScalingGroup,
}

424
heat/engine/resources/instance_group.py Normal file
View File

@ -0,0 +1,424 @@
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import copy
from heat.common import environment_format
from heat.common import exception
from heat.common.i18n import _
from heat.common import timeutils as iso8601utils
from heat.engine import attributes
from heat.engine import function
from heat.engine import properties
from heat.engine import rsrc_defn
from heat.engine import scheduler
from heat.engine import stack_resource
from heat.scaling import template
(SCALED_RESOURCE_TYPE,) = ('OS::Heat::ScaledResource',)
class InstanceGroup(stack_resource.StackResource):
PROPERTIES = (
AVAILABILITY_ZONES, LAUNCH_CONFIGURATION_NAME, SIZE,
LOAD_BALANCER_NAMES, TAGS,
) = (
'AvailabilityZones', 'LaunchConfigurationName', 'Size',
'LoadBalancerNames', 'Tags',
)
_TAG_KEYS = (
TAG_KEY, TAG_VALUE,
) = (
'Key', 'Value',
)
_ROLLING_UPDATE_SCHEMA_KEYS = (
MIN_INSTANCES_IN_SERVICE, MAX_BATCH_SIZE, PAUSE_TIME
) = (
'MinInstancesInService', 'MaxBatchSize', 'PauseTime'
)
_UPDATE_POLICY_SCHEMA_KEYS = (ROLLING_UPDATE,) = ('RollingUpdate',)
ATTRIBUTES = (
INSTANCE_LIST,
) = (
'InstanceList',
)
properties_schema = {
AVAILABILITY_ZONES: properties.Schema(
properties.Schema.LIST,
_('Not Implemented.'),
required=True
),
LAUNCH_CONFIGURATION_NAME: properties.Schema(
properties.Schema.STRING,
_('The reference to a LaunchConfiguration resource.'),
required=True,
update_allowed=True
),
SIZE: properties.Schema(
properties.Schema.INTEGER,
_('Desired number of instances.'),
required=True,
update_allowed=True
),
LOAD_BALANCER_NAMES: properties.Schema(
properties.Schema.LIST,
_('List of LoadBalancer resources.')
),
TAGS: properties.Schema(
properties.Schema.LIST,
_('Tags to attach to this group.'),
schema=properties.Schema(
properties.Schema.MAP,
schema={
TAG_KEY: properties.Schema(
properties.Schema.STRING,
required=True
),
TAG_VALUE: properties.Schema(
properties.Schema.STRING,
required=True
),
},
)
),
}
attributes_schema = {
INSTANCE_LIST: attributes.Schema(
_("A comma-delimited list of server ip addresses. "
"(Heat extension).")
),
}
rolling_update_schema = {
MIN_INSTANCES_IN_SERVICE: properties.Schema(properties.Schema.NUMBER,
default=0),
MAX_BATCH_SIZE: properties.Schema(properties.Schema.NUMBER,
default=1),
PAUSE_TIME: properties.Schema(properties.Schema.STRING,
default='PT0S')
}
update_policy_schema = {
ROLLING_UPDATE: properties.Schema(properties.Schema.MAP,
schema=rolling_update_schema)
}
def __init__(self, name, json_snippet, stack):
"""
UpdatePolicy is currently only specific to InstanceGroup and
AutoScalingGroup. Therefore, init is overridden to parse for the
UpdatePolicy.
"""
super(InstanceGroup, self).__init__(name, json_snippet, stack)
self.update_policy = self.t.update_policy(self.update_policy_schema,
self.context)
def validate(self):
"""
Add validation for update_policy
"""
super(InstanceGroup, self).validate()
if self.update_policy:
self.update_policy.validate()
policy_name = self.update_policy_schema.keys()[0]
if self.update_policy[policy_name]:
pause_time = self.update_policy[policy_name][self.PAUSE_TIME]
if iso8601utils.parse_isoduration(pause_time) > 3600:
raise ValueError('Maximum PauseTime is 1 hour.')
def validate_launchconfig(self):
# It seems to be a common error to not have a dependancy on the
# launchconfiguration. This can happen if the the actual resource
# name is used instead of {get_resource: launch_conf} and no
# depends_on is used.
conf_refid = self.properties.get(self.LAUNCH_CONFIGURATION_NAME)
if conf_refid:
conf = self.stack.resource_by_refid(conf_refid)
if conf is None:
raise ValueError(_('%(lc)s (%(ref)s)'
' reference can not be found.')
% dict(lc=self.LAUNCH_CONFIGURATION_NAME,
ref=conf_refid))
if self.name not in conf.required_by():
raise ValueError(_('%(lc)s (%(ref)s)'
' requires a reference to the'
' configuration not just the name of the'
' resource.') % dict(
lc=self.LAUNCH_CONFIGURATION_NAME,
ref=conf_refid))
def get_instance_names(self):
"""Get a list of resource names of the instances in this InstanceGroup.
Failed resources will be ignored.
"""
return [r.name for r in self.get_instances()]
def get_instances(self):
"""Get a list of all the instance resources managed by this group.
Sort the list of instances first by created_time then by name.
"""
resources = []
if self.nested():
resources = [resource for resource in self.nested().itervalues()
if resource.status != resource.FAILED]
return sorted(resources, key=lambda r: (r.created_time, r.name))
def _environment(self):
"""Return the environment for the nested stack."""
return {
environment_format.PARAMETERS: {},
environment_format.RESOURCE_REGISTRY: {
SCALED_RESOURCE_TYPE: 'AWS::EC2::Instance',
},
}
def handle_create(self):
"""Create a nested stack and add the initial resources to it."""
self.validate_launchconfig()
num_instances = self.properties[self.SIZE]
initial_template = self._create_template(num_instances)
return self.create_with_template(initial_template, self._environment())
def check_create_complete(self, task):
"""
When stack creation is done, update the load balancer.
If any instances failed to be created, delete them.
"""
done = super(InstanceGroup, self).check_create_complete(task)
if done:
self._lb_reload()
return done
def handle_update(self, json_snippet, tmpl_diff, prop_diff):
"""
If Properties has changed, update self.properties, so we
get the new values during any subsequent adjustment.
"""
if tmpl_diff:
# parse update policy
if 'UpdatePolicy' in tmpl_diff:
up = json_snippet.update_policy(self.update_policy_schema,
self.context)
self.update_policy = up
if prop_diff:
self.properties = json_snippet.properties(self.properties_schema,
self.context)
# Replace instances first if launch configuration has changed
self._try_rolling_update(prop_diff)
# Get the current capacity, we may need to adjust if
# Size has changed
if self.SIZE in prop_diff:
inst_list = self.get_instances()
if len(inst_list) != self.properties[self.SIZE]:
self.resize(self.properties[self.SIZE])
def _tags(self):
"""
Make sure that we add a tag that Ceilometer can pick up.
These need to be prepended with 'metering.'.
"""
tags = self.properties.get(self.TAGS) or []
for t in tags:
if t[self.TAG_KEY].startswith('metering.'):
# the user has added one, don't add another.
return tags
return tags + [{self.TAG_KEY: 'metering.groupname',
self.TAG_VALUE: self.FnGetRefId()}]
def handle_delete(self):
return self.delete_nested()
def _get_conf_properties(self):
conf_refid = self.properties[self.LAUNCH_CONFIGURATION_NAME]
conf = self.stack.resource_by_refid(conf_refid)
props = function.resolve(conf.properties.data)
props['Tags'] = self._tags()
return conf, props
def _get_instance_definition(self):
conf, props = self._get_conf_properties()
return rsrc_defn.ResourceDefinition(None,
SCALED_RESOURCE_TYPE,
props,
conf.t.metadata())
def _get_instance_templates(self):
"""Get templates for resource instances."""
return [(instance.name, instance.t)
for instance in self.get_instances()]
def _create_template(self, num_instances, num_replace=0,
template_version=('HeatTemplateFormatVersion',
'2012-12-12')):
"""
Create a template to represent autoscaled instances.
Also see heat.scaling.template.resource_templates.
"""
instance_definition = self._get_instance_definition()
old_resources = self._get_instance_templates()
definitions = template.resource_templates(
old_resources, instance_definition, num_instances, num_replace)
return template.make_template(definitions, version=template_version)
def _try_rolling_update(self, prop_diff):
if (self.update_policy[self.ROLLING_UPDATE] and
self.LAUNCH_CONFIGURATION_NAME in prop_diff):
policy = self.update_policy[self.ROLLING_UPDATE]
pause_sec = iso8601utils.parse_isoduration(policy[self.PAUSE_TIME])
self._replace(policy[self.MIN_INSTANCES_IN_SERVICE],
policy[self.MAX_BATCH_SIZE],
pause_sec)
def _replace(self, min_in_service, batch_size, pause_sec):
"""
Replace the instances in the group using updated launch configuration
"""
def changing_instances(tmpl):
instances = self.get_instances()
current = set((i.name, i.t) for i in instances)
updated = set(tmpl.resource_definitions(self.nested()).items())
# includes instances to be updated and deleted
affected = set(k for k, v in current ^ updated)
return set(i.FnGetRefId() for i in instances if i.name in affected)
def pause_between_batch():
while True:
try:
yield
except scheduler.Timeout:
return
capacity = len(self.nested()) if self.nested() else 0
efft_bat_sz = min(batch_size, capacity)
efft_min_sz = min(min_in_service, capacity)
# effective capacity includes temporary capacity added to accommodate
# the minimum number of instances in service during update
efft_capacity = max(capacity - efft_bat_sz, efft_min_sz) + efft_bat_sz
batch_cnt = (efft_capacity + efft_bat_sz - 1) // efft_bat_sz
if pause_sec * (batch_cnt - 1) >= self.stack.timeout_secs():
raise ValueError('The current UpdatePolicy will result '
'in stack update timeout.')
try:
remainder = capacity
while remainder > 0 or efft_capacity > capacity:
if capacity - remainder >= efft_min_sz:
efft_capacity = capacity
template = self._create_template(efft_capacity, efft_bat_sz)
self._lb_reload(exclude=changing_instances(template))
updater = self.update_with_template(template,
self._environment())
updater.run_to_completion()
self.check_update_complete(updater)
remainder -= efft_bat_sz
if ((remainder > 0 or efft_capacity > capacity) and
pause_sec > 0):
self._lb_reload()
waiter = scheduler.TaskRunner(pause_between_batch)
waiter(timeout=pause_sec)
finally:
self._lb_reload()
def resize(self, new_capacity):
"""
Resize the instance group to the new capacity.
When shrinking, the oldest instances will be removed.
"""
new_template = self._create_template(new_capacity)
try:
updater = self.update_with_template(new_template,
self._environment())
updater.run_to_completion()
self.check_update_complete(updater)
finally:
# Reload the LB in any case, so it's only pointing at healthy
# nodes.
self._lb_reload()
def _lb_reload(self, exclude=None):
'''
Notify the LoadBalancer to reload its config to include
the changes in instances we have just made.
This must be done after activation (instance in ACTIVE state),
otherwise the instances' IP addresses may not be available.
'''
exclude = exclude or []
if self.properties[self.LOAD_BALANCER_NAMES]:
id_list = [inst.FnGetRefId() for inst in self.get_instances()
if inst.FnGetRefId() not in exclude]
for lb in self.properties[self.LOAD_BALANCER_NAMES]:
lb_resource = self.stack[lb]
props = copy.copy(lb_resource.properties.data)
if 'Instances' in lb_resource.properties_schema:
props['Instances'] = id_list
elif 'members' in lb_resource.properties_schema:
props['members'] = id_list
else:
raise exception.Error(
_("Unsupported resource '%s' in LoadBalancerNames") %
(lb,))
lb_defn = rsrc_defn.ResourceDefinition(
lb_resource.name,
lb_resource.type(),
properties=props,
metadata=lb_resource.t.get('Metadata'),
deletion_policy=lb_resource.t.get('DeletionPolicy'))
scheduler.TaskRunner(lb_resource.update, lb_defn)()
def FnGetRefId(self):
return self.physical_resource_name_or_FnGetRefId()
def _resolve_attribute(self, name):
'''
heat extension: "InstanceList" returns comma delimited list of server
ip addresses.
'''
if name == self.INSTANCE_LIST:
return u','.join(inst.FnGetAtt('PublicIp')
for inst in self.get_instances()) or None
def child_template(self):
num_instances = int(self.properties[self.SIZE])
return self._create_template(num_instances)
def child_params(self):
return self._environment()
def resource_mapping():
return {
'OS::Heat::InstanceGroup': InstanceGroup,
}

167
heat/engine/resources/openstack/autoscaling_group.py Normal file
View File

@ -0,0 +1,167 @@
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from heat.common import exception
from heat.common.i18n import _
from heat.engine import attributes
from heat.engine import constraints
from heat.engine import properties
from heat.engine.resources.aws import autoscaling_group as aws_asg
from heat.engine import rsrc_defn
class AutoScalingResourceGroup(aws_asg.AutoScalingGroup):
"""An autoscaling group that can scale arbitrary resources."""
PROPERTIES = (
RESOURCE, MAX_SIZE, MIN_SIZE, COOLDOWN, DESIRED_CAPACITY,
ROLLING_UPDATES,
) = (
'resource', 'max_size', 'min_size', 'cooldown', 'desired_capacity',
'rolling_updates',
)
_ROLLING_UPDATES_SCHEMA = (
MIN_IN_SERVICE, MAX_BATCH_SIZE, PAUSE_TIME,
) = (
'min_in_service', 'max_batch_size', 'pause_time',
)
ATTRIBUTES = (
OUTPUTS, OUTPUTS_LIST, CURRENT_SIZE,
) = (
'outputs', 'outputs_list', 'current_size',
)
properties_schema = {
RESOURCE: properties.Schema(
properties.Schema.MAP,
_('Resource definition for the resources in the group, in HOT '
'format. The value of this property is the definition of a '
'resource just as if it had been declared in the template '
'itself.'),
required=True,
update_allowed=True,
),
MAX_SIZE: properties.Schema(
properties.Schema.INTEGER,
_('Maximum number of resources in the group.'),
required=True,
update_allowed=True,
constraints=[constraints.Range(min=0)],
),
MIN_SIZE: properties.Schema(
properties.Schema.INTEGER,
_('Minimum number of resources in the group.'),
required=True,
update_allowed=True,
constraints=[constraints.Range(min=0)]
),
COOLDOWN: properties.Schema(
properties.Schema.INTEGER,
_('Cooldown period, in seconds.'),
update_allowed=True
),
DESIRED_CAPACITY: properties.Schema(
properties.Schema.INTEGER,
_('Desired initial number of resources.'),
update_allowed=True
),
ROLLING_UPDATES: properties.Schema(
properties.Schema.MAP,
_('Policy for rolling updates for this scaling group.'),
required=False,
update_allowed=True,
schema={
MIN_IN_SERVICE: properties.Schema(
properties.Schema.NUMBER,
_('The minimum number of resources in service while '
'rolling updates are being executed.'),
constraints=[constraints.Range(min=0)],
default=0),
MAX_BATCH_SIZE: properties.Schema(
properties.Schema.NUMBER,
_('The maximum number of resources to replace at once.'),
constraints=[constraints.Range(min=0)],
default=1),
PAUSE_TIME: properties.Schema(
properties.Schema.NUMBER,
_('The number of seconds to wait between batches of '
'updates.'),
constraints=[constraints.Range(min=0)],
default=0),
},
),
}
attributes_schema = {
OUTPUTS: attributes.Schema(
_("A map of resource names to the specified attribute of each "
"individual resource.")
),
OUTPUTS_LIST: attributes.Schema(
_("A list of the specified attribute of each individual resource.")
),
CURRENT_SIZE: attributes.Schema(
_("The current size of AutoscalingResourceGroup.")
),
}
def _get_instance_definition(self):
rsrc = self.properties[self.RESOURCE]
return rsrc_defn.ResourceDefinition(
None, rsrc['type'],
properties=rsrc.get('properties'),
metadata=rsrc.get('metadata'))
def _lb_reload(self, exclude=None):
"""AutoScalingResourceGroup does not maintain load balancer
connections, so we just ignore calls to update the LB.
"""
pass
def _try_rolling_update(self, prop_diff):
if (self.properties[self.ROLLING_UPDATES] and
self.RESOURCE in prop_diff):
policy = self.properties[self.ROLLING_UPDATES]
self._replace(policy[self.MIN_IN_SERVICE],
policy[self.MAX_BATCH_SIZE],
policy[self.PAUSE_TIME])
def _create_template(self, num_instances, num_replace=0,
template_version=('heat_template_version',
'2013-05-23')):
"""Create a template in the HOT format for the nested stack."""
return super(AutoScalingResourceGroup,
self)._create_template(num_instances, num_replace,
template_version=template_version)
def FnGetAtt(self, key, *path):
if key == self.CURRENT_SIZE:
return len(self.get_instances())
if path:
attrs = ((rsrc.name,
rsrc.FnGetAtt(*path)) for rsrc in self.get_instances())
if key == self.OUTPUTS:
return dict(attrs)
if key == self.OUTPUTS_LIST:
return [value for name, value in attrs]
raise exception.InvalidTemplateAttribute(resource=self.name,
key=key)
def resource_mapping():
return {
'OS::Heat::AutoScalingGroup': AutoScalingResourceGroup,
}

2
heat/tests/autoscaling/test_new_capacity.py
View File

@ -11,7 +11,7 @@
# License for the specific language governing permissions and limitations
# under the License.
from heat.engine.resources import autoscaling as asc
from heat.engine.resources.aws import autoscaling_group as asc
from heat.tests import common

6
heat/tests/test_heat_autoscaling_group.py
View File

@ -21,7 +21,7 @@ from heat.common import exception
from heat.common import short_id
from heat.common import template_format
from heat.engine import resource
from heat.engine.resources import autoscaling
from heat.engine.resources.openstack import autoscaling_group as asg
from heat.engine import rsrc_defn
from heat.engine import scheduler
from heat.engine import stack_resource
@ -267,7 +267,7 @@ class AutoScalingGroupTest(common.HeatTestCase):
def test_attribute_current_size(self):
rsrc = self.create_stack(self.parsed)['my-group']
mock_instances = self.patchobject(autoscaling.AutoScalingResourceGroup,
mock_instances = self.patchobject(asg.AutoScalingResourceGroup,
'get_instances')
mock_instances.return_value = ['one', 'two', 'three']
@ -275,7 +275,7 @@ class AutoScalingGroupTest(common.HeatTestCase):
def test_attribute_current_size_with_path(self):
rsrc = self.create_stack(self.parsed)['my-group']
mock_instances = self.patchobject(autoscaling.AutoScalingResourceGroup,
mock_instances = self.patchobject(asg.AutoScalingResourceGroup,
'get_instances')
mock_instances.return_value = ['one', 'two', 'three', 'four']
self.assertEqual(4, rsrc.FnGetAtt('current_size', 'name'))

4
heat/tests/test_instance_group.py
View File

@ -21,8 +21,8 @@ from heat.common import template_format
from heat.engine import parser
from heat.engine import resource
from heat.engine import resources
from heat.engine.resources import autoscaling as asc
from heat.engine.resources import instance
from heat.engine.resources import instance_group as instgrp
from heat.engine import rsrc_defn
from heat.engine import scheduler
from heat.tests.autoscaling import inline_templates
@ -399,7 +399,7 @@ class TestChildTemplate(common.HeatTestCase):
defn = rsrc_defn.ResourceDefinition('ig', 'OS::Heat::InstanceGroup',
{'Size': 2,
'LaunchConfigurationName': 'foo'})
self.instance_group = asc.InstanceGroup('ig', defn, stack)
self.instance_group = instgrp.InstanceGroup('ig', defn, stack)
def test_child_template(self):
self.instance_group._create_template = mock.Mock(return_value='tpl')

4
heat/tests/test_notifications.py
View File

@ -21,7 +21,7 @@ from heat.engine import environment
from heat.engine import parser
from heat.engine import resource
# imports for mocking
from heat.engine.resources import autoscaling
from heat.engine.resources.aws import autoscaling_group as aws_asg
from heat.engine.resources import instance
from heat.engine.resources import loadbalancer
from heat.engine.resources import nova_keypair
@ -272,7 +272,7 @@ class ScaleNotificationTest(common.HeatTestCase):
group = self.create_autoscaling_stack_and_get_group()
err_message = 'Boooom'
m_as = self.patchobject(autoscaling.AutoScalingGroup, 'resize')
m_as = self.patchobject(aws_asg.AutoScalingGroup, 'resize')
m_as.side_effect = exception.Error(err_message)
info, error = self.expected_notifs_calls(group,