6de18f831d
The module simply calls the external tool "ping" to retriev RTT information and displays the average RTT in the widget. Addresses issue #6
98 lines
3.1 KiB
Python
98 lines
3.1 KiB
Python
from __future__ import absolute_import
|
|
|
|
import re
|
|
import time
|
|
import shlex
|
|
import threading
|
|
import subprocess
|
|
|
|
import bumblebee.module
|
|
import bumblebee.util
|
|
|
|
def description():
|
|
return "Periodically checks the RTT of a configurable IP"
|
|
|
|
def parameters():
|
|
return [
|
|
"ping.interval: Time in seconds between two RTT checks (defaults to 60)",
|
|
"ping.address: IP address to check",
|
|
"ping.warning: Threshold for warning state, in seconds (defaults to 1.0)",
|
|
"ping.critical: Threshold for critical state, in seconds (defaults to 2.0)",
|
|
"ping.timeout: Timeout for waiting for a reply (defaults to 5.0)",
|
|
"ping.probes: Number of probes to send (defaults to 5)",
|
|
]
|
|
|
|
def get_rtt(obj):
|
|
loops = obj.get("interval")
|
|
|
|
for thread in threading.enumerate():
|
|
if thread.name == "MainThread":
|
|
main = thread
|
|
|
|
interval = obj.get("interval")
|
|
while main.is_alive():
|
|
loops += 1
|
|
if loops < interval:
|
|
time.sleep(1)
|
|
continue
|
|
|
|
loops = 0
|
|
try:
|
|
res = subprocess.check_output(shlex.split("ping -n -q -c {} -W {} {}".format(
|
|
obj.get("rtt-probes"), obj.get("rtt-timeout"), obj.get("address")
|
|
)))
|
|
except Exception as e:
|
|
obj.set("rtt-unreachable", True)
|
|
break
|
|
|
|
obj.set("rtt-unreachable", False)
|
|
|
|
for line in res.decode().split("\n"):
|
|
if not line.startswith("rtt"): continue
|
|
m = re.search(r'([0-9\.]+)/([0-9\.]+)/([0-9\.]+)/([0-9\.]+)\s+(\S+)', line)
|
|
|
|
obj.set("rtt-min", float(m.group(1)))
|
|
obj.set("rtt-avg", float(m.group(2)))
|
|
obj.set("rtt-max", float(m.group(3)))
|
|
obj.set("rtt-unit", m.group(5))
|
|
|
|
class Module(bumblebee.module.Module):
|
|
def __init__(self, output, config, alias):
|
|
super(Module, self).__init__(output, config, alias)
|
|
|
|
self._counter = {}
|
|
|
|
self.set("address", self._config.parameter("address", "8.8.8.8"))
|
|
self.set("interval", self._config.parameter("interval", 60))
|
|
self.set("rtt-probes", self._config.parameter("probes", 5))
|
|
self.set("rtt-timeout", self._config.parameter("timeout", 5.0))
|
|
|
|
self._thread = threading.Thread(target=get_rtt, args=(self,))
|
|
self._thread.start()
|
|
|
|
def set(self, what, value):
|
|
self._counter[what] = value
|
|
|
|
def get(self, what):
|
|
return self._counter.get(what, 0)
|
|
|
|
def widgets(self):
|
|
text = "{}: {:.1f}{}".format(
|
|
self.get("address"),
|
|
self.get("rtt-avg"),
|
|
self.get("rtt-unit")
|
|
)
|
|
|
|
if self.get("rtt-unreachable"):
|
|
text = "{}: unreachable".format(self.get("address"))
|
|
|
|
return bumblebee.output.Widget(self, text)
|
|
|
|
def warning(self, widget):
|
|
return self.get("rtt-avg") > float(self._config.parameter("warning", 1.0))*1000.0
|
|
|
|
def critical(self, widget):
|
|
if self.get("rtt-unreachable"): return True
|
|
return self.get("rtt-avg") > float(self._config.parameter("critical", 2.0))*1000.0
|
|
|
|
# vim: tabstop=8 expandtab shiftwidth=4 softtabstop=4
|