X-Git-Url: https://gerrit.opnfv.org/gerrit/gitweb?a=blobdiff_plain;f=yardstick%2Fdispatcher%2Finfluxdb.py;h=e8c7cf57ba5cb18d415dac10a12d23f71a326c45;hb=d85cefdffe8b3ccb9b919f1cb83be42152e9e802;hp=d388d28a1c905b35db50c3fc865a1c7d97db1c05;hpb=dd8ca59fd7421a1f362e50f0d3413a903761d179;p=yardstick.git diff --git a/yardstick/dispatcher/influxdb.py b/yardstick/dispatcher/influxdb.py index d388d28a1..e8c7cf57b 100644 --- a/yardstick/dispatcher/influxdb.py +++ b/yardstick/dispatcher/influxdb.py @@ -10,43 +10,18 @@ from __future__ import absolute_import import logging -import os import time +import os import requests -import six -from oslo_config import cfg -from oslo_serialization import jsonutils +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 LOG = logging.getLogger(__name__) -CONF = cfg.CONF -influx_dispatcher_opts = [ - cfg.StrOpt('target', - default='http://127.0.0.1:8086', - help='The target where the http request will be sent. ' - 'If this is not set, no data will be posted. For ' - 'example: target = http://hostname:1234/path'), - 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 ' - 'timeout.'), -] - -CONF.register_opts(influx_dispatcher_opts, group="dispatcher_influxdb") - class InfluxdbDispatcher(DispatchBase): """Dispatcher class for posting data into an influxdb target. @@ -56,124 +31,93 @@ class InfluxdbDispatcher(DispatchBase): def __init__(self, conf): super(InfluxdbDispatcher, self).__init__(conf) - 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('NODE_NAME', 'unknown'), - "installer": os.environ.get('INSTALLER_TYPE', 'unknown'), - "deploy_scenario": os.environ.get('DEPLOY_SCENARIO', 'unknown'), - "version": os.path.basename(os.environ.get('YARDSTICK_BRANCH', - 'unknown')) - - } + db_conf = conf['dispatcher_influxdb'] + self.timeout = int(db_conf.get('timeout', 5)) + self.target = db_conf.get('target', 'http://127.0.0.1:8086') + self.db_name = db_conf.get('db_name', 'yardstick') + self.username = db_conf.get('username', 'root') + self.password = db_conf.get('password', 'root') - def _dict_key_flatten(self, data): - next_data = {} - - if not [v for v in data.values() - if type(v) == dict or type(v) == list]: - return data - - for k, v in six.iteritems(data): - if type(v) == dict: - 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): - next_data["%s%d" % (k, index)] = item - else: - next_data[k] = v + self.influxdb_url = "%s/write?db=%s" % (self.target, self.db_name) - return self._dict_key_flatten(next_data) + self.task_id = None + self.tags = None - def _get_nano_timestamp(self, results): - try: - timestamp = results["benchmark"]["timestamp"] - except Exception: - timestamp = time.time() + def flush_result_data(self, data): + LOG.debug('Test result all : %s', data) + if self.target == '': + # if the target was not set, do not do anything + LOG.error('Dispatcher target was not set, no data will be posted.') - return str(int(float(timestamp) * 1000000000)) + result = data['result'] + self.tags = result['info'] + self.task_id = result['task_id'] + self.criteria = result['criteria'] + testcases = result['testcases'] - def _get_extended_tags(self, data): - runner_info = self.runners_info[data["runner_id"]] - tags = { - "runner_id": data["runner_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"] + for case, data in testcases.items(): + tc_criteria = data['criteria'] + for record in data['tc_data']: + # skip results with no data because we influxdb encode empty dicts + if record.get("data"): + self.upload_one_record(record, case, tc_criteria) - return tags + return 0 - def _data_to_line_protocol(self, data): - msg = {} - point = {} - 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) - msg["points"] = [point] - msg["tags"] = self.static_tags + def upload_one_record(self, data, case, tc_criteria, task_id=None): + if task_id: + self.task_id = task_id - return make_lines(msg).encode('utf-8') - - def record_result_data(self, 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 - LOG.error('Dispatcher target was not set, no data will' - 'be posted.') - return -1 - - if isinstance(data, dict) and "scenario_cfg" in data: - 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.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 + 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) - 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) - return -1 - return 0 + def _data_to_line_protocol(self, data, case, criteria): + msg = {} - def flush_result_data(self): - LOG.debug('Test result all : %s', - jsonutils.dump_as_bytes(self.raw_result)) - return 0 + 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": utils.flatten_dict_key(data["data"]), + "time": self._get_nano_timestamp(data), + "tags": self._get_extended_tags(criteria), + } + msg["points"] = [point] + msg["tags"] = self.tags + + return make_lines(msg).encode('utf-8') + + def _get_nano_timestamp(self, results): + try: + timestamp = results["timestamp"] + except KeyError: + timestamp = time.time() + + return str(int(float(timestamp) * 1000000000)) + + def _get_extended_tags(self, criteria): + tags = { + "task_id": self.task_id, + "criteria": criteria + } + + return tags