Постепенная замена legacy-кода тестируемым. Миграция без big bang. Практическая стратегия
Strangler Fig Pattern — постепенная замена legacy-кода новым, тестируемым кодом. Как фиговое дерево обвивает старое дерево, пока полностью не заменит его.
Цель: Научиться безопасно мигрировать legacy-код без big bang rewrite.
# ❌ ПЛОХО: Переписать всё сразу
# 1. Остановить разработку на 6 месяцев
# 2. Переписать весь модуль с нуля
# 3. Deploy нового кода
# 4. ??? Всё сломалось
# 5. Откатить на legacyПочему плохо:
Фаза 1: Legacy код + Тесты
┌─────────────────┐
│ Legacy Code │ ← Характеризационные тесты
└─────────────────┘
Фаза 2: Частичная замена
┌─────────────────┐ ┌─────────────────┐
│ Legacy Code │ → │ New Code │ ← Новая функция
│ (старое) │ │ (тестируемая) │
└─────────────────┘ └─────────────────┘
↓ Роутинг решает куда идти
Фаза 3: Полная замена
┌─────────────────┐
│ New Code │ ← Legacy удалён
│ (тестируемый) │
└─────────────────┘
# legacy/discount.py
def calculate_discount_legacy(price, customer_type):
"""Legacy калькулятор"""
if customer_type == "gold":
return price * 0.8
elif customer_type == "silver":
return price * 0.9
else:
return price
# tests/test_discount_characterization.py
def test_legacy_behavior():
"""Зафиксировать текущее поведение"""
assert calculate_discount_legacy(100, "gold") == 80
assert calculate_discount_legacy(100, "silver") == 90# new/discount.py
from dataclasses import dataclass
@dataclass
class DiscountRule:
customer_type: str
discount_percent: float
class DiscountCalculator:
"""Новый калькулятор (тестируемый)"""
def __init__(self, rules: list[DiscountRule]):
self.rules = {r.customer_type: r for r in rules}
def calculate(self, price: float, customer_type: str) -> float:
rule = self.rules.get(customer_type)
if rule:
return price * (1 - rule.discount_percent / 100)
return price
# tests/test_discount_new.py
def test_new_calculator():
"""Тесты нового кода"""
rules = [
DiscountRule("gold", 20),
DiscountRule("silver", 10),
]
calc = DiscountCalculator(rules)
assert calc.calculate(100, "gold") == 80
assert calc.calculate(100, "silver") == 90# discount_facade.py
import os
from legacy.discount import calculate_discount_legacy
from new.discount import DiscountCalculator, DiscountRule
# Feature flag
USE_NEW_CALCULATOR = os.getenv("USE_NEW_DISCOUNT", "false") == "true"
# Инициализация нового калькулятора
new_calculator = DiscountCalculator([
DiscountRule("gold", 20),
DiscountRule("silver", 10),
])
def calculate_discount(price: float, customer_type: str) -> float:
"""Facade: роутинг между legacy и new"""
if USE_NEW_CALCULATOR:
return new_calculator.calculate(price, customer_type)
else:
return calculate_discount_legacy(price, customer_type)# Шаг 1: 10% трафика на новый код
if random.random() < 0.1:
USE_NEW_CALCULATOR = True
# Шаг 2: Сравнение результатов (shadow mode)
legacy_result = calculate_discount_legacy(price, customer_type)
new_result = new_calculator.calculate(price, customer_type)
if legacy_result != new_result:
logger.warning(f"Mismatch: legacy={legacy_result}, new={new_result}")
return legacy_result # Пока используем legacy
# Шаг 3: 50% трафика
# Шаг 4: 100% трафика
# Шаг 5: Удалить legacy код# legacy/user_service.py
class UserService:
def get_user_display_name(self, user_id):
"""Legacy метод (плохой)"""
conn = psycopg2.connect("postgresql://...")
cur = conn.cursor()
cur.execute("SELECT first_name, last_name FROM users WHERE id=%s", (user_id,))
row = cur.fetchone()
if row:
return f"{row[0]} {row[1]}"
return "Unknown User"Задание:
# monitoring.py
import time
from dataclasses import dataclass
@dataclass
class ComparisonMetrics:
legacy_time: float
new_time: float
results_match: bool
exception: Optional[Exception] = None
def compare_implementations(price, customer_type):
"""Сравнить legacy vs new"""
# Legacy
start = time.time()
try:
legacy_result = calculate_discount_legacy(price, customer_type)
legacy_time = time.time() - start
legacy_exception = None
except Exception as e:
legacy_result = None
legacy_exception = e
# New
start = time.time()
try:
new_result = new_calculator.calculate(price, customer_type)
new_time = time.time() - start
new_exception = None
except Exception as e:
new_result = None
new_exception = e
# Сравнение
match = legacy_result == new_result
# Логирование
metrics.record("strangler_fig_comparison", {
"match": match,
"legacy_time_ms": legacy_time * 1000,
"new_time_ms": new_time * 1000,
})
return new_result if USE_NEW_CALCULATOR else legacy_resultВопросы ещё не добавлены
Вопросы для этой подтемы ещё не добавлены.
Далее: Практика: Покрываем модуль платежей