SignalWaiter

Complexity

Total Complexity

6

Size/Duplication

Total Lines	50
Duplicated Lines	0 %

Importance

Changes	1
Bugs	0	Features	0

2 Methods

Rating	Name	Duplication	Size	Complexity
A	wait()	0	9	2
B	__init__()	0	22	4

# -*- coding: utf-8 -*-
"""Various helper classes.

.. moduleauthor:: Kenny Freeman <kenny.freeman@gmail.com>

"""
import imp
import sys
import os
import os.path
import time
import fcntl
import socket
import signal
import string
import inspect
import platform
import traceback
from collections import deque

import plumd

__author__ = 'Kenny Freeman'
__email__ = 'kenny.freeman@gmail.com'
__license__ = "ISCL"
__docformat__ = 'reStructuredText'

PY3 = sys.version_info > (3,)


# use CNT_LIMIT to detect counters wrapping over
# eg. ('64bit', 'ELF')
arch = platform.architecture()[0]
if arch == '64bit':
    MAX_CNT = 18446744073709551615
elif arch == '32bit':
    MAX_CNT = 4294967295
else:
    MAX_CNT = long(sys.maxint)

Undefined variable 'long'

The Module sys does not seem to have a member named maxint.

def get_hostname():
    """Return the hostname of the system the code is running on. Eventually this
    may include more checks.

    :rtype: str
    """
    return socket.gethostname().split(".")[0]


def get_file_map(fname, col, skip):
    """Return a dict representation of a file - use on small files only.

    eg. file contents of:

    8       1 sda1 284 166 2673 1724 58 18 9056 369 0 1122 2091

    with a call of col=2, skip=0 returns:

    {"sda1": [8, 1, 284, 166, 2673, 1724, 58, 18, 9056, 369, 0, 1122, 2091]}

    :param fname: Full pathname to a file.
    :type fname: str
    :param col: Column number to use as the map
    :type col: int
    :param skip: Skip this many lines of the file.
    :type skip: int
    :rtype: dict
    """
    fdat = {}
    if os.access(fname, os.R_OK):
        with open(fname, 'r') as filed:
            dat = filed.read().split("\n")
            for line in dat[skip:]:
                vals = line.split()
                if len(vals) > col:
                    map_val = vals.pop(col)
                    fdat[map_val] = deque(vals)
    return fdat


def get_file_map_list(fname, col, skip):
    """Return a dict representation of a file - use on small files only.

    eg. file contents of:

    8       1 sda1 284 166 2673 1724 58 18 9056 369 0 1122 2091

    with a call of col=2, skip=0 returns:

    {"sda1": [8, 1, 284, 166, 2673, 1724, 58, 18, 9056, 369, 0, 1122, 2091]}

    :param fname: Full pathname to a file.
    :type fname: str
    :param col: Column number to use as the map
    :type col: int
    :param skip: Skip this many lines of the file.
    :type skip: int
    :rtype: dict
    """
    fdat = {}
    if os.access(fname, os.R_OK):
        with open(fname, 'r') as filed:
            dat = filed.read().split("\n")
            for line in dat[skip:]:
                vals = line.split()
                if len(vals) > col:
                    map_val = vals.pop(col)
                    fdat[map_val] = vals
    return fdat


def get_file_list(fname):
    """Return the file as a list - use on small files only.

    :param fname: Full pathname to a file.
    :type fname: str
    :rtype: dict
    """
    if os.access(fname, os.R_OK):
        with open(fname, 'r') as filed:
            return deque(filed.read().strip().split("\n"))
    return deque([''])


def get_file(fname):
    """Return the file as a string - use on small files only.

    :param fname: Full pathname to a file.
    :type fname: str
    :rtype: dict
    """
    if os.access(fname, os.R_OK):
        with open(fname, 'r') as filed:
            return filed.read()
    return ''


class Interval(object):
    """A simple helper class to ensure a loop occours every ltime seconds
    while optionally providing a means to interupt the loop with a
    :class:`threading.Event`.


    :param ltime: The target duration of a loop
    :type ltime: int or float
    :param evt: An event object that will cause the loop to complete early if
        the event is triggered while waiting.
    :type evt: threading.Event
    :raises: ValueError if the passed loop time is not an int
    :raises: ValueError if the passed loop time is not a float
    :raises: ValueError if the passed loop time is < 0

    As an example, use it like this to loop on an interval:

    :Example:

        >>> import threading
        >>> evt = threading.Event()
        >>> while loop:
        >>>     with Interval(111.1, evt) as timer:
        >>>         do_things_here()

    .. note:: if the do_things_here() call returns before the loop timer
        completes the timer object will either evt.wait() if evt was defined or
        time.sleep() for the time remaining in the interval.
    """

    def __init__(self, ltime, evt=None):
        """A simple helper class to ensure a loop occours every ltime seconds
        while optionally providing a means to interupt the loop with a
        :class:`threading.Event`.


        :param ltime: The target duration of a loop
        :type ltime: int or float
        :param evt: An event object that will cause the loop to complete early
            if the event is triggered while waiting.
        :type evt: threading.Event
        """
        if ltime is None or ltime < 0:
            raise ValueError("an interval must be defined")
        self.ltime = float(ltime)
        self.evt = evt
        self.lstart = None

    @property
    def remaining(self):
        """Return the remaining time in the loop.

        :rtype: float
        """
        if self.lstart is not None:
            return self.ltime - (time.time() - self.lstart)
        return 0.0

    @property
    def elapsed(self):
        """Return the elapsed time of the loop.

        :rtype: float
        """
        if self.lstart is not None:
            return time.time() - self.lstart
        return 0.0

    def __str__(self):
        """Return a nicely formatted string.

        :rtype: str
        """
        elapsed = None
        if self.lstart is not None:
            elapsed = time.time() - self.lstart
        sval = "Interval {0} seconds, {1} elapsed, {2} started"
        return sval.format(self.ltime, elapsed, self.lstart)

    def __repr__(self):
        """Return a nicely formatted string.

        :rtype: str
        """
        elapsed = None
        if self.lstart is not None:
            elapsed = time.time() - self.lstart
        sval = "Interval {0} seconds, {1} elapsed, {2} started"
        return sval.format(self.ltime, elapsed, self.lstart)

    def __enter__(self):
        """Called on entrance to with() statement - simply records loop start
        time.

        :rtype: Interval
        """
        self.lstart = time.time()
        return self

    def __exit__(self, *args):
        """Called on exit from a with() statement and either waits on the
        self.evt :class:`threading.Event` event if defined or time.sleep()'s
        the remaining interval time.
        """
        if self.lstart is None:
            self.lstart = time.time()
        # block on the event for the remaining time, if any
        wtime = self.ltime - (time.time() - self.lstart)
        if wtime > 0:
            if self.evt is not None:
                self.evt.wait(wtime)
            else:
                time.sleep(wtime)


class SignalWaiter(object):
    """A helper class that makes waiting on signals very easy. Define the
    list of signals to wait for and to ignore and then call wait(). Instead of
    using signal handler functions it uses the signal.set_wakeup_fd() function
    and blocks on read().

    See http://www.pocketnix.org/doc/Fighting_set__wakeup__fd/ and others for
    more details.

    :param sig_wait: :class:`list` signals to wait for
    :type sig_wait: list
    :param sig_ignore: :class:`list` signals to ignore
    :type sig_ignore: list
    """

    def __init__(self, sig_wait, sig_ignore):
        """Create a :class:`SignalWaiter`"""
        self.sig_wait = sig_wait
        self.sig_ignore = sig_ignore

        # use a pipe to listen for signals
        self.rpipe, self.wpipe = os.pipe()
        # set non blocking mode on the write end of the pipe
        flags = fcntl.fcntl(self.wpipe, fcntl.F_GETFL, 0)
        flags = flags | os.O_NONBLOCK
        flags = fcntl.fcntl(self.wpipe, fcntl.F_SETFL, flags)

        # set the write end as a signal wakeup fd - we read signals from it
        signal.set_wakeup_fd(self.wpipe)

        # install dummy signal handlers that do nothing
        for signum in self.sig_wait:
            signal.signal(signum, lambda x, y: None)

        # install the ignore signal handler for sig_ignore signals
        for signum in self.sig_ignore:
            signal.signal(signum, signal.SIG_IGN)

        # now a call to wait() blocks the caller until a signal is received
        # only the signals we've registered handlers for will trigger

    def wait(self):
        """Block the calling thread until signal(s) are received.

        :rtype: :class:`int` or :class:`Exception`
        """
        try:
            return os.read(self.rpipe, 1)
        except Exception as exc:

Catching very general exceptions such as Exception is usually not recommended.

            return exc


class Filter(object):
    """A helper class that efficiently filters characters from strings.

    :param keep_chars: A string containing all characters to keep in strings
    :type keep_chars: str
    """

    def __init__(self, keep_chars=None):
        """Create a :class:`Filter`"""
        if not keep_chars:
            # translate is different in python3, so is string
            if PY3:
                keep_chars = string.ascii_letters + string.digits + "_-"
            else:
                keep_chars = string.letters + string.digits + "_-"

The Module string does not seem to have a member named letters.

        self.filter_chars = "".join(char for char in
                                    [chr(i) for i in range(256)]
                                    if char not in keep_chars)
        self.table = dict.fromkeys([ord(i) for i in self.filter_chars], None)
        if PY3:
            self.process = self.process_py3
        else:
            self.process = self.process_py2

    def process_py3(self, filter_str):
        """Return a filtered string, removes filter_char's from it.

        :rtype: str
        """
        return filter_str.translate(self.table)

    def process_py2(self, filter_str):
        """Return a filtered string, removes filter_char's from it.

        :rtype: str
        """
        return filter_str.translate(None, self.filter_chars)


def load_instance(log, cname, mod, cobj):
    """Return an instance of the class cname from the specified module.

    :param log: A logger
    :type log: logging.RootLogger
    :param cname: The class name to load
    :type cname: str
    :param mod: A module object from eg. imp.load_source or __import__
    :type mod: module
    :param cobj: A conf configuration helper instance to pass to the class
    :type cobj: conf
    :rtype: Object -- an instance of the class requested
    :raises: PluginLoadError: many things can go wrong during loading
    """
    obj = None
    # find the class in the module
    for n, d in inspect.getmembers(mod, inspect.isclass):
        if n == cname:
            # try to instantiate the class
            try:
                obj = d(log, cobj)
            # plugins may raise pretty much any exception
            except Exception as e:
                tb = traceback.format_exc()
                estr = "class {0}, module {1} raised: {2}, trace:{3}"
                eargs = [cname, mod.__name__, e, tb]
                raise plumd.PluginLoadError(estr.format(*eargs))

Usage of * or ** arguments should usually be done with care.

            break
    # the module does not have the expected class in it
    if obj is None:
        estr = "{0} does not define class {1}".format(mod.__name__, cname)
        raise plumd.PluginLoadError(estr)
    return obj


def load_file_instance(log, cname, fname, cobj):
    """Import the class cname from the python file fname and return
    an instance of it.

    raises:
        PluginNotFoundError if the fname doesn't exist.

    :param log: A logger
    :type log: logging.RootLogger
    :param cname: The name of the class to load
    :type cname: str
    :param fname: The full path to the python file to load from
    :type fname: str
    :param cobj: A conf configuration helper instance to pass to the class
    :type cobj: conf
    :rtype: Object -- an instance of the class requested
    :raises: PluginNotFoundError
    """
    mod = None
    if os.path.isfile(fname):
        # load the module that has the same name as the file (minus .py)
        mod = imp.load_source(cname, fname)
    else:
        raise plumd.PluginNotFoundError(
            "file {0} does not exist".format(fname))
    return load_instance(log, cname, mod, cobj)


def load_module_instance(log, cname, mname, cobj):
    """Import the class cname from the python module mname and return
    an instance of it.

    raises:
        PluginLoadError - so many things can go wrong loading plugins..

    :param log: A logger
    :type log: logging.RootLogger
    :param cname: The name of the class to load
    :type cname: str
    :param mname: The name of the module to load from
    :type fname: str
    :param cobj: A conf configuration helper instance to pass to the class
    :type cobj: conf
    :rtype: Object -- an instance of the class requested
    :raises: PluginLoadError
    """
    mod = None
    fname = None
    if "." in mname:
        fname = mname
        log.debug("importing {0} from {1}".format(mname, fname))
    else:
        log.debug("importing {0}".format(mname))
    try:
        mod = __import__(mname.strip(), fromlist=fname)
    except Exception as e:
        msg = "module {0} import failed: {1}".format(mname, e)
        raise plumd.PluginLoadError(msg)
    return load_instance(log, cname, mod, cobj)


def load_plugin(log, pconf):
    """Load the configured plugin and return an instance of it.

    The configuration file for pconf must define at least:

    name: <name for this instance>          # a uniq name to give this instance
    module: <name of module to load from>   # the module name to load
    file: <full path to file to load from>  # or, the filename to load
    pclass: <name of class to load>         # name of the class to load

    plus any plugin specific configuration. You can instantiate multiple
    instances of a plugin with different configurations however must ensure
    that they are configured with uniq names.

    Also, if both file and module are defined in the configuration the file
    configuration is used and module is ignored.

    Raises:
        ConfigError if 'file' and 'module' sources are both configured.
        ConfigError if the configuration is missing 'name'.
        PluginLoadError if the configured file/module is not found.
        PluginLoadError if it was unable to instantiate the requested class.

    :param log: A logger
    :type log: logging.RootLogger
    :param pconf: a configuration object that defines the plugin to load
    :type pconf: plumd.config.conf
    :raises: PluginLoadError, ConfigError
    """
    fname = pconf.get('file')                   # full path to python file
    mname = pconf.get('module')                 # module name
    cname = pconf.get('class', exception=True)  # class name
    instance = None

    if fname:
        instance = load_file_instance(log, cname, fname, pconf)
    elif mname:
        try:
            __import__(mname.strip())
        except ImportError as e:
            estr = "module not found: {0} : {1} : {2}"
            raise plumd.PluginLoadError(estr.format(mname, pconf.path, e))
        instance = load_module_instance(log, cname, mname, pconf)

    if instance is None:
        raise plumd.PluginLoadError("load failed for: {0}".format(pconf.path))

    return instance


def load_from_conf(log, config, pconf):
    """Load a plugin as defined by pconf configuration file and return an
    instance of it.

    raises:
        ConfigError if ptype has an incorrect configuration path set
        ConfigError if a plugin configuration is missing 'name'
        DuplicatePlugin if a plugin configuration has a duplicate 'name'
        PluginLoadError if there was an error loading the plugin

    :param log: A logger
    :type log: logging.RootLogger
    :param config: a :class:`plumd.config.Conf` instance
    :type config: plumd.config.Conf
    :param pconf: Full path to a plugin configuration.
    :type pconf: str
    :rtype: object
    :raises: ConfigError, DuplicatePlugin
    """
    # create a configuration and pass on select values from the main config
    defaults = {
        'poll.interval': config.get('poll.interval'),
        'delay.poll': config.get('delay.poll'),
        'meta': config.get('meta')
    }
    pconfig = plumd.config.Conf(pconf).defaults(defaults)
    # check to see if the plugin is disabled, if so return here
    if not pconfig.get('enabled', default=True):
        return

    pname = pconfig.get("name", exception=True)

    # load the plugin
    log.debug("loading {0} from {1}".format(pname, pconfig.path))
    return load_plugin(log, pconfig)


def get_plugins_dict(log, plugins, pclass):

The argument log seems to be unused.

    """Return a dict of plugins that are of type pclass where the dict
    keys are the plugins name and the values are the plugin object.

    raises:
        DuplicatePlugin if a plugin has a duplicate 'name'

    :param log: A logger
    :type log: logging.RootLogger
    :param plugins: A list of plugin objects.
    :type plugins: list
    :param pclas: A class eg. :class:`plumd.Reader`, :class:`plumd.Writer`
    :type pclass: type
    :rtype: dict
    :raises: DuplicatePlugin
    """
    plugs = {}
    for pobj in plugins:
        # get the list of base/super classes for the plugin
        pclasses = inspect.getmro(pobj.__class__)

        # ensure the plugin is a subclass of the requested class
        if pclass in pclasses:
            pconf = pobj.config
            pname = pconf.get('name', exception=True)
            if pname in plugs:
                err = "duplicate plugin: {0} from {1}"
                raise plumd.DuplicatePlugin(err.format(pname, pconf.path))
            plugs[pname] = pobj
    return plugs


def load_all_plugins(log, config):
    """Return a tuple of reader and writer plugin dicts.

    raises:
        ConfigError if ptype has an incorrect configuration path set
        ConfigError if a plugin configuration is missing 'name'
        DuplicatePlugin if a plugin configuration has a duplicate 'name'
        PluginLoadError if there was an error loading the plugin

    Returns a tuple of: (readers, writers) where each is a dict of:
    {'name': <plugin_instance>}. Name is the configured plugin name and
    <plugin_instance> is the corresponding instance of the plugin. Note plugins
    can be insantiated multiple times with different configurations.

    :param log: A logger
    :type log: logging.RootLogger
    :param config: a :class:`plumd.config.Conf` instance
    :type config: plumd.config.Conf
    :rtype: tuple
    :raises: ConfigError, DuplicatePlugin, PluginLoadError
    """
    plugins = []
    # get a list of configuration files
    # note the call to ls can raise an exception which we pass upwards
    for pconf in plumd.config.find(config.get('config.plugins'), 'yaml'):
        # load the plugin object
        pobj = load_from_conf(log, config, pconf)

        # check if the plugin has been disabled in configuration
        if pobj is None:
            msg = "skipping disabled plugin: {0}"
            log.info(msg.format(pconf))
            continue

        plugins.append(pobj)

    log.debug("plugins: {0}".format(
        " ".join([p.__class__.__name__ for p in plugins])))

    # we iterate over the plugin list multiple times..
    # since the list is small and there are not many types this is ok.
    readers = get_plugins_dict(log, plugins, plumd.Reader)
    writers = get_plugins_dict(log, plugins, plumd.Writer)
    log.info("readers: {0}".format(" ".join(readers.keys())))
    log.info("writers: {0}".format(" ".join(writers.keys())))

    return (readers, writers)


def config_plugin_writers(lobj):
    """Update the list of writers for each reader in the PluginLoader.

    todo: change lobj to a tuple of (readers,writers)

    :param lobj: a :class:`plumd.PluginLoader` object
    :type lobj: plumd.PluginLoader

    :raises: ConfigError
    """
    allw = [(wobj.queue_evt, wobj.queue) for wobj in lobj.writers.values()]
    # now set the PluginReaders writers
    for prname, probj in lobj.readers.items():
        wcfg = probj.config.get('writers')
        # if nothing is configured the reader writes to all
        if wcfg is None:
            probj.queues = allw
            continue

        # get list of writer names
        wnames = [w for w in wcfg if w in lobj.writers]
        if wnames:
            probj.queues = [(lobj.writers[w].queue_evt, lobj.writers[w].queue)
                            for w in wnames]
            continue

        msg = "reader {0} from {1} has no valid writers configured"
        lobj.log.error(msg.format(prname, probj.conf.path))
        sys.exit(1)


class Differential(object):
    """Compute the rate of change of a set of metrics."""

    def __init__(self):
        """Compute the rate of change of a set of metrics."""
        self.metrics = {}

    def per_second(self, name, val, timestamp):
        """Record and return the rate of change of a metric in units/second.

        Assumes values will be < MAX_CNT and detects and corrects overflow.

        :param name: The name of a metric
        :type name: str
        :param val: The value of a metric
        :type val: int or float
        :param max: The maximum possible value for val
        :type val: int or float
        :rtype: int or float
        """
        ret = type(val)(0)
        if name not in self.metrics:
            self.metrics[name] = (val, timestamp)
        else:
            # get the previous value
            pval, ptime = self.metrics[name]
            self.metrics[name] = (val, timestamp)
            # check for counter wrap
            if val < pval:
                pval = pval - MAX_CNT
            dval = val - pval
            dtime = timestamp - ptime
            if dtime > 0:
                ret = float(dval) / float(dtime)
            else:
                ret = 0
        return ret