Add task exporter to the file system

This is the first task exporter plugin. It used to export rally task
results to the file in json format.

Change-Id: I25117933ac0881453a001d96600d4c95e6064fff

etc/rally.bash_completion

@@ -36,6 +36,7 @@ _rally()
     OPTS["task_abort"]="--uuid --soft"
     OPTS["task_delete"]="--force --uuid"
     OPTS["task_detailed"]="--uuid --iterations-data"
+    OPTS["task_export"]="--uuid --connection"
     OPTS["task_list"]="--deployment --all-deployments --status --uuids-only"
     OPTS["task_report"]="--tasks --out --open --html --html-static --junit"
     OPTS["task_results"]="--uuid"
@@ -86,4 +87,4 @@ _rally()
     return 0
 }
 
-complete -o filenames -F _rally rally
+complete -o filenames -F _rally rally

rally/cli/commands/task.py

@@ -24,6 +24,7 @@ import webbrowser
 import jsonschema
 from oslo_utils import uuidutils
 import six
+from six.moves.urllib import parse as urlparse
 import yaml
 
 from rally import api
@@ -37,12 +38,15 @@ from rally.common import utils as rutils
 from rally import consts
 from rally import exceptions
 from rally import plugins
+from rally.task import exporter
 from rally.task.processing import plot
 
 
 class FailedToLoadTask(exceptions.RallyException):
     msg_fmt = _("Failed to load task")
 
+LOG = logging.getLogger(__name__)
+
 
 class TaskCommands(object):
     """Set of commands that allow you to manage benchmarking tasks and results.
@@ -699,3 +703,49 @@ class TaskCommands(object):
         print("Using task: %s" % task_id)
         api.Task.get(task_id)
         fileutils.update_globals_file("RALLY_TASK", task_id)
+
+    @cliutils.args("--uuid", dest="uuid", type=str,
+                   required=True,
+                   help="UUID of a the task.")
+    @cliutils.args("--connection", dest="connection_string", type=str,
+                   required=True,
+                   help="Connection url to the task export system.")
+    @plugins.ensure_plugins_are_loaded
+    def export(self, uuid, connection_string):
+        """Export task results to the custom task's exporting system.
+
+        :param uuid: UUID of the task
+        :param connection_string: string used to connect to the system
+        """
+
+        parsed_obj = urlparse.urlparse(connection_string)
+        try:
+            client = exporter.TaskExporter.get(parsed_obj.scheme)(
+                connection_string)
+        except exceptions.InvalidConnectionString as e:
+            if logging.is_debug():
+                LOG.exception(e)
+            print (e)
+            return 1
+        except exceptions.PluginNotFound as e:
+            if logging.is_debug():
+                LOG.exception(e)
+            msg = ("\nPlease check your connection string. The format of "
+                   "`connection` should be plugin-name://"
+                   "<user>:<pwd>@<full_address>:<port>/<path>.<type>")
+            print (str(e) + msg)
+            return 1
+
+        try:
+            client.export(uuid)
+        except (IOError, exceptions.RallyException) as e:
+            if logging.is_debug():
+                LOG.exception(e)
+            print (e)
+            return 1
+        print(_("Task %(uuid)s results was successfully exported to %("
+                "connection)s using %(name)s plugin.") % {
+                    "uuid": uuid,
+                    "connection": connection_string,
+                    "name": parsed_obj.scheme
+        })

rally/exceptions.py

@@ -81,7 +81,7 @@ class ThreadTimeoutException(RallyException):
 
 
 class PluginNotFound(NotFoundException):
-    msg_fmt = _("There is no plugin with name: %(name)s in "
+    msg_fmt = _("There is no plugin with name: `%(name)s` in "
                 "%(namespace)s namespace.")
 
 
@@ -246,3 +246,8 @@ class SSHTimeout(RallyException):
 
 class SSHError(RallyException):
     pass
+
+
+class InvalidConnectionString(RallyException):
+    msg_fmt = _("The connection string is not valid: %(message)s. Please "
+                "check your connection string.")

rally/plugins/common/exporter/file_system.py

@@ -0,0 +1,98 @@
+# Copyright 2016: Mirantis Inc.
+# All Rights Reserved.
+#
+#    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 json
+import os
+
+from six.moves.urllib import parse as urlparse
+
+from rally import api
+from rally.common import logging
+from rally.common.plugin import plugin
+from rally import exceptions
+from rally.task import exporter
+
+
+def configure(name, namespace="default"):
+    return plugin.configure(name=name, namespace=namespace)
+
+LOG = logging.getLogger(__name__)
+
+
+@configure(name="file-exporter")
+class FileExporter(exporter.TaskExporter):
+
+    def validate(self):
+        """Validate connection string.
+
+        The format of connection string in file plugin is
+        file:///<path>.<type>
+        """
+
+        parse_obj = urlparse.urlparse(self.connection_string)
+
+        available_formats = ("json",)
+        available_formats_str = ", ".join(available_formats)
+        if self.connection_string is None or parse_obj.path == "":
+            raise exceptions.InvalidConnectionString(
+                "It should be `file-exporter:///<path>.<type>`.")
+        if self.type not in available_formats:
+            raise exceptions.InvalidConnectionString(
+                "Type of the exported task is not available. The available "
+                "formats are %s." %
+                available_formats_str)
+
+    def __init__(self, connection_string):
+        super(FileExporter, self).__init__(connection_string)
+        self.path = os.path.expanduser(urlparse.urlparse(
+            connection_string).path[1:])
+        self.type = connection_string.split(".")[-1]
+        self.validate()
+
+    def export(self, uuid):
+        """Export results of the task to the file.
+
+        :param uuid: uuid of the task object
+        """
+        task = api.Task.get(uuid)
+
+        LOG.debug("Got the task object by it's uuid %s. " % uuid)
+
+        task_results = [{"key": x["key"], "result": x["data"]["raw"],
+                         "sla": x["data"]["sla"],
+                         "load_duration": x["data"]["load_duration"],
+                         "full_duration": x["data"]["full_duration"]}
+                        for x in task.get_results()]
+
+        if self.type == "json":
+            if task_results:
+                res = json.dumps(task_results, sort_keys=True, indent=4)
+                LOG.debug("Got the task %s results." % uuid)
+            else:
+                msg = ("Task %s results would be available when it will "
+                       "finish." % uuid)
+                raise exceptions.RallyException(msg)
+
+        if os.path.dirname(self.path) and (not os.path.exists(os.path.dirname(
+                self.path))):
+            raise IOError("There is no such directory: %s" %
+                          os.path.dirname(self.path))
+        with open(self.path, "w") as f:
+            LOG.debug("Writing task %s results to the %s." % (
+                uuid, self.connection_string))
+            f.write(res)
+            LOG.debug("Task %s results was written to the %s." % (
+                uuid, self.connection_string))

rally/task/exporter.py

@@ -31,15 +31,19 @@ def configure(name, namespace="default"):
 
 
 @six.add_metaclass(abc.ABCMeta)
-@configure(name="base_task_exporter")
+@configure(name="base-exporter")
 class TaskExporter(plugin.Plugin):
 
+    def __init__(self, connection_string):
+        self.connection_string = connection_string
+
     @abc.abstractmethod
-    def export(self, task_uuid, connection_string):
+    def export(self, task_uuid):
-        """
+        """Export results of the task to the task storage.
-         Export results of the task to the task storage.
 
         :param task_uuid: uuid of task results
-        :param connection_string: string used to connect
+        """
-                                  to the external system
+
-        """
+    @abc.abstractmethod
+    def validate(self):
+        """Used to validate connection string."""

tests/functional/test_cli_task.py

@@ -740,6 +740,59 @@ class TaskTestCase(unittest.TestCase):
             r"(?P<task_id>[0-9a-f\-]{36}): started", output)
         self.assertIsNotNone(result)
 
+    def test_export(self):
+        rally = utils.Rally()
+        cfg = {
+            "Dummy.dummy": [
+                {
+                    "runner": {
+                        "type": "constant",
+                        "times": 100,
+                        "concurrency": 5
+                    }
+                }
+            ]
+        }
+        config = utils.TaskConfig(cfg)
+        output = rally("task start --task %s" % config.filename)
+        uuid = re.search(
+            r"(?P<uuid>[0-9a-f\-]{36}): started", output).group("uuid")
+        connection = (
+            "file-exporter:///" + rally.gen_report_path(extension="json"))
+        output = rally("task export --uuid %s --connection %s" % (
+            uuid, connection))
+        expected = (
+            "Task %(uuid)s results was successfully exported to %("
+            "connection)s using file-exporter plugin." % {
+                "uuid": uuid,
+                "connection": connection,
+            })
+        self.assertIn(expected, output)
+
+    def test_export_with_wrong_connection(self):
+        rally = utils.Rally()
+        cfg = {
+            "Dummy.dummy": [
+                {
+                    "runner": {
+                        "type": "constant",
+                        "times": 100,
+                        "concurrency": 5
+                    }
+                }
+            ]
+        }
+        config = utils.TaskConfig(cfg)
+        output = rally("task start --task %s" % config.filename)
+        uuid = re.search(
+            r"(?P<uuid>[0-9a-f\-]{36}): started", output).group("uuid")
+        connection = (
+            "fake:///" + rally.gen_report_path(extension="json"))
+        self.assertRaises(utils.RallyCliError,
+                          rally,
+                          "task export --uuid %s --connection %s" % (
+                              uuid, connection))
+
 
 class SLATestCase(unittest.TestCase):
 

tests/unit/cli/commands/test_task.py

@@ -742,3 +742,36 @@ class TaskCommandsTestCase(test.TestCase):
         task_id = "ddc3f8ba-082a-496d-b18f-72cdf5c10a14"
         mock_task_get.side_effect = exceptions.TaskNotFound(uuid=task_id)
         self.assertRaises(exceptions.TaskNotFound, self.task.use, task_id)
+
+    @mock.patch("rally.task.exporter.TaskExporter.get")
+    def test_export(self, mock_task_exporter_get):
+        mock_client = mock.Mock()
+        mock_exporter_class = mock.Mock(return_value=mock_client)
+        mock_task_exporter_get.return_value = mock_exporter_class
+        self.task.export("fake_uuid", "file-exporter:///fake_path.json")
+        mock_task_exporter_get.assert_called_once_with("file-exporter")
+        mock_client.export.assert_called_once_with("fake_uuid")
+
+    @mock.patch("rally.task.exporter.TaskExporter.get")
+    def test_export_exception(self, mock_task_exporter_get):
+        mock_client = mock.Mock()
+        mock_exporter_class = mock.Mock(return_value=mock_client)
+        mock_task_exporter_get.return_value = mock_exporter_class
+        mock_client.export.side_effect = IOError
+        self.task.export("fake_uuid", "file-exporter:///fake_path.json")
+        mock_task_exporter_get.assert_called_once_with("file-exporter")
+        mock_client.export.assert_called_once_with("fake_uuid")
+
+    @mock.patch("rally.cli.commands.task.sys.stdout")
+    @mock.patch("rally.task.exporter.TaskExporter.get")
+    def test_export_InvalidConnectionString(self, mock_task_exporter_get,
+                                            mock_stdout):
+        mock_exporter_class = mock.Mock(
+            side_effect=exceptions.InvalidConnectionString)
+        mock_task_exporter_get.return_value = mock_exporter_class
+        self.task.export("fake_uuid", "file-exporter:///fake_path.json")
+        mock_stdout.write.assert_has_calls([
+            mock.call("The connection string is not valid: None. "
+                      "Please check your connection string."),
+            mock.call("\n")])
+        mock_task_exporter_get.assert_called_once_with("file-exporter")

tests/unit/plugins/common/exporter/test_file_system.py

@@ -0,0 +1,96 @@
+# Copyright 2016: Mirantis Inc.
+# All Rights Reserved.
+#
+#    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 ddt
+import mock
+import six
+import six.moves.builtins as __builtin__
+
+from rally import exceptions
+from rally.plugins.common.exporter import file_system
+from tests.unit import test
+
+if six.PY3:
+    import io
+    file = io.BytesIO
+
+
+@ddt.ddt
+class FileExporterTestCase(test.TestCase):
+
+    @mock.patch("rally.plugins.common.exporter.file_system.os.path.exists")
+    @mock.patch.object(__builtin__, "open", autospec=True)
+    @mock.patch("rally.plugins.common.exporter.file_system.json.dumps")
+    @mock.patch("rally.api.Task.get")
+    def test_file_exporter_export(self, mock_task_get, mock_dumps, mock_open,
+                                  mock_exists):
+        mock_task = mock.Mock()
+        mock_exists.return_value = True
+        mock_task_get.return_value = mock_task
+        mock_task.get_results.return_value = [{
+            "key": "fake_key",
+            "data": {
+                "raw": "bar_raw",
+                "sla": "baz_sla",
+                "load_duration": "foo_load_duration",
+                "full_duration": "foo_full_duration",
+            }
+        }]
+        mock_dumps.return_value = "fake_results"
+        input_mock = mock.MagicMock(spec=file)
+        mock_open.return_value = input_mock
+
+        exporter = file_system.FileExporter("file-exporter:///fake_path.json")
+        exporter.export("fake_uuid")
+
+        mock_open().__enter__().write.assert_called_once_with("fake_results")
+        mock_task_get.assert_called_once_with("fake_uuid")
+        expected_dict = [
+            {
+                "load_duration": "foo_load_duration",
+                "full_duration": "foo_full_duration",
+                "result": "bar_raw",
+                "key": "fake_key",
+                "sla": "baz_sla"
+            }
+        ]
+        mock_dumps.assert_called_once_with(expected_dict, sort_keys=True,
+                                           indent=4)
+
+    @mock.patch("rally.api.Task.get")
+    def test_file_exporter_export_running_task(self, mock_task_get):
+        mock_task = mock.Mock()
+        mock_task_get.return_value = mock_task
+        mock_task.get_results.return_value = []
+
+        exporter = file_system.FileExporter("file-exporter:///fake_path.json")
+        self.assertRaises(exceptions.RallyException, exporter.export,
+                          "fake_uuid")
+
+    @ddt.data(
+        {"connection": "",
+         "raises": exceptions.InvalidConnectionString},
+        {"connection": "file-exporter:///fake_path.json",
+         "raises": None},
+        {"connection": "file-exporter:///fake_path.fake",
+         "raises": exceptions.InvalidConnectionString},
+    )
+    @ddt.unpack
+    def test_file_exporter_validate(self, connection, raises):
+        print (connection)
+        if raises:
+            self.assertRaises(raises, file_system.FileExporter, connection)
+        else:
+            file_system.FileExporter(connection)

tests/unit/task/test_exporter.py

@@ -16,9 +16,12 @@ from rally.task import exporter
 from tests.unit import test
 
 
-@exporter.configure(name="test_exporter")
+@exporter.configure(name="test-exporter")
 class TestExporter(exporter.TaskExporter):
 
+    def validate(self):
+        pass
+
     def export(self, task, connection_string):
         pass
 
@@ -26,7 +29,7 @@ class TestExporter(exporter.TaskExporter):
 class ExporterTestCase(test.TestCase):
 
     def test_task_export(self):
-        self.assertRaises(TypeError, exporter.TaskExporter)
+        self.assertRaises(TypeError, exporter.TaskExporter, "fake_connection")
 
     def test_task_export_instantiate(self):
-        TestExporter()
+        TestExporter("fake_connection")