【系统架构设计百科】Google 基础设施：Borg、Spanner 与全球化架构

# Borg Job 配置示例（BCL 风格伪代码）
job = Job(
    name = "web-search-frontend",
    priority = PRODUCTION,
    instances = 500,

    # 资源需求
    requirements = Requirements(
        cpu = 2.0,          # 2 个 CPU 核心
        ram = 4 * GiB,      # 4 GiB 内存
        disk = 50 * GiB,    # 50 GiB 磁盘
    ),

    # 约束条件
    constraints = [
        # 分散在不同故障域
        Constraint(attribute="failure_domain", operator=SPREAD),
        # 只使用特定机型
        Constraint(attribute="machine_type", operator=IN, values=["n2-standard"]),
    ],

    # 容器配置
    container = Container(
        image = "//search/frontend:release-2026.04",
        ports = [Port(name="http", number=8080)],
        health_check = HealthCheck(
            path = "/healthz",
            interval = 10 * SECONDS,
            timeout = 3 * SECONDS,
        ),
    ),

    # 更新策略
    update_strategy = RollingUpdate(
        max_unavailable = 10,       # 最多 10 个实例同时不可用
        max_surge = 50,             # 最多额外启动 50 个实例
        canary_count = 5,           # 先更新 5 个金丝雀实例
        canary_duration = 300,      # 金丝雀观察 300 秒
    ),
)

# 资源回收算法伪代码
class ResourceReclaimer:
    def __init__(self, safety_margin: float = 0.1):
        self.safety_margin = safety_margin

    def calculate_reclaimable(self, task: Task) -> Resources:
        """计算可回收资源"""
        reserved = task.resource_request        # 申请量
        actual_peak = task.peak_usage(window=300)  # 过去 5 分钟峰值
        safe_usage = actual_peak * (1 + self.safety_margin)  # 安全用量

        reclaimable = Resources(
            cpu = max(0, reserved.cpu - safe_usage.cpu),
            ram = max(0, reserved.ram - safe_usage.ram),
        )
        return reclaimable

    def reclaim_for_cell(self, cell: Cell) -> Resources:
        """计算整个 Cell 的可回收资源"""
        total_reclaimable = Resources(cpu=0, ram=0)
        for machine in cell.machines:
            for task in machine.running_tasks:
                if task.priority >= PRODUCTION:
                    reclaimable = self.calculate_reclaimable(task)
                    total_reclaimable += reclaimable
        return total_reclaimable

# Kubernetes Label 和 Selector 示例
apiVersion: apps/v1
kind: Deployment
metadata:
  name: web-search-frontend
  labels:
    app: web-search
    tier: frontend
    version: v2
    team: search-infra
spec:
  replicas: 500
  selector:
    matchLabels:
      app: web-search
      tier: frontend
  template:
    metadata:
      labels:
        app: web-search
        tier: frontend
        version: v2
    spec:
      containers:
      - name: search-frontend
        image: gcr.io/search/frontend:release-2026.04
        ports:
        - containerPort: 8080
          name: http
        resources:
          requests:
            cpu: "2"
            memory: "4Gi"
          limits:
            cpu: "4"
            memory: "8Gi"
        readinessProbe:
          httpGet:
            path: /healthz
            port: 8080
          periodSeconds: 10
          timeoutSeconds: 3

# Kubernetes 声明式 API 示例：自定义资源定义（CRD）
apiVersion: apiextensions.k8s.io/v1
kind: CustomResourceDefinition
metadata:
  name: spannerinstances.database.google.com
spec:
  group: database.google.com
  versions:
    - name: v1
      served: true
      storage: true
      schema:
        openAPIV3Schema:
          type: object
          properties:
            spec:
              type: object
              properties:
                displayName:
                  type: string
                nodeCount:
                  type: integer
                  minimum: 1
                config:
                  type: string
  scope: Namespaced
  names:
    plural: spannerinstances
    singular: spannerinstance
    kind: SpannerInstance

# Autopilot 资源推荐算法伪代码
class AutopilotRecommender:
    def __init__(self):
        self.model = load_ml_model("resource_predictor_v3")
        self.history_window = 7 * 24 * 3600  # 7 天历史数据

    def recommend(self, task_group: TaskGroup) -> ResourceRecommendation:
        """基于历史数据和 ML 模型推荐资源配置"""
        # 收集历史资源使用数据
        history = task_group.get_usage_history(self.history_window)

        # 提取特征
        features = self.extract_features(history)

        # ML 模型预测未来资源需求
        predicted_peak = self.model.predict(features)

        # 添加安全裕度
        recommendation = ResourceRecommendation(
            cpu = predicted_peak.cpu * 1.15,      # 15% 安全裕度
            memory = predicted_peak.memory * 1.20,  # 20% 安全裕度（OOM 代价更高）
            confidence = self.model.confidence(features),
        )

        return recommendation

    def extract_features(self, history: UsageHistory) -> Features:
        """从历史数据中提取特征"""
        return Features(
            p50_cpu = history.percentile(metric="cpu", p=50),
            p95_cpu = history.percentile(metric="cpu", p=95),
            p99_cpu = history.percentile(metric="cpu", p=99),
            p50_memory = history.percentile(metric="memory", p=50),
            p95_memory = history.percentile(metric="memory", p=95),
            p99_memory = history.percentile(metric="memory", p=99),
            diurnal_pattern = history.detect_diurnal_pattern(),
            growth_rate = history.linear_growth_rate(),
            task_count = history.avg_task_count(),
        )

-- Spanner 表定义示例（DDL）
CREATE TABLE Users (
    UserId     INT64 NOT NULL,
    Email      STRING(256) NOT NULL,
    Name       STRING(128),
    CreatedAt  TIMESTAMP NOT NULL OPTIONS (allow_commit_timestamp=true),
) PRIMARY KEY (UserId);

-- Albums 是 Users 的子表，物理上交错存储
CREATE TABLE Albums (
    UserId     INT64 NOT NULL,
    AlbumId    INT64 NOT NULL,
    Title      STRING(256),
    CreatedAt  TIMESTAMP NOT NULL OPTIONS (allow_commit_timestamp=true),
) PRIMARY KEY (UserId, AlbumId),
  INTERLEAVE IN PARENT Users ON DELETE CASCADE;

-- Photos 是 Albums 的子表
CREATE TABLE Photos (
    UserId     INT64 NOT NULL,
    AlbumId    INT64 NOT NULL,
    PhotoId    INT64 NOT NULL,
    FileName   STRING(512),
    FileSize   INT64,
    UploadedAt TIMESTAMP NOT NULL OPTIONS (allow_commit_timestamp=true),
) PRIMARY KEY (UserId, AlbumId, PhotoId),
  INTERLEAVE IN PARENT Albums ON DELETE CASCADE;

-- 二级索引
CREATE INDEX PhotosByUploadTime ON Photos(UploadedAt DESC);
CREATE UNIQUE INDEX UsersByEmail ON Users(Email);

// TrueTime API 定义
type TrueTime interface {
    // Now 返回一个时间区间 [earliest, latest]
    // 保证真实时间在这个区间内
    Now() TTInterval

    // After 判断时间 t 是否确定已经过去
    After(t Timestamp) bool

    // Before 判断时间 t 是否确定还未到来
    Before(t Timestamp) bool
}

// TTInterval 表示一个时间不确定性区间
type TTInterval struct {
    Earliest Timestamp  // 真实时间不早于此
    Latest   Timestamp  // 真实时间不晚于此
}

// 不确定性范围 epsilon = (Latest - Earliest) / 2
// 通常 epsilon 在 1-7 毫秒之间，平均约 4 毫秒

# TrueTime 时间同步示例
class TrueTimeService:
    """TrueTime 客户端守护进程（每台机器运行一个）"""

    def __init__(self):
        self.time_masters = discover_time_masters()
        self.poll_interval = 30  # 每 30 秒轮询一次
        self.local_clock_drift_rate = 200e-6  # 200 ppm 晶振漂移率

    def now(self) -> TTInterval:
        """返回当前时间的不确定性区间"""
        local_time = read_local_clock()
        epsilon = self.calculate_epsilon()

        return TTInterval(
            earliest=local_time - epsilon,
            latest=local_time + epsilon,
        )

    def calculate_epsilon(self) -> Duration:
        """计算当前的时间不确定性"""
        time_since_last_sync = now() - self.last_sync_time
        drift_uncertainty = time_since_last_sync * self.local_clock_drift_rate
        sync_uncertainty = self.last_sync_epsilon  # 上次同步时的不确定性

        return sync_uncertainty + drift_uncertainty

    def sync_with_masters(self):
        """与多个 Time Master 同步"""
        readings = []
        for master in self.time_masters:
            rtt_start = read_local_clock()
            master_time = master.get_time()
            rtt_end = read_local_clock()
            rtt = rtt_end - rtt_start

            readings.append(TimeReading(
                master_time=master_time,
                rtt=rtt,
                uncertainty=rtt / 2,  # 网络延迟不确定性
            ))

        # 使用 Marzullo 算法选择最佳时间区间
        best_interval = marzullo_algorithm(readings)
        self.apply_correction(best_interval)
        self.last_sync_time = read_local_clock()
        self.last_sync_epsilon = best_interval.uncertainty

// Commit Wait 机制伪代码
func (txn *Transaction) Commit() error {
    // 1. 获取 Paxos 多数派确认
    err := txn.paxosGroup.ReplicateWrite(txn.mutations)
    if err != nil {
        return err
    }

    // 2. 分配提交时间戳：取当前 TrueTime 区间的 latest 值
    commitTS := TrueTime.Now().Latest

    // 3. Commit Wait：等待直到 TrueTime 确认 commitTS 已经过去
    // 这保证了任何后续事务看到的时间一定晚于 commitTS
    for !TrueTime.After(commitTS) {
        sleep(1 * time.Millisecond)
    }

    // 4. 提交完成，通知客户端
    txn.status = COMMITTED
    txn.commitTimestamp = commitTS
    return nil
}

# B4 流量工程带宽分配伪代码
class TrafficEngineer:
    def __init__(self, topology: NetworkTopology):
        self.topology = topology

    def allocate_bandwidth(
        self, demands: list, link_capacities: dict
    ) -> dict:
        """基于优先级的最大最小公平分配"""
        allocations = {}
        remaining_capacity = dict(link_capacities)

        # 按优先级从高到低分配
        for priority in [HIGH, MEDIUM, LOW]:
            priority_demands = [
                d for d in demands if d.priority == priority
            ]

            # 为当前优先级的流量寻找最短路径
            for demand in priority_demands:
                paths = self.find_k_shortest_paths(
                    demand.src, demand.dst, k=3
                )

                # 在多条路径上分配带宽（ECMP 扩展）
                allocated = self.split_across_paths(
                    demand, paths, remaining_capacity
                )
                allocations[demand.id] = allocated

                # 更新剩余容量
                for path, bw in allocated.items():
                    for link in path.links:
                        remaining_capacity[link] -= bw

        return allocations

    def find_k_shortest_paths(self, src, dst, k=3):
        """使用 Yen 算法找到 K 条最短路径"""
        return yen_k_shortest_paths(self.topology, src, dst, k)

    def split_across_paths(self, demand, paths, capacity):
        """将流量需求分配到多条路径上"""
        total_available = sum(
            min(capacity[link] for link in p.links) for p in paths
        )
        allocated = {}

        for path in paths:
            path_capacity = min(capacity[link] for link in path.links)
            share = min(
                demand.bandwidth * (path_capacity / total_available),
                path_capacity,
            )
            allocated[path] = share

        return allocated

# Colossus 纠删码存储示例
class ColossusErasureCoding:
    """Reed-Solomon 纠删码配置"""

    def __init__(self, data_chunks: int = 6, parity_chunks: int = 3):
        self.data_chunks = data_chunks        # 数据块数量
        self.parity_chunks = parity_chunks    # 校验块数量
        self.total_chunks = data_chunks + parity_chunks
        # 存储开销 = 9/6 = 1.5 倍（对比三副本的 3 倍）

    def encode(self, data: bytes) -> list:
        """将数据编码为数据块 + 校验块"""
        chunk_size = len(data) // self.data_chunks
        data_blocks = [
            data[i * chunk_size:(i + 1) * chunk_size]
            for i in range(self.data_chunks)
        ]
        parity_blocks = reed_solomon_encode(
            data_blocks, self.parity_chunks
        )
        return data_blocks + parity_blocks

    def decode(self, available_chunks: dict) -> bytes:
        """只要有任意 6 个块（总共 9 个），即可恢复原始数据"""
        if len(available_chunks) < self.data_chunks:
            raise InsufficientChunksError(
                f"需要至少 {self.data_chunks} 个块，"
                f"当前只有 {len(available_chunks)} 个"
            )
        return reed_solomon_decode(
            available_chunks,
            self.data_chunks,
            self.parity_chunks,
        )

# BeyondCorp 访问决策伪代码
class AccessEngine:
    def evaluate_access(self, request: AccessRequest) -> AccessDecision:
        """评估一次访问请求"""
        # 1. 验证用户身份
        user = self.authenticate_user(request.credentials)
        if not user:
            return AccessDecision.DENY("身份验证失败")

        # 2. 验证设备状态
        device = self.device_inventory.get(request.device_id)
        if not device:
            return AccessDecision.DENY("未知设备")

        # 3. 计算信任分数
        trust_score = self.trust_inferer.compute(user, device)

        # 4. 检查策略
        policy = self.get_policy(request.target_service)
        required_trust = policy.minimum_trust_level

        if trust_score >= required_trust:
            return AccessDecision.ALLOW(
                trust_level=trust_score,
                audit_log=True,
            )
        elif trust_score >= required_trust * 0.8:
            # 信任分数接近阈值，要求额外验证
            return AccessDecision.REQUIRE_MFA(
                reason="信任分数低于阈值",
                trust_level=trust_score,
            )
        else:
            return AccessDecision.DENY(
                reason=f"信任分数 {trust_score} 低于要求 {required_trust}"
            )

    def compute_trust_score(self, user, device) -> float:
        """计算综合信任分数"""
        score = 0.0

        # 设备因素
        if device.is_corp_managed:
            score += 0.3
        if device.disk_encrypted:
            score += 0.1
        if device.os_up_to_date:
            score += 0.1
        if device.screen_lock_enabled:
            score += 0.05

        # 用户因素
        if user.mfa_enabled:
            score += 0.2
        if user.recent_security_training:
            score += 0.05
        if not user.has_security_incidents:
            score += 0.1

        # 行为因素
        if self.is_normal_access_pattern(user):
            score += 0.1

        return min(score, 1.0)

-- 订单系统 Spanner 表设计
CREATE TABLE Inventory (
    ProductId    INT64 NOT NULL,
    RegionCode   STRING(16) NOT NULL,
    Quantity     INT64 NOT NULL,
    LastUpdated  TIMESTAMP NOT NULL OPTIONS (allow_commit_timestamp=true),
) PRIMARY KEY (ProductId, RegionCode);

CREATE TABLE Orders (
    OrderId      STRING(36) NOT NULL,
    UserId       INT64 NOT NULL,
    CreatedAt    TIMESTAMP NOT NULL OPTIONS (allow_commit_timestamp=true),
    Status       STRING(32) NOT NULL,
    TotalAmount  FLOAT64 NOT NULL,
) PRIMARY KEY (OrderId);

CREATE TABLE OrderItems (
    OrderId      STRING(36) NOT NULL,
    ItemId       INT64 NOT NULL,
    ProductId    INT64 NOT NULL,
    Quantity     INT64 NOT NULL,
    UnitPrice    FLOAT64 NOT NULL,
) PRIMARY KEY (OrderId, ItemId),
  INTERLEAVE IN PARENT Orders ON DELETE CASCADE;

CREATE INDEX OrdersByUser ON Orders(UserId, CreatedAt DESC);

// Spanner 全局一致性事务示例（Java）
import com.google.cloud.spanner.*;

public class OrderService {
    private final DatabaseClient dbClient;

    public OrderService(DatabaseClient dbClient) {
        this.dbClient = dbClient;
    }

    /**
     * 创建订单并扣减库存
     * 这是一个跨分片的读写事务（Read-Write Transaction）
     */
    public Order createOrder(String userId, List<OrderItem> items) {
        return dbClient
            .readWriteTransaction()
            .run(transaction -> {
                String orderId = generateOrderId();
                double totalAmount = 0;

                for (OrderItem item : items) {
                    // 1. 读取当前库存（强一致性读）
                    Struct row = transaction.readRow(
                        "Inventory",
                        Key.of(item.getProductId(), "APAC-NE"),
                        Arrays.asList("Quantity", "LastUpdated")
                    );

                    if (row == null) {
                        throw new SpannerException(
                            ErrorCode.NOT_FOUND,
                            "商品不存在: " + item.getProductId()
                        );
                    }

                    long currentQuantity = row.getLong("Quantity");

                    // 2. 检查库存是否充足
                    if (currentQuantity < item.getQuantity()) {
                        throw new SpannerException(
                            ErrorCode.ABORTED,
                            "库存不足: " + item.getProductId()
                            + " 需要 " + item.getQuantity()
                            + " 实际 " + currentQuantity
                        );
                    }

                    // 3. 扣减库存
                    transaction.buffer(
                        Mutation.newUpdateBuilder("Inventory")
                            .set("ProductId").to(item.getProductId())
                            .set("RegionCode").to("APAC-NE")
                            .set("Quantity")
                            .to(currentQuantity - item.getQuantity())
                            .set("LastUpdated")
                            .to(Value.COMMIT_TIMESTAMP)
                            .build()
                    );

                    totalAmount += item.getUnitPrice() * item.getQuantity();

                    // 4. 插入订单项
                    transaction.buffer(
                        Mutation.newInsertBuilder("OrderItems")
                            .set("OrderId").to(orderId)
                            .set("ItemId").to(item.getItemId())
                            .set("ProductId").to(item.getProductId())
                            .set("Quantity").to(item.getQuantity())
                            .set("UnitPrice").to(item.getUnitPrice())
                            .build()
                    );
                }

                // 5. 插入订单主记录
                transaction.buffer(
                    Mutation.newInsertBuilder("Orders")
                        .set("OrderId").to(orderId)
                        .set("UserId").to(Long.parseLong(userId))
                        .set("CreatedAt").to(Value.COMMIT_TIMESTAMP)
                        .set("Status").to("CREATED")
                        .set("TotalAmount").to(totalAmount)
                        .build()
                );

                return new Order(orderId, totalAmount);
            });
    }

    /**
     * 查询用户订单（强一致性读取）
     * 无论用户在哪个地区发起请求，
     * 都能看到最新提交的数据
     */
    public List<Order> getUserOrders(long userId) {
        // 使用强一致性读取（Strong Read）
        // Spanner 保证返回的数据反映了截至此刻所有已提交的事务
        try (ResultSet rs = dbClient
            .singleUse()  // 默认使用强一致性
            .executeQuery(
                Statement.newBuilder(
                    "SELECT o.OrderId, o.CreatedAt, o.Status, "
                    + "o.TotalAmount, oi.ProductId, oi.Quantity "
                    + "FROM Orders o "
                    + "JOIN OrderItems oi ON o.OrderId = oi.OrderId "
                    + "WHERE o.UserId = @userId "
                    + "ORDER BY o.CreatedAt DESC "
                    + "LIMIT 50"
                )
                .bind("userId").to(userId)
                .build()
            )) {
            List<Order> orders = new ArrayList<>();
            while (rs.next()) {
                orders.add(mapToOrder(rs));
            }
            return orders;
        }
    }

    /**
     * 查询用户订单（过期读取，降低延迟）
     * 允许读取最多 15 秒前的数据快照
     */
    public List<Order> getUserOrdersStale(long userId) {
        // 使用过期读取（Stale Read），最多容忍 15 秒的数据延迟
        // 好处：可以从最近的副本读取，显著降低跨区域延迟
        TimestampBound staleness = TimestampBound.ofMaxStaleness(
            15, TimeUnit.SECONDS
        );

        try (ResultSet rs = dbClient
            .singleUse(staleness)
            .executeQuery(
                Statement.newBuilder(
                    "SELECT OrderId, CreatedAt, Status, TotalAmount "
                    + "FROM Orders "
                    + "WHERE UserId = @userId "
                    + "ORDER BY CreatedAt DESC "
                    + "LIMIT 50"
                )
                .bind("userId").to(userId)
                .build()
            )) {
            List<Order> orders = new ArrayList<>();
            while (rs.next()) {
                orders.add(mapToOrder(rs));
            }
            return orders;
        }
    }

    private String generateOrderId() {
        return java.util.UUID.randomUUID().toString();
    }

    private Order mapToOrder(ResultSet rs) {
        return new Order(
            rs.getString("OrderId"),
            rs.getDouble("TotalAmount")
        );
    }
}

# Spanner 实例配置示例
instance_config:
  name: "asia-leader-config"
  replicas:
    - location: asia-northeast1    # 东京（Leader 候选）
      type: READ_WRITE
      default_leader: true
    - location: asia-southeast1    # 新加坡
      type: READ_WRITE
    - location: us-central1        # 爱荷华
      type: READ_WRITE
    - location: europe-west1       # 比利时
      type: READ_ONLY
    - location: us-east1           # 弗吉尼亚
      type: READ_ONLY