diff --git a/mnpw/nagios.py b/mnpw/nagios.py index 3bde99a65c961a5a923294a548960a149b2fd3c8..06b4e090e8fe073ead3d6049e8cb74945727f9a5 100644 --- a/mnpw/nagios.py +++ b/mnpw/nagios.py @@ -5,10 +5,10 @@ https://assets.nagios.com/downloads/nagioscore/docs/nagioscore/3/en/pluginapi.ht """ # pylint: disable=logging-fstring-interpolation -import collections import enum import logging import subprocess # nosec +from dataclasses import dataclass DEFAULT_TIMEOUT = 10 # In seconds. @@ -22,9 +22,66 @@ class NagiosCode(enum.IntEnum): UNKNOWN = 3 -PerfData = collections.namedtuple( - "PerfData", ["name", "value", "unit", "warning", "critical", "min", "max"] -) +@dataclass() +class PerfData: + """Performance data. + + As published by a Nagios plugin. + """ + + name: str + value: float + unit: str = "" + warning: float = None + critical: float = None + min: float = None + max: float = None + + def __init__(self, line): # noqa: MC0001 + """Initialize using a perf data string output from a plugin.""" + if len(line.splitlines()) > 1: + raise RuntimeError("Can only parse a single line at a time.") + fields = line.split(";") + self.name, self.value = fields[0].split("=") + + # Clean the name, handle a quoted name. + self.name = self.name.strip() + if self.name[0] in ["'", '"']: + self.name = self.name[1:] + if self.name[-1] in ["'", '"']: + self.name = self.name[:-1] + self.name = self.name.strip() + + # Split the unit and value. + if not self.value[-1].isdigit(): + for i in range(len(self.value) - 1, 0, -1): + if self.value[i].isdigit(): + break + self.value, self.unit = ( + float(self.value[: i + 1]), + self.value[i + 1 :], # noqa: E203 + ) + + # Get the remaining fields, if available. + try: + self.warning = float(fields[1]) + except Exception as ex: # pylint: disable=broad-except + logging.info(str(ex)) + + try: + self.critical = float(fields[2]) + except Exception as ex: # pylint: disable=broad-except + logging.info(str(ex)) + + try: + self.min = float(fields[3]) + except Exception as ex: # pylint: disable=broad-except + logging.info(str(ex)) + + try: + self.max = float(fields[4]) + except Exception as ex: # pylint: disable=broad-except + logging.info(str(ex)) class Check: diff --git a/tests/test_nagios.py b/tests/test_nagios.py index 4a7e1c58e5540cb7a12cc1c960a3996d05588042..98843901a96aba0cc0a751ef0079642a53c70469 100644 --- a/tests/test_nagios.py +++ b/tests/test_nagios.py @@ -21,17 +21,19 @@ DISK OK - free space: / 3326 MB (56%); | /=2643MB;5948;5958;0;5968 # Lifted from the Alignak project: # https://github.com/Alignak-monitoring/alignak/blob/fcc3d6499478ce67f4b91c694111f45cc8282535/tests/test_perfdata_parsing.py -PERF_DATA_1 = """ramused=90%;85;95;;""" -PERF_DATA_2 = """ramused=1009MB;;;0;1982 """ -PERF_DATA_3 = """memused=1550MB;2973;3964;0;5810""" -PERF_DATA_4 = """swapused=540MB;;;; """ -PERF_DATA_5 = """memused=90%""" -PERF_DATA_6 = """'Physical Memory Used'=12085620736Bytes; """ -PERF_DATA_7 = """Physical Memory Utilisation'=94%;80;90;""" -PERF_DATA_8 = """'C: used'=13.06452GB;22.28832;25.2601;0;29.71777 """ -PERF_DATA_9 = """'C: used %'=44%;75;85;0;100""" -PERF_DATA_10 = """time_offset-192.168.0.1=-7.22636468709e-05s;1;2;0;;""" -PERF_DATA_11 = """àéèï-192.168.0.1=-7.22636468709e-05s;1;2;0;;""" +PERF_DATA = [ + """ramused=90%;85;95;;""", + """ramused=1009MB;;;0;1982 """, + """memused=1550MB;2973;3964;0;5810""", + """swapused=540MB;;;; """, + """memused=90%""", + """'Physical Memory Used'=12085620736Bytes; """, + """Physical Memory Utilisation'=94%;80;90;""", + """'C: used'=13.06452GB;22.28832;25.2601;0;29.71777 """, + """'C: used %'=44%;75;85;0;100""", + """time_offset-192.168.0.1=-7.22636468709e-05s;1;2;0;;""", + """àéèï-192.168.0.1=-7.22636468709e-05s;1;2;0;;""", +] @pytest.mark.parametrize( @@ -106,3 +108,9 @@ def test_output_parsing(output): with _mock_run(return_value=proc): check.run() # TODO: validate something. + + +@pytest.mark.parametrize("line", PERF_DATA) +def test_data_perf(line): + """Test parsing of perfdata.""" + nagios.PerfData(line)