Record task status if running via CLI

[yardstick.git] / yardstick / dispatcher / influxdb.py

diff --git a/yardstick/dispatcher/influxdb.py b/yardstick/dispatcher/influxdb.py
index 2f3ff08..427e669 100644 (file)
--- a/yardstick/dispatcher/influxdb.py
+++ b/yardstick/dispatcher/influxdb.py
@@ -7,16 +7,19 @@
 # http://www.apache.org/licenses/LICENSE-2.0
 ##############################################################################
 
-import os
-import json
+from __future__ import absolute_import
+
 import logging
-import requests
+import os
 import time
 
+import requests
+import six
 from oslo_config import cfg
+from oslo_serialization import jsonutils
 
+from third_party.influxdb.influxdb_line_protocol import make_lines
 from yardstick.dispatcher.base import Base as DispatchBase
-from yardstick.dispatcher.influxdb_line_protocol import make_lines
 
 LOG = logging.getLogger(__name__)
 
@@ -30,6 +33,12 @@ influx_dispatcher_opts = [
     cfg.StrOpt('db_name',
                default='yardstick',
                help='The database name to store test results.'),
+    cfg.StrOpt('username',
+               default='root',
+               help='The user name to access database.'),
+    cfg.StrOpt('password',
+               default='root',
+               help='The user password to access database.'),
     cfg.IntOpt('timeout',
                default=5,
                help='The max time in seconds to wait for a request to '
@@ -50,15 +59,21 @@ class InfluxdbDispatcher(DispatchBase):
         self.timeout = CONF.dispatcher_influxdb.timeout
         self.target = CONF.dispatcher_influxdb.target
         self.db_name = CONF.dispatcher_influxdb.db_name
+        self.username = CONF.dispatcher_influxdb.username
+        self.password = CONF.dispatcher_influxdb.password
         self.influxdb_url = "%s/write?db=%s" % (self.target, self.db_name)
         self.raw_result = []
         self.case_name = ""
         self.tc = ""
         self.task_id = -1
+        self.runners_info = {}
         self.static_tags = {
-            "pod_name": os.environ.get('POD_NAME', 'unknown'),
+            "pod_name": os.environ.get('NODE_NAME', 'unknown'),
             "installer": os.environ.get('INSTALLER_TYPE', 'unknown'),
-            "version": os.environ.get('YARDSTICK_VERSION', 'unknown')
+            "deploy_scenario": os.environ.get('DEPLOY_SCENARIO', 'unknown'),
+            "version": os.path.basename(os.environ.get('YARDSTICK_BRANCH',
+                                                       'unknown'))
+
         }
 
     def _dict_key_flatten(self, data):
@@ -68,9 +83,9 @@ class InfluxdbDispatcher(DispatchBase):
                 if type(v) == dict or type(v) == list]:
             return data
 
-        for k, v in data.iteritems():
+        for k, v in six.iteritems(data):
             if type(v) == dict:
-                for n_k, n_v in v.iteritems():
+                for n_k, n_v in six.iteritems(v):
                     next_data["%s.%s" % (k, n_k)] = n_v
             elif type(v) == list:
                 for index, item in enumerate(v):
@@ -89,18 +104,23 @@ class InfluxdbDispatcher(DispatchBase):
         return str(int(float(timestamp) * 1000000000))
 
     def _get_extended_tags(self, data):
+        runner_info = self.runners_info[data["runner_id"]]
         tags = {
             "runner_id": data["runner_id"],
-            "tc": self.tc,
-            "task_id": self.task_id
+            "task_id": self.task_id,
+            "scenarios": runner_info["scenarios"]
         }
+        if "host" in runner_info:
+            tags["host"] = runner_info["host"]
+        if "target" in runner_info:
+            tags["target"] = runner_info["target"]
 
         return tags
 
     def _data_to_line_protocol(self, data):
         msg = {}
         point = {}
-        point["measurement"] = self.case_name
+        point["measurement"] = self.tc
         point["fields"] = self._dict_key_flatten(data["benchmark"]["data"])
         point["time"] = self._get_nano_timestamp(data)
         point["tags"] = self._get_extended_tags(data)
@@ -110,7 +130,7 @@ class InfluxdbDispatcher(DispatchBase):
         return make_lines(msg).encode('utf-8')
 
     def record_result_data(self, data):
-        LOG.debug('Test result : %s' % json.dumps(data))
+        LOG.debug('Test result : %s', jsonutils.dump_as_bytes(data))
         self.raw_result.append(data)
         if self.target == '':
             # if the target was not set, do not do anything
@@ -119,25 +139,34 @@ class InfluxdbDispatcher(DispatchBase):
             return -1
 
         if isinstance(data, dict) and "scenario_cfg" in data:
-            self.case_name = data["scenario_cfg"]["type"]
             self.tc = data["scenario_cfg"]["tc"]
             self.task_id = data["scenario_cfg"]["task_id"]
+            scenario_cfg = data["scenario_cfg"]
+            runner_id = data["runner_id"]
+            self.runners_info[runner_id] = {"scenarios": scenario_cfg["type"]}
+            if "host" in scenario_cfg:
+                self.runners_info[runner_id]["host"] = scenario_cfg["host"]
+            if "target" in scenario_cfg:
+                self.runners_info[runner_id]["target"] = scenario_cfg["target"]
             return 0
 
-        if self.case_name == "":
-            LOG.error('Test result : %s' % json.dumps(data))
+        if self.tc == "":
+            LOG.error('Test result : %s', jsonutils.dump_as_bytes(data))
             LOG.error('The case_name cannot be found, no data will be posted.')
             return -1
 
         try:
             line = self._data_to_line_protocol(data)
-            LOG.debug('Test result line format : %s' % line)
+            LOG.debug('Test result line format : %s', line)
             res = requests.post(self.influxdb_url,
                                 data=line,
+                                auth=(self.username, self.password),
                                 timeout=self.timeout)
             if res.status_code != 204:
                 LOG.error('Test result posting finished with status code'
                           ' %d.' % res.status_code)
+                LOG.error(res.text)
+
         except Exception as err:
             LOG.exception('Failed to record result data: %s',
                           err)
@@ -145,5 +174,6 @@ class InfluxdbDispatcher(DispatchBase):
         return 0
 
     def flush_result_data(self):
-        LOG.debug('Test result all : %s' % json.dumps(self.raw_result))
+        LOG.debug('Test result all : %s',
+                  jsonutils.dump_as_bytes(self.raw_result))
         return 0