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 third_party.influxdb.influxdb_line_protocol import make_lines
20 from yardstick.dispatcher.base import Base as DispatchBase
22 LOG = logging.getLogger(__name__)
25 class InfluxdbDispatcher(DispatchBase):
26 """Dispatcher class for posting data into an influxdb target.
29 __dispatcher_type__ = "Influxdb"
31 def __init__(self, conf):
32 super(InfluxdbDispatcher, self).__init__(conf)
33 db_conf = conf['dispatcher_influxdb']
34 self.timeout = int(db_conf.get('timeout', 5))
35 self.target = db_conf.get('target', 'http://127.0.0.1:8086')
36 self.db_name = db_conf.get('db_name', 'yardstick')
37 self.username = db_conf.get('username', 'root')
38 self.password = db_conf.get('password', 'root')
40 self.influxdb_url = "%s/write?db=%s" % (self.target, self.db_name)
44 def flush_result_data(self, data):
45 LOG.debug('Test result all : %s', data)
47 # if the target was not set, do not do anything
48 LOG.error('Dispatcher target was not set, no data will be posted.')
50 result = data['result']
51 self.tags = result['info']
52 self.task_id = result['task_id']
53 self.criteria = result['criteria']
54 testcases = result['testcases']
56 for case, data in testcases.items():
57 tc_criteria = data['criteria']
58 for record in data['tc_data']:
59 self._upload_one_record(record, case, tc_criteria)
63 def _upload_one_record(self, data, case, tc_criteria):
65 line = self._data_to_line_protocol(data, case, tc_criteria)
66 LOG.debug('Test result line format : %s', line)
67 res = requests.post(self.influxdb_url,
69 auth=(self.username, self.password),
71 if res.status_code != 204:
72 LOG.error('Test result posting finished with status code'
73 ' %d.', res.status_code)
76 except Exception as err:
77 LOG.exception('Failed to record result data: %s', err)
79 def _data_to_line_protocol(self, data, case, criteria):
83 "fields": self._dict_key_flatten(data["data"]),
84 "time": self._get_nano_timestamp(data),
85 "tags": self._get_extended_tags(criteria),
87 msg["points"] = [point]
88 msg["tags"] = self.tags
90 return make_lines(msg).encode('utf-8')
92 def _dict_key_flatten(self, data):
95 # use list, because iterable is too generic
96 if not [v for v in data.values() if
97 isinstance(v, (collections.Mapping, list))]:
100 for k, v in six.iteritems(data):
101 if isinstance(v, collections.Mapping):
102 for n_k, n_v in six.iteritems(v):
103 next_data["%s.%s" % (k, n_k)] = n_v
104 # use list because iterable is too generic
105 elif isinstance(v, list):
106 for index, item in enumerate(v):
107 next_data["%s%d" % (k, index)] = item
111 return self._dict_key_flatten(next_data)
113 def _get_nano_timestamp(self, results):
115 timestamp = results["timestamp"]
117 timestamp = time.time()
119 return str(int(float(timestamp) * 1000000000))
121 def _get_extended_tags(self, criteria):
123 "task_id": self.task_id,