haproxy.py

# haproxy-collectd-plugin - haproxy.py
#
# Author: Michael Leinartas
# Description: This is a collectd plugin which runs under the Python plugin to
# collect metrics from haproxy.
# Plugin structure and logging func taken from
# https://github.com/phrawzty/rabbitmq-collectd-plugin
#
# Modified by "Warren Turkal" <wt@signalfuse.com>, "Volodymyr Zhabiuk" <vzhabiuk@signalfx.com>

import cStringIO as StringIO
import socket
import csv
import pprint

import collectd

PLUGIN_NAME = 'haproxy'
RECV_SIZE = 1024

DEFAULT_METRICS = {
    'ConnRate': ('connection_rate', 'gauge'),
    'CumReq': ('requests', 'derive'),
    'Idle_pct': ('idle_pct', 'gauge'),
    'scur': ('session_current', 'gauge'),
    'SessRate': ('session_rate_all', 'gauge'),
    'lbtot': ('server_selected_total', 'counter'),
    'bout': ('bytes_out', 'derive'),
    'bin': ('bytes_in', 'derive'),
    'ttime': ('session_time_avg', 'gauge'),
    'req_rate': ('request_rate', 'gauge'),
    'rate': ('session_rate', 'gauge'),
    'hrsp_2xx': ('response_2xx', 'derive'),
    'hrsp_4xx': ('response_4xx', 'derive'),
    'hrsp_5xx': ('response_5xx', 'derive'),
    'ereq': ('error_request', 'derive'),
    'dreq': ('denied_request', 'derive'),
    'econ': ('error_connection', 'derive'),
    'dresp': ('denied_response', 'derive'),
    'qcur': ('queue_current', 'gauge'),
    'qtime': ('queue_time_avg', 'gauge'),
    'rtime': ('response_time_avg', 'gauge'),
    'eresp': ('error_response', 'derive'),
    'wretr': ('retries', 'derive'),
    'wredis': ('redispatched', 'derive'),
}

ENHANCED_METRICS = {
    # Metrics that are collected for the whole haproxy instance.
    # The format is  haproxy_metricname : {'signalfx_corresponding_metric': 'collectd_type'}
    # Currently signalfx_corresponding_metric match haproxy_metricname
    # Correspond to 'show info' socket command
    'MaxConn': ('max_connections', 'gauge'),
    'CumConns': ('connections', 'derive'),
    'MaxConnRate': ('max_connection_rate', 'gauge'),
    'MaxSessRate': ('max_session_rate', 'gauge'),
    'MaxSslConns': ('max_ssl_connections', 'gauge'),
    'CumSslConns': ('ssl_connections', 'derive'),
    'MaxPipes': ('max_pipes', 'gauge'),
    'Tasks': ('tasks', 'gauge'),
    'Run_queue': ('run_queue', 'gauge'),
    'PipesUsed': ('pipes_used', 'gauge'),
    'PipesFree': ('pipes_free', 'gauge'),
    'Uptime_sec': ('uptime_seconds', 'derive'),
    'CurrConns': ('current_connections', 'gauge'),
    'CurrSslConns': ('current_ssl_connections', 'gauge'),
    'SslRate': ('ssl_rate', 'gauge'),
    'SslFrontendKeyRate': ('ssl_frontend_key_rate', 'gauge'),
    'SslBackendKeyRate': ('ssl_backend_key_rate', 'gauge'),
    'SslCacheLookups': ('ssl_cache_lookups', 'derive'),
    'SslCacheMisses': ('ssl_cache_misses', 'derive'),
    'CompressBpsIn': ('compress_bps_in', 'derive'),
    'CompressBpsOut': ('compress_bps_out', 'derive'),
    'ZlibMemUsage': ('zlib_mem_usage', 'gauge'),

    # Metrics that are collected per each proxy separately.
    # Proxy name would be the dimension as well as service_name
    # Correspond to 'show stats' socket command
    'chkfail': ('failed_checks', 'derive'),
    'downtime': ('downtime', 'derive'),
    'hrsp_1xx': ('response_1xx', 'derive'),
    'hrsp_3xx': ('response_3xx', 'derive'),
    'hrsp_other': ('response_other', 'derive'),
    'qmax': ('queue_max', 'gauge'),
    'qlimit': ('queue_limit', 'gauge'),
    'rate_lim': ('session_rate_limit', 'gauge'),
    'rate_max': ('session_rate_max', 'gauge'),
    'req_rate_max': ('request_rate_max', 'gauge'),
    'stot': ('session_total', 'derive'),
    'slim': ('session_limit', 'gauge'),
    'smax': ('session_max', 'gauge'),
    'throttle': ('throttle', 'gauge'),
    'cli_abrt': ('cli_abrt', 'derive'),
    'srv_abrt': ('srv_abrt', 'derive'),
    'comp_in': ('comp_in', 'derive'),
    'comp_out': ('comp_out', 'derive'),
    'comp_byp': ('comp_byp', 'derive'),
    'comp_rsp': ('comp_rsp', 'derive'),
    'ctime': ('connect_time_avg', 'gauge'),
    'act': ('active_servers', 'gauge'),
    'bck': ('backup_servers', 'gauge'),
    'check_duration': ('health_check_duration', 'gauge'),
    'lastsess': ('last_session', 'gauge'),
    'conn_rate': ('conn_rate', 'gauge'),
    'conn_rate_max': ('conn_rate_max', 'gauge'),
    'conn_tot': ('conn_total', 'counter'),
    'intercepted': ('intercepted', 'gauge'),
    'dcon': ('denied_tcp_conn', 'gauge'),
    'dses': ('denied_tcp_sess', 'gauge'),
}

DIMENSIONS_LIST = [
    'pxname',
    'svname',
    'pid',
    'sid',
    'iid',
    'type',
    'addr',
    'cookie',
    'mode',
    'algo',
]

DEFAULT_METRICS = dict((k.lower(), v) for k, v in DEFAULT_METRICS.items())
ENHANCED_METRICS = dict((k.lower(), v) for k, v in ENHANCED_METRICS.items())
METRIC_DELIM = '.'  # for the frontend/backend stats

DEFAULT_SOCKET = '/var/run/haproxy.sock'
DEFAULT_PROXY_MONITORS = ['server', 'frontend', 'backend']


class HAProxySocket(object):
    """
            Encapsulates communication with HAProxy via the socket interface
    """

    def __init__(self, socket_file=DEFAULT_SOCKET):
        self.socket_file = socket_file

    def connect(self):
        # unix sockets all start with '/', use tcp otherwise
        is_unix = self.socket_file.startswith('/')
        if is_unix:
            stat_sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
            stat_sock.connect(self.socket_file)
            return stat_sock
        else:
            socket_host, separator, port = self.socket_file.rpartition(':')
            if socket_host is not '' and port is not '' and separator is ':':
                stat_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                stat_sock.connect((socket_host, int(port)))
                return stat_sock
            else:
                collectd.error('Could not connect to socket with host %s. Check HAProxy config.' % self.socket_file)
                return

    def communicate(self, command):
        '''Get response from single command.

        Args:
            command: string command to send to haproxy stat socket

        Returns:
            a string of the response data
        '''
        if not command.endswith('\n'):
            command += '\n'
        stat_sock = self.connect()
        if stat_sock is None:
            return ''
        stat_sock.sendall(command)
        result_buf = StringIO.StringIO()
        buf = stat_sock.recv(RECV_SIZE)
        while buf:
            result_buf.write(buf)
            buf = stat_sock.recv(RECV_SIZE)

        stat_sock.close()
        return result_buf.getvalue()

    def get_server_info(self):
        result = {}
        output = self.communicate('show info')
        for line in output.splitlines():
            try:
                key, val = line.split(':', 1)
            except ValueError:
                continue
            result[key.strip()] = val.strip()

        return result

    def get_server_stats(self):
        output = self.communicate('show stat')
        # sanitize and make a list of lines
        output = output.lstrip('# ').strip()
        output = [l.strip(',') for l in output.splitlines()]
        csvreader = csv.DictReader(output)
        result = [d.copy() for d in csvreader]
        return result


def get_stats(module_config):
    """
        Makes two calls to haproxy to fetch server info and server stats.
        Returns the dict containing metric name as the key and a tuple of metric value and the dict of dimensions if any
    """
    if module_config['socket'] is None:
        collectd.error("Socket configuration parameter is undefined. Couldn't get the stats")
        return
    stats = []
    haproxy = HAProxySocket(module_config['socket'])

    try:
        server_info = haproxy.get_server_info()
        server_stats = haproxy.get_server_stats()
    except socket.error:
        collectd.warning('status err Unable to connect to HAProxy socket at %s' % module_config['socket'])
        return stats

    # server wide stats
    for key, val in server_info.iteritems():
        try:
            stats.append((key, int(val), dict()))
        except (TypeError, ValueError):
            pass

    # proxy specific stats
    for statdict in server_stats:
        dimensions = _build_dimension_dict(statdict)
        if not (statdict['svname'].lower() in module_config['proxy_monitors'] or
                statdict['pxname'].lower() in module_config['proxy_monitors']):
            continue
        for metricname, val in statdict.items():
            try:
                stats.append((metricname, int(val), dimensions))
            except (TypeError, ValueError):
                pass

    return stats


def _build_dimension_dict(statdict):
    """
    Builds dimensions dict to send back with metrics with readable metric names
    Args:
    statdict dictionary of metrics from HAProxy to be filtered for dimensions
    """

    dimensions = {}

    for key in DIMENSIONS_LIST:
        if key in statdict and key == 'pxname':
            dimensions['proxy_name'] = statdict['pxname']
        elif key in statdict and key == 'svname':
            dimensions['service_name'] = statdict['svname']
        elif key in statdict and key == 'pid':
            dimensions['process_id'] = statdict['pid']
        elif key in statdict and key == 'sid':
            dimensions['server_id'] = statdict['sid']
        elif key in statdict and key == 'iid':
            dimensions['unique_proxy_id'] = statdict['iid']
        elif key in statdict and key == 'type':
            dimensions['type'] = _get_proxy_type(statdict['type'])
        elif key in statdict and key == 'addr':
            dimensions['address'] = statdict['addr']
        elif key in statdict and key == 'algo':
            dimensions['algorithm'] = statdict['algo']
        elif key in statdict:
            dimensions[key] = statdict[key]

    return dimensions


def config(config_values):
    """
    A callback method that  loads information from the HaProxy collectd plugin config file.
    Args:
    config_values (collectd.Config): Object containing config values
    """

    module_config = {}
    socket = DEFAULT_SOCKET
    proxy_monitors = []
    excluded_metrics = set()
    enhanced_metrics = False
    interval = None
    testing = False
    custom_dimensions = {}

    for node in config_values.children:
        if node.key == "ProxyMonitor" and node.values[0]:
            proxy_monitors.append(node.values[0])
        elif node.key == "Socket" and node.values[0]:
            socket = node.values[0]
        elif node.key == "Interval" and node.values[0]:
            interval = node.values[0]
        elif node.key == "EnhancedMetrics" and node.values[0]:
            enhanced_metrics = _str_to_bool(node.values[0])
        elif node.key == "ExcludeMetric" and node.values[0]:
            excluded_metrics.add(node.values[0])
        elif node.key == "Testing" and node.values[0]:
            testing = _str_to_bool(node.values[0])
        elif node.key == 'Dimension':
            if len(node.values) == 2:
                custom_dimensions.update({node.values[0]: node.values[1]})
            else:
                collectd.warning("WARNING: Check configuration \
                                            setting for %s" % node.key)
        else:
            collectd.warning('Unknown config key: %s' % node.key)

    if not proxy_monitors:
        proxy_monitors += DEFAULT_PROXY_MONITORS

    module_config = {
        'socket': socket,
        'proxy_monitors': proxy_monitors,
        'interval': interval,
        'enhanced_metrics': enhanced_metrics,
        'excluded_metrics': excluded_metrics,
        'custom_dimensions': custom_dimensions,
        'testing': testing,
    }
    proxys = "_".join(proxy_monitors)

    if testing:
        return module_config

    interval_kwarg = {}
    if interval:
        interval_kwarg['interval'] = interval
    collectd.register_read(collect_metrics, data=module_config,
                           name='node_' + module_config['socket'] + '_' + proxys,
                           **interval_kwarg)


def _format_dimensions(dimensions):
    """
    Formats a dictionary of dimensions to a format that enables them to be
    specified as key, value pairs in plugin_instance to signalfx. E.g.
    >>> dimensions = {'a': 'foo', 'b': 'bar'}
    >>> _format_dimensions(dimensions)
    "[a=foo,b=bar]"
    Args:
    dimensions (dict): Mapping of {dimension_name: value, ...}
    Returns:
    str: Comma-separated list of dimensions
    """

    dim_pairs = ["%s=%s" % (k, v) for k, v in dimensions.iteritems()]
    return "[%s]" % (",".join(dim_pairs))


def _get_proxy_type(type_id):
    """
        Return human readable proxy type
        Args:
        type_id: 0=frontend, 1=backend, 2=server, 3=socket/listener
    """
    proxy_types = {
        0: 'frontend',
        1: 'backend',
        2: 'server',
        3: 'socket/listener',
    }
    return proxy_types.get(int(type_id))


def _str_to_bool(val):
    '''
    Converts a true/false string to a boolean
    '''
    val = str(val).strip().lower()
    if val == 'true':
        return True
    elif val != 'false':
        collectd.warning('Warning: String (%s) could not be converted to a boolean. Returning false.' % val)

    return False


def collect_metrics(module_config):
    collectd.debug('beginning collect_metrics')
    """
        A callback method that gets metrics from HAProxy and records them to collectd.
    """

    info = get_stats(module_config)

    if not info:
        collectd.warning('%s: No data received' % PLUGIN_NAME)
        return

    for metric_name, metric_value, dimensions in info:
        # assert metric is in valid metrics lists
        if not metric_name.lower() in DEFAULT_METRICS and not metric_name.lower() in ENHANCED_METRICS:
            collectd.debug("metric %s is not in either metric list" % metric_name.lower())
            continue

        # skip metrics in enhanced metrics mode if not enabled
        if not module_config['enhanced_metrics'] and metric_name.lower() in ENHANCED_METRICS:
            continue

        # pull metric name & type from respective metrics list
        if metric_name.lower() in DEFAULT_METRICS:
            translated_metric_name, val_type = DEFAULT_METRICS[metric_name.lower()]
        else:
            translated_metric_name, val_type = ENHANCED_METRICS[metric_name.lower()]

        # skip over any exlcluded metrics
        if translated_metric_name in module_config['excluded_metrics']:
            collectd.debug("excluding metric %s" % translated_metric_name)
            continue

        # create datapoint and dispatch
        datapoint = collectd.Values()
        datapoint.type = val_type
        datapoint.type_instance = translated_metric_name
        datapoint.plugin = PLUGIN_NAME
        dimensions.update(module_config['custom_dimensions'])
        if len(dimensions) > 0:
            datapoint.plugin_instance = _format_dimensions(dimensions)
        datapoint.values = (metric_value,)
        pprint_dict = {
                    'plugin': datapoint.plugin,
                    'plugin_instance': datapoint.plugin_instance,
                    'type': datapoint.type,
                    'type_instance': datapoint.type_instance,
                    'values': datapoint.values
                }
        collectd.debug(pprint.pformat(pprint_dict))
        datapoint.dispatch()

collectd.register_config(config)