Service Discovery — Dynamic Service Location in Microservices

# discovery_patterns.py
# Service discovery patterns

def discovery_patterns():
    print("Service Discovery Patterns")
    print("=" * 40)
    print()
    
    patterns = [
        {
            "pattern": "Client-Side Discovery",
            "desc": "The client queries the registry directly and selects an instance to call.",
            "pros": "No intermediary, simple architecture, fewer network hops",
            "cons": "Client logic includes discovery, language-specific implementations needed"
        },
        {
            "pattern": "Server-Side Discovery",
            "desc": "The client calls a load balancer/router that queries the registry and forwards the request.",
            "pros": "Clients are simpler, works with any language, central control",
            "cons": "Extra network hop, load balancer can become bottleneck"
        },
        {
            "pattern": "DNS-Based Discovery",
            "desc": "Service names resolve to multiple IP addresses via DNS. Clients get a list and pick one.",
            "pros": "No registry needed, standard DNS infrastructure, simple",
            "cons": "Slow DNS propagation, no health-aware routing, limited to simple cases"
        },
    ]
    
    for p in patterns:
        print(f"{p['pattern']:30s}")
        print(f"  {p['desc']}")
        print(f"  Pros: {p['pros']}")
        print(f"  Cons: {p['cons']}")
        print()

discovery_patterns()

# client_discovery.py
# Client-side discovery implementation

def client_discovery():
    print("Client-Side Service Discovery")
    print("=" * 40)
    print()
    
    registry_code = """
import time
import json
import threading

class ServiceRegistry:
    """In-memory service registry (like Consul, Eureka)."""
    
    def __init__(self):
        self._services = {}  # service_name -> [instances]
        self._lock = threading.Lock()
    
    def register(self, service_name, instance_id, 
                 host, port, metadata=None):
        with self._lock:
            if service_name not in self._services:
                self._services[service_name] = []
            
            # Remove existing registration for this instance
            self._services[service_name] = [
                i for i in self._services[service_name]
                if i["instance_id"] != instance_id
            ]
            
            self._services[service_name].append({
                "instance_id": instance_id,
                "host": host,
                "port": port,
                "metadata": metadata or {},
                "last_heartbeat": time.time()
            })
    
    def unregister(self, service_name, instance_id):
        with self._lock:
            if service_name in self._services:
                self._services[service_name] = [
                    i for i in self._services[service_name]
                    if i["instance_id"] != instance_id
                ]
    
    def heartbeat(self, service_name, instance_id):
        with self._lock:
            for instance in self._services.get(service_name, []):
                if instance["instance_id"] == instance_id:
                    instance["last_heartbeat"] = time.time()
    
    def get_instances(self, service_name):
        """Return healthy instances for a service."""
        with self._lock:
            now = time.time()
            instances = self._services.get(service_name, [])
            return [
                i for i in instances
                if now - i["last_heartbeat"] < 30  # 30s TTL
            ]
    
    def cleanup(self):
        """Remove instances that haven't heartbeated."""
        with self._lock:
            now = time.time()
            for name in list(self._services.keys()):
                self._services[name] = [
                    i for i in self._services[name]
                    if now - i["last_heartbeat"] < 30
                ]
"""
    print("Service Registry:")
    print(registry_code)
    
    client_code = """
import random

class ServiceDiscoveryClient:
    """Client that discovers and calls services."""
    
    def __init__(self, registry_endpoint):
        self.registry_endpoint = registry_endpoint
        self.cache = {}  # Local cache to reduce registry calls
        self.cache_ttl = 5  # Refresh every 5 seconds
    
    def get_service_url(self, service_name):
        instance = self._discover(service_name)
        if instance:
            return f"http://{instance['host']}:{instance['port']}"
        raise ServiceNotFoundException(service_name)
    
    def _discover(self, service_name):
        instances = self._get_from_registry(service_name)
        if not instances:
            return None
        # Pick a random instance (load balancing)
        return random.choice(instances)
    
    def _get_from_registry(self, service_name):
        # In production: HTTP call to Consul/Eureka
        # Here: direct registry call
        return registry.get_instances(service_name)
    
    def call_service(self, service_name, path, method="GET"):
        url = self.get_service_url(service_name)
        full_url = f"{url}{path}"
        print(f"Calling {full_url}")
        # Make HTTP request
        return httpx.request(method, full_url)
"""
    print("Discovery Client:")
    print(client_code)

client_discovery()

# server_discovery.py
# Server-side discovery

def server_discovery():
    print("Server-Side Service Discovery")
    print("=" * 40)
    print()
    
    code = """
# Server-side discovery with API Gateway / Load Balancer

# The client only knows the gateway address.
# The gateway handles discovery internally.

class ApiGateway:
    """API Gateway with built-in service discovery."""
    
    def __init__(self, registry):
        self.registry = registry
        self.routes = {
            "/api/users/": "user-service",
            "/api/orders/": "order-service",
            "/api/payments/": "payment-service",
            "/api/files/": "file-service",
        }
    
    def handle_request(self, request):
        # Find which service handles this path
        for prefix, service_name in self.routes.items():
            if request.path.startswith(prefix):
                return self._forward(service_name, request)
        
        return {"error": "route not found"}, 404
    
    def _forward(self, service_name, request):
        # Use server-side discovery to find instances
        instances = self.registry.get_instances(service_name)
        if not instances:
            return {"error": "service unavailable"}, 503
        
        # Pick instance (round-robin or random)
        instance = self._select_instance(instances)
        url = f"http://{instance['host']}:{instance['port']}{request.path}"
        
        # Forward request and return response
        # In production: proxy the request
        print(f"Gateway forwarding to {url}")
        return self._proxy_request(url, request)
    
    def _select_instance(self, instances):
        # Round-robin selection
        self._rr_index = getattr(self, '_rr_index', 0) + 1
        return instances[self._rr_index % len(instances)]
    
    def _proxy_request(self, url, request):
        # HTTP proxy implementation
        pass
"""
    print(code)

server_discovery()

# health_checks.py
# Health check implementation

def health_checks():
    print("Health Checks for Service Discovery")
    print("=" * 40)
    print()
    
    code = """
import httpx
import threading
import time

class HealthChecker:
    """Periodically checks service health and removes unhealthy instances."""
    
    def __init__(self, registry, check_interval=10):
        self.registry = registry
        self.check_interval = check_interval
        self.client = httpx.Client(timeout=5.0)
    
    def start(self):
        thread = threading.Thread(target=self._run_checks, daemon=True)
        thread.start()
    
    def _run_checks(self):
        while True:
            self._check_all()
            time.sleep(self.check_interval)
    
    def _check_all(self):
        for service_name in list(self.registry._services.keys()):
            for instance in self.registry.get_instances(service_name):
                if not self._is_healthy(instance):
                    print(f"Removing unhealthy instance: "
                          f"{instance['instance_id']}")
                    self.registry.unregister(
                        service_name, instance["instance_id"]
                    )
    
    def _is_healthy(self, instance):
        try:
            url = f"http://{instance['host']}:{instance['port']}/health"
            response = self.client.get(url)
            return response.status_code == 200
        except Exception:
            return False

# Service-side health endpoint
class ServiceHealthEndpoint:
    @staticmethod
    def health_check():
        return {
            "status": "healthy",
            "service": "user-service",
            "version": "1.2.3",
            "uptime_seconds": 12345,
            "database": "connected",
            "memory_mb": 256
        }, 200
"""
    print(code)

health_checks()

def discovery_system():
    print("Service Discovery System Design")
    print("=" * 40)
    print()
    print("Components:")
    print("  Registry:   In-memory registry with TTL-based cleanup")
    print("  Heartbeat:  Services send heartbeat every 10s")
    print("  Health:     Registry probes /health every 30s")
    print("  Discovery:  Client queries registry on each request")
    print("  Cache:      Client caches results for 5s with fallback")
    print()
    print("Registration Flow:")
    print("  1. Service starts -> POST /register")
    print("  2. Service sends heartbeat -> PUT /heartbeat")
    print("  3. Client requests -> GET /discover/{service}")
    print("  4. Client selects random instance -> HTTP call")
    print()
    print("Failure Scenarios:")
    print("  Service crash:   TTL expires after 30s, auto-removed")
    print("  Registry down:   Client uses last-known cache")
    print("  No instances:    Client returns 503 Service Unavailable")

discovery_system()