Serialization of destruction dependencies
If the object was deleted through the API between deployments the only way for those who used to subscribe to its destruction to be notifyed upon next deployment is to persist destruction dependencies to the object model. This commit adds such serialization and deserialization. Also move GC class from engine to dsl Targets-blueprint: dependency-driven-resource-deallocation Closes-Bug: #1619248 Change-Id: Icd2e882be5770244aa1ecafe265aff1439ebec9e
This commit is contained in:
Stan Lagun
parent
dc050d41cb
commit
2b57eb3fca
|
|
@ -333,7 +333,7 @@ class MuranoDslExecutor(object):
|
|||
obj = objects[0]
|
||||
if obj.destroyed:
|
||||
return
|
||||
for dependency in obj.dependencies.get('onDestruction', []):
|
||||
for dependency in obj.destruction_dependencies:
|
||||
try:
|
||||
handler = dependency['handler']
|
||||
if handler:
|
||||
|
|
@ -351,7 +351,7 @@ class MuranoDslExecutor(object):
|
|||
LOG.warning(_LW(
|
||||
'Muted exception during destruction dependency '
|
||||
'execution in {0}: {1}').format(obj, e), exc_info=True)
|
||||
obj.dependencies.pop('onDestruction', None)
|
||||
obj.load_dependencies(None)
|
||||
|
||||
def destroy_objects(self, *objects):
|
||||
if not objects:
|
||||
|
|
|
|||
|
|
@ -565,6 +565,7 @@ def parse_object_definition(spec, scope_type, context):
|
|||
'id': system_data.get('id'),
|
||||
'name': system_data.get('name'),
|
||||
'destroyed': system_data.get('destroyed', False),
|
||||
'dependencies': system_data.get('dependencies', {}),
|
||||
'extra': {
|
||||
key: value for key, value in six.iteritems(system_data)
|
||||
if key.startswith('_')
|
||||
|
|
@ -577,11 +578,11 @@ def assemble_object_definition(parsed, model_format=dsl_types.DumpTypes.Mixed):
|
|||
result = {
|
||||
parsed['type']: parsed['properties'],
|
||||
'id': parsed['id'],
|
||||
'name': parsed['name']
|
||||
'name': parsed['name'],
|
||||
'dependencies': parsed['dependencies'],
|
||||
'destroyed': parsed['destroyed']
|
||||
}
|
||||
result.update(parsed['extra'])
|
||||
if parsed['destroyed']:
|
||||
result['destroyed'] = True
|
||||
return result
|
||||
result = parsed['properties']
|
||||
header = {
|
||||
|
|
|
|||
|
|
@ -19,6 +19,7 @@ from murano.dsl import dsl
|
|||
from murano.dsl import dsl_types
|
||||
from murano.dsl import exceptions
|
||||
from murano.dsl import helpers
|
||||
from murano.dsl.principal_objects import garbage_collector
|
||||
from murano.dsl import yaql_integration
|
||||
|
||||
|
||||
|
|
@ -55,7 +56,7 @@ class MuranoObject(dsl_types.MuranoObject):
|
|||
self._parents[name] = known_classes[name] = obj
|
||||
else:
|
||||
self._parents[name] = known_classes[name]
|
||||
self._dependencies = {}
|
||||
self._destruction_dependencies = []
|
||||
|
||||
@property
|
||||
def extension(self):
|
||||
|
|
@ -189,8 +190,22 @@ class MuranoObject(dsl_types.MuranoObject):
|
|||
return self._initialized
|
||||
|
||||
@property
|
||||
def dependencies(self):
|
||||
return self._dependencies
|
||||
def destruction_dependencies(self):
|
||||
return self._destruction_dependencies
|
||||
|
||||
def load_dependencies(self, dependencies):
|
||||
self._destruction_dependencies = []
|
||||
if not dependencies:
|
||||
return
|
||||
destruction_dependencies = dependencies.get('onDestruction', [])
|
||||
object_store = helpers.get_object_store()
|
||||
for record in destruction_dependencies:
|
||||
subscriber_id = record['subscriber']
|
||||
subscriber = object_store.get(subscriber_id)
|
||||
if not subscriber:
|
||||
continue
|
||||
garbage_collector.GarbageCollector.subscribe_destruction(
|
||||
self, subscriber, record.get('handler'))
|
||||
|
||||
def get_property(self, name, context=None):
|
||||
start_type, derived = self.type, False
|
||||
|
|
@ -283,7 +298,7 @@ class MuranoObject(dsl_types.MuranoObject):
|
|||
|
||||
def to_dictionary(self, include_hidden=False,
|
||||
serialization_type=dsl_types.DumpTypes.Serializable,
|
||||
allow_refs=False):
|
||||
allow_refs=False, with_destruction_dependencies=False):
|
||||
context = helpers.get_context()
|
||||
result = {}
|
||||
for parent in self._parents.values():
|
||||
|
|
@ -312,8 +327,7 @@ class MuranoObject(dsl_types.MuranoObject):
|
|||
'id': self.object_id,
|
||||
'name': self.name
|
||||
}
|
||||
if self.destroyed:
|
||||
result['destroyed'] = True
|
||||
header = result
|
||||
else:
|
||||
if serialization_type == dsl_types.DumpTypes.Mixed:
|
||||
result.update({'?': {
|
||||
|
|
@ -327,8 +341,22 @@ class MuranoObject(dsl_types.MuranoObject):
|
|||
'id': self.object_id,
|
||||
'name': self.name
|
||||
}})
|
||||
if self.destroyed:
|
||||
result['?']['destroyed'] = True
|
||||
header = result['?']
|
||||
if self.destroyed:
|
||||
header['destroyed'] = True
|
||||
if with_destruction_dependencies:
|
||||
dds = []
|
||||
for record in self.destruction_dependencies:
|
||||
subscriber = record['subscriber']()
|
||||
if not subscriber or self.executor.object_store.is_doomed(
|
||||
subscriber):
|
||||
continue
|
||||
dds.append({
|
||||
'subscriber': subscriber.object_id,
|
||||
'handler': record['handler']
|
||||
})
|
||||
if dds:
|
||||
header.setdefault('dependencies', {})['onDestruction'] = dds
|
||||
return result
|
||||
|
||||
def mark_destroyed(self, clear_data=False):
|
||||
|
|
@ -338,6 +366,7 @@ class MuranoObject(dsl_types.MuranoObject):
|
|||
self._extension = None
|
||||
self._properties = None
|
||||
self._owner = None
|
||||
self._destruction_dependencies = None
|
||||
self._this = None
|
||||
for p in six.itervalues(self._parents):
|
||||
p.mark_destroyed(clear_data)
|
||||
|
|
|
|||
|
|
@ -164,8 +164,7 @@ class ObjectStore(object):
|
|||
for obj in sentenced_objects:
|
||||
obj_subscribers = [obj.owner]
|
||||
|
||||
dds = obj.dependencies.get('onDestruction', [])
|
||||
for dd in dds:
|
||||
for dd in obj.destruction_dependencies:
|
||||
subscriber = dd['subscriber']
|
||||
if subscriber:
|
||||
subscriber = subscriber()
|
||||
|
|
@ -275,6 +274,7 @@ class InitializationObjectStore(ObjectStore):
|
|||
class_obj, owner,
|
||||
name=parsed['name'],
|
||||
object_id=object_id if self._keep_ids else None)
|
||||
obj.load_dependencies(parsed['dependencies'])
|
||||
if parsed['destroyed']:
|
||||
obj.mark_destroyed()
|
||||
self.put(obj, object_id or obj.object_id)
|
||||
|
|
|
|||
|
|
@ -13,6 +13,7 @@
|
|||
# under the License.
|
||||
|
||||
from murano.dsl.principal_objects import exception
|
||||
from murano.dsl.principal_objects import garbage_collector
|
||||
from murano.dsl.principal_objects import stack_trace
|
||||
from murano.dsl.principal_objects import sys_object
|
||||
|
||||
|
|
@ -21,3 +22,4 @@ def register(package):
|
|||
package.register_class(sys_object.SysObject)
|
||||
package.register_class(stack_trace.StackTrace)
|
||||
package.register_class(exception.DslException)
|
||||
package.register_class(garbage_collector.GarbageCollector)
|
||||
|
|
|
|||
|
|
@ -22,12 +22,12 @@ from murano.dsl import helpers
|
|||
@dsl.name('io.murano.system.GC')
|
||||
class GarbageCollector(object):
|
||||
@staticmethod
|
||||
@specs.parameter('publisher', dsl.MuranoObjectParameter())
|
||||
@specs.parameter('subscriber', dsl.MuranoObjectParameter())
|
||||
@specs.parameter('publisher', dsl.MuranoObjectParameter(decorate=False))
|
||||
@specs.parameter('subscriber', dsl.MuranoObjectParameter(decorate=False))
|
||||
@specs.parameter('handler', yaqltypes.String(nullable=True))
|
||||
def subscribe_destruction(publisher, subscriber, handler=None):
|
||||
publisher_this = publisher.object.real_this
|
||||
subscriber_this = subscriber.object.real_this
|
||||
publisher_this = publisher.real_this
|
||||
subscriber_this = subscriber.real_this
|
||||
|
||||
if handler:
|
||||
subscriber.type.find_single_method(handler)
|
||||
|
|
@ -38,21 +38,20 @@ class GarbageCollector(object):
|
|||
if not dependency:
|
||||
dependency = {'subscriber': helpers.weak_ref(subscriber_this),
|
||||
'handler': handler}
|
||||
publisher_this.dependencies.setdefault(
|
||||
'onDestruction', []).append(dependency)
|
||||
publisher_this.destruction_dependencies.append(dependency)
|
||||
|
||||
@staticmethod
|
||||
@specs.parameter('publisher', dsl.MuranoObjectParameter())
|
||||
@specs.parameter('subscriber', dsl.MuranoObjectParameter())
|
||||
@specs.parameter('publisher', dsl.MuranoObjectParameter(decorate=False))
|
||||
@specs.parameter('subscriber', dsl.MuranoObjectParameter(decorate=False))
|
||||
@specs.parameter('handler', yaqltypes.String(nullable=True))
|
||||
def unsubscribe_destruction(publisher, subscriber, handler=None):
|
||||
publisher_this = publisher.object.real_this
|
||||
subscriber_this = subscriber.object.real_this
|
||||
publisher_this = publisher.real_this
|
||||
subscriber_this = subscriber.real_this
|
||||
|
||||
if handler:
|
||||
subscriber.type.find_single_method(handler)
|
||||
|
||||
dds = publisher_this.dependencies.get('onDestruction', [])
|
||||
dds = publisher_this.destruction_dependencies
|
||||
dependency = GarbageCollector._find_dependency(
|
||||
publisher_this, subscriber_this, handler)
|
||||
|
||||
|
|
@ -61,7 +60,7 @@ class GarbageCollector(object):
|
|||
|
||||
@staticmethod
|
||||
def _find_dependency(publisher, subscriber, handler):
|
||||
dds = publisher.dependencies.get('onDestruction', [])
|
||||
dds = publisher.destruction_dependencies
|
||||
for dd in dds:
|
||||
if dd['handler'] != handler:
|
||||
continue
|
||||
|
|
@ -34,19 +34,22 @@ def serialize(obj, executor,
|
|||
make_copy=False,
|
||||
serialize_attributes=False,
|
||||
serialize_actions=False,
|
||||
serialization_type=serialization_type)['Objects']
|
||||
serialization_type=serialization_type,
|
||||
with_destruction_dependencies=False)['Objects']
|
||||
|
||||
|
||||
def _serialize_object(root_object, designer_attributes, allow_refs,
|
||||
executor, serialize_actions=True,
|
||||
serialization_type=dsl_types.DumpTypes.Serializable):
|
||||
serialization_type=dsl_types.DumpTypes.Serializable,
|
||||
with_destruction_dependencies=True):
|
||||
serialized_objects = set()
|
||||
|
||||
obj = root_object
|
||||
while True:
|
||||
obj, need_another_pass = _pass12_serialize(
|
||||
obj, None, serialized_objects, designer_attributes, executor,
|
||||
serialize_actions, serialization_type, allow_refs)
|
||||
serialize_actions, serialization_type, allow_refs,
|
||||
with_destruction_dependencies)
|
||||
if not need_another_pass:
|
||||
break
|
||||
tree = [obj]
|
||||
|
|
@ -59,7 +62,8 @@ def serialize_model(root_object, executor,
|
|||
make_copy=True,
|
||||
serialize_attributes=True,
|
||||
serialize_actions=True,
|
||||
serialization_type=dsl_types.DumpTypes.Serializable):
|
||||
serialization_type=dsl_types.DumpTypes.Serializable,
|
||||
with_destruction_dependencies=True):
|
||||
designer_attributes = executor.object_store.designer_attributes
|
||||
|
||||
if root_object is None:
|
||||
|
|
@ -70,11 +74,13 @@ def serialize_model(root_object, executor,
|
|||
with helpers.with_object_store(executor.object_store):
|
||||
tree, serialized_objects = _serialize_object(
|
||||
root_object, designer_attributes, allow_refs, executor,
|
||||
serialize_actions, serialization_type)
|
||||
serialize_actions, serialization_type,
|
||||
with_destruction_dependencies)
|
||||
|
||||
tree_copy = _serialize_object(
|
||||
root_object, None, allow_refs, executor, serialize_actions,
|
||||
serialization_type)[0] if make_copy else None
|
||||
serialization_type,
|
||||
with_destruction_dependencies)[0] if make_copy else None
|
||||
|
||||
attributes = executor.attribute_store.serialize(
|
||||
serialized_objects) if serialize_attributes else None
|
||||
|
|
@ -113,7 +119,8 @@ def _serialize_available_action(obj, current_actions, executor):
|
|||
|
||||
def _pass12_serialize(value, parent, serialized_objects,
|
||||
designer_attributes_getter, executor,
|
||||
serialize_actions, serialization_type, allow_refs):
|
||||
serialize_actions, serialization_type, allow_refs,
|
||||
with_destruction_dependencies):
|
||||
if isinstance(value, dsl.MuranoObjectInterface):
|
||||
value = value.object
|
||||
if isinstance(value, (six.string_types,
|
||||
|
|
@ -135,7 +142,8 @@ def _pass12_serialize(value, parent, serialized_objects,
|
|||
return value, False
|
||||
if isinstance(value, dsl_types.MuranoObject):
|
||||
result = value.to_dictionary(
|
||||
serialization_type=serialization_type, allow_refs=allow_refs)
|
||||
serialization_type=serialization_type, allow_refs=allow_refs,
|
||||
with_destruction_dependencies=with_destruction_dependencies)
|
||||
if designer_attributes_getter is not None:
|
||||
if serialization_type == dsl_types.DumpTypes.Inline:
|
||||
system_data = result
|
||||
|
|
@ -149,7 +157,8 @@ def _pass12_serialize(value, parent, serialized_objects,
|
|||
serialized_objects.add(value.object_id)
|
||||
return _pass12_serialize(
|
||||
result, value, serialized_objects, designer_attributes_getter,
|
||||
executor, serialize_actions, serialization_type, allow_refs)
|
||||
executor, serialize_actions, serialization_type, allow_refs,
|
||||
with_destruction_dependencies)
|
||||
elif isinstance(value, utils.MappingType):
|
||||
result = {}
|
||||
need_another_pass = False
|
||||
|
|
@ -168,7 +177,8 @@ def _pass12_serialize(value, parent, serialized_objects,
|
|||
result_value = _pass12_serialize(
|
||||
d_value, parent, serialized_objects,
|
||||
designer_attributes_getter, executor, serialize_actions,
|
||||
serialization_type, allow_refs)
|
||||
serialization_type, allow_refs,
|
||||
with_destruction_dependencies)
|
||||
result[result_key] = result_value[0]
|
||||
if result_value[1]:
|
||||
need_another_pass = True
|
||||
|
|
@ -179,7 +189,8 @@ def _pass12_serialize(value, parent, serialized_objects,
|
|||
for t in value:
|
||||
v, nmp = _pass12_serialize(
|
||||
t, parent, serialized_objects, designer_attributes_getter,
|
||||
executor, serialize_actions, serialization_type, allow_refs)
|
||||
executor, serialize_actions, serialization_type, allow_refs,
|
||||
with_destruction_dependencies)
|
||||
if nmp:
|
||||
need_another_pass = True
|
||||
result.append(v)
|
||||
|
|
|
|||
|
|
@ -15,7 +15,6 @@
|
|||
|
||||
from murano.engine.system import agent
|
||||
from murano.engine.system import agent_listener
|
||||
from murano.engine.system import garbage_collector
|
||||
from murano.engine.system import heat_stack
|
||||
from murano.engine.system import instance_reporter
|
||||
from murano.engine.system import logger
|
||||
|
|
@ -37,4 +36,3 @@ def register(package):
|
|||
package.register_class(logger.Logger)
|
||||
package.register_class(test_fixture.TestFixture)
|
||||
package.register_class(workflowclient.MistralClient)
|
||||
package.register_class(garbage_collector.GarbageCollector)
|
||||
|
|
|
|||
|
|
@ -37,6 +37,11 @@ Methods:
|
|||
|
||||
Name: TestGC
|
||||
|
||||
Properties:
|
||||
outNode:
|
||||
Usage: Out
|
||||
Contract: $.class(TestGCNode)
|
||||
|
||||
Methods:
|
||||
testObjectsCollect:
|
||||
Body:
|
||||
|
|
@ -113,3 +118,16 @@ Methods:
|
|||
Body:
|
||||
- trace(sys:GC.isDestroyed($obj))
|
||||
- $this.destroyed: $obj
|
||||
|
||||
testDestructionDependencySerialization:
|
||||
Body:
|
||||
- $model:
|
||||
:TestGCNode:
|
||||
value: A
|
||||
nodes:
|
||||
:TestGCNode:
|
||||
value: B
|
||||
- $.outNode: new($model)
|
||||
- sys:GC.subscribeDestruction($.outNode, $this, _handler)
|
||||
- sys:GC.subscribeDestruction($.outNode.nodes[0], $this, _handler)
|
||||
|
||||
|
|
|
|||
|
|
@ -13,7 +13,7 @@
|
|||
# under the License.
|
||||
|
||||
from murano.dsl import exceptions
|
||||
from murano.engine.system import garbage_collector
|
||||
from murano.dsl.principal_objects import garbage_collector
|
||||
from murano.tests.unit.dsl.foundation import object_model as om
|
||||
from murano.tests.unit.dsl.foundation import test_case
|
||||
|
||||
|
|
@ -64,6 +64,28 @@ class TestGC(test_case.DslTestCase):
|
|||
self.runner.testCallOnDestroyedObject)
|
||||
self.assertEqual(['foo', 'X'], self.traces)
|
||||
|
||||
def test_destruction_dependencies_serialization(self):
|
||||
self.runner.testDestructionDependencySerialization()
|
||||
node1 = self.runner.serialized_model['Objects']['outNode']
|
||||
node2 = node1['nodes'][0]
|
||||
|
||||
deps = {
|
||||
'onDestruction': [{
|
||||
'subscriber': self.runner.root.object_id,
|
||||
'handler': '_handler'
|
||||
}]
|
||||
}
|
||||
self.assertEqual(deps, node1['?'].get('dependencies'))
|
||||
|
||||
self.assertEqual(
|
||||
node1['?'].get('dependencies'),
|
||||
node2['?'].get('dependencies'))
|
||||
|
||||
model = self.runner.serialized_model
|
||||
model['Objects']['outNode'] = None
|
||||
self.new_runner(model)
|
||||
self.assertEqual(['Destroy A', 'Destroy B', 'B', 'A'], self.traces)
|
||||
|
||||
def test_is_doomed(self):
|
||||
self.runner.testIsDoomed()
|
||||
self.assertEqual([[], True, 'B', [True], False, 'A'], self.traces)
|
||||
|
|
|
|||
|
|
@ -17,12 +17,11 @@ import weakref
|
|||
import mock
|
||||
from testtools import matchers
|
||||
|
||||
from murano.dsl import dsl
|
||||
from murano.dsl import exceptions
|
||||
from murano.dsl import murano_method
|
||||
from murano.dsl import murano_object
|
||||
from murano.dsl import murano_type
|
||||
from murano.engine.system import garbage_collector
|
||||
from murano.dsl.principal_objects import garbage_collector
|
||||
from murano.tests.unit import base
|
||||
|
||||
|
||||
|
|
@ -30,8 +29,8 @@ class TestGarbageCollector(base.MuranoTestCase):
|
|||
def setUp(self):
|
||||
super(TestGarbageCollector, self).setUp()
|
||||
|
||||
subscriber = mock.MagicMock(spec=murano_object.MuranoObject)
|
||||
subscriber.real_this = subscriber
|
||||
self.subscriber = mock.MagicMock(spec=murano_object.MuranoObject)
|
||||
self.subscriber.real_this = self.subscriber
|
||||
|
||||
mock_class = mock.MagicMock(spec=murano_type.MuranoClass)
|
||||
mock_method = mock.MagicMock(spec=murano_method.MuranoMethod)
|
||||
|
|
@ -44,36 +43,30 @@ class TestGarbageCollector(base.MuranoTestCase):
|
|||
raise exceptions.NoMethodFound(name)
|
||||
|
||||
mock_class.find_single_method = find_single_method
|
||||
subscriber.type = mock_class
|
||||
self.subscriber = mock.MagicMock(spec=dsl.MuranoObjectInterface)
|
||||
self.subscriber.object = subscriber
|
||||
self.subscriber.type = subscriber.type
|
||||
self.subscriber.type = mock_class
|
||||
|
||||
publisher = mock.MagicMock(spec=murano_object.MuranoObject)
|
||||
publisher.real_this = publisher
|
||||
self.publisher = mock.MagicMock(spec=dsl.MuranoObjectInterface)
|
||||
self.publisher.object = publisher
|
||||
self.publisher = mock.MagicMock(spec=murano_object.MuranoObject)
|
||||
self.publisher.real_this = self.publisher
|
||||
|
||||
def test_set_dd(self):
|
||||
self.publisher.object.dependencies = {}
|
||||
self.publisher.destruction_dependencies = []
|
||||
garbage_collector.GarbageCollector.subscribe_destruction(
|
||||
self.publisher, self.subscriber, handler="mockHandler")
|
||||
dep = self.publisher.object.dependencies["onDestruction"]
|
||||
dep = self.publisher.destruction_dependencies
|
||||
self.assertThat(dep, matchers.HasLength(1))
|
||||
dep = dep[0]
|
||||
self.assertEqual("mockHandler", dep["handler"])
|
||||
self.assertEqual(self.subscriber.object, dep["subscriber"]())
|
||||
self.assertEqual(self.subscriber, dep["subscriber"]())
|
||||
|
||||
def test_unset_dd(self):
|
||||
self.publisher.object.dependencies = (
|
||||
{"onDestruction": [{
|
||||
"subscriber": weakref.ref(self.subscriber.object),
|
||||
"handler": "mockHandler"
|
||||
}]})
|
||||
self.publisher.destruction_dependencies = [{
|
||||
"subscriber": weakref.ref(self.subscriber),
|
||||
"handler": "mockHandler"
|
||||
}]
|
||||
garbage_collector.GarbageCollector.unsubscribe_destruction(
|
||||
self.publisher, self.subscriber, handler="mockHandler")
|
||||
self.assertEqual(
|
||||
[], self.publisher.object.dependencies["onDestruction"])
|
||||
[], self.publisher.destruction_dependencies)
|
||||
|
||||
def test_set_wrong_handler(self):
|
||||
self.assertRaises(
|
||||
|
|
|
|||
Loading…
Reference in New Issue