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
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 class InfluxdbDispatcher(DispatchBase):
28 """Dispatcher class for posting data into an influxdb target.
31 __dispatcher_type__ = "Influxdb"
33 def __init__(self, conf, config):
34 super(InfluxdbDispatcher, self).__init__(conf)
35 db_conf = config['yardstick'].get('dispatcher_influxdb', {})
36 self.timeout = int(db_conf.get('timeout', 5))
37 self.target = db_conf.get('target', 'http://127.0.0.1:8086')
38 self.db_name = db_conf.get('db_name', 'yardstick')
39 self.username = db_conf.get('username', 'root')
40 self.password = db_conf.get('password', 'root')
41 self.influxdb_url = "%s/write?db=%s" % (self.target, self.db_name)
46 self.runners_info = {}
48 "pod_name": os.environ.get('NODE_NAME', 'unknown'),
49 "installer": os.environ.get('INSTALLER_TYPE', 'unknown'),
50 "deploy_scenario": os.environ.get('DEPLOY_SCENARIO', 'unknown'),
51 "version": os.path.basename(os.environ.get('YARDSTICK_BRANCH',
56 def _dict_key_flatten(self, data):
59 # use list, because iterable is too generic
60 if not [v for v in data.values() if
61 isinstance(v, (collections.Mapping, list))]:
64 for k, v in six.iteritems(data):
65 if isinstance(v, collections.Mapping):
66 for n_k, n_v in six.iteritems(v):
67 next_data["%s.%s" % (k, n_k)] = n_v
68 # use list because iterable is too generic
69 elif isinstance(v, list):
70 for index, item in enumerate(v):
71 next_data["%s%d" % (k, index)] = item
75 return self._dict_key_flatten(next_data)
77 def _get_nano_timestamp(self, results):
79 timestamp = results["benchmark"]["timestamp"]
81 timestamp = time.time()
83 return str(int(float(timestamp) * 1000000000))
85 def _get_extended_tags(self, data):
86 runner_info = self.runners_info[data["runner_id"]]
88 "runner_id": data["runner_id"],
89 "task_id": self.task_id,
90 "scenarios": runner_info["scenarios"]
92 if "host" in runner_info:
93 tags["host"] = runner_info["host"]
94 if "target" in runner_info:
95 tags["target"] = runner_info["target"]
99 def _data_to_line_protocol(self, data):
102 "measurement": self.tc,
103 "fields": self._dict_key_flatten(data["benchmark"]["data"]),
104 "time": self._get_nano_timestamp(data),
105 "tags": self._get_extended_tags(data),
107 msg["points"] = [point]
108 msg["tags"] = self.static_tags
110 return make_lines(msg).encode('utf-8')
112 def record_result_data(self, data):
113 LOG.debug('Test result : %s', jsonutils.dump_as_bytes(data))
114 self.raw_result.append(data)
115 if self.target == '':
116 # if the target was not set, do not do anything
117 LOG.error('Dispatcher target was not set, no data will'
121 if isinstance(data, dict) and "scenario_cfg" in data:
122 self.tc = data["scenario_cfg"]["tc"]
123 self.task_id = data["scenario_cfg"]["task_id"]
124 scenario_cfg = data["scenario_cfg"]
125 runner_id = data["runner_id"]
126 self.runners_info[runner_id] = {"scenarios": scenario_cfg["type"]}
127 if "host" in scenario_cfg:
128 self.runners_info[runner_id]["host"] = scenario_cfg["host"]
129 if "target" in scenario_cfg:
130 self.runners_info[runner_id]["target"] = scenario_cfg["target"]
134 LOG.error('Test result : %s', jsonutils.dump_as_bytes(data))
135 LOG.error('The case_name cannot be found, no data will be posted.')
139 line = self._data_to_line_protocol(data)
140 LOG.debug('Test result line format : %s', line)
141 res = requests.post(self.influxdb_url,
143 auth=(self.username, self.password),
144 timeout=self.timeout)
145 if res.status_code != 204:
146 LOG.error('Test result posting finished with status code'
147 ' %d.', res.status_code)
150 except Exception as err:
151 LOG.exception('Failed to record result data: %s',
156 def flush_result_data(self):
157 LOG.debug('Test result all : %s',
158 jsonutils.dump_as_bytes(self.raw_result))