Merge "Add basic reporting to engine"

meta/io.murano/Classes/Environment.yaml

@@ -29,6 +29,7 @@ Workflow:
       - $this.agentListener: new(sys:AgentListener, name => $.name)
       - $this.stack: new(sys:HeatStack, name => $.name)
       - $this.instanceNotifier: new(sys:InstanceNotifier, environment => $this)
+      - $this.reporter: new(sys:StatusReporter, environment => $this)
 
   deploy:
     Body:

muranoapi/common/engine.py

@@ -27,6 +27,7 @@ from muranoapi.dsl import results_serializer
 from muranoapi.engine import environment
 from muranoapi.engine import package_class_loader
 from muranoapi.engine import package_loader
+from muranoapi.engine.system import status_reporter
 import muranoapi.engine.system.system_objects as system_objects
 from muranoapi.openstack.common.gettextutils import _  # noqa
 from muranoapi.openstack.common import log as logging
@@ -46,20 +47,35 @@ class TaskProcessingEndpoint(object):
         env = environment.Environment()
         env.token = task['token']
         env.tenant_id = task['tenant_id']
+        LOG.debug('Processing new task: {0}'.format(task))
+        try:
+            with package_loader.ApiPackageLoader(env.token, env.tenant_id) as \
+                    pkg_loader:
+                class_loader = package_class_loader.PackageClassLoader(
+                    pkg_loader)
+                system_objects.register(class_loader, pkg_loader)
 
-        with package_loader.ApiPackageLoader(env.token, env.tenant_id) as \
-                pkg_loader:
-            class_loader = package_class_loader.PackageClassLoader(pkg_loader)
-            system_objects.register(class_loader, pkg_loader)
+                exc = executor.MuranoDslExecutor(class_loader, env)
+                obj = exc.load(task['model'])
 
-            exc = executor.MuranoDslExecutor(class_loader, env)
-            obj = exc.load(task['model'])
-
-            if obj is not None:
-                obj.type.invoke('deploy', exc, obj, {})
-
-            s_res = results_serializer.serialize(obj, exc)
-            rpc.api().process_result(s_res)
+                try:
+                    if obj is not None:
+                        obj.type.invoke('deploy', exc, obj, {})
+                except Exception as e:
+                    reporter = status_reporter.StatusReporter()
+                    reporter.initialize(obj)
+                    reporter.report_error(obj, '{0}'.format(e))
+                finally:
+                    s_res = results_serializer.serialize(obj, exc)
+                    rpc.api().process_result(s_res)
+        except Exception as e:
+            # TODO(gokrokve) report error here
+            # TODO(slagun) code below needs complete rewrite and redesign
+            msg_env = Environment(task['model']['Objects']['?']['id'])
+            reporter = status_reporter.StatusReporter()
+            reporter.initialize(msg_env)
+            reporter.report_error(msg_env, '{0}'.format(e))
+            rpc.api().process_result(task['model'])
 
 
 def _prepare_rpc_service(server_id):
@@ -76,3 +92,8 @@ def get_rpc_service():
     if RPC_SERVICE is None:
         RPC_SERVICE = _prepare_rpc_service(str(uuid.uuid4()))
     return RPC_SERVICE
+
+
+class Environment:
+    def __init__(self, object_id):
+        self.object_id = object_id

muranoapi/engine/system/status_reporter.py

@@ -0,0 +1,55 @@
+# Copyright (c) 2014 Mirantis Inc.
+#
+# 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
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+# implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from oslo import messaging
+
+from muranoapi.common import config
+from muranoapi.common import uuidutils
+from muranoapi.dsl import murano_class
+from muranoapi.openstack.common import log as logging
+
+
+LOG = logging.getLogger(__name__)
+
+
+@murano_class.classname('io.murano.system.StatusReporter')
+class StatusReporter(object):
+    transport = None
+
+    def initialize(self, environment):
+        if StatusReporter.transport is None:
+            StatusReporter.transport = \
+                messaging.get_transport(config.CONF)
+        self._notifier = messaging.Notifier(
+            StatusReporter.transport,
+            publisher_id=uuidutils.generate_uuid(),
+            topic='murano')
+        self._environment_id = environment.object_id
+
+    def _report(self, instance, msg, details=None, level='info'):
+        body = {
+            'id': instance.object_id,
+            'text': msg,
+            'details': details,
+            'level': level,
+            'environment_id': self._environment_id
+        }
+        self._notifier.info({}, 'murano.report_notification', body)
+
+    def report(self, instance, msg):
+        self._report(instance, msg)
+
+    def report_error(self, instance, msg):
+        self._report(instance, msg, None, 'error')

muranoapi/engine/system/system_objects.py

@@ -21,6 +21,7 @@ from muranoapi.engine.system import agent_listener
 from muranoapi.engine.system import heat_stack
 from muranoapi.engine.system import instance_reporter
 from muranoapi.engine.system import resource_manager
+from muranoapi.engine.system import status_reporter
 
 
 def _auto_register(class_loader):
@@ -48,3 +49,4 @@ def register(class_loader, package_loader):
     class_loader.import_class(heat_stack.HeatStack)
     class_loader.import_class(ResourceManagerWrapper)
     class_loader.import_class(instance_reporter.InstanceReportNotifier)
+    class_loader.import_class(status_reporter.StatusReporter)