1 ##############################################################################
2 # Copyright (c) 2015 Huawei Technologies Co.,Ltd and others.
4 # All rights reserved. This program and the accompanying materials
5 # are made available under the terms of the Apache License, Version 2.0
6 # which accompanies this distribution, and is available at
7 # http://www.apache.org/licenses/LICENSE-2.0
8 ##############################################################################
10 from __future__ import absolute_import
18 from oslo_config import cfg
19 from oslo_serialization import jsonutils
21 from third_party.influxdb.influxdb_line_protocol import make_lines
22 from yardstick.dispatcher.base import Base as DispatchBase
24 LOG = logging.getLogger(__name__)
27 influx_dispatcher_opts = [
29 default='http://127.0.0.1:8086',
30 help='The target where the http request will be sent. '
31 'If this is not set, no data will be posted. For '
32 'example: target = http://hostname:1234/path'),
35 help='The database name to store test results.'),
36 cfg.StrOpt('username',
38 help='The user name to access database.'),
39 cfg.StrOpt('password',
41 help='The user password to access database.'),
44 help='The max time in seconds to wait for a request to '
48 CONF.register_opts(influx_dispatcher_opts, group="dispatcher_influxdb")
51 class InfluxdbDispatcher(DispatchBase):
52 """Dispatcher class for posting data into an influxdb target.
55 __dispatcher_type__ = "Influxdb"
57 def __init__(self, conf):
58 super(InfluxdbDispatcher, self).__init__(conf)
59 self.timeout = CONF.dispatcher_influxdb.timeout
60 self.target = CONF.dispatcher_influxdb.target
61 self.db_name = CONF.dispatcher_influxdb.db_name
62 self.username = CONF.dispatcher_influxdb.username
63 self.password = CONF.dispatcher_influxdb.password
64 self.influxdb_url = "%s/write?db=%s" % (self.target, self.db_name)
69 self.runners_info = {}
71 "pod_name": os.environ.get('NODE_NAME', 'unknown'),
72 "installer": os.environ.get('INSTALLER_TYPE', 'unknown'),
73 "deploy_scenario": os.environ.get('DEPLOY_SCENARIO', 'unknown'),
74 "version": os.path.basename(os.environ.get('YARDSTICK_BRANCH',
79 def _dict_key_flatten(self, data):
82 if not [v for v in data.values()
83 if type(v) == dict or type(v) == list]:
86 for k, v in six.iteritems(data):
88 for n_k, n_v in six.iteritems(v):
89 next_data["%s.%s" % (k, n_k)] = n_v
91 for index, item in enumerate(v):
92 next_data["%s%d" % (k, index)] = item
96 return self._dict_key_flatten(next_data)
98 def _get_nano_timestamp(self, results):
100 timestamp = results["benchmark"]["timestamp"]
102 timestamp = time.time()
104 return str(int(float(timestamp) * 1000000000))
106 def _get_extended_tags(self, data):
107 runner_info = self.runners_info[data["runner_id"]]
109 "runner_id": data["runner_id"],
110 "task_id": self.task_id,
111 "scenarios": runner_info["scenarios"]
113 if "host" in runner_info:
114 tags["host"] = runner_info["host"]
115 if "target" in runner_info:
116 tags["target"] = runner_info["target"]
120 def _data_to_line_protocol(self, data):
123 point["measurement"] = self.tc
124 point["fields"] = self._dict_key_flatten(data["benchmark"]["data"])
125 point["time"] = self._get_nano_timestamp(data)
126 point["tags"] = self._get_extended_tags(data)
127 msg["points"] = [point]
128 msg["tags"] = self.static_tags
130 return make_lines(msg).encode('utf-8')
132 def record_result_data(self, data):
133 LOG.debug('Test result : %s', jsonutils.dump_as_bytes(data))
134 self.raw_result.append(data)
135 if self.target == '':
136 # if the target was not set, do not do anything
137 LOG.error('Dispatcher target was not set, no data will'
141 if isinstance(data, dict) and "scenario_cfg" in data:
142 self.tc = data["scenario_cfg"]["tc"]
143 self.task_id = data["scenario_cfg"]["task_id"]
144 scenario_cfg = data["scenario_cfg"]
145 runner_id = data["runner_id"]
146 self.runners_info[runner_id] = {"scenarios": scenario_cfg["type"]}
147 if "host" in scenario_cfg:
148 self.runners_info[runner_id]["host"] = scenario_cfg["host"]
149 if "target" in scenario_cfg:
150 self.runners_info[runner_id]["target"] = scenario_cfg["target"]
154 LOG.error('Test result : %s', jsonutils.dump_as_bytes(data))
155 LOG.error('The case_name cannot be found, no data will be posted.')
159 line = self._data_to_line_protocol(data)
160 LOG.debug('Test result line format : %s', line)
161 res = requests.post(self.influxdb_url,
163 auth=(self.username, self.password),
164 timeout=self.timeout)
165 if res.status_code != 204:
166 LOG.error('Test result posting finished with status code'
167 ' %d.' % res.status_code)
170 except Exception as err:
171 LOG.exception('Failed to record result data: %s',
176 def flush_result_data(self):
177 LOG.debug('Test result all : %s',
178 jsonutils.dump_as_bytes(self.raw_result))