Skip to content
Snippets Groups Projects
Commit 5eb9cfe1 authored by nimrod's avatar nimrod
Browse files

Replace the PerfData dataclass with a named tuple.

- Works on Python 3.6.
- The upsides of dataclass don't help in this case.
- Is faster and uses less memory (although in this case it would hard to
  even measure the difference).
parent 9873c3ab
No related branches found
No related tags found
No related merge requests found
...@@ -20,7 +20,6 @@ build-executable: ...@@ -20,7 +20,6 @@ build-executable:
extends: .python3-build extends: .python3-build
before_script: before_script:
- poetry install --no-dev - poetry install --no-dev
- poetry run pip freeze
script: script:
- poetry run make - poetry run make
artifacts: artifacts:
......
...@@ -2,7 +2,7 @@ ...@@ -2,7 +2,7 @@
all: dist/mnpw all: dist/mnpw
dist/mnpw: mnpw/*.py pyproject.toml *.rst LICENSE.txt dist/mnpw: mnpw/*.py pyproject.toml *.rst LICENSE.txt
pyinstaller --noconfirm --onefile "--name=$$(basename $@)" --hidden-import dataclasses mnpw/__init__.py pyinstaller --noconfirm --onefile "--name=$$(basename $@)" mnpw/__init__.py
.PHONY: clean .PHONY: clean
clean: clean:
......
...@@ -5,10 +5,10 @@ https://assets.nagios.com/downloads/nagioscore/docs/nagioscore/3/en/pluginapi.ht ...@@ -5,10 +5,10 @@ https://assets.nagios.com/downloads/nagioscore/docs/nagioscore/3/en/pluginapi.ht
""" """
# pylint: disable=logging-fstring-interpolation # pylint: disable=logging-fstring-interpolation
import collections
import enum import enum
import logging import logging
import subprocess # nosec import subprocess # nosec
from dataclasses import dataclass
DEFAULT_TIMEOUT = 10 # In seconds. DEFAULT_TIMEOUT = 10 # In seconds.
...@@ -22,67 +22,64 @@ class NagiosCode(enum.IntEnum): ...@@ -22,67 +22,64 @@ class NagiosCode(enum.IntEnum):
UNKNOWN = 3 UNKNOWN = 3
@dataclass() PerfData = collections.namedtuple(
class PerfData: "PerfData", ["name", "value", "unit", "warning", "critical", "min", "max"]
"""Performance data. )
PerfData.__new__.__defaults__ = tuple([None] * 7)
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 def parse_perf_data(line): # noqa: MC0001
"""Initialize using a perf data string output from a plugin.""" """Convert a single-line string to a DataPerf tuple."""
if len(line.splitlines()) > 1: if len(line.splitlines()) > 1:
raise RuntimeError("Can only parse a single line at a time.") raise RuntimeError("Can only parse a single line at a time.")
fields = line.split(";") fields = line.split(";")
self.name, self.value = fields[0].split("=") name, value = fields[0].split("=")
# Clean the name, handle a quoted name. # Clean the name, handle a quoted name.
self.name = self.name.strip() name = name.strip()
if self.name[0] in ["'", '"']: if name[0] in ["'", '"']:
self.name = self.name[1:] name = name[1:]
if self.name[-1] in ["'", '"']: if name[-1] in ["'", '"']:
self.name = self.name[:-1] name = name[:-1]
self.name = self.name.strip() name = name.strip()
pd = {"name": name} # pylint: disable=invalid-name
# Split the unit and value. # Split the unit and value.
if not self.value[-1].isdigit(): if not value[-1].isdigit():
for i in range(len(self.value) - 1, 0, -1): for i in range(len(value) - 1, 0, -1):
if self.value[i].isdigit(): if value[i].isdigit():
break break
self.value, self.unit = ( value, unit = (
float(self.value[: i + 1]), float(value[: i + 1]),
self.value[i + 1 :], # noqa: E203 value[i + 1 :], # noqa: E203
) )
pd["value"] = value
pd["unit"] = unit
# Get the remaining fields, if available. # Get the remaining fields, if available.
try: try:
self.warning = float(fields[1]) pd["warning"] = float(fields[1])
except Exception as ex: # pylint: disable=broad-except except Exception as ex: # pylint: disable=broad-except
logging.info(str(ex)) logging.info(str(ex))
try: try:
self.critical = float(fields[2]) pd["critical"] = float(fields[2])
except Exception as ex: # pylint: disable=broad-except except Exception as ex: # pylint: disable=broad-except
logging.info(str(ex)) logging.info(str(ex))
try: try:
self.min = float(fields[3]) pd["min"] = float(fields[3])
except Exception as ex: # pylint: disable=broad-except except Exception as ex: # pylint: disable=broad-except
logging.info(str(ex)) logging.info(str(ex))
try: try:
self.max = float(fields[4]) pd["max"] = float(fields[4])
except Exception as ex: # pylint: disable=broad-except except Exception as ex: # pylint: disable=broad-except
logging.info(str(ex)) logging.info(str(ex))
return PerfData(**pd)
class Check: class Check:
# pylint: disable=invalid-name # pylint: disable=invalid-name
...@@ -121,7 +118,7 @@ class Check: ...@@ -121,7 +118,7 @@ class Check:
if "|" in lines[0]: if "|" in lines[0]:
p1, p2 = lines[0].split("|") p1, p2 = lines[0].split("|")
self.Output = p1.strip() self.Output = p1.strip()
self.PerfData.append(PerfData(p2)) self.PerfData.append(parse_perf_data(p2))
else: else:
self.Output = lines[0].strip() self.Output = lines[0].strip()
...@@ -131,12 +128,12 @@ class Check: ...@@ -131,12 +128,12 @@ class Check:
processing_perfdata = False processing_perfdata = False
for line in lines[1:]: for line in lines[1:]:
if processing_perfdata: if processing_perfdata:
self.PerfData.append(PerfData(line)) self.PerfData.append(parse_perf_data(line))
else: else:
if "|" in line: if "|" in line:
self.AdditionalOutput.append(line.split("|")[0].strip()) self.AdditionalOutput.append(line.split("|")[0].strip())
processing_perfdata = True processing_perfdata = True
self.PerfData.append(PerfData(line.split("|")[1])) self.PerfData.append(parse_perf_data(line.split("|")[1]))
else: else:
self.AdditionalOutput.append(line) self.AdditionalOutput.append(line)
......
...@@ -28,7 +28,6 @@ exclude = [ ...@@ -28,7 +28,6 @@ exclude = [
[tool.poetry.dependencies] [tool.poetry.dependencies]
python = "^3.6.1" python = "^3.6.1"
requests = "^2.25.1" requests = "^2.25.1"
dataclasses = { version = '^0.8', python = '<3.7'}
[tool.poetry.dev-dependencies] [tool.poetry.dev-dependencies]
pre-commit = "^2.13.0" pre-commit = "^2.13.0"
......
...@@ -114,7 +114,7 @@ def test_output_parsing(output): ...@@ -114,7 +114,7 @@ def test_output_parsing(output):
@pytest.mark.parametrize("line", PERF_DATA) @pytest.mark.parametrize("line", PERF_DATA)
def test_data_perf(line): def test_data_perf(line):
"""Test parsing of perfdata.""" """Test parsing of perfdata."""
pd = nagios.PerfData(line) pd = nagios.parse_perf_data(line)
assert isinstance(pd.name, str) assert isinstance(pd.name, str)
assert len(pd.name) > 0 assert len(pd.name) > 0
assert isinstance(pd.value, float) assert isinstance(pd.value, float)
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment