Meta 大规模数据摄入系统迁移实录：保障社交图谱快照可靠性的架构演进

import json
import logging
from enum import Enum
from dataclasses import dataclass

# 假设的底层客户端，实际使用时替换为真实的 Kafka Producer 或 HTTP Client
class LegacyIngestionClient:
    def send(self, topic: str, payload: dict):
        logging.info(f"[Legacy] Sending to {topic}: {json.dumps(payload)[:50]}...")

class NewIngestionClient:
    def send(self, topic: str, payload: dict):
        # 新系统可能有不同的序列化协议或路由规则
        logging.info(f"[New] Sending to {topic}: {json.dumps(payload)[:50]}...")

class MigrationMode(Enum):
    LEGACY_ONLY = 1      # 仅旧系统（初始状态）
    SHADOW = 2           # 影子模式：旧系统为主，新系统并行接收（验证期）
    NEW_ONLY = 3         # 仅新系统（迁移完成）

@dataclass
class IngestionConfig:
    mode: MigrationMode = MigrationMode.LEGACY_ONLY
    shadow_failure_threshold: int = 5  # 影子模式连续失败多少次后报警/熔断

class IngestionGateway:
    def __init__(self, config: IngestionConfig):
        self.config = config
        self.legacy_client = LegacyIngestionClient()
        self.new_client = NewIngestionClient()
        self.shadow_fail_count = 0

    def ingest(self, topic: str, payload: dict):
        """根据当前配置路由数据摄入流量"""
        if self.config.mode == MigrationMode.LEGACY_ONLY:
            self.legacy_client.send(topic, payload)

        elif self.config.mode == MigrationMode.SHADOW:
            # 1. 主流量依然发往旧系统，确保业务不受影响
            self.legacy_client.send(topic, payload)

            # 2. 影子流量发往新系统，捕获异常但不影响主流程
            try:
                self.new_client.send(topic, payload)
                self.shadow_fail_count = 0
            except Exception as e:
                self.shadow_fail_count += 1
                logging.error(f"Shadow ingestion failed: {e}. Fail count: {self.shadow_fail_count}")
                if self.shadow_fail_count >= self.config.shadow_failure_threshold:
                    # 触发熔断报警：新系统存在严重问题，需人工介入
                    logging.critical("Shadow mode circuit breaker triggered! Halting shadow ingestion.")
                    # 可以在此处自动将 config.mode 切回 LEGACY_ONLY 或停止影子发送
                    self.config.mode = MigrationMode.LEGACY_ONLY

        elif self.config.mode == MigrationMode.NEW_ONLY:
            self.new_client.send(topic, payload)

# --- 运行示例 ---
if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)

    # 初始配置：仅旧系统
    config = IngestionConfig(mode=MigrationMode.LEGACY_ONLY)
    gateway = IngestionGateway(config)

    sample_data = {"user_id": "u123", "action": "add_friend", "target_id": "u456"}

    # 阶段 1: 业务正常运转
    gateway.ingest("social_graph_updates", sample_data)

    # 阶段 2: 开启影子模式验证新系统
    print("\n--- Switching to SHADOW mode ---")
    config.mode = MigrationMode.SHADOW
    gateway.ingest("social_graph_updates", sample_data)

    # 阶段 3: 验证通过，全量切到新系统
    print("\n--- Switching to NEW_ONLY mode ---")
    config.mode = MigrationMode.NEW_ONLY
    gateway.ingest("social_graph_updates", sample_data)