Mailing-List: contact dev-help@ariatosca.incubator.apache.org; run by ezmlm
Precedence: bulk
Reply-To: dev@ariatosca.incubator.apache.org
Content-Type: text/plain; charset="us-ascii"
MIME-Version: 1.0
Content-Transfer-Encoding: 7bit
From: mxmrlv@apache.org
To: dev@ariatosca.incubator.apache.org
Date: Wed, 10 May 2017 16:11:32 -0000
Message-Id: <5d972e16d4364b17b003ec2eaaa3e2f1@git.apache.org>
In-Reply-To: <fc0fdefb067148919d9b7bb68faffa10@git.apache.org>
References: <fc0fdefb067148919d9b7bb68faffa10@git.apache.org>
Subject: [2/2] incubator-ariatosca git commit: moved to the proper event,
 only extraction of all of the models remains
archived-at: Wed, 10 May 2017 16:11:36 -0000

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/6e54fef3
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/tree/6e54fef3
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/diff/6e54fef3

Branch: refs/heads/ARIA-213-Sporadic-tests-failures-over-locked-database-issue
Commit: 6e54fef392abf014d8f2ea30cb0da008f90485f7
Parents: 9dba620
Author: max-orlov <maxim@gigaspaces.com>
Authored: Wed May 10 19:11:26 2017 +0300
Committer: max-orlov <maxim@gigaspaces.com>
Committed: Wed May 10 19:11:26 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/6e54fef3/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/6e54fef3/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)