DevOps NetworkDevOpsNetwork
Join Free
Practical infra coding

Debug a health check that always reports unhealthy

mediumDebug and review code

Problem statement

A node agent decides whether to restart a service by running its check command, which prints OK when the service is fine. The team reports that every service is marked unhealthy, even though running the check by hand prints OK. On one host the agent also stopped doing anything at all. Find the bugs and fix service_healthy.

health.py (broken)

Python
import subprocess
def service_healthy(check_cmd):
out = subprocess.run(check_cmd, capture_output=True).stdout
return out.strip() == "OK"

The test harness runs four fake check commands, built with the current Python interpreter so they work on any OS:

Check Behaviour
healthy prints OK, exits 0
degraded prints OK, then exits 2
crashing prints db unreachable to stderr, exits 1
hung sleeps for 30 seconds

The fixed version should return a readable verdict string instead of a bare boolean, so the agent can log why it restarted something.

Symptom (actual output of the broken version; hung has to be skipped or the run blocks for 30 s):

◈ DIAGRAM
healthy -> False
degraded -> False
crashing -> False
hung (skipped: would block for 30 s)

Examples

Example 1

Input: The same harness with the fixed `service_healthy` (timeout 5 s, nothing skipped)

Output: healthy -> healthy degraded -> unhealthy (exit 2: OK) crashing -> unhealthy (exit 1: db unreachable) hung -> unhealthy (timed out)

Explanation: degraded prints OK but exits 2; only the exit code reveals it. hung is killed after 5 s instead of blocking the agent.

Hints

Approach

Optimal

Bug 1: bytes compared with str. capture_output=True without text=True returns stdout as bytes. b"OK".strip() == "OK" is always False in Python 3 (no error, just False), so every service looks unhealthy. Fix: text=True, which decodes the output with the locale encoding (pass encoding="utf-8" to be explicit).

Bug 2: no timeout. A check that hangs (a stuck database connection, a blocking NFS mount) blocks subprocess.run forever, and the whole agent stops checking every other service. That is the host where "nothing happened". Fix: timeout=5. On expiry subprocess.run kills the child and raises TimeoutExpired, which becomes a clear verdict.

Bug 3: exit code ignored. The check's exit status is its real verdict; the text is secondary. degraded prints OK and exits 2, which a text-only comparison would call healthy once bug 1 is fixed. Fix: treat any non-zero returncode as unhealthy, and include stderr (or stdout) in the reason.

Bug 4: a missing command crashes the agent. If the check binary is not installed, subprocess.run raises FileNotFoundError (an OSError), which would take down the agent loop. Fix: catch OSError and report it as unhealthy with the reason.

Keep the argument list. The broken code already passes a list, not a string with shell=True. Keep it that way: building a shell string from config values invites quoting bugs and command injection.

The fixed function returns a descriptive string, so the agent's log explains every restart decision.

ComplexityTime O(output size)Space O(output size)
Python
import subprocess
def service_healthy(check_cmd, timeout=5):
try:
result = subprocess.run(check_cmd, capture_output=True, text=True, timeout=timeout)
except subprocess.TimeoutExpired:
return "unhealthy (timed out)"
except OSError as e: # command not found, not executable
return f"unhealthy (cannot run: {e.strerror})"
if result.returncode != 0:
reason = result.stderr.strip() or result.stdout.strip() or "no output"
return f"unhealthy (exit {result.returncode}: {reason})"
if result.stdout.strip() != "OK":
return f"unhealthy (unexpected output {result.stdout.strip()!r})"
return "healthy"
# ---- test harness: fake health-check commands built from the current Python ----
import sys
PY = sys.executable
CHECKS = {
"healthy": [PY, "-c", "print('OK')"],
"degraded": [PY, "-c", "import sys; print('OK'); sys.exit(2)"],
"crashing": [PY, "-c", "raise SystemExit('db unreachable')"],
"hung": [PY, "-c", "import time; time.sleep(30)"],
}
skip = set(sys.argv[1:]) # the broken version hangs on "hung"
for name, cmd in CHECKS.items():
if name in skip:
print(f"{name:<9} (skipped: would block for 30 s)")
continue
print(f"{name:<9} -> {service_healthy(cmd)}")

Follow-up questions

  • Restart a service only after 3 consecutive failed checks, to avoid flapping.
  • Run the checks for 50 services in parallel with a limit of 8 at a time.
  • Capture at most the first 4 KB of output, so a check that prints megabytes cannot fill the agent's memory.

Frequently asked questions

Agents, deploy scripts and CI steps run external commands constantly. The bytes/str mix-up, missing timeouts and ignored exit codes are classic subprocess bugs in ops code, and each one fails quietly rather than with a traceback.

check=True raises CalledProcessError on a non-zero exit, which is the right choice when failure should stop your program. A health checker expects failures as a normal outcome, so reading returncode directly and turning it into a verdict is clearer than catching an exception for it.

Not necessarily. subprocess.run kills the direct child. If the check is a shell script that started its own children, those can survive. For that case, start the check in its own process group (start_new_session=True on POSIX) and kill the whole group on timeout.