A/B Testing ML Technical Debt Management —

ML Technical Debt

A/B Testing ML Technical Debt Management Feature Flag Experiment Lifecycle Cleanup Pipeline Production

Experiment Lifecycle

# === Experiment Lifecycle Management ===



from dataclasses import dataclass

from datetime import datetime, timedelta

from enum import Enum



class ExperimentStatus(Enum):

    DRAFT = "draft"

    RUNNING = "running"

    ANALYZING = "analyzing"

    DECIDED = "decided"

    CLEANUP = "cleanup"

    ARCHIVED = "archived"



@dataclass

class Experiment:

    id: str

    name: str

    owner: str

    status: ExperimentStatus

    start_date: str

    end_date: str

    cleanup_deadline: str

    feature_flags: list

    models: list

    decision: str

    debt_items: list



experiments = [

    Experiment("exp-042", "New Ranking Model v3",

        "ml-team@company.com",

        ExperimentStatus.DECIDED,

        "2025-01-15", "2025-02-15", "2025-03-01",

        ["ranking_v3_enabled", "ranking_v3_percentage"],

        ["ranking-model-v2 (control)", "ranking-model-v3 (treatment)"],

        "SHIP: v3 wins +2.3% conversion",

        ["Remove v2 model", "Remove feature flags", "Update dashboard"]),

    Experiment("exp-043", "Recommendation Algorithm Update",

        "data-team@company.com",

        ExperimentStatus.RUNNING,

        "2025-02-01", "2025-03-01", "2025-03-15",

        ["rec_algo_v2_enabled"],

        ["rec-model-v1 (control)", "rec-model-v2 (treatment)"],

        "PENDING",

        []),

    Experiment("exp-038", "Search Relevance Boost",

        "search-team@company.com",

        ExperimentStatus.ARCHIVED,

        "2024-11-01", "2024-12-01", "2024-12-15",

        [],

        ["search-model-v5 (shipped)"],

        "SHIPPED: +5% relevance score",

        ["All cleanup completed 2024-12-10"]),

]



print("=== Experiment Registry ===")

for e in experiments:

    print(f"  [{e.id}] {e.name}")

    print(f"    Owner: {e.owner} | Status: {e.status.value}")

    print(f"    Period: {e.start_date} → {e.end_date}")

    print(f"    Cleanup Deadline: {e.cleanup_deadline}")

    print(f"    Flags: {e.feature_flags}")

    print(f"    Decision: {e.decision}")

    if e.debt_items:

        print(f"    Debt Items: {e.debt_items}")

Cleanup Automation

# === Automated Cleanup Pipeline ===



# Feature Flag Cleanup Script

# import requests

# from datetime import datetime, timedelta

#

# UNLEASH_URL = "http://unleash:4242/api"

# HEADERS = {"Authorization": "Bearer token"}

#

# # Get all feature flags

# flags = requests.get(f"{UNLEASH_URL}/admin/features", headers=HEADERS).json()

#

# for flag in flags["features"]:

#     created = datetime.fromisoformat(flag["createdAt"])

#     age_days = (datetime.now() - created).days

#     

#     # Flag older than 90 days and stale

#     if age_days > 90 and flag.get("stale", False):

#         print(f"STALE FLAG: {flag['name']} ({age_days} days old)")

#         # Archive flag

#         requests.delete(f"{UNLEASH_URL}/admin/features/{flag['name']}",

#             headers=HEADERS)



@dataclass

class CleanupTask:

    task: str

    trigger: str

    automation: str

    verification: str



tasks = [

    CleanupTask("Remove Feature Flags",

        "Experiment decided + 2 weeks",

        "Script: Unleash API delete stale flags",

        "ตรวจว่า Flag ไม่ถูกอ้างอิงใน Code"),

    CleanupTask("Remove Dead Code Paths",

        "Feature Flag removed",

        "PR: Remove if/else branches for old flags",

        "Unit Test + Integration Test pass"),

    CleanupTask("Decommission Old Models",

        "New model shipped + 4 weeks",

        "Script: Delete model from registry + serving",

        "ตรวจว่าไม่มี Traffic ไปยัง Old Model"),

    CleanupTask("Delete Test Pipelines",

        "Experiment archived",

        "Terraform destroy test-specific resources",

        "ตรวจ Cloud Cost ลดลง"),

    CleanupTask("Archive Dashboards",

        "Experiment archived",

        "Script: Move dashboards to Archive folder",

        "ตรวจว่า Active Dashboard ยังทำงาน"),

    CleanupTask("Update Documentation",

        "Cleanup completed",

        "Manual: Update wiki decision log",

        "Review by Experiment Owner"),

]



print("=== Cleanup Tasks ===")

for t in tasks:

    print(f"  [{t.task}] Trigger: {t.trigger}")

    print(f"    Automation: {t.automation}")

    print(f"    Verify: {t.verification}")

Debt Metrics Dashboard

# === Technical Debt Metrics ===



@dataclass

class DebtMetric:

    metric: str

    current: str

    target: str

    trend: str

    action: str



metrics = [

    DebtMetric("Active Feature Flags",

        "23 flags", "< 10 flags",

        "เพิ่มขึ้น +3 จากเดือนก่อน",

        "Cleanup exp-039 exp-040 exp-041"),

    DebtMetric("Stale Experiments (จบ > 2 wk ไม่ cleanup)",

        "5 experiments", "0",

        "ลดลง -2 จากเดือนก่อน",

        "Enforce cleanup deadline"),

    DebtMetric("Dead Code Paths",

        "12 branches", "0",

        "เท่าเดิม",

        "Sprint task: remove dead branches"),

    DebtMetric("Old Model Versions in Serving",

        "3 old models", "0",

        "เพิ่มขึ้น +1",

        "Decommission ranking-v1 rec-v1"),

    DebtMetric("Monthly Cleanup Rate",

        "65%", "> 90%",

        "เพิ่มขึ้น จาก 50%",

        "Automate cleanup reminders"),

    DebtMetric("Debt Score (composite)",

        "43 points", "< 10",

        "ลดลง จาก 52",

        "Focus on Feature Flag cleanup"),

]



print("=== Debt Metrics ===")

for m in metrics:

    print(f"  [{m.metric}]")

    print(f"    Current: {m.current} | Target: {m.target}")

    print(f"    Trend: {m.trend}")

    print(f"    Action: {m.action}")

เคล็ดลับ

• Deadline: ทุก Experiment ต้องมี Cleanup Deadline 2 สัปดาห์หลังจบ
• Registry: บันทึกทุก Experiment ใน Registry ติดตาม Status ได้
• Automate: ใช้ Script ลบ Stale Feature Flags อัตโนมัติ
• Review: ทุก Quarter Review Debt Metrics ทำ Cleanup Sprint
• Owner: ทุก Experiment ต้องมี Owner รับผิดชอบ Cleanup

การนำไปใช้งานจริงในองค์กร

สำหรับองค์กรขนาดกลางถึงใหญ่ แนะนำให้ใช้หลัก Three-Tier Architecture คือ Core Layer ที่เป็นแกนกลางของระบบ Distribution Layer ที่ทำหน้าที่กระจาย Traffic และ Access Layer ที่เชื่อมต่อกับผู้ใช้โดยตรง การแบ่ง Layer ชัดเจนช่วยให้การ Troubleshoot ง่ายขึ้นและสามารถ Scale ระบบได้ตามความต้องการ

เรื่อง Network Security ก็สำคัญไม่แพ้กัน ควรติดตั้ง Next-Generation Firewall ที่สามารถ Deep Packet Inspection ได้ ใช้ Network Segmentation แยก VLAN สำหรับแต่ละแผนก ติดตั้ง IDS/IPS เพื่อตรวจจับการโจมตี และทำ Regular Security Audit อย่างน้อยปีละ 2 ครั้ง

สรุป

A/B Testing ML Technical Debt Feature Flag Experiment Lifecycle Cleanup Automation Registry Debt Metrics Dashboard Production Pipeline Management