metrics py module

This commit is contained in:
Jörn-Michael Miehe 2023-08-31 11:17:19 +00:00
parent 63dc67f84e
commit 06f8ba1316
6 changed files with 97 additions and 84 deletions

View file

@ -1,4 +1,4 @@
from .metrics import cpu_metric, disk_metric from . import metrics
from .settings import SETTINGS from .settings import SETTINGS
@ -7,12 +7,12 @@ def main() -> None:
print(SETTINGS.model_dump()) print(SETTINGS.model_dump())
# CPU metric # CPU metric
print(cpu_metric()) print(metrics.cpu())
# MEM metric # MEM metric
# DISK metric # DISK metric
print(disk_metric()) print(metrics.disk())
if __name__ == "__main__": if __name__ == "__main__":

View file

@ -1,81 +0,0 @@
import os
from dataclasses import dataclass, field
import psutil
from .settings import SETTINGS, MetricSettings
@dataclass(slots=True, frozen=True)
class Report:
result: str
failed: bool = field(default=False, kw_only=True)
def _report(
*,
settings: MetricSettings,
name: str,
value: float,
) -> Report:
result = settings.report.format(name=name, value=value)
if (
value > settings.threshold and not settings.inverted
or value < settings.threshold and settings.inverted
):
return Report(result, failed=True)
else:
return Report(result)
def cpu_metric() -> Report | None:
if not SETTINGS.cpu.enabled:
return None
value = psutil.cpu_percent(interval=1)
return _report(
settings=SETTINGS.cpu,
name=SETTINGS.cpu.name,
value=value,
)
def disk_metric() -> Report | None:
if not SETTINGS.disk.enabled:
return None
def path_to_free_percent(path: os.PathLike) -> float:
try:
sv = os.statvfs(path)
return sv.f_bavail / sv.f_blocks * 100
except ZeroDivisionError:
return 0
data = sorted([
(str(path), path_to_free_percent(path))
for path in SETTINGS.disk.paths
], key=lambda d: d[1])
reports = [_report(
settings=SETTINGS.disk,
name=path,
value=percent,
) for path, percent in data]
report_inner = ", ".join(
report.result
for report in reports[:SETTINGS.disk.count]
)
return Report(
SETTINGS.disk.report_outer.format(
name=SETTINGS.disk.name,
inner=report_inner,
),
failed=any(
report.failed
for report in reports
),
)

View file

@ -0,0 +1,7 @@
from .cpu import cpu
from .disk import disk
__all__ = [
"cpu",
"disk",
]

View file

@ -0,0 +1,28 @@
from dataclasses import dataclass, field
from typing import Self
from ..settings import MetricSettings
@dataclass(slots=True, frozen=True)
class Report:
result: str
failed: bool = field(default=False, kw_only=True)
@classmethod
def new(
cls, *,
settings: MetricSettings,
name: str,
value: float,
) -> Self:
result = settings.report.format(name=name, value=value)
if (
value > settings.threshold and not settings.inverted
or value < settings.threshold and settings.inverted
):
return cls(result, failed=True)
else:
return cls(result)

View file

@ -0,0 +1,16 @@
import psutil
from ..settings import SETTINGS
from ._report import Report
def cpu() -> Report | None:
if not SETTINGS.cpu.enabled:
return None
value = psutil.cpu_percent(interval=1)
return Report.new(
settings=SETTINGS.cpu,
name=SETTINGS.cpu.name,
value=value,
)

View file

@ -0,0 +1,43 @@
import os
from ..settings import SETTINGS
from ._report import Report
def disk() -> Report | None:
if not SETTINGS.disk.enabled:
return None
def path_to_free_percent(path: os.PathLike) -> float:
try:
sv = os.statvfs(path)
return sv.f_bavail / sv.f_blocks * 100
except ZeroDivisionError:
return 0
data = sorted([
(str(path), path_to_free_percent(path))
for path in SETTINGS.disk.paths
], key=lambda d: d[1])
reports = [Report.new(
settings=SETTINGS.disk,
name=path,
value=percent,
) for path, percent in data]
report_inner = ", ".join(
report.result
for report in reports[:SETTINGS.disk.count]
)
return Report(
SETTINGS.disk.report_outer.format(
name=SETTINGS.disk.name,
inner=report_inner,
),
failed=any(
report.failed
for report in reports
),
)