api/utils/influx.py

   1 import logging
   2 from urlparse import urlsplit
   3
   4 from influxdb import InfluxDBClient
   5 import ConfigParser
   6
   7 from api import conf
   8
   9 logger = logging.getLogger(__name__)
  10
  11
  12 def get_data_db_client():
  13     parser = ConfigParser.ConfigParser()
  14     try:
  15         parser.read(conf.OUTPUT_CONFIG_FILE_PATH)
  16         dispatcher = parser.get('DEFAULT', 'dispatcher')
  17
  18         if 'influxdb' != dispatcher:
  19             raise RuntimeError
  20
  21         ip = _get_ip(parser.get('dispatcher_influxdb', 'target'))
  22         username = parser.get('dispatcher_influxdb', 'username')
  23         password = parser.get('dispatcher_influxdb', 'password')
  24         db_name = parser.get('dispatcher_influxdb', 'db_name')
  25         return InfluxDBClient(ip, conf.PORT, username, password, db_name)
  26     except ConfigParser.NoOptionError:
  27         logger.error('can not find the key')
  28         raise
  29
  30
  31 def _get_ip(url):
  32     return urlsplit(url).hostname
  33
  34
  35 def _write_data(measurement, field, timestamp, tags):
  36     point = {
  37         'measurement': measurement,
  38         'fields': field,
  39         'time': timestamp,
  40         'tags': tags
  41     }
  42
  43     try:
  44         client = get_data_db_client()
  45
  46         logger.debug('Start to write data: %s', point)
  47         client.write_points([point])
  48     except RuntimeError:
  49         logger.debug('dispatcher is not influxdb')
  50
  51
  52 def write_data_tasklist(task_id, timestamp, status, error=''):
  53     field = {'status': status, 'error': error}
  54     tags = {'task_id': task_id}
  55     _write_data('tasklist', field, timestamp, tags)