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 os
  14 import time
  15
  16 import collections
  17 import requests
  18 import six
  19 from oslo_serialization import jsonutils
  20
  21 from third_party.influxdb.influxdb_line_protocol import make_lines
  22 from yardstick.dispatcher.base import Base as DispatchBase
  23
  24 LOG = logging.getLogger(__name__)
  25
  26
  27 class InfluxdbDispatcher(DispatchBase):
  28     """Dispatcher class for posting data into an influxdb target.
  29     """
  30
  31     __dispatcher_type__ = "Influxdb"
  32
  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)
  42         self.raw_result = []
  43         self.case_name = ""
  44         self.tc = ""
  45         self.task_id = -1
  46         self.runners_info = {}
  47         self.static_tags = {
  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',
  52                                                        'unknown'))
  53
  54         }
  55
  56     def _dict_key_flatten(self, data):
  57         next_data = {}
  58
  59         # use list, because iterable is too generic
  60         if not [v for v in data.values() if
  61                 isinstance(v, (collections.Mapping, list))]:
  62             return data
  63
  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
  72             else:
  73                 next_data[k] = v
  74
  75         return self._dict_key_flatten(next_data)
  76
  77     def _get_nano_timestamp(self, results):
  78         try:
  79             timestamp = results["benchmark"]["timestamp"]
  80         except Exception:
  81             timestamp = time.time()
  82
  83         return str(int(float(timestamp) * 1000000000))
  84
  85     def _get_extended_tags(self, data):
  86         runner_info = self.runners_info[data["runner_id"]]
  87         tags = {
  88             "runner_id": data["runner_id"],
  89             "task_id": self.task_id,
  90             "scenarios": runner_info["scenarios"]
  91         }
  92         if "host" in runner_info:
  93             tags["host"] = runner_info["host"]
  94         if "target" in runner_info:
  95             tags["target"] = runner_info["target"]
  96
  97         return tags
  98
  99     def _data_to_line_protocol(self, data):
 100         msg = {}
 101         point = {
 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),
 106         }
 107         msg["points"] = [point]
 108         msg["tags"] = self.static_tags
 109
 110         return make_lines(msg).encode('utf-8')
 111
 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'
 118                       'be posted.')
 119             return -1
 120
 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"]
 131             return 0
 132
 133         if self.tc == "":
 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.')
 136             return -1
 137
 138         try:
 139             line = self._data_to_line_protocol(data)
 140             LOG.debug('Test result line format : %s', line)
 141             res = requests.post(self.influxdb_url,
 142                                 data=line,
 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)
 148                 LOG.error(res.text)
 149
 150         except Exception as err:
 151             LOG.exception('Failed to record result data: %s',
 152                           err)
 153             return -1
 154         return 0
 155
 156     def flush_result_data(self):
 157         LOG.debug('Test result all : %s',
 158                   jsonutils.dump_as_bytes(self.raw_result))
 159         return 0