Interview Question

"Design a cloud migration strategy for a legacy monolith with 10 million users. How do you handle the 7 Rs, Strangler Fig pattern, and parallel run?"

Complete Migration Architecture

Architecture Overview

┌─────────────────────────────────────────────────────────────────────────┐
│                    CLOUD MIGRATION ARCHITECTURE                          │
├─────────────────────────────────────────────────────────────────────────┤
│                                                                          │
│  ┌───────────────── MIGRATION PHASES ────────────────┐                 │
│  │                                                       │              │
│  │  ┌──────────┐  ┌──────────┐  ┌──────────┐          │              │
│  │  │ Assess   │  │ Mobilize │  │ Migrate  │          │              │
│  │  │ (Rs)     │  │ (Plan)   │  │ (Execute)│          │              │
│  │  └──────────┘  └──────────┘  └──────────┘          │              │
│  │                                                       │              │
│  │  ┌──────────┐  ┌──────────┐  ┌──────────┐          │              │
│  │  │Modernize │  │Optimize  │  │Operate   │          │              │
│  │  │ (Refactor)│  │(Tune)   │  │(Run)    │          │              │
│  │  └──────────┘  └──────────┘  └──────────┘          │              │
│  │                                                       │              │
│  └──────────────────────┬──────────────────────────────┘              │
│                         │                                               │
│  ┌───────────────── MIGRATION PATTERNS ──────────────┐                │
│  │                                                       │              │
│  │  ┌─────────────────────────────────────────────┐    │              │
│  │  │           Strangler Fig Pattern              │    │              │
│  │  │                                               │    │              │
│  │  │  ┌──────────┐  ┌──────────┐  ┌──────────┐  │    │              │
│  │  │  │ Monolith │  │  Facade  │  │ Cloud    │  │    │              │
│  │  │  │ (Legacy) │  │ (Router) │  │ Services │  │    │              │
│  │  │  └──────────┘  └──────────┘  └──────────┘  │    │              │
│  │  │                                               │    │              │
│  │  └─────────────────────────────────────────────┘    │              │
│  │                                                       │              │
│  └──────────────────────┬──────────────────────────────┘              │
│                         │                                               │
│  ┌───────────────── TARGET ARCHITECTURE ─────────────┐                │
│  │  Microservices │ Serverless │ Cloud-Native          │                │
│  └─────────────────────────────────────────────────────┘              │
│                                                                          │
└─────────────────────────────────────────────────────────────────────────┘

Mathematical Foundation: Migration Planning

7 Rs Assessment:

• Total applications: A = 100
• Rehost: R = 40% = 40 apps
• Replatform: P = 30% = 30 apps
• Refactor: F = 15% = 15 apps
• Retire: T = 10% = 10 apps
• Retain: N = 5% = 5 apps
• Relocate: L = 0% = 0 apps
• Reimburse: (Not applicable)

Migration Timeline:

• Assessment phase: 2 months
• Mobilize phase: 1 month
• Migrate phase: 12 months
• Modernize phase: 6 months
• Total timeline: 21 months

Cost Analysis:

• Current infrastructure cost: $500,000/month
• Projected cloud cost: $300,000/month (40% reduction)
• Migration cost: $2,000,000
• Monthly savings: $200,000
• Payback period: 10 months

Risk Assessment:

• Total risk score: R_total = Σ(risk × probability × impact)
• High-risk applications: H = 10%
• Medium-risk applications: M = 30%
• Low-risk applications: L = 60%

7 Rs Framework Implementation

# 7 Rs assessment framework
from typing import Dict, Any, List
from dataclasses import dataclass
from enum import Enum

class MigrationStrategy(Enum):
    REHOST = "rehost"
    REPLATFORM = "replatform"
    REFACTOR = "refactor"
    RETIRE = "retire"
    RETAIN = "retain"
    RELOCATE = "relocate"
    REPURCHASE = "repurchase"

@dataclass
class ApplicationAssessment:
    app_id: str
    name: str
    business_value: str
    technical_complexity: str
    dependencies: List[str]
    recommended_strategy: MigrationStrategy
    estimated_effort: int  # person-months
    estimated_cost: float
    risks: List[str]

class SevenRsAssessor:
    """7 Rs assessment framework"""

    def __init__(self):
        self.applications: List[ApplicationAssessment] = []

    def assess_application(self, app_info: Dict[str, Any]) -> ApplicationAssessment:
        """Assess application for migration strategy"""
        # Analyze application characteristics
        business_value = self._assess_business_value(app_info)
        technical_complexity = self._assess_technical_complexity(app_info)
        dependencies = self._assess_dependencies(app_info)

        # Determine strategy
        strategy = self._determine_strategy(
            business_value,
            technical_complexity,
            dependencies
        )

        # Estimate effort and cost
        effort, cost = self._estimate_effort_cost(app_info, strategy)

        # Identify risks
        risks = self._identify_risks(app_info, strategy)

        return ApplicationAssessment(
            app_id=app_info['app_id'],
            name=app_info['name'],
            business_value=business_value,
            technical_complexity=technical_complexity,
            dependencies=dependencies,
            recommended_strategy=strategy,
            estimated_effort=effort,
            estimated_cost=cost,
            risks=risks
        )

    def _assess_business_value(self, app_info: Dict[str, Any]) -> str:
        """Assess business value"""
        revenue_impact = app_info.get('revenue_impact', 0)
        user_impact = app_info.get('user_impact', 0)

        if revenue_impact > 1000000 or user_impact > 1000000:
            return "high"
        elif revenue_impact > 100000 or user_impact > 100000:
            return "medium"
        else:
            return "low"

    def _assess_technical_complexity(self, app_info: Dict[str, Any]) -> str:
        """Assess technical complexity"""
        code_size = app_info.get('code_size_lines', 0)
        dependencies_count = len(app_info.get('dependencies', []))
        database_count = app_info.get('database_count', 0)

        complexity_score = (code_size / 100000) + (dependencies_count * 0.1) + (database_count * 0.5)

        if complexity_score > 10:
            return "high"
        elif complexity_score > 5:
            return "medium"
        else:
            return "low"

    def _assess_dependencies(self, app_info: Dict[str, Any]) -> List[str]:
        """Assess application dependencies"""
        return app_info.get('dependencies', [])

    def _determine_strategy(self, business_value: str, 
                           technical_complexity: str,
                           dependencies: List[str]) -> MigrationStrategy:
        """Determine migration strategy"""
        if business_value == "low" and technical_complexity == "low":
            return MigrationStrategy.RETIRE
        elif business_value == "low" and technical_complexity == "high":
            return MigrationStrategy.RETAIN
        elif business_value == "high" and technical_complexity == "low":
            return MigrationStrategy.REHOST
        elif business_value == "high" and technical_complexity == "medium":
            return MigrationStrategy.REPLATFORM
        elif business_value == "high" and technical_complexity == "high":
            return MigrationStrategy.REFACTOR
        else:
            return MigrationStrategy.REHOST

    def _estimate_effort_cost(self, app_info: Dict[str, Any], 
                             strategy: MigrationStrategy) -> tuple:
        """Estimate effort and cost"""
        base_effort = app_info.get('code_size_lines', 0) / 10000

        effort_multipliers = {
            MigrationStrategy.REHOST: 1,
            MigrationStrategy.REPLATFORM: 2,
            MigrationStrategy.REFACTOR: 4,
            MigrationStrategy.RETIRE: 0.5,
            MigrationStrategy.RETAIN: 0,
            MigrationStrategy.RELOCATE: 1.5,
            MigrationStrategy.REPURCHASE: 1
        }

        effort = base_effort * effort_multipliers.get(strategy, 1)
        cost = effort * 15000  # $15,000 per person-month

        return int(effort), cost

    def _identify_risks(self, app_info: Dict[str, Any], 
                       strategy: MigrationStrategy) -> List[str]:
        """Identify migration risks"""
        risks = []

        if strategy in [MigrationStrategy.REFACTOR, MigrationStrategy.REPLATFORM]:
            risks.append("Complex refactoring may introduce bugs")

        if len(app_info.get('dependencies', [])) > 10:
            risks.append("High dependency count increases complexity")

        if app_info.get('downtime_tolerance', 'low') == 'low':
            risks.append("Low downtime tolerance requires careful planning")

        return risks

    def generate_migration_plan(self) -> Dict[str, Any]:
        """Generate migration plan"""
        plan = {
            'total_apps': len(self.applications),
            'by_strategy': {},
            'total_effort': 0,
            'total_cost': 0,
            'timeline_months': 0
        }

        for app in self.applications:
            strategy = app.recommended_strategy.value
            plan['by_strategy'][strategy] = plan['by_strategy'].get(strategy, 0) + 1
            plan['total_effort'] += app.estimated_effort
            plan['total_cost'] += app.estimated_cost

        plan['timeline_months'] = plan['total_effort'] / 10  # 10 people available

        return plan

Strangler Fig Pattern

# Strangler Fig pattern implementation
from typing import Dict, Any, Callable, List
from dataclasses import dataclass
from enum import Enum

class RequestType(Enum):
    LEGACY = "legacy"
    CLOUD = "cloud"
    HYBRID = "hybrid"

@dataclass
class RouteRule:
    path_pattern: str
    target: RequestType
    percentage: int = 100  # For gradual migration
    enabled: bool = True

class StranglerFigRouter:
    """Strangler Fig pattern router"""

    def __init__(self):
        self.routes: List[RouteRule] = []
        self.legacy_handler: Callable = None
        self.cloud_handlers: Dict[str, Callable] = {}

    def add_route(self, rule: RouteRule):
        """Add routing rule"""
        self.routes.append(rule)

    def set_legacy_handler(self, handler: Callable):
        """Set legacy application handler"""
        self.legacy_handler = handler

    def add_cloud_handler(self, path: str, handler: Callable):
        """Add cloud service handler"""
        self.cloud_handlers[path] = handler

    def route_request(self, path: str, request: Dict[str, Any]) -> Any:
        """Route request based on rules"""
        for rule in self.routes:
            if rule.enabled and self._match_pattern(rule.path_pattern, path):
                if rule.target == RequestType.CLOUD:
                    handler = self.cloud_handlers.get(path)
                    if handler:
                        return handler(request)
                elif rule.target == RequestType.LEGACY:
                    return self.legacy_handler(request)
                elif rule.target == RequestType.HYBRID:
                    # Handle hybrid routing
                    return self._handle_hybrid(path, request)

        # Default to legacy
        return self.legacy_handler(request)

    def _match_pattern(self, pattern: str, path: str) -> bool:
        """Match path pattern"""
        import re
        pattern = pattern.replace('*', '.*')
        return bool(re.match(pattern, path))

    def _handle_hybrid(self, path: str, request: Dict[str, Any]) -> Any:
        """Handle hybrid routing"""
        # Route some traffic to cloud, some to legacy
        import random
        if random.randint(1, 100) <= 50:  # 50% to cloud
            handler = self.cloud_handlers.get(path)
            if handler:
                return handler(request)
        return self.legacy_handler(request)

    def migrate_route(self, path_pattern: str, target: RequestType):
        """Migrate a route to cloud"""
        for rule in self.routes:
            if rule.path_pattern == path_pattern:
                rule.target = target
                rule.percentage = 100
                return

        # Add new rule
        self.routes.append(RouteRule(
            path_pattern=path_pattern,
            target=target
        ))

    def gradual_migration(self, path_pattern: str, percentage: int):
        """Gradually migrate traffic"""
        for rule in self.routes:
            if rule.path_pattern == path_pattern:
                rule.percentage = percentage
                return

class FeatureToggle:
    """Feature toggle for gradual migration"""

    def __init__(self):
        self.toggles: Dict[str, bool] = {}

    def enable(self, feature: str):
        """Enable feature"""
        self.toggles[feature] = True

    def disable(self, feature: str):
        """Disable feature"""
        self.toggles[feature] = False

    def is_enabled(self, feature: str) -> bool:
        """Check if feature is enabled"""
        return self.toggles.get(feature, False)

    def set_percentage(self, feature: str, percentage: int, user_id: str):
        """Set feature percentage for user"""
        import hashlib
        hash_value = int(hashlib.md5(f"{feature}:{user_id}".encode()).hexdigest(), 16)
        enabled = (hash_value % 100) < percentage
        self.toggles[feature] = enabled

Parallel Run Implementation

# Parallel run for migration validation
import boto3
import json
from typing import Dict, Any, Callable
from dataclasses import dataclass
from datetime import datetime

@dataclass
class ParallelRunConfig:
    legacy_endpoint: str
    cloud_endpoint: str
    comparison_strategy: str  # 'exact', 'semantic', 'custom'
    traffic_split: int = 100  # Percentage to cloud
    alert_on_mismatch: bool = True

class ParallelRunManager:
    """Parallel run manager for migration validation"""

    def __init__(self, config: ParallelRunConfig):
        self.config = config
        self.sns = boto3.client('sns')
        self.cloudwatch = boto3.client('cloudwatch')

    def execute_parallel_run(self, request: Dict[str, Any]) -> Dict[str, Any]:
        """Execute parallel run"""
        # Call legacy
        legacy_response = self._call_legacy(request)

        # Call cloud
        cloud_response = self._call_cloud(request)

        # Compare responses
        comparison = self._compare_responses(legacy_response, cloud_response)

        # Log metrics
        self._log_metrics(comparison)

        # Alert on mismatch
        if not comparison['match'] and self.config.alert_on_mismatch:
            self._send_alert(comparison)

        # Return response based on traffic split
        import random
        if random.randint(1, 100) <= self.config.traffic_split:
            return cloud_response
        else:
            return legacy_response

    def _call_legacy(self, request: Dict[str, Any]) -> Dict[str, Any]:
        """Call legacy endpoint"""
        # In production, make HTTP call
        return {'status': 'success', 'source': 'legacy'}

    def _call_cloud(self, request: Dict[str, Any]) -> Dict[str, Any]:
        """Call cloud endpoint"""
        # In production, make HTTP call
        return {'status': 'success', 'source': 'cloud'}

    def _compare_responses(self, legacy: Dict[str, Any], 
                          cloud: Dict[str, Any]) -> Dict[str, Any]:
        """Compare responses"""
        match = legacy == cloud

        return {
            'match': match,
            'legacy': legacy,
            'cloud': cloud,
            'differences': self._find_differences(legacy, cloud),
            'timestamp': datetime.utcnow().isoformat()
        }

    def _find_differences(self, legacy: Dict[str, Any], 
                         cloud: Dict[str, Any]) -> Dict[str, Any]:
        """Find differences between responses"""
        differences = {}

        for key in set(list(legacy.keys()) + list(cloud.keys())):
            if key not in legacy:
                differences[key] = {'type': 'missing_in_legacy', 'cloud_value': cloud[key]}
            elif key not in cloud:
                differences[key] = {'type': 'missing_in_cloud', 'legacy_value': legacy[key]}
            elif legacy[key] != cloud[key]:
                differences[key] = {
                    'type': 'value_mismatch',
                    'legacy_value': legacy[key],
                    'cloud_value': cloud[key]
                }

        return differences

    def _log_metrics(self, comparison: Dict[str, Any]):
        """Log comparison metrics"""
        self.cloudwatch.put_metric_data(
            Namespace='Migration/ParallelRun',
            MetricData=[
                {
                    'MetricName': 'ResponseMatch',
                    'Value': 1 if comparison['match'] else 0,
                    'Unit': 'None'
                },
                {
                    'MetricName': 'DifferencesCount',
                    'Value': len(comparison['differences']),
                    'Unit': 'Count'
                }
            ]
        )

    def _send_alert(self, comparison: Dict[str, Any]):
        """Send alert on mismatch"""
        message = {
            'title': 'Parallel Run Mismatch Detected',
            'differences': comparison['differences'],
            'timestamp': comparison['timestamp']
        }

        self.sns.publish(
            TopicArn='arn:aws:sns:us-east-1:123456789012:migration-alerts',
            Message=json.dumps(message, default=str),
            Subject='Migration Alert'
        )

class MigrationValidator:
    """Migration validation framework"""

    def __init__(self):
        self.validations: list = []

    def add_validation(self, name: str, check_func: Callable, severity: str = 'error'):
        """Add validation check"""
        self.validations.append({
            'name': name,
            'check': check_func,
            'severity': severity
        })

    def validate(self, source_data: Dict[str, Any], 
                target_data: Dict[str, Any]) -> Dict[str, Any]:
        """Validate migration"""
        results = {
            'passed': True,
            'violations': []
        }

        for validation in self.validations:
            try:
                if not validation['check'](source_data, target_data):
                    results['passed'] = False
                    results['violations'].append({
                        'validation': validation['name'],
                        'severity': validation['severity']
                    })
            except Exception as e:
                results['passed'] = False
                results['violations'].append({
                    'validation': validation['name'],
                    'error': str(e),
                    'severity': 'error'
                })

        return results

Summary