Move updating provision progress to conductor

Changes:
 * Move updating provision progress to conductor;
 * Remove redundant conductor ops;
 * Improve all updates operations.

This change would improve complexity of all event log operations
and decrease number of updates of provisioning steps.
Also we remove unneeded fields in base, because we can calculate
it very easy from UI.

partially implements: bp event-log

Change-Id: I63c593d5ba27879edc2142a1e2ab310628ee045c

sahara/conductor/api.py

@@ -442,31 +442,24 @@ class LocalApi(object):
     # Events ops
 
     def cluster_provision_step_add(self, context, cluster_id, values):
-        """Create a cluster assigned ProvisionStep
+        """Create a provisioning step assigned to cluster from values dict."""
-
-        from the values dictionary
-        """
         return self._manager.cluster_provision_step_add(
             context, cluster_id, values)
 
-    def cluster_provision_step_update(self, context, provision_step, values):
+    def cluster_provision_step_update(self, context, provision_step):
-        """Update the ProvisionStep from the values dictionary."""
+        """Update the cluster provisioning step."""
-        self._manager.cluster_provision_step_update(
+        return self._manager.cluster_provision_step_update(
-            context, provision_step, values)
-
-    def cluster_provision_step_get_events(self, context, provision_step):
-        """Return all events from the specified ProvisionStep."""
-        return self._manager.cluster_provision_step_get_events(
             context, provision_step)
 
-    def cluster_provision_step_remove_events(self, context, provision_step):
+    def cluster_provision_progress_update(self, context, cluster_id):
-        """Delete all event from the specified ProvisionStep."""
+        """Return cluster with provision progress updated field."""
-        self._manager.cluster_provision_step_remove_events(
+        return self._manager.cluster_provision_progress_update(
-            context, provision_step)
+            context, cluster_id)
 
     def cluster_event_add(self, context, provision_step, values):
-        """Assign new event to the specified ProvisionStep."""
+        """Assign new event to the specified provision step."""
-        self._manager.cluster_event_add(context, provision_step, values)
+        return self._manager.cluster_event_add(
+            context, provision_step, values)
 
 
 class RemoteApi(LocalApi):

sahara/conductor/manager.py

@@ -473,25 +473,17 @@ class ConductorManager(db_base.Base):
     # Events ops
 
     def cluster_provision_step_add(self, context, cluster_id, values):
-        """Create a cluster assigned ProvisionStep
+        """Create a provisioning step assigned to cluster from values dict."""
-
-        from the values dictionary
-        """
         return self.db.cluster_provision_step_add(context, cluster_id, values)
 
-    def cluster_provision_step_update(self, context, provision_step, values):
+    def cluster_provision_step_update(self, context, provision_step):
-        """Update the ProvisionStep from the values dictionary."""
+        """Update the cluster provisioning step."""
-        self.db.cluster_provision_step_update(context, provision_step, values)
+        return self.db.cluster_provision_step_update(context, provision_step)
 
-    def cluster_provision_step_get_events(self, context, provision_step):
+    def cluster_provision_progress_update(self, context, cluster_id):
-        """Return all events from the specified ProvisionStep."""
+        """Return cluster with provision progress updated field."""
-        return self.db.cluster_provision_step_get_events(
+        return self.db.cluster_provision_progress_update(context, cluster_id)
-            context, provision_step)
-
-    def cluster_provision_step_remove_events(self, context, provision_step):
-        """Delete all event from the specified ProvisionStep."""
-        self.db.cluster_provision_step_remove_events(context, provision_step)
 
     def cluster_event_add(self, context, provision_step, values):
-        """Assign new event to the specified ProvisionStep."""
+        """Assign new event to the specified provision step."""
-        self.db.cluster_event_add(context, provision_step, values)
+        return self.db.cluster_event_add(context, provision_step, values)

sahara/conductor/objects.py

@@ -293,11 +293,8 @@ class ClusterProvisionStep(object):
     tenant_id
     step_name
     step_type
-    completed
     total
     successful
-    started_at
-    completed_at
     events - list of Events objects assigned to the cluster
     """
 

sahara/context.py

@@ -274,12 +274,13 @@ def sleep(seconds=0):
 
 class InstanceInfo(object):
     def __init__(self, cluster_id=None, instance_id=None, instance_name=None,
-                 node_group_id=None, step_type=None):
+                 node_group_id=None, step_type=None, step_id=None):
         self.cluster_id = cluster_id
         self.instance_id = instance_id
         self.instance_name = instance_name
         self.node_group_id = node_group_id
         self.step_type = step_type
+        self.step_id = step_id
 
 
 def set_step_type(step_type):
@@ -291,6 +292,8 @@ class InstanceInfoManager(object):
         self.prev_instance_info = current().current_instance_info
         if not instance_info.step_type:
             instance_info.step_type = self.prev_instance_info.step_type
+        if not instance_info.step_id:
+            instance_info.step_id = self.prev_instance_info.step_id
         current().current_instance_info = instance_info
 
     def __enter__(self):

sahara/db/api.py

@@ -110,7 +110,10 @@ def to_dict(func):
 
 def cluster_get(context, cluster, show_progress=False):
     """Return the cluster or None if it does not exist."""
-    cluster = IMPL.cluster_get(context, cluster)
+    if show_progress:
+        cluster = IMPL.cluster_provision_progress_update(context, cluster)
+    else:
+        cluster = IMPL.cluster_get(context, cluster)
     if cluster:
         return cluster.to_dict(show_progress)
     return None
@@ -446,27 +449,24 @@ def job_binary_internal_get_raw_data(context, job_binary_internal_id):
     return IMPL.job_binary_internal_get_raw_data(context,
                                                  job_binary_internal_id)
 
+# Events ops
+
 
 def cluster_provision_step_add(context, cluster_id, values):
     """Create a cluster assigned ProvisionStep from the values dictionary."""
     return IMPL.cluster_provision_step_add(context, cluster_id, values)
 
 
-def cluster_provision_step_update(context, provision_step, values):
+def cluster_provision_step_update(context, step_id):
-    """Update the ProvisionStep from the values dictionary."""
+    """Updates provision step."""
-    IMPL.cluster_provision_step_update(context, provision_step, values)
+    return IMPL.cluster_provision_step_update(context, step_id)
 
 
-def cluster_provision_step_get_events(context, provision_step):
+def cluster_provision_progress_update(context, cluster_id):
-    """Return all events from the specified ProvisionStep."""
+    """Return cluster with provision progress updated field."""
-    return IMPL.cluster_provision_step_get_events(context, provision_step)
+    return IMPL.cluster_provision_progress_update(context, cluster_id)
-
-
-def cluster_provision_step_remove_events(context, provision_step):
-    """Delete all event from the specified ProvisionStep."""
-    IMPL.cluster_provision_step_remove_events(context, provision_step)
 
 
 def cluster_event_add(context, provision_step, values):
-    """Assign new event to the specified ProvisionStep."""
+    """Assign new event to the specified provision step."""
-    IMPL.cluster_event_add(context, provision_step, values)
+    return IMPL.cluster_event_add(context, provision_step, values)

sahara/db/migration/alembic_migrations/versions/020_remove_redandunt_progress_ops.py

@@ -0,0 +1,44 @@
+# Copyright 2015 OpenStack Foundation.
+#
+# 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.
+
+"""remove redandunt progress ops
+
+Revision ID: 020
+Revises: 019
+Create Date: 2015-02-26 15:01:41.015076
+
+"""
+
+# revision identifiers, used by Alembic.
+revision = '020'
+down_revision = '019'
+
+from alembic import op
+import sqlalchemy as sa
+
+
+def upgrade():
+    op.drop_column('cluster_provision_steps', 'completed_at')
+    op.drop_column('cluster_provision_steps', 'completed')
+    op.drop_column('cluster_provision_steps', 'started_at')
+
+
+def downgrade():
+    op.add_column('cluster_provision_steps',
+                  sa.Column('completed', sa.Integer(), nullable=True))
+    op.add_column('cluster_provision_steps',
+                  sa.Column('started_at', sa.DateTime(), nullable=True))
+    op.add_column('cluster_provision_steps',
+                  sa.Column('completed_at', sa.DateTime(), nullable=True))

sahara/db/sqlalchemy/api.py

@@ -1114,6 +1114,22 @@ def _cluster_provision_step_get(context, session, provision_step_id):
     return query.filter_by(id=provision_step_id).first()
 
 
+def _cluster_provision_step_update(context, session, step_id):
+    step = _cluster_provision_step_get(context, session, step_id)
+
+    if step is None:
+        raise ex.NotFoundException(
+            step_id,
+            _("Cluster Provision Step id '%s' not found!"))
+
+    if step.successful is not None:
+        return
+    if len(step.events) == step.total:
+        for event in step.events:
+            session.delete(event)
+        step.update({'successful': True})
+
+
 def cluster_provision_step_add(context, cluster_id, values):
     session = get_session()
 
@@ -1132,64 +1148,29 @@ def cluster_provision_step_add(context, cluster_id, values):
     return provision_step.id
 
 
-def cluster_provision_step_update(context, provision_step_id, values):
+def cluster_provision_step_update(context, step_id):
-    session = get_session()
+    if CONF.disable_event_log:
-
+        return
-    with session.begin():
-        provision_step = _cluster_provision_step_get(
-            context, session, provision_step_id)
-
-        if not provision_step:
-            raise ex.NotFoundException(
-                provision_step_id,
-                _("Cluster Provision Step id '%s' not found!"))
-
-        provision_step.update(values)
-
-
-def cluster_provision_step_get_events(context, provision_step_id):
     session = get_session()
     with session.begin():
-        provision_step = _cluster_provision_step_get(
+        _cluster_provision_step_update(context, session, step_id)
-            context, session, provision_step_id)
-
-        if not provision_step:
-            raise ex.NotFoundException(
-                provision_step_id,
-                _("Cluster Provision Step id '%s' not found!"))
-
-    return provision_step.events
 
 
-def cluster_provision_step_remove_events(context, provision_step_id):
+def cluster_provision_progress_update(context, cluster_id):
+    if CONF.disable_event_log:
+        return _cluster_get(context, get_session(), cluster_id)
     session = get_session()
-
     with session.begin():
-        provision_step = _cluster_provision_step_get(
+        cluster = _cluster_get(context, session, cluster_id)
-            context, session, provision_step_id)
 
-        if not provision_step:
+        if cluster is None:
-            raise ex.NotFoundException(
+            raise ex.NotFoundException(cluster_id,
-                provision_step_id,
+                                       _("Cluster id '%s' not found!"))
-                _("Cluster Provision Step id '%s' not found!"))
+        for step in cluster.provision_progress:
-
+            if step.successful is None:
-        for event in provision_step.events:
+                _cluster_provision_step_update(context, session, step.id)
-            session.delete(event)
+        result_cluster = _cluster_get(context, session, cluster_id)
-
+    return result_cluster
-
-def cluster_provision_step_remove(context, provision_step_id):
-    session = get_session()
-    cluster_provision_step_remove_events(context, provision_step_id)
-    with session.begin():
-        provision_step = _cluster_provision_step_get(
-            context, session, provision_step_id)
-
-        if not provision_step:
-            raise ex.NotFoundException(
-                provision_step_id,
-                _("Cluster Provision Step id '%s' not found!"))
-
-        session.delete(provision_step)
 
 
 def cluster_event_add(context, step_id, values):
@@ -1206,6 +1187,8 @@ def cluster_event_add(context, step_id, values):
 
         event = m.ClusterEvent()
         values['step_id'] = step_id
+        if not values['successful']:
+            provision_step.update({'successful': False})
         event.update(values)
         session.add(event)
 

sahara/db/sqlalchemy/models.py

@@ -416,11 +416,8 @@ class ClusterProvisionStep(mb.SaharaBase):
     tenant_id = sa.Column(sa.String(36))
     step_name = sa.Column(sa.String(80))
     step_type = sa.Column(sa.String(36))
-    completed = sa.Column(sa.Integer)
     total = sa.Column(sa.Integer)
     successful = sa.Column(sa.Boolean, nullable=True)
-    started_at = sa.Column(sa.DateTime())
-    completed_at = sa.Column(sa.DateTime())
     events = relationship('ClusterEvent', cascade="all,delete",
                           backref='ClusterProvisionStep',
                           lazy='joined')

sahara/service/periodic.py

@@ -147,6 +147,8 @@ def _make_periodic_tasks():
                     continue
 
                 terminate_cluster(ctx, cluster, description='transient')
+                # Add event log info cleanup
+                context.ctx().current_instance_info = context.InstanceInfo()
             context.set_ctx(None)
 
         @periodic_task.periodic_task(spacing=zombie_task_spacing)
@@ -185,7 +187,8 @@ def _make_periodic_tasks():
                     continue
 
                 terminate_cluster(ctx, cluster, description='incomplete')
-
+                # Add event log info cleanup
+                context.ctx().current_instance_info = context.InstanceInfo()
             context.set_ctx(None)
 
     return SaharaPeriodicTasks()

sahara/tests/unit/conductor/test_api.py

@@ -13,8 +13,12 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+
+import testtools
+
 from sahara import conductor
 from sahara import context
+from sahara import exceptions
 from sahara.tests.unit import base
 from sahara.utils import general as gu
 
@@ -116,6 +120,19 @@ class TestConductorClusterApi(base.SaharaWithDbTestCase):
         ng = gu.get_by_id(cluster.node_groups, ng_id)
         self.assertEqual(ng.instances[0].instance_name, 'tst123')
 
+    def _get_events(self, ctx, cluster_id, step_id=None):
+        cluster = self.api.cluster_get(ctx, cluster_id, show_progress=True)
+        events = []
+        for step in cluster.provision_progress:
+            if step_id == step['id']:
+                return step['events']
+            else:
+                events += step['events']
+        if step_id:
+            return events
+        else:
+            return []
+
     def test_events_ops(self):
         ctx, cluster = self._make_sample()
 
@@ -138,20 +155,6 @@ class TestConductorClusterApi(base.SaharaWithDbTestCase):
         self.assertEqual(st_type, provision_step['step_type'])
         self.assertEqual(cluster.id, provision_step['cluster_id'])
 
-        # test provision step updating
-
-        self.api.cluster_provision_step_update(ctx, step_id, {
-            'total': 100,
-            'completed': 59,
-        })
-
-        ncluster = self.api.cluster_get(ctx, cluster.id)
-        self.assertEqual(1, len(ncluster['provision_progress']))
-        provision_step = ncluster['provision_progress'][0]
-
-        self.assertEqual(100, provision_step['total'])
-        self.assertEqual(59, provision_step['completed'])
-
         # test adding event to step and getting events from step
 
         self.api.cluster_event_add(ctx, step_id, {
@@ -162,14 +165,13 @@ class TestConductorClusterApi(base.SaharaWithDbTestCase):
             'successful': True
         })
 
-        events = self.api.cluster_provision_step_get_events(ctx, step_id)
+        events = self._get_events(ctx, cluster.id, step_id)
         self.assertEqual(1, len(events))
         self.assertEqual(st_name, events[0].instance_name)
         self.assertEqual(True, events[0].successful)
         self.assertEqual(st_info, events[0].event_info)
 
-        # test removing events from step
+        self.api.cluster_destroy(ctx, cluster.id)
 
-        self.api.cluster_provision_step_remove_events(ctx, step_id)
+        with testtools.ExpectedException(exceptions.NotFoundException):
-        events = self.api.cluster_provision_step_get_events(ctx, step_id)
+            self._get_events(ctx, cluster.id, step_id)
-        self.assertEqual(0, len(events))

sahara/tests/unit/db/migration/test_migrations.py

@@ -456,6 +456,14 @@ class SaharaMigrationsCheckers(object):
         self.assertColumnExists(engine, 'node_group_templates', 'is_default')
         self.assertColumnExists(engine, 'cluster_templates', 'is_default')
 
+    def _check_020(self, engine, data):
+        self.assertColumnNotExists(engine, 'cluster_provision_steps',
+                                   'completed')
+        self.assertColumnNotExists(engine, 'cluster_provision_steps',
+                                   'completed_at')
+        self.assertColumnNotExists(engine, 'cluster_provision_steps',
+                                   'started_at')
+
 
 class TestMigrationsMySQL(SaharaMigrationsCheckers,
                           base.BaseWalkMigrationTestCase,

sahara/tests/unit/service/test_volumes.py

@@ -83,15 +83,13 @@ class TestAttachVolume(base.SaharaWithDbTestCase):
     @mock.patch('sahara.service.volumes._await_attach_volumes')
     @mock.patch('sahara.service.volumes._create_attach_volume')
     @mock.patch('sahara.utils.cluster_progress_ops.add_successful_event')
-    @mock.patch('sahara.utils.cluster_progress_ops.update_provisioning_steps')
     @mock.patch('sahara.utils.cluster_progress_ops.add_provisioning_step')
-    def test_attach(self, add_step, update_step, add_event,
+    def test_attach(self, add_step, add_event,
                     p_create_attach_vol, p_await, p_mount):
         p_create_attach_vol.side_effect = ['/dev/vdb', '/dev/vdc'] * 2
         p_await.return_value = None
         p_mount.return_value = None
         add_event.return_value = None
-        update_step.return_value = None
         add_step.return_value = None
 
         instance1 = {'id': '1',

sahara/tests/unit/utils/test_cluster_progress_ops.py

@@ -54,7 +54,7 @@ class ClusterProgressOpsTest(base.SaharaWithDbTestCase):
             "successful": True
         })
 
-        cpo.update_provisioning_steps(cluster.id)
+        self.api.cluster_provision_progress_update(ctx, cluster.id)
 
         # check that we have correct provision step
 
@@ -62,7 +62,6 @@ class ClusterProgressOpsTest(base.SaharaWithDbTestCase):
         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
 
@@ -71,13 +70,12 @@ class ClusterProgressOpsTest(base.SaharaWithDbTestCase):
             "successful": True
         })
 
-        cpo.update_provisioning_steps(cluster.id)
+        self.api.cluster_provision_progress_update(ctx, 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
 
@@ -91,7 +89,7 @@ class ClusterProgressOpsTest(base.SaharaWithDbTestCase):
             "successful": False,
         })
 
-        cpo.update_provisioning_steps(cluster.id)
+        self.api.cluster_provision_progress_update(ctx, cluster.id)
 
         result_cluster = self.api.cluster_get(ctx, cluster.id)
 
@@ -100,11 +98,7 @@ class ClusterProgressOpsTest(base.SaharaWithDbTestCase):
                 self.assertEqual(False, step.successful)
 
         # check that it's possible to add provision step after failed step
-
+        step_id3 = cpo.add_provisioning_step(cluster.id, "some_name", 2)
-        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))
@@ -114,9 +108,11 @@ class ClusterProgressOpsTest(base.SaharaWithDbTestCase):
 
         step_id1 = self.api.cluster_provision_step_add(ctx, cluster.id, {
             'step_name': "some_name1",
+            'total': 3,
         })
         step_id2 = self.api.cluster_provision_step_add(ctx, cluster.id, {
             'step_name': "some_name",
+            'total': 2,
         })
 
         self.api.cluster_event_add(ctx, step_id1, {

sahara/utils/cluster_progress_ops.py

@@ -39,6 +39,9 @@ CONF.register_opts(event_log_opts)
 
 
 def add_successful_event(instance):
+    if CONF.disable_event_log:
+        return
+
     cluster_id = instance.cluster_id
     step_id = get_current_provisioning_step(cluster_id)
     if step_id:
@@ -49,10 +52,12 @@ def add_successful_event(instance):
             'instance_name': instance.instance_name,
             'event_info': None,
         })
-        update_provisioning_steps(cluster_id)
 
 
 def add_fail_event(instance, exception):
+    if CONF.disable_event_log:
+        return
+
     cluster_id = instance.cluster_id
     step_id = get_current_provisioning_step(cluster_id)
     event_info = six.text_type(exception)
@@ -65,16 +70,18 @@ def add_fail_event(instance, exception):
             'instance_name': instance.instance_name,
             'event_info': event_info,
         })
-        update_provisioning_steps(cluster_id)
 
 
 def add_provisioning_step(cluster_id, step_name, total):
     if CONF.disable_event_log or not g.check_cluster_exists(cluster_id):
         return
 
-    update_provisioning_steps(cluster_id)
+    prev_step = get_current_provisioning_step(cluster_id)
+    if prev_step:
+        conductor.cluster_provision_step_update(context.ctx(), prev_step)
+
     step_type = context.ctx().current_instance_info.step_type
-    return conductor.cluster_provision_step_add(
+    new_step = conductor.cluster_provision_step_add(
         context.ctx(), cluster_id, {
             'step_name': step_name,
             'step_type': step_type,
@@ -82,64 +89,15 @@ def add_provisioning_step(cluster_id, step_name, total):
             'total': total,
             'started_at': timeutils.utcnow(),
         })
+    context.current().current_instance_info.step_id = new_step
+    return new_step
 
 
 def get_current_provisioning_step(cluster_id):
     if CONF.disable_event_log or not g.check_cluster_exists(cluster_id):
         return None
-
+    current_instance_info = context.ctx().current_instance_info
-    update_provisioning_steps(cluster_id)
+    return current_instance_info.step_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):
-    if CONF.disable_event_log or not g.check_cluster_exists(cluster_id):
-        return
-
-    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 event_wrapper(mark_successful_on_exit, **spec):

sahara/utils/general.py

@@ -102,6 +102,7 @@ def change_cluster_status(cluster, status, status_description=None):
         update_dict["status_description"] = status_description
 
     cluster = conductor.cluster_update(ctx, cluster, update_dict)
+    conductor.cluster_provision_progress_update(ctx, cluster.id)
 
     LOG.info(_LI("Cluster status has been changed: id={id}, New status="
                  "{status}").format(id=cluster.id, status=cluster.status))