Merge "Use context name instead of key_uuid"

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

diff --git a/yardstick/dispatcher/influxdb.py b/yardstick/dispatcher/influxdb.py
index 373aae1..e8c7cf5 100644 (file)
--- a/yardstick/dispatcher/influxdb.py
+++ b/yardstick/dispatcher/influxdb.py
@@ -11,11 +11,12 @@ from __future__ import absolute_import
 
 import logging
 import time
+import os
 
-import collections
 import requests
-import six
+from requests import ConnectionError
 
+from yardstick.common import utils
 from third_party.influxdb.influxdb_line_protocol import make_lines
 from yardstick.dispatcher.base import Base as DispatchBase
 
@@ -39,7 +40,8 @@ class InfluxdbDispatcher(DispatchBase):
 
         self.influxdb_url = "%s/write?db=%s" % (self.target, self.db_name)
 
-        self.task_id = -1
+        self.task_id = None
+        self.tags = None
 
     def flush_result_data(self, data):
         LOG.debug('Test result all : %s', data)
@@ -56,31 +58,46 @@ class InfluxdbDispatcher(DispatchBase):
         for case, data in testcases.items():
             tc_criteria = data['criteria']
             for record in data['tc_data']:
-                self._upload_one_record(record, case, tc_criteria)
+                # skip results with no data because we influxdb encode empty dicts
+                if record.get("data"):
+                    self.upload_one_record(record, case, tc_criteria)
 
         return 0
 
-    def _upload_one_record(self, data, case, tc_criteria):
+    def upload_one_record(self, data, case, tc_criteria, task_id=None):
+        if task_id:
+            self.task_id = task_id
+
+        line = self._data_to_line_protocol(data, case, tc_criteria)
+        LOG.debug('Test result line format : %s', line)
+
         try:
-            line = self._data_to_line_protocol(data, case, tc_criteria)
-            LOG.debug('Test result line format : %s', line)
             res = requests.post(self.influxdb_url,
                                 data=line,
                                 auth=(self.username, self.password),
                                 timeout=self.timeout)
+        except ConnectionError as err:
+            LOG.exception('Failed to record result data: %s', err)
+        else:
             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)
-
     def _data_to_line_protocol(self, data, case, criteria):
         msg = {}
+
+        if not self.tags:
+            self.tags = {
+                'deploy_scenario': os.environ.get('DEPLOY_SCENARIO', 'unknown'),
+                'installer': os.environ.get('INSTALLER_TYPE', 'unknown'),
+                'pod_name': os.environ.get('NODE_NAME', 'unknown'),
+                'version': os.environ.get('YARDSTICK_BRANCH', 'unknown')
+            }
+
         point = {
             "measurement": case,
-            "fields": self._dict_key_flatten(data["data"]),
+            "fields": utils.flatten_dict_key(data["data"]),
             "time": self._get_nano_timestamp(data),
             "tags": self._get_extended_tags(criteria),
         }
@@ -89,31 +106,10 @@ class InfluxdbDispatcher(DispatchBase):
 
         return make_lines(msg).encode('utf-8')
 
-    def _dict_key_flatten(self, data):
-        next_data = {}
-
-        # use list, because iterable is too generic
-        if not [v for v in data.values() if
-                isinstance(v, (collections.Mapping, list))]:
-            return data
-
-        for k, v in six.iteritems(data):
-            if isinstance(v, collections.Mapping):
-                for n_k, n_v in six.iteritems(v):
-                    next_data["%s.%s" % (k, n_k)] = n_v
-            # use list because iterable is too generic
-            elif isinstance(v, list):
-                for index, item in enumerate(v):
-                    next_data["%s%d" % (k, index)] = item
-            else:
-                next_data[k] = v
-
-        return self._dict_key_flatten(next_data)
-
     def _get_nano_timestamp(self, results):
         try:
             timestamp = results["timestamp"]
-        except Exception:
+        except KeyError:
             timestamp = time.time()
 
         return str(int(float(timestamp) * 1000000000))