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 ##############################################################################
16 from oslo_config import cfg
18 from yardstick.dispatcher.base import Base as DispatchBase
19 from yardstick.dispatcher.influxdb_line_protocol import make_lines
21 LOG = logging.getLogger(__name__)
24 influx_dispatcher_opts = [
26 default='http://127.0.0.1:8086',
27 help='The target where the http request will be sent. '
28 'If this is not set, no data will be posted. For '
29 'example: target = http://hostname:1234/path'),
32 help='The database name to store test results.'),
35 help='The max time in seconds to wait for a request to '
39 CONF.register_opts(influx_dispatcher_opts, group="dispatcher_influxdb")
42 class InfluxdbDispatcher(DispatchBase):
43 """Dispatcher class for posting data into an influxdb target.
46 __dispatcher_type__ = "Influxdb"
48 def __init__(self, conf):
49 super(InfluxdbDispatcher, self).__init__(conf)
50 self.timeout = CONF.dispatcher_influxdb.timeout
51 self.target = CONF.dispatcher_influxdb.target
52 self.db_name = CONF.dispatcher_influxdb.db_name
53 self.influxdb_url = "%s/write?db=%s" % (self.target, self.db_name)
59 "pod_name": os.environ.get('POD_NAME', 'unknown'),
60 "installer": os.environ.get('INSTALLER_TYPE', 'unknown'),
61 "version": os.environ.get('YARDSTICK_VERSION', 'unknown')
64 def _dict_key_flatten(self, data):
67 if not [v for v in data.values()
68 if type(v) == dict or type(v) == list]:
71 for k, v in data.iteritems():
73 for n_k, n_v in v.iteritems():
74 next_data["%s.%s" % (k, n_k)] = n_v
76 for index, item in enumerate(v):
77 next_data["%s%d" % (k, index)] = item
81 return self._dict_key_flatten(next_data)
83 def _get_nano_timestamp(self, results):
85 timestamp = results["benchmark"]["timestamp"]
87 timestamp = time.time()
89 return str(int(float(timestamp) * 1000000000))
91 def _get_extended_tags(self, data):
93 "runner_id": data["runner_id"],
95 "task_id": self.task_id
100 def _data_to_line_protocol(self, data):
103 point["measurement"] = self.case_name
104 point["fields"] = self._dict_key_flatten(data["benchmark"]["data"])
105 point["time"] = self._get_nano_timestamp(data)
106 point["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' % json.dumps(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.case_name = data["scenario_cfg"]["type"]
123 self.tc = data["scenario_cfg"]["tc"]
124 self.task_id = data["scenario_cfg"]["task_id"]
127 if self.case_name == "":
128 LOG.error('Test result : %s' % json.dumps(data))
129 LOG.error('The case_name cannot be found, no data will be posted.')
133 line = self._data_to_line_protocol(data)
134 LOG.debug('Test result line format : %s' % line)
135 res = requests.post(self.influxdb_url,
137 timeout=self.timeout)
138 if res.status_code != 204:
139 LOG.error('Test result posting finished with status code'
140 ' %d.' % res.status_code)
141 except Exception as err:
142 LOG.exception('Failed to record result data: %s',
147 def flush_result_data(self):
148 LOG.debug('Test result all : %s' % json.dumps(self.raw_result))