How to use the vaping.plugin.register function in vaping

class EmitPluginStore(plugins.EmitBase):
    def init(self):
        super(EmitPluginStore, self).init()
        self.store = []
    def emit(self, message):
        self.store.append(message)


@plugin.register('queue_tester')
class QueueTester(plugins.ProbeBase):
    def probe(self):
        msg = self.new_message()
        return msg


@plugin.register('probe0')
class TimedPlugin0(plugins.TimedProbe):
    default_config = {
        'interval': '1m',
        'count': 5,
        'period': 20,
    }

    def probe(self):
        return []


@plugin.register('probe1')
class ProbePlugin1(plugins.ProbeBase):
    def probe(self):
        return []

'var0': 999,
}


@plugin.register('plugin0')
class Plugin0(plugins.PluginBase):
    pass


@plugin.register('emit0')
class EmitPlugin0(plugins.EmitBase):
    def emit(self, message):
        pass


@plugin.register('emit_abc')
class EmitPluginABC(plugins.EmitBase):
    # emit not defined to test TypeError
    pass

@plugin.register('emit_store')
class EmitPluginStore(plugins.EmitBase):
    def init(self):
        super(EmitPluginStore, self).init()
        self.store = []
    def emit(self, message):
        self.store.append(message)


@plugin.register('queue_tester')
class QueueTester(plugins.ProbeBase):
    def probe(self):

}
    ],
}

anon_config = {
    'type': 'plugin0',
    'var0': 999,
}


@plugin.register('plugin0')
class Plugin0(plugins.PluginBase):
    pass


@plugin.register('emit0')
class EmitPlugin0(plugins.EmitBase):
    def emit(self, message):
        pass


@plugin.register('emit_abc')
class EmitPluginABC(plugins.EmitBase):
    # emit not defined to test TypeError
    pass

@plugin.register('emit_store')
class EmitPluginStore(plugins.EmitBase):
    def init(self):
        super(EmitPluginStore, self).init()
        self.store = []
    def emit(self, message):

return msg


@plugin.register('probe0')
class TimedPlugin0(plugins.TimedProbe):
    default_config = {
        'interval': '1m',
        'count': 5,
        'period': 20,
    }

    def probe(self):
        return []


@plugin.register('probe1')
class ProbePlugin1(plugins.ProbeBase):
    def probe(self):
        return []


def test_plugin_registry():
    assert Plugin0 == plugin.get_plugin_class('plugin0')
    with pytest.raises(ValueError):
        plugin.get_plugin_class('nonexistant')

    with pytest.raises(ValueError):
        @plugin.register('plugin0')
        class p0(plugins.PluginBase):
            pass

from builtins import str

try:
    import zmq.green as zmq
except ImportError:
    zmq = None

import vaping
import vaping.plugins


@vaping.plugin.register('zeromq')
class ZeroMQ(vaping.plugins.EmitBase):
    """
    plugin to emit json encoded messages via zeromq

    # Instanced Attributes

    - ctx (`zmq Context`)
    - sock (`zmq socket`)
    """

    def __init__(self, config, ctx):
        super(ZeroMQ, self).__init__(config, ctx)

        self.sock = None

    def init(self):

from __future__ import absolute_import

import vaping
import vaping.config

try:
    import rrdtool
except ImportError:
    rrdtool = None

@vaping.plugin.register('rrd')
class RRDToolPlugin(vaping.plugins.TimeSeriesDB):

    """
    RRDTool plugin that allows vaping to persist data
    in a rrdtool database
    """

    def __init__(self, config, ctx):
        if not rrdtool:
            raise ImportError("rrdtool not found")
        super(RRDToolPlugin, self).__init__(config, ctx)


    def init(self):
        # rrdtool specific config
        self.data_sources = self.config.get("data_sources", [])

def parse_line(line):
    if line.startswith('CNTR'):
        keys = (x[0] for x in _KEYDEF['CNTR'])
        typs = (x[1] for x in _KEYDEF['CNTR'])
        return {k: t(d) for (d, k, t) in zip(line.split(','), keys, typs)}


def calc_rate(last, cur):
    now = datetime.datetime.utcnow()
    time_delta = (now - last['ts']).total_seconds()
    in_delta = cur['in_oct'] - last['data']['in_oct']
    in_bps = in_delta * 8 / time_delta
    print("time_delta=%f in_delta=%d Gbps=%f" % (time_delta, in_delta, old_div(in_bps, 1000000000)))


@vaping.plugin.register('sflowtool')
class SflowTool(vaping.plugins.TimedProbe):

    default_config={
        'interval': '1m',
        'count': 5,
    }

    def init(self):
        self.stdout_queue = Queue()
        self.stderr_queue = Queue()
        args = [
            'sflowtool',
            '-l',
        ]
        self.spawn_process(args)

# automatic group setup for fping

    for group_name, group_config in list(probe.groups.items()):
        if "hosts" not in group_config:
            continue

        r = {}
        for host in group_config.get("hosts"):
            if isinstance(host, dict):
                r[host["host"]] = host
            else:
                r[host] = {"host":host}
        graphsrv.group.add(probe.name, group_name, r, **group_config)


@vaping.plugin.register('vodka')
class VodkaPlugin(vaping.plugins.EmitBase):

    """
    Plugin that emits to vodka data
    """

    def init(self):
        self._is_started = False

    def start(self):

        if self._is_started:
            return
        self._is_started = True
        vodka.run(self.config, self.vaping.config)

import vaping.config
import requests

try:
    import graphyte
except ImportError:
    graphyte = None


def munge_filename(filename):
    filename = filename.replace('.', '_')
    filename = filename.replace('-', '.')
    return filename


@vaping.plugin.register('graphite')
class GraphitePlugin(vaping.plugins.TimeSeriesDB):

    """
    Graphite plugin that allows vaping to persist data
    to a graphite via line protocole
    """

    def __init__(self, config, ctx):
        super(GraphitePlugin, self).__init__(config, ctx)

        if not graphyte:
            self.log.critical("missing graphyte, install it with `pip install graphyte`")
            raise RuntimeError("graphyte not found")

        # get configs
        self.proto = self.config.get("proto", "http")

from __future__ import absolute_import

import collections
import datetime
import munge
import shlex

import vaping
from vaping.io import subprocess


@vaping.plugin.register('command')
class CommandProbe(vaping.plugins.TimedProbe):

    """
    Probe type plugin that allows you to run an arbitrary
    command for each host and return the command output as
    data

    # Config

    - command (`str`): command to run (use `{host}` to reference the host)
    - interval (`float`): time between probes

    # Instanced Attributes

    - command (`str`): command to run
    """