fluxios.py

#!/usr/bin/python 
# vim: set ts=4 sw=4 et :

from ConfigParser import SafeConfigParser
from optparse import OptionParser
from StringIO import StringIO
from threading import Thread
from ast import literal_eval
import logging
import logging.handlers
import os
import os.path
import re
import sys
import time
import shlex
import signal
from timeit import default_timer as timer

# ##########################################################
# ###  Do not edit this file, edit fluxios.cfg    #####

shutdown_flag = False # graceful shutdown

# initialize logger, and add console handler while we start up
log = logging.getLogger('log')
console_handler = logging.StreamHandler(sys.stderr)
log.setLevel(logging.INFO)
console_formatter = logging.Formatter('%(message)s')
console_handler.setFormatter(console_formatter)
log.addHandler(console_handler)
log.info("Fluxios starting...")

SIGNALS_TO_NAMES_DICT = dict((getattr(signal, n), n) \
    for n in dir(signal) if n.startswith('SIG') and '_' not in n )


# default configuration
default_cfg = StringIO("""\
[fluxios]
spool_directory = /var/spool/nagios/fluxios
log_file = /var/log/nagios/fluxios.log
log_max_size = 24
log_keep = 4
log_level = logging.INFO
interval = 15
measurement_prefix =
batch_size = 500
extra_tags = 

[influxdb]
host = 127.0.0.1
port = 8086
proxies = None

cluster = False
hosts = 
shuffle = True
healing_delay = 900

ssl = False
verify_ssl = True
timeout = 15
database = nagios
username = fluxios
password = 
use_udp = False
udp_port = 4444
""")

config_file = '/etc/fluxios/fluxios.cfg'

# config dictionary
cfg = {}

# influxdb client
db = 0

# options parsing
parser = OptionParser("usage: %prog [options] sends nagios performance data to InfluxDB.")
parser.add_option("-c", "--config", dest="config_file", default=config_file,
                    help="Set custom config file location.")
parser.add_option("--show-defaults", dest="show_defaults", action="store_true", default=False,
                    help="Print default configuration.")
parser.add_option("-D", "--debug", dest="debug", action="store_true", default=False,
                    help="Set log_level=logging.DEBUG")


def convert_config_value(value):
    """
    param value: string containing either "true" or "false", case insensitive
    return: boolean True or False, or the value if it's neither
    """
    if (value.lower() == "true"):
        return True
    elif (value.lower() == "false"):
        return False
    elif (value.lower() == "none" or value.strip() == ""):
        return None
    try:
        return literal_eval(value)
    except (ValueError, SyntaxError) as e:
        return value

def read_config(config_file, defaults):
    """
    param config_file: full path of file to read
    param defaults: string containing default options
    return: dict of configuration file sections and options
    """
    # initialize
    config = SafeConfigParser()

    # first, read config from defaults
    config.readfp(defaults)
        
    config_dict = {}

    # then read config from file
    if os.path.isfile(config_file):
        config.read(config_file)
    else:
        log.info("Could not read config file, using defaults: {0}".format(config_file))

    for section in config.sections():
        config_dict[section] = {}
        for name, value in config.items(section):
            config_dict[section][name] = convert_config_value(value)
            log.debug("Parsed config option [{0}] {1} = {2}"
                .format(section, name, config_dict[section][name]))

    return config_dict


def config_logger():
    """
    sets up fluxios config
    """

    try:
        cfg['fluxios']['log_max_size'] = int(cfg['fluxios']['log_max_size'])
    except ValueError:
        print "log_max_size needs to be an integer"
        sys.exit(1)
    try:
        cfg['fluxios']['log_keep'] = int(cfg['fluxios']['log_keep'])
    except ValueError:
        print "log_keep needs to be an integer"
        sys.exit(1)

    log_max_bytes = cfg['fluxios']['log_max_size']*1024*1024

    try:
        file_handler = logging.handlers.RotatingFileHandler( \
            cfg['fluxios']['log_file'], \
            maxBytes=log_max_bytes, \
            backupCount=int(cfg['fluxios']['log_keep']), \
        )
    except IOError as e:
        print "IOError while configuring RotatingFileHandler: {0}".format(e)
        sys.exit(1)

    formatter = logging.Formatter(
        "%(asctime)s %(filename)s[%(process)d] %(levelname)s (%(funcName)s) %(message)s",
        "%B %d %H:%M:%S")
    file_handler.setFormatter(formatter)
    log.addHandler(file_handler)
    log.info("Added file log ({0}), removing console log handler" \
        .format(cfg['fluxios']['log_file']))
    log.removeHandler(console_handler)
    log.setLevel(getattr(logging, str(cfg['fluxios']['log_level']), logging.INFO))

def init_influxdb_client():
    global db

    if not cfg['influxdb']['cluster']:
        try:
            from influxdb import InfluxDBClient
        except ImportError as e:
            log.exception("Could not import influxdb module: {0}".format(e))
            sys.exit(1) 
        db = InfluxDBClient(
            host=cfg['influxdb']['host'],
            port=cfg['influxdb']['port'],
            username=cfg['influxdb']['username'],
            password=cfg['influxdb']['password'],
            database=cfg['influxdb']['database'],
            ssl=cfg['influxdb']['ssl'],
            verify_ssl=cfg['influxdb']['verify_ssl'],
            timeout=float(cfg['influxdb']['timeout']),
            use_udp=cfg['influxdb']['use_udp'],
            udp_port=cfg['influxdb']['udp_port'],
            proxies=cfg['influxdb']['proxies']
        )
    else:
        try:
            from influxdb import InfluxDBClusterClient
        except ImportError as e:
            log.exception("Could not import influxdb module: {0}".format(e))
        db = InfluxDBClusterClient(
            hosts=cfg['influxdb']['hosts'],
            username=cfg['influxdb']['username'],
            password=cfg['influxdb']['password'],
            database=cfg['influxdb']['database'],
            ssl=cfg['influxdb']['ssl'],
            verify_ssl=cfg['influxdb']['verify_ssl'],
            timeout=float(cfg['influxdb']['timeout']),
            use_udp=cfg['influxdb']['use_udp'],
            udp_port=cfg['influxdb']['udp_port'],
            shuffle=cfg['influxdb']['shuffle'],
            healing_delay=cfg['influxdb']['healing_delay']
        )
   
 
def process_perfdata_file(file_name):
    '''
    param file: full path perfdata file to extract points from
    return: list of influxdb points
    '''

    processed_lines = 0  # number of perfdata lines processed
    skipped_lines = 0 # number of lines skipped because we couldn't massage them
    points = []
    
    perfdata_re = \
        "^'?([^=']+)'?=(U|[\d\.\-]+)([\w\/%]*);?([\d\.\-:~@]+)?;?([\d\.\-:~@]+)?;?([\d\.\-]+)?;?([\d\.\-]+)?;?\s*"

    start = timer()
    try:
        file = open(file_name, "r")
        file_array = file.readlines()
        file.close()
    except (IOError, OSError) as ex:
        log.critical("Can't open file: {0} error: {1}".format(file, ex))
        return False
    # parse each line
    for line in file_array:
        processed_lines += 1
        try:
            line_dict = dict(re.split('::', x, 1) for x in line.split('\t'))
        except Exception as e:
            skipped_lines += 1
            log.warn("{0}: Could not parse perfdata line into key::value pairs in file {1}, skipping: {2}"
                .format(e, file_name, line))
            continue

        # pick out values from the line
        if line_dict['DATATYPE'] == "SERVICEPERFDATA":
            service_description = line_dict['SERVICEDESC']
            perfdata = line_dict['SERVICEPERFDATA']
            if not perfdata:
                skipped_lines += 1
                log.debug("perfdata string is empty while reading line file {0}: {1}"
                    .format(file_name, line))
                continue
            check_command = line_dict['SERVICECHECKCOMMAND'].split('!')[0]
        elif line_dict['DATATYPE'] == "HOSTPERFDATA":
            service_description = "__host__"
            perfdata = line_dict['HOSTPERFDATA']
            check_command = line_dict['HOSTCHECKCOMMAND'].split('!')[0]
        else:
            skipped_lines += 1
            log.warn("Unknown DATATYPE, skipping: '{0}'"
                .format(line_dict['DATATYPE']))
            continue
        
        host_name = line_dict['HOSTNAME']
        timestamp = line_dict['TIMET']

        # extract individual metrics from the perfdata string
        for metric in re.findall("(.*?=.+?)\s", perfdata + ' '):
            m = re.search(perfdata_re, metric)
            if m:
                (label, value, uom, warn, crit, min, max) = m.groups()
                numeric_fields = {
                    "value": value,
                    "warn": warn,
                    "crit": crit,
                    "min": min,
                    "max": max
                }
                fields = {"label": label, "uom": uom}
                for field, value in numeric_fields.items():
                    if value is not None and value.strip():
                        value = re.sub('[^0-9.]','', value)
                        try:
                            fields[field] = float(value)
                        except ValueError:
                            log.error("Failed to float() '{0}' = '{1}' for '{2}'@'{3}'"
                                    .format(field, value, service_description, host_name))

                tags = {
                    "service_description": service_description,
                    "host_name": host_name,
                    "metric": label
                }
                if isinstance(cfg['fluxios']['extra_tags'], dict):
                    tags.update(cfg['fluxios']['extra_tags'])

                point = {
                    "measurement": check_command,
                    "timestamp": timestamp,
                    "fields": fields,
                    "tags": tags
                }
                log.debug("Processed perfdata into point: {0}".format(point))
                points.append(point)
            else:
                log.debug("perfdata metric from file {0} did not match, skipping: {1}"
                    .format(file, metric))
        # END: for
    end = timer()
    log.info("Processed {0}/{1} lines into {2} points in {3:.2f} seconds "
        "({4:.2f} lines/sec, {5:.2f} pts/sec)"
        .format(processed_lines,processed_lines+skipped_lines, len(points),
            round(end-start, 2), round(processed_lines/(end-start), 2), 
            round(len(points)/(end-start), 2)))
    return points


def rm_file(file):
    """
    param file: File to be deleted
    """
    try:
        os.remove(file)
        return True
    except (OSError, IOError) as e:
        log.critical("Could not remove file {0} error: {1}".format(file, e))
        return False


def process_spool_dir(dir):
    """
    param dir: Directory containing perfdata files to be processed
    """
    log.info("Processing spool dir {0}".format(dir))
    num_files = 0

    try:
        files = os.listdir(dir)
    except (IOError, OSError) as e:
        log.error("Exception reading spool dir({0}): {1}".format(dir, e))
        return False
    for file in files:
        if not shutdown_flag:
            file_path = dir + '/' + file
            log.debug("Processing file: {0}".format(file_path))
            if check_skip_file(file_path):
                log.info("Skipping file: {0}".format(file))
                continue

            num_files += 1
            points = process_perfdata_file(file_path)
            if send_points(points):
                log.info(("Successfully wrote {0} points to InfluxDB")
                    .format(len(points)))
            else:
                log.error(("Losing {0} points due to error")
                    .format(len(points)))

            if rm_file(file_path):
                log.debug(("Deleted file: {0}").format(file_path))
            else:
                log.debug(("Could not delete file: {0}").format(file_path))
        else:
            log.info("fluxios shutting down")
            print "fluxios shutting down"
            sys.exit(0) 
    log.info("Processed {0} files".format(num_files))

def send_points(points):
    """
    param points: list of points to send to influxdb
    return: True on success, False otherwise
    """
    start = timer()
    try:
        db.write_points(
            points, 
            time_precision='s', 
            tags=cfg['fluxios']['extra_tags'],
            batch_size=int(cfg['fluxios']['batch_size'])
        )
    except Exception as e:
        log.error(("Exception while trying to write points: {0}")
            .format(e))
        return False
    end = timer()
    log.info("Wrote {0} points in {1:.2f} seconds ({2:.2f} pts/s)"
        .format(len(points), round(end - start, 2),
            round(len(points)/(end - start), 2)))

    return True


def check_skip_file(file):
    """
    param file: Full path to file to check
    return: True if the file should be skipped, False otherwise
    """
    if (
        file == "host-perfdata" or
        file == "service-perfdata"
    ):
        return True
    elif re.match('^_', file):
        return True

    if os.stat(file)[6] == 0:
        log.info("Found empty file, deleting it: {0}".format(file))
        rm_file(file)
        return True

    if os.path.isdir(file):
        return True

    return False

def sighandler(signum, frame):
    log.info('Received {0}, going to shutdown...'
        .format(SIGNALS_TO_NAMES_DICT[signum]))
    print "Received {0}, going to shutdown...".format(SIGNALS_TO_NAMES_DICT[signum])
    global shutdown_flag
    shutdown_flag = True
    sys.exit()


def loop():
    log.info("Starting main loop")
    # loop as long as we are not told to shut down
    while not shutdown_flag:
        try:
            process_spool_dir(cfg['fluxios']['spool_directory'])
            log.debug("sleeping for {0} seconds".format(cfg['fluxios']['interval']))
            time.sleep(float(cfg['fluxios']['interval']))
        except Exception as e:
            log.exception("Caught exception in loop()")
            time.sleep(float(cfg['fluxios']['interval']))

    log.info("fluxios shutting down")
    print "fluxios shutting down"
    sys.exit(0)


if __name__ == '__main__':
    (options, args) = parser.parse_args()
    if options.show_defaults:
        print "config_file = {0}\n".format(config_file)
        print default_cfg.getvalue()
        sys.exit(0)
    cfg = read_config(options.config_file, default_cfg)
 
    config_logger()
    if options.debug:
        log.info("Overriding log_level to logging.DEBUG")
        log.setLevel(logging.DEBUG)
    
    log.info("Configuration: {0}".format(cfg))
    
    # set up sighandler()
    signal.signal(signal.SIGTERM, sighandler)
    signal.signal(signal.SIGINT, sighandler)
    signal.signal(signal.SIGHUP, sighandler)

    init_influxdb_client()

    try:
        Thread(target=loop, args=()).start()
    except Exception as e:
        log.exception("Exception while trying to start thread")
        sys.exit(1)
   
    while True:
        time.sleep(1) 

    sys.exit(0)