How to Mock a Liveness Check and Restart a Process in Python
Simulate a failing process and restart it after a liveness check fails, using a mock class and a liveness loop.
Python code
48 linesimport subprocess
import sys
import time
import os
class ProcessMock:
def __init__(self, name, fail_after_seconds=3):
self.name = name
self.fail_after = fail_after_seconds
self.start_time = None
self.is_running = False
def start(self):
self.start_time = time.time()
self.is_running = True
print(f"[{self.name}] started")
def is_alive(self):
if not self.is_running:
return False
return time.time() - self.start_time < self.fail_after
def stop(self):
self.is_running = False
print(f"[{self.name}] stopped")
def restart(self):
print(f"[{self.name}] restarting...")
self.stop()
self.start()
def run_liveness_loop(process, check_interval=1.0):
attempts = 0
max_attempts = 3
while attempts < max_attempts:
time.sleep(check_interval)
if process.is_alive():
print(f"[{process.name}] healthy")
else:
print(f"[{process.name}] FAILED liveness check")
process.restart()
attempts += 1
print(f"[{process.name}] giving up after {max_attempts} restart attempts")
if __name__ == "__main__":
process = ProcessMock(name="worker", fail_after_seconds=2)
process.start()
run_liveness_loop(process)
Output
[worker] started
[worker] healthy
[worker] healthy
[worker] FAILED liveness check
[worker] restarting...
[worker] stopped
[worker] started
[worker] healthy
[worker] healthy
[worker] FAILED liveness check
[worker] restarting...
[worker] stopped
[worker] started
[worker] healthy
[worker] healthy
[worker] FAILED liveness check
[worker] restarting...
[worker] stopped
[worker] started
[worker] giving up after 3 restart attempts
How it works
The ProcessMock class simulates a long-running process with a hard-coded lifetime (fail_after_seconds). The is_alive() method checks whether the current time minus the start time is less than the failure threshold, mimicking a health probe. The run_liveness_loop polls with time.sleep, and on failure it restarts the process and increments an attempt counter. After a maximum number of restarts, it gives up, which is a common pattern in production supervisors. The loop uses simple polling rather than threads or asyncio to keep the mock self-contained and easy to test.
Common mistakes
- Forgetting to sleep before the first liveness check, causing immediate false failures.
- Not resetting a start time on restart, so the process never appears alive again.
- Infinite restart loops without a max attempt count, leading to resource exhaustion.
- Using `is_alive()` with a fixed timestamp instead of tracking start time per start call.
Variations
- Use `threading.Event.wait(timeout)` instead of `time.sleep` for a responsive check.
- Implement liveness as a context manager to automatically stop/restart in a `with` block.
Real-world use cases
- Testing orchestrator logic that restarts a background worker after a health check fails.
- Simulating a crash in a bulk job to validate that retry counters and give-up behavior work end-to-end.
- Writing unit tests for a container orchestrator's restart policy before deploy to production.
Sponsored
More from Reliability & rate limiting
- At Least Once with Idempotent Consumer in Python medium
- Build a Rate Limiter Decorator in Python easy
- Build a queue-based admission control system in Python easy
- Chaos Inject Random Failures in Python easy
- Circuit breaker failure threshold count in Python medium
- Exactly Once Processing Dedupe Mock in Python easy
Keep learning
Related tutorials and quizzes for this topic.