Strangler Fig 패턴 완벽 가이드: 모놀리스에서 마이크로서비스로의 무중단 마이그레이션

Phase 1: 마이그레이션 시작
┌──────────────┐     ┌──────────────┐
│   Facade     │────▶│  Monolith    │
│  (Proxy)     │     │  ┌────────┐  │
│              │     │  │Service │  │
│              │     │  │   A    │  │
│              │     │  ├────────┤  │
│              │     │  │Service │  │
│              │     │  │   B    │  │
│              │     │  ├────────┤  │
│              │     │  │Service │  │
│              │     │  │   C    │  │
│              │     │  └────────┘  │
└──────────────┘     └──────────────┘

Phase 2: 점진적 추출
┌──────────────┐     ┌──────────────┐
│   Facade     │────▶│  Monolith    │
│  (Proxy)     │     │  ┌────────┐  │
│              │     │  │Service │  │
│              │─┐   │  │   B    │  │
│              │ │   │  ├────────┤  │
│              │ │   │  │Service │  │
│              │ │   │  │   C    │  │
└──────────────┘ │   └──────────────┘
                 │
                 │   ┌──────────────┐
                 └──▶│ Microservice │
                     │     A        │
                     └──────────────┘

Phase 3: 완전 전환
┌──────────────┐     ┌──────────────┐
│   Facade     │────▶│ Microservice │
│  (Proxy)     │     │     A        │
│              │     └──────────────┘
│              │     ┌──────────────┐
│              │────▶│ Microservice │
│              │     │     B        │
│              │     └──────────────┘
│              │     ┌──────────────┐
│              │────▶│ Microservice │
│              │     │     C        │
└──────────────┘     └──────────────┘

// strangler-fig-router.ts
// Strangler Fig 패턴 - API Gateway 라우팅 핸들러

interface RouteConfig {
  path: string
  target: 'legacy' | 'microservice'
  serviceUrl: string
  migrationStatus: 'not_started' | 'canary' | 'partial' | 'complete'
  canaryPercentage?: number // 신규 서비스로 보낼 트래픽 비율 (0-100)
  fallbackToLegacy: boolean
  healthCheckUrl: string
}

interface ServiceHealth {
  isHealthy: boolean
  lastChecked: Date
  consecutiveFailures: number
}

const routeConfigs: RouteConfig[] = [
  {
    path: '/api/orders/*',
    target: 'microservice',
    serviceUrl: 'http://order-service:8080',
    migrationStatus: 'complete',
    fallbackToLegacy: true,
    healthCheckUrl: 'http://order-service:8080/health',
  },
  {
    path: '/api/products/*',
    target: 'microservice',
    serviceUrl: 'http://product-service:8081',
    migrationStatus: 'canary',
    canaryPercentage: 20,
    fallbackToLegacy: true,
    healthCheckUrl: 'http://product-service:8081/health',
  },
  {
    path: '/api/users/*',
    target: 'legacy',
    serviceUrl: 'http://monolith:3000',
    migrationStatus: 'not_started',
    fallbackToLegacy: false,
    healthCheckUrl: 'http://monolith:3000/health',
  },
]

const LEGACY_BASE_URL = 'http://monolith:3000'
const healthCache = new Map<string, ServiceHealth>()

async function checkServiceHealth(config: RouteConfig): Promise<boolean> {
  const cached = healthCache.get(config.path)
  if (cached && Date.now() - cached.lastChecked.getTime() < 5000) {
    return cached.isHealthy
  }

  try {
    const response = await fetch(config.healthCheckUrl, {
      signal: AbortSignal.timeout(2000),
    })
    const isHealthy = response.ok
    healthCache.set(config.path, {
      isHealthy,
      lastChecked: new Date(),
      consecutiveFailures: isHealthy ? 0 : (cached?.consecutiveFailures ?? 0) + 1,
    })
    return isHealthy
  } catch {
    const failures = (cached?.consecutiveFailures ?? 0) + 1
    healthCache.set(config.path, {
      isHealthy: false,
      lastChecked: new Date(),
      consecutiveFailures: failures,
    })
    return false
  }
}

function shouldRouteToMicroservice(config: RouteConfig, requestId: string): boolean {
  if (config.migrationStatus === 'complete') return true
  if (config.migrationStatus === 'not_started') return false

  if (config.migrationStatus === 'canary' && config.canaryPercentage) {
    // 요청 ID 기반 결정적 라우팅 (동일 사용자는 항상 동일 서비스로)
    const hash = simpleHash(requestId)
    return hash % 100 < config.canaryPercentage
  }

  return false
}

function simpleHash(str: string): number {
  let hash = 0
  for (let i = 0; i < str.length; i++) {
    const char = str.charCodeAt(i)
    hash = (hash << 5) - hash + char
    hash = hash & hash // 32bit 정수 변환
  }
  return Math.abs(hash)
}

async function routeRequest(path: string, requestId: string, request: Request): Promise<Response> {
  const config = routeConfigs.find((r) => path.startsWith(r.path.replace('/*', '')))

  if (!config) {
    // 매칭되는 라우트 없음 - 레거시로 전달
    return fetch(`${LEGACY_BASE_URL}${path}`, { ...request })
  }

  const useNewService = shouldRouteToMicroservice(config, requestId)

  if (useNewService) {
    const isHealthy = await checkServiceHealth(config)

    if (isHealthy) {
      try {
        const response = await fetch(`${config.serviceUrl}${path}`, { ...request })
        // 성공 메트릭 기록
        recordMetric('route_to_microservice', config.path, 'success')
        return response
      } catch (error) {
        recordMetric('route_to_microservice', config.path, 'error')

        if (config.fallbackToLegacy) {
          console.warn(`Microservice 실패, 레거시 폴백: ${config.path}`)
          recordMetric('fallback_to_legacy', config.path, 'triggered')
          return fetch(`${LEGACY_BASE_URL}${path}`, { ...request })
        }
        throw error
      }
    } else if (config.fallbackToLegacy) {
      recordMetric('fallback_to_legacy', config.path, 'health_check_failed')
      return fetch(`${LEGACY_BASE_URL}${path}`, { ...request })
    }
  }

  return fetch(`${LEGACY_BASE_URL}${path}`, { ...request })
}

function recordMetric(type: string, path: string, status: string): void {
  // Prometheus, Datadog 등 메트릭 시스템으로 전송
  console.log(`[METRIC] ${type} | path=${path} | status=${status} | ts=${Date.now()}`)
}

# anti_corruption_layer.py
# 주문 서비스 Anti-Corruption Layer

from dataclasses import dataclass
from datetime import datetime
from decimal import Decimal
from enum import Enum
from typing import Optional
import httpx


# === 신규 마이크로서비스 도메인 모델 ===
class OrderStatus(Enum):
    PENDING = "pending"
    CONFIRMED = "confirmed"
    SHIPPED = "shipped"
    DELIVERED = "delivered"
    CANCELLED = "cancelled"


@dataclass
class OrderItem:
    product_id: str
    product_name: str
    quantity: int
    unit_price: Decimal
    total_price: Decimal


@dataclass
class Order:
    order_id: str
    customer_id: str
    items: list[OrderItem]
    status: OrderStatus
    total_amount: Decimal
    currency: str
    created_at: datetime
    updated_at: datetime


# === ACL: 레거시 모놀리스 데이터 변환 ===
class LegacyOrderACL:
    """
    레거시 모놀리스의 주문 데이터 모델을 신규 마이크로서비스의
    도메인 모델로 변환하는 Anti-Corruption Layer.

    레거시 시스템은 다음과 같은 문제를 가지고 있다:
    - 주문 상태가 숫자 코드 (0, 1, 2, 3, 4)
    - 금액이 정수형 (센트 단위)
    - 날짜가 Unix timestamp
    - 상품 정보가 주문과 같은 테이블에 JSON 문자열로 저장
    """

    # 레거시 숫자 코드 -> 신규 도메인 상태 매핑
    STATUS_MAP: dict[int, OrderStatus] = {
        0: OrderStatus.PENDING,
        1: OrderStatus.CONFIRMED,
        2: OrderStatus.SHIPPED,
        3: OrderStatus.DELIVERED,
        4: OrderStatus.CANCELLED,
    }

    def __init__(self, legacy_api_url: str):
        self.legacy_api_url = legacy_api_url
        self.client = httpx.AsyncClient(
            base_url=legacy_api_url,
            timeout=10.0,
        )

    async def get_order(self, order_id: str) -> Optional[Order]:
        """레거시 API에서 주문을 조회하고 신규 도메인 모델로 변환한다."""
        response = await self.client.get(f"/legacy/orders/{order_id}")
        if response.status_code == 404:
            return None

        legacy_data = response.json()
        return self._translate_order(legacy_data)

    def _translate_order(self, legacy: dict) -> Order:
        """레거시 주문 데이터를 신규 Order 도메인 객체로 변환한다."""
        items = self._translate_items(legacy.get("items_json", "[]"))
        total = Decimal(legacy["total_cents"]) / 100

        return Order(
            order_id=str(legacy["id"]),
            customer_id=str(legacy["cust_id"]),
            items=items,
            status=self.STATUS_MAP.get(legacy["stat_cd"], OrderStatus.PENDING),
            total_amount=total,
            currency=legacy.get("curr", "KRW"),
            created_at=datetime.fromtimestamp(legacy["created_ts"]),
            updated_at=datetime.fromtimestamp(legacy["modified_ts"]),
        )

    def _translate_items(self, items_json: str) -> list[OrderItem]:
        """레거시 JSON 문자열 형태의 상품 목록을 OrderItem 리스트로 변환한다."""
        import json
        raw_items = json.loads(items_json)
        return [
            OrderItem(
                product_id=str(item["pid"]),
                product_name=item.get("pname", "Unknown"),
                quantity=item["qty"],
                unit_price=Decimal(item["price_cents"]) / 100,
                total_price=Decimal(item["price_cents"] * item["qty"]) / 100,
            )
            for item in raw_items
        ]

    def translate_to_legacy(self, order: Order) -> dict:
        """신규 Order를 레거시 형식으로 역변환한다 (레거시 시스템 업데이트 시 사용)."""
        reverse_status = {v: k for k, v in self.STATUS_MAP.items()}
        return {
            "id": int(order.order_id),
            "cust_id": int(order.customer_id),
            "stat_cd": reverse_status.get(order.status, 0),
            "total_cents": int(order.total_amount * 100),
            "curr": order.currency,
            "created_ts": int(order.created_at.timestamp()),
            "modified_ts": int(order.updated_at.timestamp()),
        }

    async def close(self):
        await self.client.aclose()

// feature-flag-migration.ts
// Feature Flag 기반 마이그레이션 제어

interface MigrationFlag {
  name: string
  enabled: boolean
  rolloutPercentage: number // 0-100
  allowedUserIds?: string[] // 화이트리스트 (내부 테스터)
  excludedUserIds?: string[] // 블랙리스트 (VIP 고객 등 보수적 전환)
  enabledRegions?: string[] // 지역별 롤아웃
  createdAt: string
  updatedAt: string
  metadata: Record<string, string>
}

class MigrationFeatureFlagService {
  private flags: Map<string, MigrationFlag> = new Map()
  private refreshIntervalMs = 30_000 // 30초마다 갱신

  constructor(private flagSource: string) {
    this.startPeriodicRefresh()
  }

  async loadFlags(): Promise<void> {
    try {
      const response = await fetch(this.flagSource)
      const data: MigrationFlag[] = await response.json()
      this.flags = new Map(data.map((f) => [f.name, f]))
    } catch (error) {
      console.error('Feature flag 로드 실패, 캐시된 값 유지:', error)
      // 로드 실패 시 기존 캐시 유지 - 안전한 기본 동작
    }
  }

  isEnabled(
    flagName: string,
    context: {
      userId: string
      region?: string
      sessionId?: string
    }
  ): boolean {
    const flag = this.flags.get(flagName)
    if (!flag || !flag.enabled) return false

    // 블랙리스트 확인 (최우선 - VIP 고객 보호)
    if (flag.excludedUserIds?.includes(context.userId)) {
      return false
    }

    // 화이트리스트 확인 (내부 테스터 우선 적용)
    if (flag.allowedUserIds?.includes(context.userId)) {
      return true
    }

    // 지역별 제한 확인
    if (flag.enabledRegions && context.region) {
      if (!flag.enabledRegions.includes(context.region)) {
        return false
      }
    }

    // 비율 기반 롤아웃 (사용자 ID 기반 결정적 분배)
    if (flag.rolloutPercentage < 100) {
      const hash = this.consistentHash(context.userId + flagName)
      return hash % 100 < flag.rolloutPercentage
    }

    return true
  }

  private consistentHash(input: string): number {
    let hash = 5381
    for (let i = 0; i < input.length; i++) {
      hash = (hash << 5) + hash + input.charCodeAt(i)
      hash = hash & hash
    }
    return Math.abs(hash)
  }

  private startPeriodicRefresh(): void {
    setInterval(() => this.loadFlags(), this.refreshIntervalMs)
  }
}

// 사용 예시: 주문 서비스 마이그레이션
const flagService = new MigrationFeatureFlagService('http://config-server:8888/flags/migration')

async function processOrder(userId: string, orderData: unknown) {
  const useNewOrderService = flagService.isEnabled('migration.order-service', {
    userId,
    region: 'ap-northeast-2',
  })

  if (useNewOrderService) {
    // 신규 마이크로서비스로 주문 처리
    return await newOrderService.createOrder(orderData)
  } else {
    // 레거시 모놀리스로 주문 처리
    return await legacyMonolith.createOrder(orderData)
  }
}

# docker-compose.yml - Debezium CDC 구성 예시
version: '3.8'
services:
  # 레거시 모놀리스 DB (MySQL)
  legacy-db:
    image: mysql:8.0
    environment:
      MYSQL_ROOT_PASSWORD: legacy_password
      MYSQL_DATABASE: monolith
    ports:
      - '3306:3306'
    volumes:
      - legacy-data:/var/lib/mysql
    command: >
      --server-id=1
      --log-bin=mysql-bin
      --binlog-format=ROW
      --binlog-row-image=FULL
      --gtid-mode=ON
      --enforce-gtid-consistency=ON

  # Kafka - CDC 이벤트 스트리밍
  kafka:
    image: confluentinc/cp-kafka:7.6.0
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'true'
    ports:
      - '9092:9092'
    depends_on:
      - zookeeper

  # Debezium Connect - CDC 커넥터
  debezium:
    image: debezium/connect:2.5
    environment:
      BOOTSTRAP_SERVERS: kafka:9092
      GROUP_ID: strangler-fig-cdc
      CONFIG_STORAGE_TOPIC: cdc_configs
      OFFSET_STORAGE_TOPIC: cdc_offsets
      STATUS_STORAGE_TOPIC: cdc_status
    ports:
      - '8083:8083'
    depends_on:
      - kafka
      - legacy-db

  # 신규 마이크로서비스 DB (PostgreSQL)
  order-service-db:
    image: postgres:16
    environment:
      POSTGRES_DB: orders
      POSTGRES_USER: order_service
      POSTGRES_PASSWORD: secure_password
    ports:
      - '5432:5432'
    volumes:
      - order-data:/var/lib/postgresql/data

volumes:
  legacy-data:
  order-data:

# cdc_consumer.py
# Debezium CDC 이벤트를 소비하여 마이크로서비스 DB에 동기화

import json
from kafka import KafkaConsumer
from datetime import datetime
from decimal import Decimal


class CDCEventConsumer:
    """
    레거시 DB의 변경사항을 Debezium CDC로 캡처하여
    신규 마이크로서비스 DB에 동기화하는 컨슈머.
    """

    def __init__(self, kafka_servers: str, topic: str, order_repository):
        self.consumer = KafkaConsumer(
            topic,
            bootstrap_servers=kafka_servers,
            group_id='order-service-cdc-sync',
            auto_offset_reset='earliest',
            enable_auto_commit=False,  # 수동 커밋으로 exactly-once 보장
            value_deserializer=lambda m: json.loads(m.decode('utf-8')),
        )
        self.order_repo = order_repository
        self.processed_count = 0
        self.error_count = 0

    def start(self):
        """CDC 이벤트 소비 루프를 시작한다."""
        print(f"CDC Consumer 시작 - 레거시 DB 변경 감지 중...")

        for message in self.consumer:
            try:
                self._process_event(message.value)
                self.consumer.commit()
                self.processed_count += 1

                if self.processed_count % 1000 == 0:
                    print(f"처리 완료: {self.processed_count}건, "
                          f"오류: {self.error_count}건")
            except Exception as e:
                self.error_count += 1
                print(f"CDC 이벤트 처리 실패: {e}")
                # Dead Letter Queue로 전송
                self._send_to_dlq(message.value, str(e))

    def _process_event(self, event: dict):
        """Debezium CDC 이벤트를 파싱하고 적절한 핸들러로 전달한다."""
        operation = event.get('op')  # c=create, u=update, d=delete, r=read(snapshot)
        after = event.get('after')   # 변경 후 데이터
        before = event.get('before') # 변경 전 데이터

        if operation in ('c', 'r'):
            # INSERT 또는 스냅샷
            self._handle_create(after)
        elif operation == 'u':
            # UPDATE
            self._handle_update(before, after)
        elif operation == 'd':
            # DELETE
            self._handle_delete(before)

    def _handle_create(self, data: dict):
        """레거시 DB INSERT를 신규 서비스 DB에 반영한다."""
        order = self._transform_legacy_to_new(data)
        self.order_repo.upsert(order)

    def _handle_update(self, before: dict, after: dict):
        """레거시 DB UPDATE를 신규 서비스 DB에 반영한다."""
        order = self._transform_legacy_to_new(after)
        self.order_repo.upsert(order)

    def _handle_delete(self, data: dict):
        """레거시 DB DELETE를 신규 서비스 DB에 반영한다."""
        order_id = str(data['id'])
        self.order_repo.soft_delete(order_id)

    def _transform_legacy_to_new(self, legacy: dict) -> dict:
        """레거시 데이터 형식을 신규 마이크로서비스 형식으로 변환한다."""
        status_map = {0: 'pending', 1: 'confirmed', 2: 'shipped',
                      3: 'delivered', 4: 'cancelled'}

        return {
            'order_id': str(legacy['id']),
            'customer_id': str(legacy['cust_id']),
            'status': status_map.get(legacy.get('stat_cd', 0), 'pending'),
            'total_amount': Decimal(legacy.get('total_cents', 0)) / 100,
            'currency': legacy.get('curr', 'KRW'),
            'created_at': datetime.fromtimestamp(legacy.get('created_ts', 0)),
            'updated_at': datetime.fromtimestamp(legacy.get('modified_ts', 0)),
            'sync_source': 'cdc_legacy',
            'synced_at': datetime.utcnow(),
        }

    def _send_to_dlq(self, event: dict, error_msg: str):
        """처리 실패한 이벤트를 Dead Letter Queue로 전송한다."""
        # DLQ 토픽으로 전송하여 수동 재처리 가능하도록 보관
        print(f"DLQ 전송: order_id={event.get('after', {}).get('id')}, "
              f"error={error_msg}")

# rollback-playbook.yaml
# 마이크로서비스 마이그레이션 롤백 플레이북

rollback_triggers:
  - condition: '신규 서비스 에러율 > 1% (5분 지속)'
    severity: warning
    action: '카나리 비율을 이전 단계로 축소'

  - condition: '신규 서비스 P99 레이턴시 > 레거시 대비 200% (10분 지속)'
    severity: critical
    action: '즉시 전체 트래픽을 레거시로 전환'

  - condition: '데이터 불일치 건수 > 10건/분'
    severity: critical
    action: 'CDC 동기화 중단, 트래픽 레거시 전환, 데이터 정합성 검증'

rollback_procedures:
  instant_rollback:
    description: 'API Gateway 라우팅 변경으로 즉시 롤백 (소요시간: ~10초)'
    steps:
      - 'API Gateway 라우팅 규칙에서 대상 서비스를 legacy로 변경'
      - 'Feature Flag를 disabled로 설정'
      - '모니터링 대시보드에서 레거시 서비스 정상 동작 확인'
      - '장애 원인 분석 및 포스트모템 시작'
    estimated_time: '10초 ~ 1분'

  gradual_rollback:
    description: '카나리 비율을 단계적으로 축소 (소요시간: ~30분)'
    steps:
      - '카나리 비율을 현재 값의 50%로 축소'
      - '5분 대기 후 에러율/레이턴시 확인'
      - '정상이면 유지, 비정상이면 카나리 비율을 0%로 설정'
      - '데이터 정합성 검증 스크립트 실행'
    estimated_time: '30분 ~ 1시간'

  data_rollback:
    description: '데이터 불일치 발생 시 복구 절차'
    steps:
      - 'CDC 파이프라인 즉시 중단'
      - '트래픽을 레거시로 전환'
      - '신규 서비스 DB의 변경 사항을 레거시 DB와 비교'
      - '불일치 데이터 식별 및 보정 스크립트 실행'
      - '정합성 검증 후 CDC 파이프라인 재시작'
    estimated_time: '1시간 ~ 4시간'

monitoring_during_rollback:
  metrics:
    - '에러율 (5xx responses)'
    - 'P50/P95/P99 레이턴시'
    - '초당 요청 수 (RPS)'
    - '데이터베이스 커넥션 풀 사용률'
    - 'CDC 컨슈머 지연 (lag)'
  alerts:
    - '롤백 시작/완료 Slack 알림'
    - '데이터 정합성 검증 결과 보고'
    - '포스트모템 일정 자동 생성'