API now gets and handles conductor exceptions

Change-Id: I843c0004c93873343bf64340975a6ecd02d75814

muranoapi/api/v1/deployments.py

@@ -31,7 +31,8 @@ class Controller(object):
             .filter_by(environment_id=environment_id) \
             .order_by(desc(Deployment.created))
         result = query.all()
-        deployments = [deployment.to_dict() for deployment in result]
+        deployments = [set_dep_state(deployment, unit).to_dict() for deployment
+                       in result]
         return {'deployments': deployments}
 
     def statuses(self, request, environment_id, deployment_id):
@@ -86,3 +87,23 @@ def verify_and_get_deployment(db_session, environment_id, deployment_id):
 
 def create_resource():
     return wsgi.Resource(Controller())
+
+
+def set_dep_state(deployment, unit):
+    num_errors = unit.query(Status).filter_by(level='error').count()
+    num_warnings = unit.query(Status).filter_by(level='warning').count()
+    if deployment.finished:
+        if num_errors:
+            deployment.state = 'completed_w_errors'
+        elif num_warnings:
+            deployment.state = 'completed_w_warnings'
+        else:
+            deployment.state = 'success'
+    else:
+        if num_errors:
+            deployment.state = 'running_w_errors'
+        elif num_warnings:
+            deployment.state = 'running_w_warnings'
+        else:
+            deployment.state = 'running'
+    return deployment

muranoapi/common/service.py

@@ -55,10 +55,12 @@ class TaskResultHandlerService(service.Service):
             try:
                 with self._create_mq_client() as mqClient:
                     mqClient.declare(conf.results_exchange, conf.results_queue)
-                    with mqClient.open(conf.results_queue) as results_sb:
+                    with mqClient.open(conf.results_queue,
+                                       prefetch_count=100) as results_sb:
                         while True:
-                            result = results_sb.get_message()
-                            eventlet.spawn(handle_result, result)
+                            result = results_sb.get_message(timeout=1)
+                            if result:
+                                eventlet.spawn(handle_result, result)
             except Exception as ex:
                 log.exception(ex)
 
@@ -67,10 +69,12 @@ class TaskResultHandlerService(service.Service):
             try:
                 with self._create_mq_client() as mqClient:
                     mqClient.declare(conf.reports_exchange, conf.reports_queue)
-                    with mqClient.open(conf.reports_queue) as reports_sb:
+                    with mqClient.open(conf.reports_queue,
+                                       prefetch_count=100) as reports_sb:
                         while True:
-                            report = reports_sb.get_message()
-                            eventlet.spawn(handle_report, report)
+                            report = reports_sb.get_message(timeout=1)
+                            if report:
+                                eventlet.spawn(handle_report, report)
             except Exception as ex:
                 log.exception(ex)
 
@@ -108,9 +112,21 @@ def handle_result(message):
         #close deployment
         deployment = get_last_deployment(session, environment.id)
         deployment.finished = timeutils.utcnow()
+
+        num_errors = session.query(Status).filter_by(level='error').count()
+        num_warnings = session.query(Status).filter_by(level='warning').count()
+
+        final_status_text = "Deployment finished"
+        if num_errors:
+            final_status_text += " with errors"
+
+        elif num_warnings:
+            final_status_text += " with warnings"
+
         status = Status()
         status.deployment_id = deployment.id
-        status.text = "Deployment finished"
+        status.text = final_status_text
+        status.level = 'info'
         deployment.statuses.append(status)
         deployment.save(session)
     except Exception as ex:

muranoapi/db/migrate_repo/versions/012_add_details_and_level_columns_to_status.py

@@ -0,0 +1,34 @@
+#    Copyright (c) 2013 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 sqlalchemy.schema import MetaData, Table, Column
+from sqlalchemy.types import Text
+
+meta = MetaData()
+
+
+def upgrade(migrate_engine):
+    meta.bind = migrate_engine
+    status = Table('status', meta, autoload=True)
+    details = Column('details', Text(), nullable=True)
+    level = Column('level', Text(), nullable=False, server_default='info')
+    details.create(status)
+    level.create(status)
+
+
+def downgrade(migrate_engine):
+    meta.bind = migrate_engine
+    status = Table('status', meta, autoload=True)
+    status.c.details.drop()
+    status.c.level.drop()

muranoapi/db/models.py

@@ -166,6 +166,8 @@ class Status(BASE, ModelBase):
     entity = Column(String(10), nullable=True)
     deployment_id = Column(String(32), ForeignKey('deployment.id'))
     text = Column(Text(), nullable=False)
+    level = Column(Text(), nullable=False)
+    details = Column(Text(), nullable=True)
 
     def to_dict(self):
         dictionary = super(Status, self).to_dict()

muranoapi/db/services/sessions.py

@@ -130,6 +130,7 @@ class SessionServices(object):
             del deployment.description['token']
         status = Status()
         status.text = "Deployment scheduled"
+        status.level = "info"
         deployment.statuses.append(status)
         with unit.begin():
             unit.add(session)