You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@ariatosca.apache.org by mx...@apache.org on 2017/05/11 11:44:52 UTC
[11/12] incubator-ariatosca git commit: moved to the proper event,
only extraction of all of the models remains
moved to the proper event, only extraction of all of the models remains
Project: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/commit/f523dfb2
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/tree/f523dfb2
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/diff/f523dfb2
Branch: refs/heads/ARIA-213-Sporadic-tests-failures-over-locked-database-issue
Commit: f523dfb2ad56fd7630f9d9f9306fd44dfa7e9c0e
Parents: 41cd1be
Author: max-orlov <ma...@gigaspaces.com>
Authored: Wed May 10 19:11:26 2017 +0300
Committer: max-orlov <ma...@gigaspaces.com>
Committed: Thu May 11 14:28:49 2017 +0300
----------------------------------------------------------------------
aria/orchestrator/workflows/executor/process.py | 15 ++++-----
aria/storage/instrumentation.py | 33 ++++++++++++--------
2 files changed, 26 insertions(+), 22 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/f523dfb2/aria/orchestrator/workflows/executor/process.py
----------------------------------------------------------------------
diff --git a/aria/orchestrator/workflows/executor/process.py b/aria/orchestrator/workflows/executor/process.py
index 52da26d..04f0172 100644
--- a/aria/orchestrator/workflows/executor/process.py
+++ b/aria/orchestrator/workflows/executor/process.py
@@ -308,7 +308,7 @@ class _Messenger(object):
'task_id': self.task_id,
'exception': exceptions.wrap_if_needed(exception),
'traceback': exceptions.get_exception_as_string(*sys.exc_info()),
- 'tracked_changes': tracked_changes,
+ 'tracked_changes': tracked_changes or {},
'new_instances': new_instances or {}
})
response = _recv_message(sock)
@@ -341,16 +341,10 @@ def _patch_ctx(ctx, messenger, instrument):
# Rollback is performed on parent process when commit fails
pass
- def patched_put(_):
- # TODO: currently we need to add signal to the put event (or commit per model), but currently we just use the init event.
- pass
-
# when autoflush is set to true (the default), refreshing an object will trigger
# an auto flush by sqlalchemy, this autoflush will attempt to commit changes made so
# far on the session. this is not the desired behavior in the subprocess
session.autoflush = False
- # for model_name in instrument.new_instances:
- ctx.model.log.put = patched_put
session.commit = patched_commit
session.rollback = patched_rollback
@@ -377,11 +371,14 @@ def _main():
# This is required for the instrumentation work properly.
# See docstring of `remove_mutable_association_listener` for further details
modeling_types.remove_mutable_association_listener()
+ try:
+ ctx = context_dict['context_cls'].deserialize_from_dict(**context_dict['context'])
+ except BaseException as e:
+ messenger.failed(exception=e, tracked_changes=None, new_instances=None)
- with instrumentation.track_changes() as instrument:
+ with instrumentation.track_changes(ctx) as instrument:
try:
messenger.started()
- ctx = context_dict['context_cls'].deserialize_from_dict(**context_dict['context'])
_patch_ctx(ctx=ctx, messenger=messenger, instrument=instrument)
task_func = imports.load_attribute(implementation)
aria.install_aria_extensions()
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/f523dfb2/aria/storage/instrumentation.py
----------------------------------------------------------------------
diff --git a/aria/storage/instrumentation.py b/aria/storage/instrumentation.py
index 6ca79ce..b1e9cd7 100644
--- a/aria/storage/instrumentation.py
+++ b/aria/storage/instrumentation.py
@@ -36,7 +36,7 @@ _INSTRUMENTED = {
_NEW_INSTANCE = 'NEW_INSTANCE'
-def track_changes(instrumented=None):
+def track_changes(ctx, instrumented=None):
"""Track changes in the specified model columns
This call will register event listeners using sqlalchemy's event mechanism. The listeners
@@ -59,16 +59,16 @@ def track_changes(instrumented=None):
:param instrumented: A dict from model columns to their python native type
:return: The instrumentation context
"""
- return _Instrumentation(instrumented or _INSTRUMENTED)
+ return _Instrumentation(ctx, instrumented or _INSTRUMENTED)
class _Instrumentation(object):
- def __init__(self, instrumented):
+ def __init__(self, ctx, instrumented):
self.tracked_changes = {}
self.new_instances = {}
self.listeners = []
- self._track_changes(instrumented)
+ self._track_changes(ctx, instrumented)
self._new_instance_index = 0
@property
@@ -76,7 +76,7 @@ class _Instrumentation(object):
self._new_instance_index += 1
return '{prefix}_{index}'.format(prefix=_NEW_INSTANCE, index=self._new_instance_index - 1)
- def _track_changes(self, instrumented):
+ def _track_changes(self, ctx, instrumented):
instrumented_attribute_classes = {}
for instrumented_attribute, attribute_type in instrumented.get('modified', {}).items():
self._register_set_attribute_listener(
@@ -93,15 +93,22 @@ class _Instrumentation(object):
# instrument creation of new instances
for instrumented_class in instrumented.get('new', {}):
- self._register_instance_listener(instrumented_class)
-
- def _register_instance_listener(self, instrumented_class):
- def listener(target, *args, **_):
- mapi_name = target.__modelname__
- tracked_instances = self.new_instances.setdefault(mapi_name, {})
+ self._register_instance_listener(ctx, instrumented_class)
+
+ def _register_instance_listener(self, ctx, instrumented_class):
+ def listener(session, instance):
+ if not isinstance(instance, instrumented_class):
+ return
+ session.expunge(instance)
+ tracked_instances = self.new_instances.setdefault(instance.__modelname__, {})
tracked_attributes = tracked_instances.setdefault(self._new_instance_id, {})
- tracked_attributes.update(**args[1])
- listener_args = (instrumented_class, 'init', listener)
+ # TODO: we need a better way to get all of the attributes
+ tracked_attributes.update(execution_fk=instance.execution_fk,
+ task_fk=instance.task_fk,
+ **instance.to_dict())
+ listener_args = (getattr(ctx.model, instrumented_class.__tablename__)._session,
+ 'after_attach',
+ listener)
sqlalchemy.event.listen(*listener_args)
self.listeners.append(listener_args)