api/utils/influx.py

   1 ##############################################################################
   2 # Copyright (c) 2016 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 import logging
  10 from urlparse import urlsplit
  11
  12 from influxdb import InfluxDBClient
  13 import ConfigParser
  14
  15 from api import conf
  16
  17 logger = logging.getLogger(__name__)
  18
  19
  20 def get_data_db_client():
  21     parser = ConfigParser.ConfigParser()
  22     try:
  23         parser.read(conf.OUTPUT_CONFIG_FILE_PATH)
  24         dispatcher = parser.get('DEFAULT', 'dispatcher')
  25
  26         if 'influxdb' != dispatcher:
  27             raise RuntimeError
  28
  29         ip = _get_ip(parser.get('dispatcher_influxdb', 'target'))
  30         username = parser.get('dispatcher_influxdb', 'username')
  31         password = parser.get('dispatcher_influxdb', 'password')
  32         db_name = parser.get('dispatcher_influxdb', 'db_name')
  33         return InfluxDBClient(ip, conf.PORT, username, password, db_name)
  34     except ConfigParser.NoOptionError:
  35         logger.error('can not find the key')
  36         raise
  37
  38
  39 def _get_ip(url):
  40     return urlsplit(url).hostname
  41
  42
  43 def _write_data(measurement, field, timestamp, tags):
  44     point = {
  45         'measurement': measurement,
  46         'fields': field,
  47         'time': timestamp,
  48         'tags': tags
  49     }
  50
  51     try:
  52         client = get_data_db_client()
  53
  54         logger.debug('Start to write data: %s', point)
  55         client.write_points([point])
  56     except RuntimeError:
  57         logger.debug('dispatcher is not influxdb')
  58
  59
  60 def write_data_tasklist(task_id, timestamp, status, error=''):
  61     field = {'status': status, 'error': error}
  62     tags = {'task_id': task_id}
  63     _write_data('tasklist', field, timestamp, tags)
  64
  65
  66 def query(query_sql):
  67     try:
  68         client = get_data_db_client()
  69         logger.debug('Start to query: %s', query_sql)
  70         return list(client.query(query_sql).get_points())
  71     except RuntimeError:
  72         logger.error('dispatcher is not influxdb')
  73         raise