REL04

REL04-BP02 - Implement loosely coupled dependencies

REL04-BP02: Implement loosely coupled dependencies

Overview

Design and implement loosely coupled dependencies between distributed system components to minimize the impact of failures and enable independent evolution of services. Loose coupling reduces cascading failures, improves system resilience, and allows services to operate independently even when dependencies are unavailable or degraded.

Implementation Steps

1. Design Asynchronous Communication Patterns

Implement message queues and event-driven architectures
Use publish-subscribe patterns for service communication
Design fire-and-forget messaging for non-critical operations
Implement event sourcing and CQRS patterns where appropriate

2. Implement Service Interface Abstraction

Create abstraction layers between services and their dependencies
Use dependency injection and interface-based programming
Implement adapter patterns for external service integration
Design service contracts that hide implementation details

3. Establish Temporal Decoupling

Implement asynchronous processing for time-consuming operations
Use message queues to buffer requests during peak loads
Design batch processing for non-real-time operations
Implement eventual consistency patterns where appropriate

4. Implement Spatial Decoupling

Use service discovery mechanisms instead of hard-coded endpoints
Implement load balancers and service meshes for routing
Design location-transparent service communication
Use content-based routing and message transformation

5. Design Failure Isolation Mechanisms

Implement bulkhead patterns to isolate failures
Use circuit breakers to prevent cascading failures
Design graceful degradation and fallback mechanisms
Implement timeout and retry strategies with exponential backoff

6. Establish Data Decoupling Strategies

Avoid shared databases between services
Implement data replication and synchronization patterns
Use event-driven data consistency mechanisms
Design service-specific data models and storage

Implementation Examples

Example 1: Loosely Coupled Architecture Implementation Framework

View code

import boto3
import json
import logging
import asyncio
import time
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Any, Callable
from dataclasses import dataclass, asdict
from enum import Enum
from abc import ABC, abstractmethod
import concurrent.futures
import threading
from contextlib import asynccontextmanager

class CouplingType(Enum):
    TEMPORAL = "temporal"
    SPATIAL = "spatial"
    PLATFORM = "platform"
    DATA = "data"

class CommunicationPattern(Enum):
    SYNCHRONOUS = "synchronous"
    ASYNCHRONOUS = "asynchronous"
    EVENT_DRIVEN = "event_driven"
    STREAMING = "streaming"

@dataclass
class ServiceDependency:
    service_name: str
    dependency_name: str
    coupling_type: CouplingType
    communication_pattern: CommunicationPattern
    criticality: str
    timeout_ms: int
    retry_config: Dict[str, Any]
    fallback_strategy: str

class ServiceInterface(ABC):
    """Abstract interface for service dependencies"""
    
    @abstractmethod
    async def call(self, request: Dict[str, Any]) -> Dict[str, Any]:
        pass
    
    @abstractmethod
    async def health_check(self) -> bool:
        pass

class CircuitBreaker:
    """Circuit breaker implementation for fault tolerance"""
    
    def __init__(self, failure_threshold: int = 5, timeout_seconds: int = 60, 
                 half_open_max_calls: int = 3):
        self.failure_threshold = failure_threshold
        self.timeout_seconds = timeout_seconds
        self.half_open_max_calls = half_open_max_calls
        
        self.failure_count = 0
        self.last_failure_time = None
        self.state = "CLOSED"  # CLOSED, OPEN, HALF_OPEN
        self.half_open_calls = 0
        self.lock = threading.Lock()
    
    async def call(self, func: Callable, *args, **kwargs):
        """Execute function with circuit breaker protection"""
        with self.lock:
            if self.state == "OPEN":
                if self._should_attempt_reset():
                    self.state = "HALF_OPEN"
                    self.half_open_calls = 0
                else:
                    raise Exception("Circuit breaker is OPEN")
            
            if self.state == "HALF_OPEN":
                if self.half_open_calls >= self.half_open_max_calls:
                    raise Exception("Circuit breaker HALF_OPEN limit exceeded")
                self.half_open_calls += 1
        
        try:
            result = await func(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            self._on_failure()
            raise e
    
    def _should_attempt_reset(self) -> bool:
        """Check if circuit breaker should attempt reset"""
        if self.last_failure_time is None:
            return True
        return time.time() - self.last_failure_time >= self.timeout_seconds
    
    def _on_success(self):
        """Handle successful call"""
        with self.lock:
            self.failure_count = 0
            if self.state == "HALF_OPEN":
                self.state = "CLOSED"
    
    def _on_failure(self):
        """Handle failed call"""
        with self.lock:
            self.failure_count += 1
            self.last_failure_time = time.time()
            
            if self.failure_count >= self.failure_threshold:
                self.state = "OPEN"

class AsyncServiceClient(ServiceInterface):
    """Asynchronous service client with loose coupling patterns"""
    
    def __init__(self, service_config: Dict[str, Any]):
        self.service_name = service_config['name']
        self.endpoint_url = service_config.get('endpoint_url')
        self.timeout_ms = service_config.get('timeout_ms', 5000)
        self.retry_config = service_config.get('retry_config', {
            'max_retries': 3,
            'backoff_factor': 2,
            'base_delay_ms': 100
        })
        
        # Initialize circuit breaker
        self.circuit_breaker = CircuitBreaker(
            failure_threshold=service_config.get('circuit_breaker', {}).get('failure_threshold', 5),
            timeout_seconds=service_config.get('circuit_breaker', {}).get('timeout_seconds', 60)
        )
        
        # Initialize AWS clients
        self.sqs = boto3.client('sqs')
        self.sns = boto3.client('sns')
        self.eventbridge = boto3.client('events')
        
        # Message queue for async communication
        self.request_queue_url = service_config.get('request_queue_url')
        self.response_topic_arn = service_config.get('response_topic_arn')
    
    async def call(self, request: Dict[str, Any]) -> Dict[str, Any]:
        """Make service call with circuit breaker protection"""
        return await self.circuit_breaker.call(self._make_request, request)
    
    async def _make_request(self, request: Dict[str, Any]) -> Dict[str, Any]:
        """Make actual service request with retry logic"""
        last_exception = None
        
        for attempt in range(self.retry_config['max_retries'] + 1):
            try:
                if self.request_queue_url:
                    # Asynchronous communication via SQS
                    return await self._send_async_request(request)
                else:
                    # Synchronous HTTP request (with timeout)
                    return await self._send_sync_request(request)
                    
            except Exception as e:
                last_exception = e
                if attempt < self.retry_config['max_retries']:
                    delay = self._calculate_backoff_delay(attempt)
                    await asyncio.sleep(delay / 1000)  # Convert to seconds
                    logging.warning(f"Request failed, retrying in {delay}ms: {str(e)}")
        
        raise last_exception
    
    async def _send_async_request(self, request: Dict[str, Any]) -> Dict[str, Any]:
        """Send asynchronous request via message queue"""
        try:
            # Add correlation ID for response tracking
            correlation_id = f"{self.service_name}_{int(time.time() * 1000)}"
            request['correlation_id'] = correlation_id
            request['response_topic'] = self.response_topic_arn
            request['timestamp'] = datetime.utcnow().isoformat()
            
            # Send message to SQS queue
            response = self.sqs.send_message(
                QueueUrl=self.request_queue_url,
                MessageBody=json.dumps(request),
                MessageAttributes={
                    'CorrelationId': {
                        'StringValue': correlation_id,
                        'DataType': 'String'
                    },
                    'ServiceName': {
                        'StringValue': self.service_name,
                        'DataType': 'String'
                    }
                }
            )
            
            # For async calls, return immediately with correlation ID
            return {
                'status': 'accepted',
                'correlation_id': correlation_id,
                'message_id': response['MessageId']
            }
            
        except Exception as e:
            logging.error(f"Async request failed: {str(e)}")
            raise
    
    async def _send_sync_request(self, request: Dict[str, Any]) -> Dict[str, Any]:
        """Send synchronous HTTP request"""
        import aiohttp
        
        try:
            timeout = aiohttp.ClientTimeout(total=self.timeout_ms / 1000)
            
            async with aiohttp.ClientSession(timeout=timeout) as session:
                async with session.post(self.endpoint_url, json=request) as response:
                    if response.status >= 400:
                        raise Exception(f"HTTP {response.status}: {await response.text()}")
                    
                    return await response.json()
                    
        except asyncio.TimeoutError:
            raise Exception(f"Request timeout after {self.timeout_ms}ms")
        except Exception as e:
            logging.error(f"Sync request failed: {str(e)}")
            raise
    
    def _calculate_backoff_delay(self, attempt: int) -> int:
        """Calculate exponential backoff delay"""
        base_delay = self.retry_config['base_delay_ms']
        backoff_factor = self.retry_config['backoff_factor']
        return int(base_delay * (backoff_factor ** attempt))
    
    async def health_check(self) -> bool:
        """Perform health check on the service"""
        try:
            if self.endpoint_url:
                # HTTP health check
                import aiohttp
                timeout = aiohttp.ClientTimeout(total=5)
                
                async with aiohttp.ClientSession(timeout=timeout) as session:
                    health_url = f"{self.endpoint_url}/health"
                    async with session.get(health_url) as response:
                        return response.status == 200
            else:
                # Queue-based health check
                return await self._check_queue_health()
                
        except Exception as e:
            logging.warning(f"Health check failed for {self.service_name}: {str(e)}")
            return False
    
    async def _check_queue_health(self) -> bool:
        """Check health of message queue"""
        try:
            if self.request_queue_url:
                # Check queue attributes
                response = self.sqs.get_queue_attributes(
                    QueueUrl=self.request_queue_url,
                    AttributeNames=['ApproximateNumberOfMessages']
                )
                return True  # If we can get attributes, queue is healthy
            return False
            
        except Exception:
            return False

class EventDrivenService:
    """Service implementation using event-driven patterns"""
    
    def __init__(self, service_config: Dict[str, Any]):
        self.service_name = service_config['name']
        self.event_bus_name = service_config.get('event_bus_name', 'default')
        
        # Initialize AWS clients
        self.eventbridge = boto3.client('events')
        self.sqs = boto3.client('sqs')
        self.sns = boto3.client('sns')
        
        # Event handlers registry
        self.event_handlers: Dict[str, Callable] = {}
        
        # Dead letter queue for failed events
        self.dlq_url = service_config.get('dead_letter_queue_url')
    
    def register_event_handler(self, event_type: str, handler: Callable):
        """Register handler for specific event type"""
        self.event_handlers[event_type] = handler
        logging.info(f"Registered handler for event type: {event_type}")
    
    async def publish_event(self, event_type: str, event_data: Dict[str, Any]):
        """Publish event to EventBridge"""
        try:
            event_entry = {
                'Source': self.service_name,
                'DetailType': event_type,
                'Detail': json.dumps({
                    **event_data,
                    'timestamp': datetime.utcnow().isoformat(),
                    'service_name': self.service_name
                }),
                'EventBusName': self.event_bus_name
            }
            
            response = self.eventbridge.put_events(Entries=[event_entry])
            
            if response['FailedEntryCount'] > 0:
                raise Exception(f"Failed to publish event: {response['Entries'][0].get('ErrorMessage')}")
            
            logging.info(f"Published event {event_type} to {self.event_bus_name}")
            
        except Exception as e:
            logging.error(f"Failed to publish event {event_type}: {str(e)}")
            raise
    
    async def process_event(self, event: Dict[str, Any]):
        """Process incoming event"""
        try:
            event_type = event.get('DetailType')
            event_data = json.loads(event.get('Detail', '{}'))
            
            if event_type in self.event_handlers:
                handler = self.event_handlers[event_type]
                await handler(event_data)
                logging.info(f"Successfully processed event {event_type}")
            else:
                logging.warning(f"No handler registered for event type: {event_type}")
                
        except Exception as e:
            logging.error(f"Failed to process event: {str(e)}")
            
            # Send to dead letter queue if configured
            if self.dlq_url:
                await self._send_to_dlq(event, str(e))
            
            raise
    
    async def _send_to_dlq(self, event: Dict[str, Any], error_message: str):
        """Send failed event to dead letter queue"""
        try:
            dlq_message = {
                'original_event': event,
                'error_message': error_message,
                'failed_at': datetime.utcnow().isoformat(),
                'service_name': self.service_name
            }
            
            self.sqs.send_message(
                QueueUrl=self.dlq_url,
                MessageBody=json.dumps(dlq_message)
            )
            
            logging.info("Sent failed event to dead letter queue")
            
        except Exception as e:
            logging.error(f"Failed to send event to DLQ: {str(e)}")

class LooseCouplingOrchestrator:
    """Orchestrator for managing loosely coupled services"""
    
    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.services: Dict[str, AsyncServiceClient] = {}
        self.event_service = EventDrivenService(config.get('event_service', {}))
        
        # Initialize services
        for service_config in config.get('services', []):
            service_name = service_config['name']
            self.services[service_name] = AsyncServiceClient(service_config)
    
    async def execute_workflow(self, workflow_config: Dict[str, Any]) -> Dict[str, Any]:
        """Execute workflow with loose coupling patterns"""
        workflow_id = f"workflow_{int(time.time() * 1000)}"
        workflow_result = {
            'workflow_id': workflow_id,
            'status': 'started',
            'steps': [],
            'started_at': datetime.utcnow().isoformat()
        }
        
        try:
            steps = workflow_config.get('steps', [])
            
            for step_config in steps:
                step_result = await self._execute_step(step_config)
                workflow_result['steps'].append(step_result)
                
                # Check if step failed and handle accordingly
                if not step_result.get('success', False):
                    if step_config.get('required', True):
                        workflow_result['status'] = 'failed'
                        break
                    else:
                        # Continue with optional step failure
                        logging.warning(f"Optional step failed: {step_config.get('name')}")
            
            if workflow_result['status'] != 'failed':
                workflow_result['status'] = 'completed'
            
            workflow_result['completed_at'] = datetime.utcnow().isoformat()
            
            # Publish workflow completion event
            await self.event_service.publish_event(
                'WorkflowCompleted',
                {
                    'workflow_id': workflow_id,
                    'status': workflow_result['status'],
                    'duration_ms': self._calculate_duration(workflow_result)
                }
            )
            
            return workflow_result
            
        except Exception as e:
            workflow_result['status'] = 'error'
            workflow_result['error'] = str(e)
            workflow_result['completed_at'] = datetime.utcnow().isoformat()
            
            # Publish workflow error event
            await self.event_service.publish_event(
                'WorkflowFailed',
                {
                    'workflow_id': workflow_id,
                    'error': str(e)
                }
            )
            
            return workflow_result
    
    async def _execute_step(self, step_config: Dict[str, Any]) -> Dict[str, Any]:
        """Execute individual workflow step"""
        step_name = step_config.get('name', 'unknown')
        service_name = step_config.get('service')
        
        step_result = {
            'step_name': step_name,
            'service_name': service_name,
            'started_at': datetime.utcnow().isoformat(),
            'success': False
        }
        
        try:
            if service_name in self.services:
                service_client = self.services[service_name]
                
                # Execute service call
                request_data = step_config.get('request', {})
                response = await service_client.call(request_data)
                
                step_result['response'] = response
                step_result['success'] = True
                
            else:
                raise Exception(f"Service {service_name} not found")
            
            step_result['completed_at'] = datetime.utcnow().isoformat()
            return step_result
            
        except Exception as e:
            step_result['error'] = str(e)
            step_result['completed_at'] = datetime.utcnow().isoformat()
            logging.error(f"Step {step_name} failed: {str(e)}")
            return step_result
    
    def _calculate_duration(self, workflow_result: Dict[str, Any]) -> int:
        """Calculate workflow duration in milliseconds"""
        try:
            start_time = datetime.fromisoformat(workflow_result['started_at'].replace('Z', '+00:00'))
            end_time = datetime.fromisoformat(workflow_result['completed_at'].replace('Z', '+00:00'))
            return int((end_time - start_time).total_seconds() * 1000)
        except:
            return 0

Example 2: Loose Coupling Implementation Script