From ae6170965716e7eaef8500f0fadc85f0ec4e4687 Mon Sep 17 00:00:00 2001 From: Julien Riou Date: Sun, 29 Oct 2023 09:25:55 +0100 Subject: [PATCH] refactor: Use python packages Signed-off-by: Julien Riou --- {lib => check_ovhcloud}/__init__.py | 0 check_ovhcloud/ping/__init__.py | 86 +++++++++++++++++ check_ovhcloud/voip/__init__.py | 139 ++++++++++++++++++++++++++++ check_ping | 83 +---------------- check_voip | 136 +-------------------------- 5 files changed, 227 insertions(+), 217 deletions(-) rename {lib => check_ovhcloud}/__init__.py (100%) create mode 100644 check_ovhcloud/ping/__init__.py create mode 100644 check_ovhcloud/voip/__init__.py diff --git a/lib/__init__.py b/check_ovhcloud/__init__.py similarity index 100% rename from lib/__init__.py rename to check_ovhcloud/__init__.py diff --git a/check_ovhcloud/ping/__init__.py b/check_ovhcloud/ping/__init__.py new file mode 100644 index 0000000..d19390c --- /dev/null +++ b/check_ovhcloud/ping/__init__.py @@ -0,0 +1,86 @@ +#!/usr/bin/env python3 + +import argparse +import logging +import sys +import ovh + +from check_ovhcloud import setup_logging, show_version, BooleanContext + +from nagiosplugin import ( + Check, + Metric, + Resource, + Summary, +) +from nagiosplugin.state import Unknown + +logger = logging.getLogger(__name__) + + +def parse_arguments(): + parser = argparse.ArgumentParser() + parser.add_argument( + "-v", + "--verbose", + dest="loglevel", + action="store_const", + const=logging.INFO, + help="Print more output", + ) + parser.add_argument( + "-d", + "--debug", + dest="loglevel", + action="store_const", + const=logging.DEBUG, + default=logging.WARNING, + help="Print even more output", + ) + + parser.add_argument( + "--version", + dest="show_version", + action="store_true", + help="Print version and exit", + ) + + args = parser.parse_args() + return args + + +class Ping(Resource): + def __init__(self, client): + self.client = client + + def probe(self): + try: + me = self.client.get(f"/me") + return [Metric(me["firstname"], True, context="ping")] + except Exception as err: + return [Metric(str(err), False, context="ping")] + + +def main(): + args = parse_arguments() + setup_logging(args) + + if args.show_version: + show_version() + return + + try: + check = Check( + Ping(client=ovh.Client()), + BooleanContext("ping", expected=True, critical=True), + Summary(), + ) + check.main() + except Exception as err: + print(f"Failed to execute check: {str(err)}") + logger.debug(err, exc_info=True) + sys.exit(Unknown.code) + + +if __name__ == "__main__": + main() diff --git a/check_ovhcloud/voip/__init__.py b/check_ovhcloud/voip/__init__.py new file mode 100644 index 0000000..dee7ef4 --- /dev/null +++ b/check_ovhcloud/voip/__init__.py @@ -0,0 +1,139 @@ +#!/usr/bin/env python3 + +import argparse +import logging +import sys +import ovh +from datetime import datetime, timezone + +from check_ovhcloud import setup_logging, show_version + +from nagiosplugin import ( + Check, + Metric, + Resource, + ScalarContext, + Summary, +) +from nagiosplugin.state import Unknown + +logger = logging.getLogger(__name__) + + +def parse_arguments(): + parser = argparse.ArgumentParser() + parser.add_argument( + "-v", + "--verbose", + dest="loglevel", + action="store_const", + const=logging.INFO, + help="Print more output", + ) + parser.add_argument( + "-d", + "--debug", + dest="loglevel", + action="store_const", + const=logging.DEBUG, + default=logging.WARNING, + help="Print even more output", + ) + + parser.add_argument( + "--version", + dest="show_version", + action="store_true", + help="Print version and exit", + ) + + parser.add_argument( + "--registration-warning", + "-w", + dest="registration_warning", + type=int, + default=7200, + help="Raise warning if last line registration time is higher than this threshold (in seconds)", + ) + + parser.add_argument( + "--registration-critical", + "-c", + dest="registration_critical", + type=int, + default=86400, + help="Raise critical if last line registration time is higher than this threshold (in seconds)", + ) + + args = parser.parse_args() + return args + + +class Voip(Resource): + def __init__(self, client): + self.client = client + + def probe(self): + metrics = [] + billing_accounts = self.client.get(f"/telephony") + for billing_account in billing_accounts: + service_names = self.client.get(f"/telephony/{billing_account}/line") + for service_name in service_names: + last_registration = self.client.get( + f"/telephony/{billing_account}/line/{service_name}/lastRegistrations" + )[0] + last_registration_delta = int( + ( + datetime.now(tz=timezone.utc) + - datetime.fromisoformat(last_registration["datetime"]) + ).total_seconds() + ) + metrics.append( + Metric( + f"{service_name} last registration", + last_registration_delta, + context="last_registration", + uom="s", + ) + ) + return metrics + + +class VoipSummary(Summary): + def problem(self, results): + return ", ".join( + [ + f"{result.metric.name} {result.state}: {result.hint}" + for result in results + if str(result.state) != "ok" + ] + ) + + +def main(): + args = parse_arguments() + setup_logging(args) + + if args.show_version: + show_version() + return + + try: + check = Check( + Voip(client=ovh.Client()), + ScalarContext( + "last_registration", + warning=args.registration_warning, + critical=args.registration_critical, + ), + VoipSummary(), + ) + check.main() + except Exception as err: + print(f"Failed to execute check: {str(err)}") + logger.debug(err, exc_info=True) + sys.exit(Unknown.code) + + +if __name__ == "__main__": + main() diff --git a/check_ping b/check_ping index a5d0ebc..f4d4f4e 100755 --- a/check_ping +++ b/check_ping @@ -1,86 +1,5 @@ #!/usr/bin/env python3 - -import argparse -import logging -import sys -import ovh - -from lib import setup_logging, show_version, BooleanContext - -from nagiosplugin import ( - Check, - Metric, - Resource, - Summary, -) -from nagiosplugin.state import Unknown - -logger = logging.getLogger(__name__) - - -def parse_arguments(): - parser = argparse.ArgumentParser() - parser.add_argument( - "-v", - "--verbose", - dest="loglevel", - action="store_const", - const=logging.INFO, - help="Print more output", - ) - parser.add_argument( - "-d", - "--debug", - dest="loglevel", - action="store_const", - const=logging.DEBUG, - default=logging.WARNING, - help="Print even more output", - ) - - parser.add_argument( - "--version", - dest="show_version", - action="store_true", - help="Print version and exit", - ) - - args = parser.parse_args() - return args - - -class Ping(Resource): - def __init__(self, client): - self.client = client - - def probe(self): - try: - me = self.client.get(f"/me") - return [Metric(me["firstname"], True, context="ping")] - except Exception as err: - return [Metric(str(err), False, context="ping")] - - -def main(): - args = parse_arguments() - setup_logging(args) - - if args.show_version: - show_version() - return - - try: - check = Check( - Ping(client=ovh.Client()), - BooleanContext("ping", expected=True, critical=True), - Summary(), - ) - check.main() - except Exception as err: - print(f"Failed to execute check: {str(err)}") - logger.debug(err, exc_info=True) - sys.exit(Unknown.code) - +from check_ovhcloud.ping import main if __name__ == "__main__": main() diff --git a/check_voip b/check_voip index 56d89f5..e827820 100755 --- a/check_voip +++ b/check_voip @@ -1,139 +1,5 @@ #!/usr/bin/env python3 - -import argparse -import logging -import sys -import ovh -from datetime import datetime, timezone - -from lib import setup_logging, show_version - -from nagiosplugin import ( - Check, - Metric, - Resource, - ScalarContext, - Summary, -) -from nagiosplugin.state import Unknown - -logger = logging.getLogger(__name__) - - -def parse_arguments(): - parser = argparse.ArgumentParser() - parser.add_argument( - "-v", - "--verbose", - dest="loglevel", - action="store_const", - const=logging.INFO, - help="Print more output", - ) - parser.add_argument( - "-d", - "--debug", - dest="loglevel", - action="store_const", - const=logging.DEBUG, - default=logging.WARNING, - help="Print even more output", - ) - - parser.add_argument( - "--version", - dest="show_version", - action="store_true", - help="Print version and exit", - ) - - parser.add_argument( - "--registration-warning", - "-w", - dest="registration_warning", - type=int, - default=7200, - help="Raise warning if last line registration time is higher than this threshold (in seconds)", - ) - - parser.add_argument( - "--registration-critical", - "-c", - dest="registration_critical", - type=int, - default=86400, - help="Raise critical if last line registration time is higher than this threshold (in seconds)", - ) - - args = parser.parse_args() - return args - - -class Voip(Resource): - def __init__(self, client): - self.client = client - - def probe(self): - metrics = [] - billing_accounts = self.client.get(f"/telephony") - for billing_account in billing_accounts: - service_names = self.client.get(f"/telephony/{billing_account}/line") - for service_name in service_names: - last_registration = self.client.get( - f"/telephony/{billing_account}/line/{service_name}/lastRegistrations" - )[0] - last_registration_delta = int( - ( - datetime.now(tz=timezone.utc) - - datetime.fromisoformat(last_registration["datetime"]) - ).total_seconds() - ) - metrics.append( - Metric( - f"{service_name} last registration", - last_registration_delta, - context="last_registration", - uom="s", - ) - ) - return metrics - - -class VoipSummary(Summary): - def problem(self, results): - return ", ".join( - [ - f"{result.metric.name} {result.state}: {result.hint}" - for result in results - if str(result.state) != "ok" - ] - ) - - -def main(): - args = parse_arguments() - setup_logging(args) - - if args.show_version: - show_version() - return - - try: - check = Check( - Voip(client=ovh.Client()), - ScalarContext( - "last_registration", - warning=args.registration_warning, - critical=args.registration_critical, - ), - VoipSummary(), - ) - check.main() - except Exception as err: - print(f"Failed to execute check: {str(err)}") - logger.debug(err, exc_info=True) - sys.exit(Unknown.code) - +from check_ovhcloud.voip import main if __name__ == "__main__": main()