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 <[email protected]> Authored: Wed May 10 19:11:26 2017 +0300 Committer: max-orlov <[email protected]> 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)
