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 import os
  11 import json
  12 import logging
  13 import requests
  14 import time
  15
  16 from oslo_config import cfg
  17
  18 from yardstick.dispatcher.base import Base as DispatchBase
  19 from yardstick.dispatcher.influxdb_line_protocol import make_lines
  20
  21 LOG = logging.getLogger(__name__)
  22
  23 CONF = cfg.CONF
  24 influx_dispatcher_opts = [
  25     cfg.StrOpt('target',
  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'),
  30     cfg.StrOpt('db_name',
  31                default='yardstick',
  32                help='The database name to store test results.'),
  33     cfg.IntOpt('timeout',
  34                default=5,
  35                help='The max time in seconds to wait for a request to '
  36                     'timeout.'),
  37 ]
  38
  39 CONF.register_opts(influx_dispatcher_opts, group="dispatcher_influxdb")
  40
  41
  42 class InfluxdbDispatcher(DispatchBase):
  43     """Dispatcher class for posting data into an influxdb target.
  44     """
  45
  46     __dispatcher_type__ = "Influxdb"
  47
  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)
  54         self.raw_result = []
  55         self.case_name = ""
  56         self.tc = ""
  57         self.task_id = -1
  58         self.runners_info = {}
  59         self.static_tags = {
  60             "pod_name": os.environ.get('POD_NAME', 'unknown'),
  61             "installer": os.environ.get('INSTALLER_TYPE', 'unknown'),
  62             "version": os.environ.get('YARDSTICK_VERSION', 'unknown')
  63         }
  64
  65     def _dict_key_flatten(self, data):
  66         next_data = {}
  67
  68         if not [v for v in data.values()
  69                 if type(v) == dict or type(v) == list]:
  70             return data
  71
  72         for k, v in data.iteritems():
  73             if type(v) == dict:
  74                 for n_k, n_v in v.iteritems():
  75                     next_data["%s.%s" % (k, n_k)] = n_v
  76             elif type(v) == list:
  77                 for index, item in enumerate(v):
  78                     next_data["%s%d" % (k, index)] = item
  79             else:
  80                 next_data[k] = v
  81
  82         return self._dict_key_flatten(next_data)
  83
  84     def _get_nano_timestamp(self, results):
  85         try:
  86             timestamp = results["benchmark"]["timestamp"]
  87         except Exception:
  88             timestamp = time.time()
  89
  90         return str(int(float(timestamp) * 1000000000))
  91
  92     def _get_extended_tags(self, data):
  93         runner_info = self.runners_info[data["runner_id"]]
  94         tags = {
  95             "runner_id": data["runner_id"],
  96             "task_id": self.task_id,
  97             "scenarios": runner_info["scenarios"]
  98         }
  99         if "host" in runner_info:
 100             tags["host"] = runner_info["host"]
 101         if "target" in runner_info:
 102             tags["target"] = runner_info["target"]
 103
 104         return tags
 105
 106     def _data_to_line_protocol(self, data):
 107         msg = {}
 108         point = {}
 109         point["measurement"] = self.tc
 110         point["fields"] = self._dict_key_flatten(data["benchmark"]["data"])
 111         point["time"] = self._get_nano_timestamp(data)
 112         point["tags"] = self._get_extended_tags(data)
 113         msg["points"] = [point]
 114         msg["tags"] = self.static_tags
 115
 116         return make_lines(msg).encode('utf-8')
 117
 118     def record_result_data(self, data):
 119         LOG.debug('Test result : %s' % json.dumps(data))
 120         self.raw_result.append(data)
 121         if self.target == '':
 122             # if the target was not set, do not do anything
 123             LOG.error('Dispatcher target was not set, no data will'
 124                       'be posted.')
 125             return -1
 126
 127         if isinstance(data, dict) and "scenario_cfg" in data:
 128             self.tc = data["scenario_cfg"]["tc"]
 129             self.task_id = data["scenario_cfg"]["task_id"]
 130             scenario_cfg = data["scenario_cfg"]
 131             runner_id = data["runner_id"]
 132             self.runners_info[runner_id] = {"scenarios": scenario_cfg["type"]}
 133             if "host" in scenario_cfg:
 134                 self.runners_info[runner_id]["host"] = scenario_cfg["host"]
 135             if "target" in scenario_cfg:
 136                 self.runners_info[runner_id]["target"] = scenario_cfg["target"]
 137             return 0
 138
 139         if self.tc == "":
 140             LOG.error('Test result : %s' % json.dumps(data))
 141             LOG.error('The case_name cannot be found, no data will be posted.')
 142             return -1
 143
 144         try:
 145             line = self._data_to_line_protocol(data)
 146             LOG.debug('Test result line format : %s' % line)
 147             res = requests.post(self.influxdb_url,
 148                                 data=line,
 149                                 timeout=self.timeout)
 150             if res.status_code != 204:
 151                 LOG.error('Test result posting finished with status code'
 152                           ' %d.' % res.status_code)
 153         except Exception as err:
 154             LOG.exception('Failed to record result data: %s',
 155                           err)
 156             return -1
 157         return 0
 158
 159     def flush_result_data(self):
 160         LOG.debug('Test result all : %s' % json.dumps(self.raw_result))
 161         return 0