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 import os
  15
  16 import requests
  17 from requests import ConnectionError
  18
  19 from yardstick.common import utils
  20 from third_party.influxdb.influxdb_line_protocol import make_lines
  21 from yardstick.dispatcher.base import Base as DispatchBase
  22
  23 LOG = logging.getLogger(__name__)
  24
  25
  26 class InfluxdbDispatcher(DispatchBase):
  27     """Dispatcher class for posting data into an influxdb target.
  28     """
  29
  30     __dispatcher_type__ = "Influxdb"
  31
  32     def __init__(self, conf):
  33         super(InfluxdbDispatcher, self).__init__(conf)
  34         db_conf = conf['dispatcher_influxdb']
  35         self.timeout = int(db_conf.get('timeout', 5))
  36         self.target = db_conf.get('target', 'http://127.0.0.1:8086')
  37         self.db_name = db_conf.get('db_name', 'yardstick')
  38         self.username = db_conf.get('username', 'root')
  39         self.password = db_conf.get('password', 'root')
  40
  41         self.influxdb_url = "%s/write?db=%s" % (self.target, self.db_name)
  42
  43         self.task_id = None
  44         self.tags = None
  45
  46     def flush_result_data(self, data):
  47         LOG.debug('Test result all : %s', data)
  48         if self.target == '':
  49             # if the target was not set, do not do anything
  50             LOG.error('Dispatcher target was not set, no data will be posted.')
  51
  52         result = data['result']
  53         self.tags = result['info']
  54         self.task_id = result['task_id']
  55         self.criteria = result['criteria']
  56         testcases = result['testcases']
  57
  58         for case, data in testcases.items():
  59             tc_criteria = data['criteria']
  60             for record in data['tc_data']:
  61                 # skip results with no data because we influxdb encode empty dicts
  62                 if record.get("data"):
  63                     self.upload_one_record(record, case, tc_criteria)
  64
  65         return 0
  66
  67     def upload_one_record(self, data, case, tc_criteria, task_id=None):
  68         if task_id:
  69             self.task_id = task_id
  70
  71         line = self._data_to_line_protocol(data, case, tc_criteria)
  72         LOG.debug('Test result line format : %s', line)
  73
  74         try:
  75             res = requests.post(self.influxdb_url,
  76                                 data=line,
  77                                 auth=(self.username, self.password),
  78                                 timeout=self.timeout)
  79         except ConnectionError as err:
  80             LOG.exception('Failed to record result data: %s', err)
  81         else:
  82             if res.status_code != 204:
  83                 LOG.error('Test result posting finished with status code'
  84                           ' %d.', res.status_code)
  85                 LOG.error(res.text)
  86
  87     def _data_to_line_protocol(self, data, case, criteria):
  88         msg = {}
  89
  90         if not self.tags:
  91             self.tags = {
  92                 'deploy_scenario': os.environ.get('DEPLOY_SCENARIO', 'unknown'),
  93                 'installer': os.environ.get('INSTALLER_TYPE', 'unknown'),
  94                 'pod_name': os.environ.get('NODE_NAME', 'unknown'),
  95                 'version': os.environ.get('YARDSTICK_BRANCH', 'unknown')
  96             }
  97
  98         point = {
  99             "measurement": case,
 100             "fields": utils.flatten_dict_key(data["data"]),
 101             "time": self._get_nano_timestamp(data),
 102             "tags": self._get_extended_tags(criteria),
 103         }
 104         msg["points"] = [point]
 105         msg["tags"] = self.tags
 106
 107         return make_lines(msg).encode('utf-8')
 108
 109     def _get_nano_timestamp(self, results):
 110         try:
 111             timestamp = results["timestamp"]
 112         except KeyError:
 113             timestamp = time.time()
 114
 115         return str(int(float(timestamp) * 1000000000))
 116
 117     def _get_extended_tags(self, criteria):
 118         tags = {
 119             "task_id": self.task_id,
 120             "criteria": criteria
 121         }
 122
 123         return tags