GPU Health Checks + Auto-Remediation

Goal: a single health_check() function that returns a structured report:

{
    'status': 'healthy' | 'degraded' | 'critical',
    'gpu_id': 0,
    'checks': {'memory_ok': True, 'temperature_ok': True, ...},
    'issues': ['rogue process holding 2048 MiB', ...],
    'memory_used_mib': 734, 'memory_pct': 9.3, 'temperature_c': 52,
    ...
}

Each individual *_ok check is a boolean. The aggregate status is critical if any hard-fail check trips, degraded for soft warnings, healthy otherwise. This separation matters — your alert rules and your dashboards want different views of the same data.

The two kinds of checks you'll combine:

1. Threshold checks — memory %, temperature, ECC errors, Xid events. The classic production alarms.
2. Rogue-process detection — any compute process not on your protected allowlist is suspicious. On a dedicated training node, only your training PID should hold a CUDA context. Anything else — orphaned Jupyter kernel, leaked Ray worker, someone's ad-hoc debug session — is an anomaly regardless of how much VRAM it holds. This catches the 2 GB leak on an 80 GB H100 where the threshold wouldn't trigger.

Thresholds used below (tuned for H100-class production — an 80% VRAM trip won't fire on a 2 GB rogue, which is why the rogue-process check matters):

Thresholds — production-realistic alerting levels

We define thresholds once at the top so the whole probe reads declaratively. These are the values we'd send to PagerDuty in production:

• Memory 80% warn / 90% critical — below 80% is normal under load.
• Temp 80°C warn / 90°C critical — H100 thermal limit is ~87°C.
• Xid events — a small set of truly fatal codes (13, 43, 44, 48, 62, 79). Others are informational.
• Throttle mask — only real throttling (power cap, thermal slowdown) counts; an idle GPU reporting IDLE is not a problem.
• Rogue PID — any non-protected CUDA process holding ≥1 GiB is treated as critical (hoarding leak).

import os, pynvml

pynvml.nvmlInit()
handle = pynvml.nvmlDeviceGetHandleByIndex(0)

# Production-realistic thresholds
MEM_WARN_PCT, MEM_CRIT_PCT = 80.0, 90.0
TEMP_WARN_C, TEMP_CRIT_C = 80, 90
ROGUE_CRIT_MIB = 1024  # a single non-protected PID holding >1 GB is critical
FATAL_XIDS = {13, 43, 44, 48, 62, 79}

# Only these throttle bits mean 'something is wrong'. IDLE (0x1), APPLICATIONS_CLOCKS,
# SYNC_BOOST, DISPLAY_CLOCK are normal operating reasons. An idle GPU reports 0x1 — not
# a problem, don't alert on it.
REAL_THROTTLE_MASK = (
    0x4    # SW_POWER_CAP
    | 0x8  # HW_SLOWDOWN
    | 0x20 # SW_THERMAL_SLOWDOWN
    | 0x40 # HW_THERMAL_SLOWDOWN
    | 0x80 # HW_POWER_BRAKE_SLOWDOWN
)

# Status priority — always take the worst finding across all checks.
_STATUS_RANK = {'healthy': 0, 'degraded': 1, 'critical': 2}
def _worst(a, b):
    return a if _STATUS_RANK[a] >= _STATUS_RANK[b] else b

Reading GPU metrics via NVML

Why NVML (Python binding to NVIDIA Management Library) and not nvidia-smi --query-gpu=...?

1. Structured — you get a dict back, not parsed CSV.
2. No subprocess cost — a 30ms call vs 200ms for nvidia-smi.
3. Access to events — Xid errors, ECC counters, clock throttle reasons aren't exposed by the CLI at all.

Notes on edge cases:

• ECC may not be supported on consumer cards — we catch NVMLError and treat as zero.
• bad_throttle ANDs the mask, so IDLE doesn't fire a false alarm.

def _read_metrics():
    mem = pynvml.nvmlDeviceGetMemoryInfo(handle)
    temp = pynvml.nvmlDeviceGetTemperature(handle, pynvml.NVML_TEMPERATURE_GPU)
    power_w = pynvml.nvmlDeviceGetPowerUsage(handle) / 1000.0
    power_limit_w = pynvml.nvmlDeviceGetEnforcedPowerLimit(handle) / 1000.0
    try:
        ecc_dbe = pynvml.nvmlDeviceGetTotalEccErrors(
            handle,
            pynvml.NVML_MEMORY_ERROR_TYPE_UNCORRECTED,
            pynvml.NVML_VOLATILE_ECC,
        )
    except pynvml.NVMLError:
        ecc_dbe = 0  # not all GPUs support ECC — treat as zero errors
    throttle = pynvml.nvmlDeviceGetCurrentClocksThrottleReasons(handle)
    return {
        'memory_used_mib': mem.used // (1024 * 1024),
        'memory_total_mib': mem.total // (1024 * 1024),
        'memory_pct': 100.0 * mem.used / mem.total,
        'temperature_c': temp,
        'power_w': power_w,
        'power_limit_w': power_limit_w,
        'ecc_dbe': int(ecc_dbe),
        'throttle_mask': int(throttle),
        'bad_throttle': int(throttle) & REAL_THROTTLE_MASK,
    }

def _read_xid_events():
    # Xid events are kernel-level GPU fault codes. Some (like 13, 43) indicate
    # memory errors bad enough that you should drain the node before the next job.
    try:
        event_set = pynvml.nvmlEventSetCreate()
        pynvml.nvmlDeviceRegisterEvents(
            handle, pynvml.nvmlEventTypeXidCriticalError, event_set
        )
        events = []
        for _ in range(10):
            try:
                data = pynvml.nvmlEventSetWait_v2(event_set, 10)
                events.append(int(data.eventData))
            except pynvml.NVMLError_Timeout:
                break
        pynvml.nvmlEventSetFree(event_set)
        return events
    except pynvml.NVMLError:
        return []

Rogue process detection + protected PIDs

The JupyterLab kernel you're running right now is a CUDA process. If our check naively flagged every CUDA PID as rogue, it would kill itself.

DEFAULT_PROTECTED holds the PIDs we must never treat as rogue: this notebook's own process, its parent (the Jupyter server), and PID 1 (the container init). In a real watchdog you'd also add the training job's parent supervisor PID.

DEFAULT_PROTECTED = {os.getpid(), os.getppid(), 1}

def _list_gpu_processes(protected):
    """Return non-protected compute processes as [{pid, used_mib}, ...].
    NVML sometimes reports usedGpuMemory=None inside a container — we keep the
    PID and treat unknown memory as 0 (the PID is still an anomaly)."""
    raw = pynvml.nvmlDeviceGetComputeRunningProcesses(handle)
    return [
        {'pid': p.pid, 'used_mib': (p.usedGpuMemory or 0) // (1024 * 1024)}
        for p in raw if p.pid not in protected
    ]

The main health_check — compose the findings

Now assemble everything: read the metrics, read Xid events, list non-protected processes, run each check against its threshold, and return a single report. The status is always the worst finding across all checks — one critical issue beats ten healthy ones.

def health_check(protected=None):
    if protected is None:
        protected = DEFAULT_PROTECTED
    m = _read_metrics()
    xids = _read_xid_events()
    rogue = _list_gpu_processes(protected)
    big_rogue = [r for r in rogue if r['used_mib'] >= ROGUE_CRIT_MIB]

    checks = {
        'memory_ok': m['memory_pct'] < MEM_CRIT_PCT,
        'temperature_ok': m['temperature_c'] < TEMP_CRIT_C,
        'ecc_ok': m['ecc_dbe'] == 0,
        'power_ok': m['bad_throttle'] == 0,
        'xid_errors_ok': not any(x in FATAL_XIDS for x in xids),
        'no_rogue_processes': len(rogue) == 0,
    }
    issues = []
    status = 'healthy'
    if m['memory_pct'] >= MEM_CRIT_PCT:
        issues.append(f"memory {m['memory_pct']:.1f}% >= {MEM_CRIT_PCT}% (critical)"); status = _worst(status, 'critical')
    elif m['memory_pct'] >= MEM_WARN_PCT:
        issues.append(f"memory {m['memory_pct']:.1f}% >= {MEM_WARN_PCT}% (warn)"); status = _worst(status, 'degraded')
    if m['temperature_c'] >= TEMP_CRIT_C:
        issues.append(f"temp {m['temperature_c']}C >= {TEMP_CRIT_C}C (critical)"); status = _worst(status, 'critical')
    elif m['temperature_c'] >= TEMP_WARN_C:
        issues.append(f"temp {m['temperature_c']}C >= {TEMP_WARN_C}C (warn)"); status = _worst(status, 'degraded')
    if m['ecc_dbe'] > 0:
        issues.append(f"{m['ecc_dbe']} uncorrected ECC error(s)"); status = _worst(status, 'critical')
    if m['bad_throttle']:
        issues.append(f"thermal/power throttling active (mask={m['bad_throttle']:#x})"); status = _worst(status, 'degraded')
    fatal_xids = [x for x in xids if x in FATAL_XIDS]
    if fatal_xids:
        issues.append(f"fatal Xid(s): {fatal_xids}"); status = _worst(status, 'critical')
    if big_rogue:
        issues.append(f"{len(big_rogue)} rogue process(es) >={ROGUE_CRIT_MIB} MiB: {big_rogue}"); status = _worst(status, 'critical')
    elif rogue:
        issues.append(f"{len(rogue)} non-protected CUDA process(es): {rogue}"); status = _worst(status, 'degraded')

    return {
        'status': status, 'gpu_id': 0,
        'checks': checks, 'issues': issues,
        'rogue_processes': rogue,
        **m,
        'xid_events': xids,
    }

Run it — what does a healthy idle GPU look like?

On a freshly provisioned lab pod with no training job running, you should see:

• status: healthy
• memory_pct near zero
• temperature_c in the 30–50°C range
• rogue_processes: [] (no one else is on the GPU)
• issues: [] (nothing triggered any threshold)

Step 2 will simulate a leak and you'll watch this exact report flip to critical.

health_report = health_check()
for k, v in health_report.items():
    if k not in ('checks', 'issues', 'rogue_processes'):
        print(f"  {k}: {v}")
print(f"  checks: {health_report['checks']}")
print(f"  rogue_processes: {health_report['rogue_processes']}")
print(f"  issues: {health_report['issues']}")