ram check, configurable loglevel

src/lego_monitoring/__init__.py

@@ -7,6 +7,7 @@ import signal
 from . import checks
 from .alerting import alerts
 from .checks.temp.sensors import print_readings
+from .config import enums as config_enums
 from .config import load_config
 from .core import cvars
 from .core.checkers import interval_checker
@@ -20,8 +21,6 @@ def stop_gracefully(signum, frame):
 
 
 def main() -> None:
-    logging.basicConfig(level=logging.INFO)
-
     asyncio.run(async_main())
 
 
@@ -46,17 +45,20 @@ async def async_main():
     if not args.config:
         raise RuntimeError("--config must be specified in standard operating mode")
 
+    logging.basicConfig(level=config.log_level)
+
     tg_client = await alerts.get_client()
     cvars.tg_client.set(tg_client)
 
+    check_sets = config_enums.CheckSet
+
     checker_sets = {
-        "start": [
-            alerts.send_start_alert(),
-        ],
-        "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))],
+        check_sets.START: [alerts.send_start_alert()],
+        check_sets.STOP: [],  # this is checked later
+        check_sets.CPU: [interval_checker(checks.cpu_check, datetime.timedelta(minutes=5))],
+        check_sets.RAM: [interval_checker(checks.ram_check, datetime.timedelta(minutes=1))],
+        check_sets.TEMP: [interval_checker(checks.temp_check, datetime.timedelta(minutes=5))],
+        check_sets.VULNIX: [interval_checker(checks.vulnix_check, datetime.timedelta(days=3))],
     }
 
     checkers = []

src/lego_monitoring/alerting/enum.py

@@ -5,10 +5,10 @@ class AlertType(StrEnum):
     BOOT = "BOOT"
     CPU = "CPU"
     ERROR = "ERROR"
+    RAM = "RAM"
     TEMP = "TEMP"
     TEST = "TEST"
     VULN = "VULN"
-    # RAM = "RAM"
     # LOGIN = "LOGIN"
     # SMART = "SMART"  # TODO
     # RAID = "RAID"

src/lego_monitoring/checks/__init__.py

@@ -1,3 +1,4 @@
 from .cpu import cpu_check
+from .ram import ram_check
 from .temp import temp_check
 from .vulnix import vulnix_check

src/lego_monitoring/checks/ram.py

@@ -0,0 +1,30 @@
+from psutil import virtual_memory
+
+from lego_monitoring.alerting import alerts
+from lego_monitoring.alerting.enum import AlertType, Severity
+from lego_monitoring.core import cvars
+
+IS_TESTING = False
+
+
+def ram_check() -> list[alerts.Alert]:
+    percentage = virtual_memory().percent
+    config = cvars.config.get().checks.ram
+    if config.critical_percentage and (IS_TESTING or percentage > config.critical_percentage):
+        return [
+            alerts.Alert(
+                alert_type=AlertType.RAM,
+                message=f"RAM usage: {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.RAM,
+                message=f"RAM usage: {percentage:.2f}% > {config.warning_percentage:.2f}%",
+                severity=Severity.WARNING,
+            )
+        ]
+    else:
+        return []

src/lego_monitoring/config/__init__.py

@@ -4,16 +4,19 @@ from typing import Optional
 
 from alt_utils import NestedDeserializableDataclass
 
+from . import enums
 from .checks.cpu import CpuCheckConfig
+from .checks.ram import RamCheckConfig
 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
+    cpu: CpuCheckConfig = field(default_factory=CpuCheckConfig)
+    ram: RamCheckConfig = field(default_factory=RamCheckConfig)
+    temp: TempCheckConfig = field(default_factory=TempCheckConfig)
+    vulnix: Optional[VulnixCheckConfig] = None  # vulnix check WILL raise if this config section is None
 
 
 @dataclass
@@ -26,7 +29,8 @@ class TelegramConfig:
 class Config(NestedDeserializableDataclass):
     checks: ChecksConfig
     telegram: TelegramConfig
-    enabled_check_sets: list[str] = field(default_factory=list)
+    enabled_check_sets: list[enums.CheckSet] = field(default_factory=list)
+    log_level: enums.LogLevelName = enums.LogLevelName.INFO
 
 
 def load_config(filepath: str) -> Config:

src/lego_monitoring/config/checks/ram.py

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

src/lego_monitoring/config/enums.py

@@ -0,0 +1,20 @@
+from enum import StrEnum
+
+
+class CheckSet(StrEnum):
+    START = "start"
+    STOP = "stop"
+
+    CPU = "cpu"
+    RAM = "ram"
+    TEMP = "temp"
+
+    VULNIX = "vulnix"
+
+
+class LogLevelName(StrEnum):
+    CRITICAL = "CRITICAL"
+    ERROR = "ERROR"
+    WARNING = "WARNING"
+    INFO = "INFO"
+    DEBUG = "DEBUG"