add cpu check

src/lego_monitoring/__init__.py

@@ -3,7 +3,6 @@ import asyncio
 import datetime
 import logging
 import signal
-import time
 
 from . import checks
 from .alerting import alerts
@@ -57,6 +56,7 @@ async def async_main():
         "stop": [],  # this is checked later
         "temp": [interval_checker(checks.temp_check, datetime.timedelta(minutes=5))],
         "vulnix": [interval_checker(checks.vulnix_check, datetime.timedelta(days=3))],
+        "cpu": [interval_checker(checks.cpu_check, datetime.timedelta(minutes=5))],
     }
 
     checkers = []

src/lego_monitoring/alerting/enum.py

@@ -3,12 +3,12 @@ from enum import StrEnum
 
 class AlertType(StrEnum):
     BOOT = "BOOT"
+    CPU = "CPU"
+    ERROR = "ERROR"
     TEMP = "TEMP"
     TEST = "TEST"
     VULN = "VULN"
-    ERROR = "ERROR"
     # RAM = "RAM"
-    # CPU = "CPU"
     # LOGIN = "LOGIN"
     # SMART = "SMART"  # TODO
     # RAID = "RAID"

src/lego_monitoring/checks/__init__.py

@@ -1,2 +1,3 @@
+from .cpu import cpu_check
 from .temp import temp_check
 from .vulnix import vulnix_check

src/lego_monitoring/checks/cpu.py

@@ -0,0 +1,30 @@
+from psutil import cpu_percent
+
+from lego_monitoring.alerting import alerts
+from lego_monitoring.alerting.enum import AlertType, Severity
+from lego_monitoring.core import cvars
+
+IS_TESTING = False
+
+
+def cpu_check() -> list[alerts.Alert]:
+    percentage = cpu_percent()
+    config = cvars.config.get().checks.cpu
+    if config.critical_percentage and (IS_TESTING or percentage > config.critical_percentage):
+        return [
+            alerts.Alert(
+                alert_type=AlertType.CPU,
+                message=f"CPU load: {percentage:.2f}% > {config.critical_percentage:.2f}%",
+                severity=Severity.CRITICAL,
+            )
+        ]
+    elif config.warning_percentage and (IS_TESTING or percentage > config.warning_percentage):
+        return [
+            alerts.Alert(
+                alert_type=AlertType.CPU,
+                message=f"CPU load: {percentage:.2f}% > {config.warning_percentage:.2f}%",
+                severity=Severity.WARNING,
+            )
+        ]
+    else:
+        return []

src/lego_monitoring/config/__init__.py

@@ -1,15 +1,17 @@
 import json
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from typing import Optional
 
 from alt_utils import NestedDeserializableDataclass
 
+from .checks.cpu import CpuCheckConfig
 from .checks.temp import TempCheckConfig
 from .checks.vulnix import VulnixCheckConfig
 
 
 @dataclass
 class ChecksConfig(NestedDeserializableDataclass):
+    cpu: Optional[CpuCheckConfig] = None
     temp: Optional[TempCheckConfig] = None
     vulnix: Optional[VulnixCheckConfig] = None
 
@@ -22,9 +24,9 @@ class TelegramConfig:
 
 @dataclass
 class Config(NestedDeserializableDataclass):
-    enabled_check_sets: list[str]
     checks: ChecksConfig
     telegram: TelegramConfig
+    enabled_check_sets: list[str] = field(default_factory=list)
 
 
 def load_config(filepath: str) -> Config:

src/lego_monitoring/config/checks/cpu.py

@@ -0,0 +1,8 @@
+from dataclasses import dataclass
+from typing import Optional
+
+
+@dataclass
+class CpuCheckConfig:
+    warning_percentage: Optional[float] = 80
+    critical_percentage: Optional[float] = 90

src/lego_monitoring/config/checks/temp.py

@@ -1,4 +1,4 @@
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from typing import Optional
 
 from alt_utils import NestedDeserializableDataclass
@@ -6,19 +6,19 @@ from alt_utils import NestedDeserializableDataclass
 
 @dataclass
 class TempReadingConfig:
-    label: Optional[str]
-    enabled: bool
-    warning_temp: Optional[float]
-    critical_temp: Optional[float]
+    label: Optional[str] = None
+    enabled: bool = True
+    warning_temp: Optional[float] = None
+    critical_temp: Optional[float] = None
 
 
 @dataclass
 class TempSensorConfig(NestedDeserializableDataclass):
-    name: Optional[str]
-    enabled: bool
-    readings: dict[str, TempReadingConfig]
+    name: Optional[str] = None
+    enabled: bool = True
+    readings: dict[str, TempReadingConfig] = field(default_factory=dict)
 
 
 @dataclass
 class TempCheckConfig(NestedDeserializableDataclass):
-    sensors: dict[str, TempSensorConfig]
+    sensors: dict[str, TempSensorConfig] = field(default_factory=dict)