inital version of speedtest

checks/check_speedtest.py

@@ -0,0 +1,117 @@
+#!/usr/bin/env python3
+"""speedtest.net python check for icinga"""
+"""dependencys:
+- pip3 install nagiosplugin
+- pip3 install argparse
+- pip3 install speedtest-cli 
+or
+system package on debian: 
+- apt install speedtest
+"""
+
+__version__     = '0.1.0'
+__author__      = 'anima'
+
+# imports
+import logging
+import argparse
+import nagiosplugin
+import speedtest
+
+# log settings
+logging.basicConfig(format='[%(asctime)s] %(levelname)s %(message)s', level=logging.INFO)
+
+class SpeedtestResource(nagiosplugin.Resource):
+    def __init__(self) -> None:
+        self.test = speedtest.Speedtest(secure=True)
+
+    def probe(self) -> list:
+        """speedtest check for up & download
+        
+        Returns:
+            Generator[nagisplugin.Metric]: multiple metric elements (yield)
+        """
+        def to_mb(byte) -> float:
+            kb = byte / 1024
+            mb = kb / 1024
+            return mb
+        
+        logging.debug('start download speedtest...')
+        self.test.download()
+        logging.debug(f'download speed: {self.test.results.download}')
+        
+        logging.debug('start upload speedtest...')
+        self.test.upload()
+        logging.debug(f'upload speed: {self.test.results.upload}')
+
+        logging.debug(f'used server: {self.test.results.server}')
+
+        yield nagiosplugin.Metric(name='download', value=round(to_mb(self.test.results.download), 2), uom=' mb/s', context='download_scalar_context')
+        yield nagiosplugin.Metric(name='upload', value=round(to_mb(self.test.results.upload), 2), uom=' mb/s', context='upload_scalar_context')
+        yield nagiosplugin.Metric(name='ping', value=self.test.results.ping, uom=' ms', context='ping_scalar_context')
+
+
+def parse_args() -> argparse.Namespace:
+    """evaluates given arguments
+
+    Returns:
+        argsparse.Namespace: Namespace Object with all arguments insert (use: args.long_name_of_argument)
+    """
+    argp = argparse.ArgumentParser(description=__doc__)
+    # Default args
+    argp.add_argument('-v', '--verbose', action='count', default=0,
+        help='increase output verbosity (use up to 3 times)')
+    argp.add_argument('-H', '--hostname', default='localhost',
+        help='IP address or hostname of device to query')
+    argp.add_argument('-t', '--timeout', default=60,
+        help='abort execution after TIMEOUT seconds')
+    argp.add_argument('-m', '--check_mode',
+        choices=[
+            'speedtest',
+        ], default='speedtest',
+        help='check mode to run')
+
+    # Nagios args / see https://nagios-plugins.org/doc/guidelines.html#THRESHOLDFORMAT
+    argp.add_argument('-w', '--warning', default='80:',
+        help='not used for this check') # for compatibility not removed
+    argp.add_argument('-c', '--critical', default='90:',
+        help='not used for this check') # for compatibility not removed
+    argp.add_argument('-wd', '--warning-download', default='90:',
+        help='warning threshold for download in mb')
+    argp.add_argument('-cd', '--critical-download', default='25:',
+        help='critical threshold for download in mb')
+    argp.add_argument('-wu', '--warning-upload', default='25:',
+        help='warning threshold for upload in mb')
+    argp.add_argument('-cu', '--critical-upload', default='10:',
+        help='critical threshold for upload in mb')
+    argp.add_argument('-wp', '--warning-ping', default=':50',
+        help='warning threshold for ping in ms')
+    argp.add_argument('-cp', '--critical-ping', default=':100',
+        help='critical threshold for ping in ms')
+
+    args = argp.parse_args()
+    return args
+
+# @nagiosplugin.guarded(verbose=0)
+def main():
+    args = parse_args()
+    if args.verbose >= 3:
+        logging.getLogger().setLevel(logging.DEBUG)
+
+    # dice which check will be run bases on check_mode
+    match args.check_mode:
+        case 'speedtest':
+            check = nagiosplugin.Check(
+                SpeedtestResource(),
+                nagiosplugin.ScalarContext(name='download_scalar_context', warning=args.warning_download, critical=args.critical_download),
+                nagiosplugin.ScalarContext(name='upload_scalar_context', warning=args.warning_upload, critical=args.critical_upload),
+                nagiosplugin.ScalarContext(name='ping_scalar_context', warning=args.warning_ping, critical=args.critical_ping),
+                nagiosplugin.Summary())
+            check.name = "speedtest"
+        case _:
+            raise nagiosplugin.CheckError(f'Unknown check mode: {args.check_mode}')
+
+    check.main(args.verbose, args.timeout)
+
+if __name__ == '__main__':
+    main()