Implementing a Self-Healing Distributed System with Python

Distributed systems are powerful, but they come with inherent complexities: network partitions, node failures, and resource exhaustion can bring down even the most carefully designed architectures. A self-healing system automatically detects failures and recovers without human intervention, ensuring high availability and resilience. In this post, we'll build a practical self-healing distributed system in Python, complete with health checks, auto-restart mechanisms, and fallback strategies.

The Core Principles

A self-healing system revolves around three key mechanisms:

1. Health Monitoring: Continuously check if components are alive and responsive.
2. Automated Recovery: Restart or replace failed components without manual intervention.
3. Graceful Degradation: When primary services fail, fall back to secondary services to maintain partial functionality.

We'll implement these using asyncio for concurrency, aiohttp for HTTP-based health checks, and Python's subprocess module for managing child processes.

System Architecture

Our system will consist of:

• A Service Manager that orchestrates multiple worker processes.
• Worker Services that perform actual work (simulated with HTTP servers).
• A Health Checker that probes workers periodically.
• A Fallback Service that kicks in when primary workers fail.

+------------------+       +------------------+
|  Service Manager |       |  Health Checker  |
|  (orchestrator)  |<----->|  (monitoring)    |
+--------+---------+       +--------+---------+
         |                           |
         v                           v
+------------------+       +------------------+
|  Worker Process 1|       |  Worker Process 2|
|  (primary)       |       |  (fallback)      |
+------------------+       +------------------+

Implementation

1. The Worker Service

First, let's create a simple HTTP-based worker that simulates occasional failures.

# worker.py
import asyncio
import random
from aiohttp import web

async def handle_health(request):
    """Health check endpoint"""
    # Simulate random failure (20% chance)
    if random.random() < 0.2:
        return web.Response(status=500, text="Internal Error")
    return web.json_response({"status": "healthy"})

async def handle_work(request):
    """Main work endpoint"""
    await asyncio.sleep(0.1)  # Simulate processing
    return web.json_response({"result": "work_done", "worker_id": request.app['worker_id']})

def create_app(worker_id, port):
    app = web.Application()
    app['worker_id'] = worker_id
    app.router.add_get('/health', handle_health)
    app.router.add_get('/work', handle_work)
    return app

if __name__ == '__main__':
    import sys
    worker_id = sys.argv[1] if len(sys.argv) > 1 else 'unknown'
    port = int(sys.argv[2]) if len(sys.argv) > 2 else 8080
    web.run_app(create_app(worker_id, port), port=port)

2. The Service Manager

The manager is the brain of our self-healing system. It spawns workers, monitors them, and takes corrective action.

# service_manager.py
import asyncio
import subprocess
import signal
import time
from typing import Dict, Optional
from dataclasses import dataclass
from datetime import datetime, timedelta

@dataclass
class WorkerInstance:
    process: subprocess.Popen
    port: int
    worker_id: str
    last_healthy: datetime
    restart_count: int = 0

class ServiceManager:
    def __init__(self):
        self.workers: Dict[str, WorkerInstance] = {}
        self.next_worker_id = 0
        self.base_port = 9000
        self.max_restarts = 3
        self.restart_window = timedelta(minutes=5)

    async def spawn_worker(self) -> WorkerInstance:
        """Start a new worker process"""
        worker_id = f"worker-{self.next_worker_id}"
        self.next_worker_id += 1
        port = self.base_port + len(self.workers)

        print(f"[Manager] Spawning {worker_id} on port {port}")

        process = subprocess.Popen(
            ['python', 'worker.py', worker_id, str(port)],
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE
        )

        instance = WorkerInstance(
            process=process,
            port=port,
            worker_id=worker_id,
            last_healthy=datetime.now()
        )
        self.workers[worker_id] = instance

        # Give worker time to start
        await asyncio.sleep(1)
        return instance

    async def check_worker_health(self, worker_id: str) -> bool:
        """Check if a worker is healthy via HTTP"""
        import aiohttp

        worker = self.workers.get(worker_id)
        if not worker:
            return False

        try:
            async with aiohttp.ClientSession() as session:
                url = f"http://localhost:{worker.port}/health"
                async with session.get(url, timeout=2) as response:
                    if response.status == 200:
                        worker.last_healthy = datetime.now()
                        return True
                    else:
                        print(f"[Health] {worker_id} returned status {response.status}")
                        return False
        except Exception as e:
            print(f"[Health] {worker_id} check failed: {e}")
            return False

    def should_restart(self, worker_id: str) -> bool:
        """Check if worker hasn't exceeded restart limits"""
        worker = self.workers.get(worker_id)
        if not worker:
            return False

        # Reset count if outside window
        now = datetime.now()
        if now - worker.last_healthy > self.restart_window:
            worker.restart_count = 0

        return worker.restart_count < self.max_restarts

    async def restart_worker(self, worker_id: str):
        """Restart a failed worker"""
        worker = self.workers.get(worker_id)
        if not worker:
            return

        print(f"[Manager] Restarting {worker_id} (attempt {worker.restart_count + 1})")

        # Kill old process
        worker.process.terminate()
        try:
            worker.process.wait(timeout=5)
        except subprocess.TimeoutExpired:
            worker.process.kill()
            worker.process.wait()

        # Spawn new one
        new_worker = await self.spawn_worker()
        new_worker.restart_count = worker.restart_count + 1
        self.workers[worker_id] = new_worker

    async def health_monitor_loop(self):
        """Continuously monitor all workers"""
        while True:
            for worker_id in list(self.workers.keys()):
                is_healthy = await self.check_worker_health(worker_id)

                if not is_healthy:
                    if self.should_restart(worker_id):
                        await self.restart_worker(worker_id)
                    else:
                        print(f"[Manager] {worker_id} exceeded restart limit. Escalating...")
                        # In production, this would trigger alerts or scale up fallback
                        await self.activate_fallback(worker_id)

            await asyncio.sleep(5)  # Check every 5 seconds

    async def activate_fallback(self, failed_worker_id: str):
        """Activate fallback mechanism"""
        print(f"[Fallback] Activating fallback for {failed_worker_id}")

        # Simulate fallback: create a new worker with different config
        fallback_id = f"fallback-{self.next_worker_id}"
        self.next_worker_id += 1
        port = self.base_port + len(self.workers) + 100  # Different port range

        process = subprocess.Popen(
            ['python', 'worker.py', fallback_id, str(port)],
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE
        )

        instance = WorkerInstance(
            process=process,
            port=port,
            worker_id=fallback_id,
            last_healthy=datetime.now(),
            restart_count=0
        )
        self.workers[fallback_id] = instance
        print(f"[Fallback] {fallback_id} started on port {port}")

    async def run(self):
        """Main entry point"""
        # Start initial workers
        initial_workers = 3
        print(f"[Manager] Starting {initial_workers} workers...")

        for _ in range(initial_workers):
            await self.spawn_worker()

        # Start health monitoring
        await self.health_monitor_loop()

if __name__ == '__main__':
    manager = ServiceManager()
    try:
        asyncio.run(manager.run())
    except KeyboardInterrupt:
        print("[Manager] Shutting down...")
        for worker in manager.workers.values():
            worker.process.terminate()

3. The Client with Automatic Fallback

No self-healing system