Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_plugin_legacy_groups():
"""
test legacy plugin groups
TODO: remove with vaping 2.0
"""
probe = plugin.get_probe({
"type" : "fping",
"name" : "probe_b",
"interval" : "3s",
"dns": {
"hosts": [{
"host" : "1.1.1.1",
"name" : "Cloudflare"
}]
}
}, {})
expected = {
"dns": {
"hosts": [{
"host" : "1.1.1.1",
"name": "Cloudflare"
def test_filename_format():
"""
test filename formatting from data
"""
inst = plugin.get_instance(config, None)
assert inst.format_filename({}, {"a":"first", "b":"second"}) == "first-second-test"
from __future__ import absolute_import
import vaping
import vaping.config
try:
import whisper
except ImportError:
whisper = None
@vaping.plugin.register('whisper')
class WhisperPlugin(vaping.plugins.TimeSeriesDB):
"""
Whisper plugin that allows vaping to persist data
in a whisper database
"""
def __init__(self, config, ctx):
if not whisper:
raise ImportError("whisper not found")
super(WhisperPlugin, self).__init__(config, ctx)
# whisper specific config
self.retention = self.config.get("retention", ['3s:1d'])
self.x_files_factor = float(self.config.get("x_files_factor", 0.5))
self.aggregation_method = self.config.get("aggregation_method", "average")
from __future__ import absolute_import
import logging
import re
import datetime
import vaping
import vaping.config
@vaping.plugin.register("logparse")
class LogParse(vaping.plugins.FileProbe):
"""
Log parse plugin base
Will parse a log line by line and probe to emit data
over a specified interval.
# Config
- path (`str`): log file path
- fields (`dict`): field definition
field name as key
`parser` regex pattern to parse field value, needs to
one group in it