Skip to content
Open
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
158 changes: 133 additions & 25 deletions tools/health_check.py
Original file line number Diff line number Diff line change
Expand Up @@ -150,6 +150,36 @@ def check_disk_usage(path: str = "/") -> Tuple[str, str, float]:


def check_memory_usage() -> Tuple[str, str, float]:
"""Check memory usage with cross-platform fallbacks."""
try:
total, available = _get_memory_info()
if total == 0:
return "WARNING", "Memory info unavailable on this platform", 0

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 as e:
return "WARNING", f"Cannot check: {e}", 0


def _get_memory_info() -> Tuple[int, int]:
"""Get total and available memory in bytes, with cross-platform fallbacks.

Returns (total_bytes, available_bytes). Returns (0, 0) if no method works.

Platform support:
- Linux: /proc/meminfo (primary)
- macOS: subprocess vm_stat + sysctl hw.memsize (fallback)
- Any: psutil library if installed (universal fallback)
"""
# Try /proc/meminfo (Linux)
try:
with open("/proc/meminfo") as f:
meminfo = {}
Expand All @@ -162,43 +192,121 @@ def check_memory_usage() -> Tuple[str, str, float]:
meminfo[key] = int(value) * 1024
except ValueError:
pass
total = meminfo.get("MemTotal", 0)
available = meminfo.get("MemAvailable", 0)
if total > 0:
return total, available
except (FileNotFoundError, PermissionError):
pass

# Try psutil (universal, if installed)
try:
import psutil
mem = psutil.virtual_memory()
return mem.total, mem.available
except ImportError:
pass

total = meminfo.get("MemTotal", 0)
available = meminfo.get("MemAvailable", 0)
used = total - available
pct = (used / total) * 100 if total > 0 else 0
# Try macOS vm_stat + sysctl
try:
import subprocess
# Get total memory from sysctl
sysctl_result = subprocess.run(
["sysctl", "-n", "hw.memsize"],
capture_output=True, text=True, timeout=5
)
if sysctl_result.returncode == 0:
total = int(sysctl_result.stdout.strip())

# Get page size and vm_stat info
page_size_result = subprocess.run(
["sysctl", "-n", "vm.pagesize"],
capture_output=True, text=True, timeout=5
)
if page_size_result.returncode == 0:
page_size = int(page_size_result.stdout.strip())

vm_stat = subprocess.run(
["vm_stat"],
capture_output=True, text=True, timeout=5
)
if vm_stat.returncode == 0:
lines = vm_stat.stdout.strip().split("
")
free_pages = 0
inactive_pages = 0
purgeable_pages = 0
for line in lines:
if "Pages free" in line:
free_pages = int(line.split(":")[1].strip().rstrip("."))
elif "Pages inactive" in line:
inactive_pages = int(line.split(":")[1].strip().rstrip("."))
elif "Pages purgeable" in line:
purgeable_pages = int(line.split(":")[1].strip().rstrip("."))

available = (free_pages + inactive_pages + purgeable_pages) * page_size
return total, available
except (FileNotFoundError, subprocess.TimeoutExpired, ValueError, IndexError):
pass

# No method worked
return 0, 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

def check_load_average() -> Tuple[str, str, float]:
"""Check system load average with cross-platform fallbacks."""
try:
load = _get_load_average()
if load is None:
return "WARNING", "Load average unavailable on this platform", 0

cpu_count = os.cpu_count() or 1
load_pct = (load / cpu_count) * 100

if load_pct < 70:
return "OK", f"Load: {load} ({load_pct:.0f}% of {cpu_count} cores)", load
elif load_pct < 90:
return "WARNING", f"Load: {load} ({load_pct:.0f}% of {cpu_count} cores)", load
else:
return "CRITICAL", f"{pct:.1f}% used", pct
return "CRITICAL", f"Load: {load} ({load_pct:.0f}% of {cpu_count} cores)", load
except Exception as e:
return "WARNING", f"Cannot check: {e}", 0


def check_load_average() -> Tuple[str, str, float]:
def _get_load_average() -> Optional[float]:
"""Get 1-minute load average with cross-platform fallbacks.

Returns the 1-minute load average as a float, or None if unavailable.

Platform support:
- Linux: /proc/loadavg (primary)
- Any: os.getloadavg() (fallback, works on most Unix systems)
- Windows: not available (returns None)
"""
# Try /proc/loadavg (Linux)
try:
with open("/proc/loadavg") as f:
parts = f.read().strip().split()
load = float(parts[0])
cpu_count = os.cpu_count() or 1
load_pct = (load / cpu_count) * 100

if load_pct < 70:
return "OK", f"Load: {load} ({load_pct:.0f}% of {cpu_count} cores)", load
elif load_pct < 90:
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
return float(parts[0])
except (FileNotFoundError, PermissionError, IndexError, ValueError):
pass

# Try os.getloadavg() (works on most Unix-like systems including macOS)
try:
load = os.getloadavg()
return load[0] # 1-minute average
except (OSError, AttributeError):
pass

# Try psutil (universal, if installed)
try:
import psutil
load = psutil.getloadavg()
return load[0]
except (ImportError, AttributeError):
pass

# ---------------------------------------------------------------------------
# HEALTH CHECK RUNNER
# ---------------------------------------------------------------------------
return None

def run_health_checks(service: Optional[str] = None, json_output: bool = False) -> Dict[str, Any]:
results: Dict[str, Any] = {
Expand Down