Merge "Add API to get the status of async task"

8 files changed, 174 insertions, 28 deletions

diff --git a/api/database/__init__.py b/api/database/__init__.py
index bc2708bc7..5b0bb05a2 100644
--- a/api/database/__init__.py
+++ b/api/database/__init__.py
@@ -21,9 +21,3 @@ db_session = scoped_session(sessionmaker(autocommit=False,
                                          bind=engine))
 Base = declarative_base()
 Base.query = db_session.query_property()
-
-
-def init_db():
-    subclasses = [subclass.__name__ for subclass in Base.__subclasses__()]
-    logger.debug('Import models: %s', subclasses)
-    Base.metadata.create_all(bind=engine)
diff --git a/api/database/handler.py b/api/database/handler.py
new file mode 100644
index 000000000..f6a22578f
--- /dev/null
+++ b/api/database/handler.py
@@ -0,0 +1,30 @@
+# ############################################################################
+# Copyright (c) 2017 Huawei Technologies Co.,Ltd and others.
+#
+# All rights reserved. This program and the accompanying materials
+# are made available under the terms of the Apache License, Version 2.0
+# which accompanies this distribution, and is available at
+# http://www.apache.org/licenses/LICENSE-2.0
+# ############################################################################
+from api.database import db_session
+from api.database.models import AsyncTasks
+
+
+class AsyncTaskHandler(object):
+    def insert(self, kwargs):
+        task = AsyncTasks(**kwargs)
+        db_session.add(task)
+        db_session.commit()
+        return task
+
+    def update_status(self, task, status):
+        task.status = status
+        db_session.commit()
+
+    def update_error(self, task, error):
+        task.error = error
+        db_session.commit()
+
+    def get_task_by_taskid(self, task_id):
+        task = AsyncTasks.query.filter_by(task_id=task_id).first()
+        return task
diff --git a/api/database/models.py b/api/database/models.py
index 25e323842..2fc141c1f 100644
--- a/api/database/models.py
+++ b/api/database/models.py
@@ -23,3 +23,14 @@ class Tasks(Base):
 
     def __repr__(self):
         return '<Task %r>' % Tasks.task_id
+
+
+class AsyncTasks(Base):
+    __tablename__ = 'asynctasks'
+    id = Column(Integer, primary_key=True)
+    task_id = Column(String(30))
+    status = Column(Integer)
+    error = Column(String(120))
+
+    def __repr__(self):
+        return '<Task %r>' % AsyncTasks.task_id
diff --git a/api/resources/asynctask.py b/api/resources/asynctask.py
new file mode 100644
index 000000000..dd2a71003
--- /dev/null
+++ b/api/resources/asynctask.py
@@ -0,0 +1,35 @@
+# ############################################################################
+# Copyright (c) 2017 Huawei Technologies Co.,Ltd and others.
+#
+# All rights reserved. This program and the accompanying materials
+# are made available under the terms of the Apache License, Version 2.0
+# which accompanies this distribution, and is available at
+# http://www.apache.org/licenses/LICENSE-2.0
+# ############################################################################
+import uuid
+
+from api.utils import common as common_utils
+from api.database.models import AsyncTasks
+
+
+def default(args):
+    return _get_status(args)
+
+
+def _get_status(args):
+    try:
+        task_id = args['task_id']
+        uuid.UUID(task_id)
+    except KeyError:
+        message = 'measurement and task_id must be provided'
+        return common_utils.error_handler(message)
+
+    asynctask = AsyncTasks.query.filter_by(task_id=task_id).first()
+
+    try:
+        status = asynctask.status
+        error = asynctask.error if asynctask.error else []
+
+        return common_utils.result_handler(status, error)
+    except AttributeError:
+        return common_utils.error_handler('no such task')
diff --git a/api/resources/env_action.py b/api/resources/env_action.py
index 59a1692a1..7e2487158 100644
--- a/api/resources/env_action.py
+++ b/api/resources/env_action.py
@@ -10,6 +10,7 @@ import logging
 import threading
 import subprocess
 import time
+import uuid
 import json
 import os
 import errno
@@ -23,17 +24,24 @@ from yardstick.common.httpClient import HttpClient
 from api import conf as api_conf
 from api.utils import influx
 from api.utils.common import result_handler
+from api.database.handler import AsyncTaskHandler
 
 logger = logging.getLogger(__name__)
+logger.setLevel(logging.DEBUG)
 
 
 def createGrafanaContainer(args):
-    thread = threading.Thread(target=_create_grafana)
+    task_id = str(uuid.uuid4())
+
+    thread = threading.Thread(target=_create_grafana, args=(task_id,))
     thread.start()
-    return result_handler('success', [])
 
+    return result_handler('success', {'task_id': task_id})
+
+
+def _create_grafana(task_id):
+    _create_task(task_id)
 
-def _create_grafana():
     client = Client(base_url=config.DOCKER_URL)
 
     try:
@@ -48,7 +56,10 @@ def _create_grafana():
         _create_data_source()
 
         _create_dashboard()
+
+        _update_task_status(task_id)
     except Exception as e:
+        _update_task_error(task_id, str(e))
         logger.debug('Error: %s', e)
 
 
@@ -96,12 +107,17 @@ def _check_image_exist(client, t):
 
 
 def createInfluxDBContainer(args):
-    thread = threading.Thread(target=_create_influxdb)
+    task_id = str(uuid.uuid4())
+
+    thread = threading.Thread(target=_create_influxdb, args=(task_id,))
     thread.start()
-    return result_handler('success', [])
+
+    return result_handler('success', {'task_id': task_id})
 
 
-def _create_influxdb():
+def _create_influxdb(task_id):
+    _create_task(task_id)
+
     client = Client(base_url=config.DOCKER_URL)
 
     try:
@@ -116,7 +132,10 @@ def _create_influxdb():
         time.sleep(5)
 
         _config_influxdb()
+
+        _update_task_status(task_id)
     except Exception as e:
+        _update_task_error(task_id, str(e))
         logger.debug('Error: %s', e)
 
 
@@ -160,34 +179,44 @@ def _change_output_to_influxdb():
 
 
 def prepareYardstickEnv(args):
-    thread = threading.Thread(target=_prepare_env_daemon)
+    task_id = str(uuid.uuid4())
+
+    thread = threading.Thread(target=_prepare_env_daemon, args=(task_id,))
     thread.start()
-    return result_handler('success', [])
 
+    return result_handler('success', {'task_id': task_id})
 
-def _prepare_env_daemon():
+
+def _prepare_env_daemon(task_id):
+    _create_task(task_id)
 
     installer_ip = os.environ.get('INSTALLER_IP', 'undefined')
     installer_type = os.environ.get('INSTALLER_TYPE', 'undefined')
 
-    _check_variables(installer_ip, installer_type)
+    try:
+        _check_variables(installer_ip, installer_type)
 
-    _create_directories()
+        _create_directories()
 
-    rc_file = config.OPENSTACK_RC_FILE
+        rc_file = config.OPENSTACK_RC_FILE
 
-    _get_remote_rc_file(rc_file, installer_ip, installer_type)
+        _get_remote_rc_file(rc_file, installer_ip, installer_type)
 
-    _source_file(rc_file)
+        _source_file(rc_file)
 
-    _append_external_network(rc_file)
+        _append_external_network(rc_file)
 
-    # update the external_network
-    _source_file(rc_file)
+        # update the external_network
+        _source_file(rc_file)
 
-    _clean_images()
+        _clean_images()
 
-    _load_images()
+        _load_images()
+
+        _update_task_status(task_id)
+    except Exception as e:
+        _update_task_error(task_id, str(e))
+        logger.debug('Error: %s', e)
 
 
 def _check_variables(installer_ip, installer_type):
@@ -257,3 +286,27 @@ def _load_images():
                          cwd=config.YARDSTICK_REPOS_DIR)
     output = p.communicate()[0]
     logger.debug('The result is: %s', output)
+
+
+def _create_task(task_id):
+    async_handler = AsyncTaskHandler()
+    task_dict = {
+        'task_id': task_id,
+        'status': 0
+    }
+    async_handler.insert(task_dict)
+
+
+def _update_task_status(task_id):
+    async_handler = AsyncTaskHandler()
+
+    task = async_handler.get_task_by_taskid(task_id)
+    async_handler.update_status(task, 1)
+
+
+def _update_task_error(task_id, error):
+    async_handler = AsyncTaskHandler()
+
+    task = async_handler.get_task_by_taskid(task_id)
+    async_handler.update_status(task, 2)
+    async_handler.update_error(task, error)
diff --git a/api/server.py b/api/server.py
index fac821b00..8cce4de87 100644
--- a/api/server.py
+++ b/api/server.py
@@ -7,13 +7,17 @@
 # http://www.apache.org/licenses/LICENSE-2.0
 ##############################################################################
 import logging
+from itertools import ifilter
+import inspect
 
 from flask import Flask
 from flask_restful import Api
 from flasgger import Swagger
 
-from api.database import init_db
+from api.database import Base
+from api.database import engine
 from api.database import db_session
+from api.database import models
 from api.urls import urlpatterns
 from yardstick import _init_logging
 
@@ -21,8 +25,6 @@ logger = logging.getLogger(__name__)
 
 app = Flask(__name__)
 
-init_db()
-
 Swagger(app)
 
 api = Api(app)
@@ -33,6 +35,21 @@ def shutdown_session(exception=None):
     db_session.remove()
 
 
+def init_db():
+    def func(a):
+        try:
+            if issubclass(a[1], Base):
+                return True
+        except TypeError:
+            pass
+        return False
+
+    subclses = ifilter(func, inspect.getmembers(models, inspect.isclass))
+    logger.debug('Import models: %s', [a[1] for a in subclses])
+    Base.metadata.create_all(bind=engine)
+
+
+init_db()
 reduce(lambda a, b: a.add_resource(b.resource, b.url,
                                    endpoint=b.endpoint) or a, urlpatterns, api)
 
diff --git a/api/urls.py b/api/urls.py
index 0fffd12db..273fb40f8 100644
--- a/api/urls.py
+++ b/api/urls.py
@@ -11,6 +11,7 @@ from api.utils.common import Url
 
 
 urlpatterns = [
+    Url('/yardstick/asynctask', views.Asynctask, 'asynctask'),
     Url('/yardstick/testcases/release/action', views.ReleaseAction, 'release'),
     Url('/yardstick/testcases/samples/action', views.SamplesAction, 'samples'),
     Url('/yardstick/results', views.Results, 'results'),
diff --git a/api/views.py b/api/views.py
index ee13b47a9..69ca89186 100644
--- a/api/views.py
+++ b/api/views.py
@@ -24,6 +24,11 @@ TestCaseActionArgsOptsModel = models.TestCaseActionArgsOptsModel
 TestCaseActionArgsOptsTaskArgModel = models.TestCaseActionArgsOptsTaskArgModel
 
 
+class Asynctask(ApiResource):
+    def get(self):
+        return self._dispatch_get()
+
+
 class ReleaseAction(ApiResource):
     @swag_from(os.getcwd() + '/swagger/docs/testcases.yaml')
     def post(self):