yardstick/dispatcher/influxdb.py

   1 ##############################################################################
   2 # Copyright (c) 2015 Huawei Technologies Co.,Ltd and others.
   3 #
   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 ##############################################################################
   9
  10 from __future__ import absolute_import
  11
  12 import logging
  13 import time
  14
  15 import collections
  16 import requests
  17 import six
  18
  19 from third_party.influxdb.influxdb_line_protocol import make_lines
  20 from yardstick.dispatcher.base import Base as DispatchBase
  21
  22 LOG = logging.getLogger(__name__)
  23
  24
  25 class InfluxdbDispatcher(DispatchBase):
  26     """Dispatcher class for posting data into an influxdb target.
  27     """
  28
  29     __dispatcher_type__ = "Influxdb"
  30
  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')
  39
  40         self.influxdb_url = "%s/write?db=%s" % (self.target, self.db_name)
  41
  42         self.task_id = -1
  43
  44     def flush_result_data(self, data):
  45         LOG.debug('Test result all : %s', data)
  46         if self.target == '':
  47             # if the target was not set, do not do anything
  48             LOG.error('Dispatcher target was not set, no data will be posted.')
  49
  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']
  55
  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)
  60
  61         return 0
  62
  63     def _upload_one_record(self, data, case, tc_criteria):
  64         try:
  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,
  68                                 data=line,
  69                                 auth=(self.username, self.password),
  70                                 timeout=self.timeout)
  71             if res.status_code != 204:
  72                 LOG.error('Test result posting finished with status code'
  73                           ' %d.', res.status_code)
  74                 LOG.error(res.text)
  75
  76         except Exception as err:
  77             LOG.exception('Failed to record result data: %s', err)
  78
  79     def _data_to_line_protocol(self, data, case, criteria):
  80         msg = {}
  81         point = {
  82             "measurement": case,
  83             "fields": self._dict_key_flatten(data["data"]),
  84             "time": self._get_nano_timestamp(data),
  85             "tags": self._get_extended_tags(criteria),
  86         }
  87         msg["points"] = [point]
  88         msg["tags"] = self.tags
  89
  90         return make_lines(msg).encode('utf-8')
  91
  92     def _dict_key_flatten(self, data):
  93         next_data = {}
  94
  95         # use list, because iterable is too generic
  96         if not [v for v in data.values() if
  97                 isinstance(v, (collections.Mapping, list))]:
  98             return data
  99
 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
 108             else:
 109                 next_data[k] = v
 110
 111         return self._dict_key_flatten(next_data)
 112
 113     def _get_nano_timestamp(self, results):
 114         try:
 115             timestamp = results["timestamp"]
 116         except Exception:
 117             timestamp = time.time()
 118
 119         return str(int(float(timestamp) * 1000000000))
 120
 121     def _get_extended_tags(self, criteria):
 122         tags = {
 123             "task_id": self.task_id,
 124             "criteria": criteria
 125         }
 126
 127         return tags