Added endpoint and utils to work with events

Change-Id: I5d4c37cf6d5dc328f92b75d08c672d29953e6041
Implements: blueprint event-log

etc/sahara/policy.json

@@ -60,5 +60,7 @@
     "job-binary-internals:create": "",
     "job-binary-internals:get": "",
     "job-binary-internals:delete": "",
-    "job-binary-internals:get_data": ""
+    "job-binary-internals:get_data": "",
+
+    "clusters:progress": ""
 }

sahara/api/v11.py

@@ -23,6 +23,7 @@ from sahara.service.validations.edp import job_binary as v_j_b
 from sahara.service.validations.edp import job_binary_internal as v_j_b_i
 from sahara.service.validations.edp import job_execution as v_j_e
 import sahara.utils.api as u
+from sahara.utils import cluster_progress_ops as cpo
 
 
 LOG = logging.getLogger(__name__)
@@ -229,3 +230,14 @@ def job_binary_internal_delete(job_binary_internal_id):
 @v.check_exists(api.get_job_binary_internal, 'job_binary_internal_id')
 def job_binary_internal_data(job_binary_internal_id):
     return api.get_job_binary_internal_data(job_binary_internal_id)
+
+# Events ops
+
+
+@rest.get('/clusters/<cluster_id>/progress')
+@acl.enforce("clusters:progress")
+def get_cluster_events(cluster_id):
+    data = u.get_request_args()
+    provision_step = data.get('provision_step')
+    response = cpo.get_cluster_events(cluster_id, provision_step)
+    return u.render(events=[event.to_dict() for event in response])

sahara/tests/unit/utils/test_cluster_progress_ops.py

@@ -0,0 +1,123 @@
+# 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 sahara import conductor
+from sahara import context
+from sahara.tests.unit import base
+from sahara.tests.unit.conductor import test_api
+from sahara.utils import cluster_progress_ops as cpo
+
+
+class ClusterProgressOpsTest(base.SaharaWithDbTestCase):
+    def setUp(self):
+        super(ClusterProgressOpsTest, self).setUp()
+        self.api = conductor.API
+
+    def _make_sample(self):
+        ctx = context.ctx()
+        cluster = self.api.cluster_create(ctx, test_api.SAMPLE_CLUSTER)
+        return ctx, cluster
+
+    def test_update_provisioning_steps(self):
+        ctx, cluster = self._make_sample()
+
+        step_id1 = self.api.cluster_provision_step_add(ctx, cluster.id, {
+            "step_name": "some_name1",
+            "total": 2,
+        })
+
+        self.api.cluster_event_add(ctx, step_id1, {
+            "event_info": "INFO",
+            "successful": True
+        })
+
+        cpo.update_provisioning_steps(cluster.id)
+
+        # check that we have correct provision step
+
+        result_cluster = self.api.cluster_get(ctx, cluster.id)
+        result_step = result_cluster.provision_progress[0]
+
+        self.assertEqual(None, result_step.successful)
+        self.assertEqual(1, result_step.completed)
+
+        # check updating in case of successful provision step
+
+        self.api.cluster_event_add(ctx, step_id1, {
+            "event_info": "INFO",
+            "successful": True
+        })
+
+        cpo.update_provisioning_steps(cluster.id)
+
+        result_cluster = self.api.cluster_get(ctx, cluster.id)
+        result_step = result_cluster.provision_progress[0]
+
+        self.assertEqual(True, result_step.successful)
+        self.assertEqual(2, result_step.completed)
+
+        # check updating in case of failed provision step
+
+        step_id2 = self.api.cluster_provision_step_add(ctx, cluster.id, {
+            "step_name": "some_name1",
+            "total": 2,
+        })
+
+        self.api.cluster_event_add(ctx, step_id2, {
+            "event_info": "INFO",
+            "successful": False,
+        })
+
+        cpo.update_provisioning_steps(cluster.id)
+
+        result_cluster = self.api.cluster_get(ctx, cluster.id)
+
+        for step in result_cluster.provision_progress:
+            if step.id == step_id2:
+                self.assertEqual(False, step.successful)
+
+        # check that it's possible to add provision step after failed step
+
+        step_id3 = self.api.cluster_provision_step_add(ctx, cluster.id, {
+            "step_name": "some_name",
+            "total": 2,
+        })
+
+        self.assertEqual(
+            step_id3, cpo.get_current_provisioning_step(cluster.id))
+
+    def test_get_cluster_events(self):
+        ctx, cluster = self._make_sample()
+
+        step_id1 = self.api.cluster_provision_step_add(ctx, cluster.id, {
+            'step_name': "some_name1",
+        })
+        step_id2 = self.api.cluster_provision_step_add(ctx, cluster.id, {
+            'step_name': "some_name",
+        })
+
+        self.api.cluster_event_add(ctx, step_id1, {
+            "event_info": "INFO",
+            'successful': True,
+        })
+
+        self.api.cluster_event_add(ctx, step_id2, {
+            "event_info": "INFO",
+            'successful': True,
+        })
+
+        self.assertEqual(2, len(cpo.get_cluster_events(cluster.id)))
+        self.assertEqual(1, len(cpo.get_cluster_events(cluster.id, step_id1)))
+        self.assertEqual(1, len(cpo.get_cluster_events(cluster.id, step_id2)))

sahara/utils/cluster_progress_ops.py

@@ -0,0 +1,169 @@
+# 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.
+
+import functools
+
+from oslo.utils import excutils
+from oslo.utils import timeutils
+import six
+
+from sahara import conductor as c
+from sahara.conductor import resource
+from sahara import context
+from sahara import exceptions
+from sahara.i18n import _
+
+conductor = c.API
+
+
+def add_successful_event(instance):
+    cluster_id = instance.node_group.cluster_id
+    step_id = get_current_provisioning_step(cluster_id)
+    if step_id:
+        conductor.cluster_event_add(context.ctx(), step_id, {
+            'successful': True,
+            'node_group_id': instance.node_group_id,
+            'instance_id': instance.id,
+            'instance_name': instance.instance_name,
+            'event_info': None,
+        })
+
+
+def add_fail_event(instance, exception):
+    cluster_id = instance.node_group.cluster_id
+    step_id = get_current_provisioning_step(cluster_id)
+    event_info = six.text_type(exception)
+
+    if step_id:
+        conductor.cluster_event_add(context.ctx(), step_id, {
+            'successful': False,
+            'node_group_id': instance.node_group_id,
+            'instance_id': instance.id,
+            'instance_name': instance.instance_name,
+            'event_info': event_info,
+        })
+
+
+def add_provisioning_step(cluster_id, step_name, total):
+    update_provisioning_steps(cluster_id)
+    return conductor.cluster_provision_step_add(context.ctx(), cluster_id, {
+        'step_name': step_name,
+        'completed': 0,
+        'total': total,
+        'started_at': timeutils.utcnow(),
+    })
+
+
+def get_current_provisioning_step(cluster_id):
+    update_provisioning_steps(cluster_id)
+    ctx = context.ctx()
+    cluster = conductor.cluster_get(ctx, cluster_id)
+    for step in cluster.provision_progress:
+        if step.successful is not None:
+            continue
+
+        return step.id
+
+    return None
+
+
+def update_provisioning_steps(cluster_id):
+    ctx = context.ctx()
+    cluster = conductor.cluster_get(ctx, cluster_id)
+
+    for step in cluster.provision_progress:
+        if step.successful is not None:
+            continue
+
+        has_failed = False
+        successful_events_count = 0
+        events = conductor.cluster_provision_step_get_events(
+            ctx, step.id)
+        for event in events:
+            if event.successful:
+                successful_events_count += 1
+            else:
+                has_failed = True
+
+        successful = None
+        if has_failed:
+            successful = False
+        elif successful_events_count == step.total:
+            successful = True
+
+        completed_at = None
+        if successful and not step.completed_at:
+            completed_at = timeutils.utcnow()
+
+        conductor.cluster_provision_step_update(ctx, step.id, {
+            'completed': successful_events_count,
+            'successful': successful,
+            'completed_at': completed_at,
+        })
+
+        if successful:
+            conductor.cluster_provision_step_remove_events(
+                ctx, step.id)
+
+
+def get_cluster_events(cluster_id, provision_step=None):
+    update_provisioning_steps(cluster_id)
+    if provision_step:
+        return conductor.cluster_provision_step_get_events(
+            context.ctx(), provision_step)
+    else:
+        cluster = conductor.cluster_get(context.ctx(), cluster_id)
+        events = []
+        for step in cluster['provision_progress']:
+            step_id = step['id']
+            events += conductor.cluster_provision_step_get_events(
+                context.ctx(), step_id)
+        return events
+
+
+def event_wrapper(mark_successful_on_exit):
+    def decorator(func):
+        @functools.wraps(func)
+        def handler(*args, **kwargs):
+            # NOTE (vgridnev): We should know information about instance,
+            #                  so we should find instance in args or kwargs.
+            #                  Also, we import sahara.conductor.resource
+            #                  to check some object is Instance
+
+            instance = None
+            for arg in args:
+                if isinstance(arg, resource.InstanceResource):
+                    instance = arg
+
+            for kw_arg in kwargs.values():
+                if isinstance(kw_arg, resource.InstanceResource):
+                    instance = kw_arg
+
+            if instance is None:
+                raise exceptions.InvalidDataException(
+                    _("Function should have an Instance as argument"))
+
+            try:
+                value = func(*args, **kwargs)
+            except Exception as e:
+                with excutils.save_and_reraise_exception():
+                    add_fail_event(instance, e)
+
+            if mark_successful_on_exit:
+                add_successful_event(instance)
+
+            return value
+        return handler
+    return decorator