1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
|
##############################################################################
# Copyright (c) 2015 Huawei Technologies Co.,Ltd and others.
#
# All rights reserved. This program and the accompanying materials
# are made available under the terms of the Apache License, Version 2.0
# which accompanies this distribution, and is available at
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
import os
import json
import logging
import requests
import time
from oslo_config import cfg
from yardstick.dispatcher.base import Base as DispatchBase
from yardstick.dispatcher.influxdb_line_protocol import make_lines
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.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.
"""
__dispatcher_type__ = "Influxdb"
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.influxdb_url = "%s/write?db=%s" % (self.target, self.db_name)
self.raw_result = []
self.case_name = ""
self.static_tags = {
"pod_name": os.environ.get('POD_NAME', 'unknown'),
"installer": os.environ.get('INSTALLER_TYPE', 'unknown'),
"version": os.environ.get('YARDSTICK_VERSION', 'unknown')
}
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 data.iteritems():
if type(v) == dict:
for n_k, n_v in v.iteritems():
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
return self._dict_key_flatten(next_data)
def _get_nano_timestamp(self, results):
try:
timestamp = results["benchmark"]["timestamp"]
except Exception:
timestamp = time.time()
return str(int(float(timestamp) * 1000000000))
def _data_to_line_protocol(self, data):
msg = {}
point = {}
point["measurement"] = self.case_name
point["fields"] = self._dict_key_flatten(data["benchmark"]["data"])
point["time"] = self._get_nano_timestamp(data)
msg["points"] = [point]
msg["tags"] = self.static_tags
return make_lines(msg).encode('utf-8')
def record_result_data(self, data):
LOG.debug('Test result : %s' % json.dumps(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.case_name = data["scenario_cfg"]["type"]
return 0
if self.case_name == "":
LOG.error('Test result : %s' % json.dumps(data))
LOG.error('The case_name cannot be found, no data will be posted.')
return -1
try:
line = self._data_to_line_protocol(data)
LOG.debug('Test result line format : %s' % line)
res = requests.post(self.influxdb_url,
data=line,
timeout=self.timeout)
if res.status_code != 204:
LOG.error('Test result posting finished with status code'
' %d.' % res.status_code)
except Exception as err:
LOG.exception('Failed to record result data: %s',
err)
return -1
return 0
def flush_result_data(self):
LOG.debug('Test result all : %s' % json.dumps(self.raw_result))
return 0
|