OpenStack Orchestration (Heat)
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

146 lines
4.4 KiB

# 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
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import ast
import eventlet
import random
import six
from oslo_log import log as logging
from heat.common import exception
from heat.objects import sync_point as sync_point_object
LOG = logging.getLogger(__name__)
def _dump_list(items, separator=', '):
return separator.join(map(str, items))
def make_key(*components):
assert len(components) >= 2
return _dump_list(components, KEY_SEPERATOR)
def create(context, entity_id, traversal_id, is_update, stack_id):
"""Creates a sync point entry in DB."""
values = {'entity_id': entity_id, 'traversal_id': traversal_id,
'is_update': is_update, 'atomic_key': 0,
'stack_id': stack_id, 'input_data': {}}
return sync_point_object.SyncPoint.create(context, values)
def get(context, entity_id, traversal_id, is_update):
"""Retrieves a sync point entry from DB."""
sync_point = sync_point_object.SyncPoint.get_by_key(context, entity_id,
if sync_point is None:
key = (entity_id, traversal_id, is_update)
raise exception.EntityNotFound(entity='Sync Point', name=key)
return sync_point
def delete_all(context, stack_id, traversal_id):
"""Deletes all sync points of a stack associated with a traversal_id."""
return sync_point_object.SyncPoint.delete_all_by_stack_and_traversal(
context, stack_id, traversal_id
def update_input_data(context, entity_id, current_traversal,
is_update, atomic_key, input_data):
rows_updated = sync_point_object.SyncPoint.update_input_data(
context, entity_id, current_traversal, is_update, atomic_key,
return rows_updated
def str_pack_tuple(t):
return u'tuple:' + str(tuple(t))
def _str_unpack_tuple(s):
s = s[s.index(':') + 1:]
return ast.literal_eval(s)
def _deserialize(d):
d2 = {}
for k, v in d.items():
if isinstance(k, six.string_types) and k.startswith(u'tuple:('):
k = _str_unpack_tuple(k)
if isinstance(v, dict):
v = _deserialize(v)
d2[k] = v
return d2
def _serialize(d):
d2 = {}
for k, v in d.items():
if isinstance(k, tuple):
k = str_pack_tuple(k)
if isinstance(v, dict):
v = _serialize(v)
d2[k] = v
return d2
def deserialize_input_data(db_input_data):
db_input_data = db_input_data.get('input_data')
if not db_input_data:
return {}
return dict(_deserialize(db_input_data))
def serialize_input_data(input_data):
return {'input_data': _serialize(input_data)}
def sync(cnxt, entity_id, current_traversal, is_update, propagate,
predecessors, new_data):
rows_updated = None
sync_point = None
input_data = None
nconflicts = max(0, len(predecessors) - 2)
# limit to 10 seconds
max_wt = min(nconflicts * 0.01, 10)
while not rows_updated:
sync_point = get(cnxt, entity_id, current_traversal, is_update)
input_data = deserialize_input_data(sync_point.input_data)
rows_updated = update_input_data(
cnxt, entity_id, current_traversal, is_update,
sync_point.atomic_key, serialize_input_data(input_data))
# don't aggressively spin; induce some sleep
if not rows_updated:
eventlet.sleep(random.uniform(0, max_wt))
waiting = predecessors - set(input_data)
key = make_key(entity_id, current_traversal, is_update)
if waiting:
LOG.debug('[%s] Waiting %s: Got %s; still need %s',
key, entity_id, _dump_list(input_data), _dump_list(waiting))
LOG.debug('[%s] Ready %s: Got %s',
key, entity_id, _dump_list(input_data))
propagate(entity_id, serialize_input_data(input_data))