diff --git a/kiwi_simple_metrics/metrics/_report.py b/kiwi_simple_metrics/metrics/_report.py index 38101cb..cf9a394 100644 --- a/kiwi_simple_metrics/metrics/_report.py +++ b/kiwi_simple_metrics/metrics/_report.py @@ -4,25 +4,56 @@ from typing import Self from ..settings import MetricSettings +@dataclass(slots=True, kw_only=True) +class ReportData: + name: str + value: float + + @classmethod + def from_free_total(cls, *, name: str, free: float, total: float) -> Self: + return cls( + name=name, + value=(total - free) / total * 100, + ) + + def report(self, settings: MetricSettings) -> "Report": + result = settings.report.format( + name=self.name, + value=self.value, + ) + + return Report(result, failed=( + self.value > settings.threshold and not settings.inverted + or self.value < settings.threshold and settings.inverted + )) + + @dataclass(slots=True, frozen=True) class Report: result: str failed: bool = field(default=False, kw_only=True) @classmethod - def new( + def aggregate( cls, *, settings: MetricSettings, - name: str, - value: float, + data: list[ReportData], ) -> Self: - result = settings.report.format(name=name, value=value) + reports = [ + data.report(settings) + for data in data + ] - if ( - value > settings.threshold and not settings.inverted - or value < settings.threshold and settings.inverted - ): - return cls(result, failed=True) - - else: - return cls(result) + return cls( + settings.report_outer.format( + name=settings.name, + inner=", ".join( + report.result + for report in reports[:settings.count] + ), + ), + failed=any( + report.failed + for report in reports + ), + ) diff --git a/kiwi_simple_metrics/metrics/cpu.py b/kiwi_simple_metrics/metrics/cpu.py index 8ab4a73..b44f011 100644 --- a/kiwi_simple_metrics/metrics/cpu.py +++ b/kiwi_simple_metrics/metrics/cpu.py @@ -1,16 +1,20 @@ import psutil from ..settings import SETTINGS -from ._report import Report +from ._report import Report, ReportData + + +def _hwdata() -> ReportData: + return ReportData( + name=SETTINGS.cpu.name, + value=psutil.cpu_percent(interval=1), + ) 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, - ) + data = _hwdata() + + return data.report(SETTINGS.cpu) diff --git a/kiwi_simple_metrics/metrics/disk.py b/kiwi_simple_metrics/metrics/disk.py index c30ee0c..a87b2b2 100644 --- a/kiwi_simple_metrics/metrics/disk.py +++ b/kiwi_simple_metrics/metrics/disk.py @@ -1,43 +1,32 @@ import os from ..settings import SETTINGS -from ._report import Report +from ._report import Report, ReportData + + +def _hwdata() -> list[ReportData]: + def get_path_statvfs(path: os.PathLike) -> dict[str, float]: + sv = os.statvfs(path) + return { + "free": sv.f_bavail, + "total": sv.f_blocks, + } + + return sorted([ + ReportData.from_free_total( + name=str(path), + **get_path_statvfs(path), + ) for path in SETTINGS.disk.paths + ], key=lambda d: d.value, reverse=True) def disk() -> Report | None: if not SETTINGS.disk.enabled: return None - def path_to_used_percent(path: os.PathLike) -> float: - try: - sv = os.statvfs(path) - return (1 - sv.f_bavail / sv.f_blocks) * 100 - except ZeroDivisionError: - return 0 + data = _hwdata() - data = sorted([ - (str(path), path_to_used_percent(path)) - for path in SETTINGS.disk.paths - ], key=lambda d: d[1], reverse=True) - - reports = [Report.new( + return Report.aggregate( 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 - ), + data=data, ) diff --git a/kiwi_simple_metrics/metrics/memory.py b/kiwi_simple_metrics/metrics/memory.py index c23cd4f..0dd60d2 100644 --- a/kiwi_simple_metrics/metrics/memory.py +++ b/kiwi_simple_metrics/metrics/memory.py @@ -1,56 +1,43 @@ import psutil from ..settings import SETTINGS -from ._report import Report +from ._report import Report, ReportData + + +def _hwdata() -> list[ReportData]: + vmem = psutil.virtual_memory() + swap = psutil.swap_memory() + + if SETTINGS.memory.swap == "exclude": + return [ReportData( + name=SETTINGS.memory.name_ram, + value=vmem.percent, + )] + + elif SETTINGS.memory.swap == "combine": + return [ReportData.from_free_total( + name=SETTINGS.memory.name, + free=vmem.available + swap.free, + total=vmem.total + swap.total, + )] + + else: # SETTINGS.memory.swap == "include" + return [ReportData( + name=SETTINGS.memory.name_ram, + value=vmem.percent, + ), ReportData( + name=SETTINGS.memory.name_swap, + value=swap.percent, + )] def memory() -> Report | None: if not SETTINGS.memory.enabled: return None - def get_used_percent(free: float, total: float) -> float: - return (total - free) / total * 100 + data = _hwdata() - vmem = psutil.virtual_memory() - swap = psutil.swap_memory() - - if SETTINGS.memory.swap == "exclude": - data = { - SETTINGS.memory.name_ram: vmem.percent, - } - - elif SETTINGS.memory.swap == "combine": - data = { - SETTINGS.memory.name: get_used_percent( - vmem.available + swap.free, - vmem.total + swap.total, - ) - } - - else: # SETTINGS.memory.swap == "include" - data = { - SETTINGS.memory.name_ram: vmem.percent, - SETTINGS.memory.name_swap: swap.percent, - } - - reports = [Report.new( + return Report.aggregate( settings=SETTINGS.memory, - name=name, - value=value, - ) for name, value in data.items()] - - report_inner = ", ".join( - report.result - for report in reports - ) - - return Report( - SETTINGS.memory.report_outer.format( - name=SETTINGS.memory.name, - inner=report_inner, - ), - failed=any( - report.failed - for report in reports - ), + data=data, ) diff --git a/kiwi_simple_metrics/settings.py b/kiwi_simple_metrics/settings.py index 81dd5e4..18544d9 100644 --- a/kiwi_simple_metrics/settings.py +++ b/kiwi_simple_metrics/settings.py @@ -12,27 +12,28 @@ class MetricSettings(BaseModel): # metric will be reported enabled: bool = True - # format string to report the metric - report: str = "{name}: {value:.2f}%" - # if the metric value exceeds this percentage, the report fails threshold: float # if True, this metric fails when the value falls below the `threshold` inverted: bool = False + # per-value format string for reporting + report: str = "{name}: {value:.2f}%" + + # per-metric format string for reporting + report_outer: str = "{name}: [{inner}]" + + # include only `count` many items (None: include all) + count: int | None = None + class CpuMS(MetricSettings): name: str = "CPU" threshold: float = math.inf -class MultiMS(MetricSettings): - # outer format string for reporting - report_outer: str = "{name}: [{inner}]" - - -class MemoryMS(MultiMS): +class MemoryMS(MetricSettings): name: str = "Memory" threshold: float = 90 report_outer: str = "{inner}" @@ -48,16 +49,14 @@ class MemoryMS(MultiMS): name_swap: str = "Swap" -class DiskMS(MultiMS): +class DiskMS(MetricSettings): name: str = "Disk Used" threshold: float = 85 + count: int = 1 # paths to check for disk space paths: list[DirectoryPath] = Field(default_factory=list) - # include only `count` many of the paths with the least free space - count: int = 1 - class Settings(BaseSettings): model_config = SettingsConfigDict(