Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
115 changes: 109 additions & 6 deletions tools/health_check.py
Original file line number Diff line number Diff line change
Expand Up @@ -149,7 +149,7 @@ def check_disk_usage(path: str = "/") -> Tuple[str, str, float]:
return "WARNING", f"Cannot check: {e}", 0


def check_memory_usage() -> Tuple[str, str, float]:
def _check_memory_linux() -> Optional[Tuple[str, str, float]]:
try:
with open("/proc/meminfo") as f:
meminfo = {}
Expand All @@ -174,11 +174,86 @@ def check_memory_usage() -> Tuple[str, str, float]:
return "WARNING", f"{pct:.1f}% used", pct
else:
return "CRITICAL", f"{pct:.1f}% used", pct
except Exception as e:
return "WARNING", f"Cannot check: {e}", 0
except Exception:
return None


def check_load_average() -> Tuple[str, str, float]:
def _check_memory_psutil() -> Optional[Tuple[str, str, float]]:
try:
import psutil
mem = psutil.virtual_memory()
pct = mem.percent
used_gb = mem.used / (1024 ** 3)
total_gb = mem.total / (1024 ** 3)

if pct < MEMORY_THRESHOLD_WARNING:
return "OK", f"{pct:.1f}% used ({used_gb:.1f}GB/{total_gb:.1f}GB)", pct
elif pct < MEMORY_THRESHOLD_CRITICAL:
return "WARNING", f"{pct:.1f}% used", pct
else:
return "CRITICAL", f"{pct:.1f}% used", pct
except Exception:
return None


def _check_memory_platform() -> Tuple[str, str, float]:
try:
import subprocess
result = subprocess.run(
["sysctl", "-n", "hw.memsize"],
capture_output=True, text=True, timeout=5
)
total = int(result.stdout.strip())

result = subprocess.run(
["vm_stat"],
capture_output=True, text=True, timeout=5
)
lines = result.stdout.strip().split("\n")
page_size = 4096
for line in lines:
if "page size of" in line:
page_size = int(line.split()[-2])
break

free_pages = 0
inactive_pages = 0
speculative_pages = 0
for line in lines[1:]:
if line.startswith("Pages free"):
free_pages = int(line.split()[-1].rstrip("."))
elif line.startswith("Pages inactive"):
inactive_pages = int(line.split()[-1].rstrip("."))
elif line.startswith("Pages speculative"):
speculative_pages = int(line.split()[-1].rstrip("."))

available = (free_pages + inactive_pages + speculative_pages) * page_size
used = total - available
pct = (used / total) * 100 if total > 0 else 0

if pct < MEMORY_THRESHOLD_WARNING:
return "OK", f"{pct:.1f}% used ({used // (1024**3)}GB/{total // (1024**3)}GB)", pct
elif pct < MEMORY_THRESHOLD_CRITICAL:
return "WARNING", f"{pct:.1f}% used", pct
else:
return "CRITICAL", f"{pct:.1f}% used", pct
except Exception:
return "WARNING", "Cannot check memory: no fallback available", 0


def check_memory_usage() -> Tuple[str, str, float]:
result = _check_memory_linux()
if result is not None:
return result

result = _check_memory_psutil()
if result is not None:
return result

return _check_memory_platform()


def _check_load_linux() -> Optional[Tuple[str, str, float]]:
try:
with open("/proc/loadavg") as f:
parts = f.read().strip().split()
Expand All @@ -192,8 +267,36 @@ def check_load_average() -> Tuple[str, str, float]:
return "WARNING", f"Load: {load} ({load_pct:.0f}% of {cpu_count} cores)", load
else:
return "CRITICAL", f"Load: {load} ({load_pct:.0f}% of {cpu_count} cores)", load
except Exception as e:
return "WARNING", f"Cannot check: {e}", 0
except Exception:
return None


def _check_load_getloadavg() -> Optional[Tuple[str, str, float]]:
try:
load1, load5, load15 = os.getloadavg()
cpu_count = os.cpu_count() or 1
load_pct = (load1 / cpu_count) * 100

if load_pct < 70:
return "OK", f"Load: {load1} ({load_pct:.0f}% of {cpu_count} cores)", load1
elif load_pct < 90:
return "WARNING", f"Load: {load1} ({load_pct:.0f}% of {cpu_count} cores)", load1
else:
return "CRITICAL", f"Load: {load1} ({load_pct:.0f}% of {cpu_count} cores)", load1
except Exception:
return None


def check_load_average() -> Tuple[str, str, float]:
result = _check_load_linux()
if result is not None:
return result

result = _check_load_getloadavg()
if result is not None:
return result

return "WARNING", "Cannot check load average: no fallback available", 0


# ---------------------------------------------------------------------------
Expand Down
150 changes: 150 additions & 0 deletions tools/test_health_check.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,150 @@
#!/usr/bin/env python3
"""Tests for cross-platform health check fallbacks."""

import os
import sys
import unittest
from unittest.mock import patch, mock_open

sys.path.insert(0, os.path.dirname(__file__))

from health_check import (
check_memory_usage,
check_load_average,
_check_memory_linux,
_check_load_linux,
MEMORY_THRESHOLD_WARNING,
MEMORY_THRESHOLD_CRITICAL,
)


class TestCheckMemoryLinux(unittest.TestCase):
def test_linux_meminfo_parses_correctly(self):
meminfo_content = (
"MemTotal: 16384000 kB\n"
"MemAvailable: 8192000 kB\n"
"MemFree: 4096000 kB\n"
)
with patch("builtins.open", mock_open(read_data=meminfo_content)):
result = _check_memory_linux()
self.assertIsNotNone(result)
status, detail, pct = result
self.assertEqual(status, "OK")
self.assertIn("50.0% used", detail)

def test_linux_meminfo_critical(self):
meminfo_content = (
"MemTotal: 16384000 kB\n"
"MemAvailable: 819200 kB\n"
)
with patch("builtins.open", mock_open(read_data=meminfo_content)):
result = _check_memory_linux()
self.assertIsNotNone(result)
status, detail, pct = result
self.assertEqual(status, "CRITICAL")

def test_linux_meminfo_file_missing(self):
with patch("builtins.open", side_effect=FileNotFoundError):
result = _check_memory_linux()
self.assertIsNone(result)

def test_fallback_when_proc_missing(self):
with patch("builtins.open", side_effect=FileNotFoundError):
result = check_memory_usage()
self.assertIsNotNone(result)
status, detail, pct = result
self.assertIn(status, ("OK", "WARNING", "CRITICAL"))


class TestCheckLoadLinux(unittest.TestCase):
def test_linux_loadavg_parses_correctly(self):
loadavg_content = "1.50 2.00 3.00 1/400 12345\n"
with patch("builtins.open", mock_open(read_data=loadavg_content)):
result = _check_load_linux()
self.assertIsNotNone(result)
status, detail, load = result
self.assertEqual(status, "OK")
self.assertIn("Load: 1.5", detail)

def test_linux_loadavg_critical(self):
loadavg_content = "16.00 15.00 14.00 1/400 12345\n"
with patch("builtins.open", mock_open(read_data=loadavg_content)):
result = _check_load_linux()
self.assertIsNotNone(result)
status, detail, load = result
self.assertEqual(status, "CRITICAL")

def test_linux_loadavg_file_missing(self):
with patch("builtins.open", side_effect=FileNotFoundError):
result = _check_load_linux()
self.assertIsNone(result)

def test_fallback_when_proc_missing(self):
with patch("builtins.open", side_effect=FileNotFoundError):
with patch("os.getloadavg", return_value=(2.0, 3.0, 4.0)):
result = check_load_average()
self.assertIsNotNone(result)
status, detail, load = result
self.assertIn(status, ("OK", "WARNING", "CRITICAL"))
self.assertIn("Load: 2.0", detail)


class TestCheckMemoryPsutil(unittest.TestCase):
def test_psutil_fallback(self):
mock_mem = unittest.mock.MagicMock()
mock_mem.percent = 45.0
mock_mem.used = 7 * 1024 ** 3
mock_mem.total = 16 * 1024 ** 3

with patch("builtins.open", side_effect=FileNotFoundError):
with patch.dict("sys.modules", {"psutil": unittest.mock.MagicMock(psutil=unittest.mock.MagicMock(virtual_memory=mock_mem))}):
import importlib
import health_check
original = health_check._check_memory_psutil
health_check._check_memory_psutil = lambda: ("OK", "45.0% used (7.0GB/16.0GB)", 45.0)
try:
result = check_memory_usage()
self.assertEqual(result[0], "OK")
finally:
health_check._check_memory_psutil = original


class TestCheckLoadGetloadavg(unittest.TestCase):
def test_getloadavg_fallback(self):
with patch("builtins.open", side_effect=FileNotFoundError):
with patch("os.getloadavg", return_value=(1.0, 2.0, 3.0)):
result = check_load_average()
self.assertIsNotNone(result)
status, detail, load = result
self.assertIn(status, ("OK", "WARNING", "CRITICAL"))
self.assertIn("Load: 1.0", detail)

def test_getloadavg_critical(self):
with patch("builtins.open", side_effect=FileNotFoundError):
with patch("os.getloadavg", return_value=(20.0, 18.0, 16.0)):
result = check_load_average()
self.assertEqual(result[0], "CRITICAL")


class TestBackwardCompatibility(unittest.TestCase):
def test_json_output_format(self):
meminfo_content = (
"MemTotal: 16384000 kB\n"
"MemAvailable: 8192000 kB\n"
)
with patch("builtins.open", mock_open(read_data=meminfo_content)):
status, detail, pct = check_memory_usage()
self.assertIsInstance(pct, float)
self.assertGreaterEqual(pct, 0)
self.assertLessEqual(pct, 100)

def test_load_returns_float(self):
loadavg_content = "1.50 2.00 3.00 1/400 12345\n"
with patch("builtins.open", mock_open(read_data=loadavg_content)):
status, detail, load = check_load_average()
self.assertIsInstance(load, float)
self.assertGreaterEqual(load, 0)


if __name__ == "__main__":
unittest.main()