Perf data. (e36d6620) · Commits · shore / My Nagios Plugin Wrapper

mnpw/nagios.py

+61 −4

Original line number	Diff line number	Diff line
		@@ -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,10 +22,67 @@ 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:
		# pylint: disable=invalid-name

tests/test_nagios.py

+19 −11

Original line number	Diff line number	Diff line
		@@ -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)