Carbon-Aware 云计算调度：从绿色承诺到生产级碳感知工程实践

引言：当算力遇见碳排放

2024 年，全球数据中心用电量已突破 400 TWh，占全球电力消耗的 1.5%—2%。国际能源署（IEA）预测到 2030 年，数据中心用电可能翻倍。与此同时，Google 在 2024 年发布白皮书，宣布将"24/7 碳免费能源"目标推迟至 2030 年，原因并非技术懈怠，而是因为碳感知调度的实际工程复杂度远超预期。

本文不谈环保愿景，直击生产级 Carbon-Aware 调度的工程本质：如何利用实时碳强度信号（Carbon Intensity Signal）、液冷系统热力学模型、以及 Kubernetes 调度器扩展，在保证 SLA 的前提下，将 AI 训练任务和批处理负载迁移到电网清洁时段运行，实现碳排放降低 15%—40%。

一、碳感知调度的理论基础

1.1 从 Energy-Proportional Computing 到 Carbon-Aware

Roderolf Proposi 在 2007 年提出 Energy-Proportional Computing 理想：服务器在空闲时零功耗、满载时 100% 能效。现实是——服务器空闲功耗通常是满载的 50%—70%。

Carbon-Aware 调度在此基础上更进一步：不仅关注能耗量，还关注能耗的"碳排放强度"（gCO₂/kWh）。电网每时每刻的碳强度不同——晴天午时光伏大发时碳强度可能 < 50 gCO₂/kWh，而夜晚煤电主导时可达 > 800 gCO₂/kWh。

关键指标定义：

1.2 碳感知 vs. 能耗优化的核心区别

传统能耗优化（DVFS、CPU C-state 调频）的目标是降低总能耗。而 Carbon-Aware 调度可能做出看似矛盾的决定——在高碳时段主动降负载、在低碳时段猛增负载。

举例：某训练任务在 2 小时低碳时段（CI = 100）以 1000W 运行，总排放 = 200g。若强行在 4 小时高碳时段（CI = 600）以 500W 运行（节省 50% 能耗），总排放却 = 1200g。碳感知调度的精髓：时间转移 > 能耗压缩。

二、碳强度数据源工程

2.1 实时碳强度 API 对比

生产级 Carbon-Aware 调度需要可靠的碳强度数据源。以下是主流方案对比：

┌─────────────────────────────────────────────────────────────────────┐
│ 数据源             │ 分辨率  │ 地理粒度   │ 预测能力      │ 免费额度  │
├─────────────────────────────────────────────────────────────────────┤
│ Electricity Maps    │ 1小时   │ 国家级/地区 │ 24小时预测    │ 有限免费  │
│ WattTime            │ 5分钟   │ 区域级     │ 实时+预测     │ API Key   │
│ Carbon Intensity UK │ 30分钟  │ 英国全国   │ 48小时预测    │ 完全免费  │
│ Climatiq            │ 1小时   │ 全球       │ 无预测       │ 按量付费  │
│ Kube CI HUB         │ 推估    │ 节点级     │ 无           │ 开源      │
└─────────────────────────────────────────────────────────────────────┘

2.2 节点级碳强度推估（Kube CI HUB 方案）

现实中，大多数数据中心没有实时碳强度电表。Kubernetes 生态的解决方案是 Kube CI HUB——通过节点所在区域电网的碳强度数据，结合节点的实时功耗（通过 RAPL/ACPI）推导碳排放。

Kube CI HUB 的核心数据流：

# Kube CI HUB 简化逻辑：节点碳排放计算
import requests

class CarbonIntensityEstimator:
    def __init__(self, region: str, data_source: str = "watttime"):
        self.region = region
        self.data_source = data_source

    def get_current_ci(self) -> float:
        """获取当前电网碳强度 (gCO₂/kWh)"""
        if self.data_source == "watttime":
            resp = requests.get(
                "https://api.watttime.org/v3/forecast",
                headers={"Authorization": f"Bearer {self.token}"},
                params={"region": self.region}
            )
            return resp.json()["data"][0]["value"]
        elif self.data_source == "electricitymap":
            resp = requests.get(
                "https://api.electricitymap.org/v3/carbon-intensity/latest",
                headers={"auth-token": self.token},
                params={"zone": self.region}
            )
            return resp.json()["carbonIntensity"]

    def calculate_node_emission(self, 
                                 node_power_watts: float,
                                 time_hours: float = 1.0) -> dict:
        """计算节点碳排放"""
        ci = self.get_current_ci()
        energy_kwh = node_power_watts * time_hours / 1000.0
        emission_g = energy_kwh * ci

        return {
            "node_power_w": node_power_watts,
            "energy_kwh": energy_kwh,
            "carbon_intensity_gco2_kwh": ci,
            "emission_gco2": emission_g,
            "region": self.region,
            "timestamp": time.time()
        }

2.3 数据中心液冷系统热力学模型

现代 AI 数据中心（NVIDIA DGX H100 系统）普遍采用液冷。液冷系统本身也构成碳感知调度的约束条件：

• 冷却延迟：液冷系统热惯量高，关机后仍需持续冷却 5—15 分钟
• PUE 动态变化：室外温度变化导致 PUE 从 1.05 波动到 1.15+
• 与自然冷却结合：北欧数据中心冬季可启用 Free Cooling，PUE 接近 1.01

碳感知调度器必须建模热约束：不能在短时间内频繁启停高密度工作负载，否则冷却系统追赶不上热负荷变化，反而导致整体 PUE 恶化。

三、Kubernetes 碳感知调度器架构

3.1 调度器扩展点

Kubernetes 提供了多个调度器扩展点，Carbon-Aware 调度可组合使用：

┌──────────────────────────────────────────────────────┐
│  Kubernetes 调度器扩展点                               │
├──────────────────────────────────────────────────────┤
│  Score Plugin:     按碳强度给节点打分                   │
│  PreFilter Plugin: 预检查碳数据是否可用                 │
│  PreScore Plugin:  获取当前碳强度数据                   │
│  Reserve Plugin:   预留碳配额                          │
│  Permit Plugin:    延迟调度至低碳窗口                   │
│  PostBind Plugin:  绑定后更新碳消耗记录                 │
└──────────────────────────────────────────────────────┘

3.2 核心调度策略实现

以下是一个完整的 Carbon-Aware Score Plugin 实现：

// pkg/scheduler/carbon_score.go

package carbon

import (
    "context"
    "math"
    "time"
)

// CarbonScorePlugin 为节点按碳强度打分
type CarbonScorePlugin struct {
    ciProvider CarbonIntensityProvider
    weights    CarbonSchedulingWeights
}

type CarbonSchedulingWeights struct {
    CarbonIntensityWeight float64 // 碳强度权重 (默认 0.4)
    PowerBudgetWeight     float64 // 功耗预算权重 (默认 0.3)
    SLAFlexWeight         float64 // SLA弹性权重 (默认 0.3)
}

// Score 实现 scheduler.Plugin 接口
func (p *CarbonScorePlugin) Score(ctx context.Context, 
    state *framework.CycleState,
    pod *v1.Pod, 
    nodeName string) (int64, *framework.Status) {

    nodeInfo := p.nodeInfoLister.Get(nodeName)
    if nodeInfo == nil {
        return 0, framework.AsStatus(
            fmt.Errorf("node %q not found", nodeName))
    }

    // 1. 获取节点所在区域的实时碳强度
    region := nodeInfo.Node().Labels["topology.kubernetes.io/zone"]
    ci, err := p.ciProvider.GetCurrentCI(region)
    if err != nil {
        // 碳数据不可用，降级为功耗感知
        return p.powerOnlyScore(nodeInfo), nil
    }

    // 2. 计算碳强度得分 (越低越好 → 分数越高)
    carbonScore := normalizeCI(ci)

    // 3. 计算功耗预算得分
    powerScore := p.calculatePowerBudgetScore(nodeInfo)

    // 4. 计算 SLA 弹性得分
    slaScore := p.calculateSLAFlexibility(pod, nodeInfo)

    // 5. 加权综合得分
    totalScore := int64(
        carbonScore * p.weights.CarbonIntensityWeight * 100 +
        powerScore * p.weights.PowerBudgetWeight * 100 +
        slaScore * p.weights.SLAFlexWeight * 100)

    return totalScore, framework.NewStatus(framework.Success)
}

// normalizeCI 将碳强度转换为 0-100 得分（越低碳排放得分越高）
func normalizeCI(ci float64) float64 {
    const (
        minCI = 20.0   // 接近纯可再生能源
        maxCI = 800.0  // 高碳电网
    )
    if ci <= minCI {
        return 1.0
    }
    if ci >= maxCI {
        return 0.0
    }
    // 反向线性归一化：低 CI → 高得分
    return 1.0 - (ci - minCI) / (maxCI - minCI)
}

// calculateSLAFlexibility 根据 Pod 的碳耐受度标签打分
func (p *CarbonScorePlugin) calculateSLAFlexibility(
    pod *v1.Pod, 
    nodeInfo *framework.NodeInfo) float64 {

    // carbon-tolerance 标签：immediate | daily | weekly | monthly
    tolerance := pod.Labels["carbon-tolerance"]

    switch tolerance {
    case "immediate":
        // 不可延迟，忽略碳强度
        return 0.5
    case "daily":
        // 允许 24 小时内灵活调度
        return 0.8
    case "weekly":
        // 允许一周内灵活调度
        return 0.9
    case "monthly":
        // 月度级批处理任务，极端灵活
        return 1.0
    default:
        return 0.5
    }
}

// NormalizeScore 确保所有节点得分在 0-100 之间
func (p *CarbonScorePlugin) NormalizeScore(ctx context.Context,
    state *framework.CycleStatus,
    pod *v1.Pod, scores framework.NodeScoreList) *framework.Status {

    var maxScore int64 = 0
    for _, score := range scores {
        if score.Score > maxScore {
            maxScore = score.Score
        }
    }

    if maxScore == 0 {
        return framework.NewStatus(framework.Success)
    }

    for i := range scores {
        scores[i].Score = scores[i].Score * 100 / maxScore
    }
    return framework.NewStatus(framework.Success)
}

3.3 碳强度感知的 KEDA Scaler

KEDA（Kubernetes Event-Driven Autoscaling）可以通过自定义 Scaler 基于碳强度自动扩缩容：

# carbon-aware-scaledobject.yaml
apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: ml-training-scaler
  namespace: ai-workloads
spec:
  scaleTargetRef:
    name: ml-training-deployment
  minReplicaCount: 1
  maxReplicaCount: 20
  triggers:
  # 基于碳强度的自定义 Metric
  - type: metrics-api
    metadata:
      targetValue: "300"            # 碳强度阈值 (gCO₂/kWh)
      url: "http://carbon-exporter.monitoring.svc/api/v1/carbon-intensity?region=us-west"
      valueLocation: "data.carbon_intensity"
  # 批处理任务增加时间窗口偏移
  - type: cron
    metadata:
      timezone: America/Los_Angeles
      start: 0 6 * * *     # 早 6 点（低谷碳时段）
      end: 0 18 * * *       # 晚 6 点（高碳时段）
      desiredReplicas: "10"

四、AI 训练任务的碳感知时间转移

4.1 Checkpoint + Resume 模式

大规模 AI 训练任务的碳感知调度关键在于 Checkpoint/Resume 基础设施。以下是与 PyTorch Distributed Checkpoint 集成的方案：

# carbon_aware_trainer.py

import torch
import torch.distributed.checkpoint as dcp
from datetime import datetime, timedelta
import pytz
from typing import Optional

class CarbonAwareTrainer:
    def __init__(self, 
                 model, 
                 optimizer,
                 scheduler,
                 carbon_provider: 'CarbonDataProvider',
                 checkpoint_dir: str = "/mnt/checkpoints",
                 region: str = "us-west"):
        self.model = model
        self.optimizer = optimizer
        self.scheduler = scheduler
        self.carbon_provider = carbon_provider
        self.checkpoint_dir = checkpoint_dir
        self.region = region
        self.accumulated_carbon_g = 0.0

    def train_with_carbon_awareness(self, 
                                     dataloader,
                                     max_hours: float = 48.0,
                                     carbon_budget_g: Optional[float] = None):
        """带碳感知的训练循环"""

        start_time = datetime.now(pytz.UTC)
        carbon_start = self.carbon_provider.get_current_ci(self.region)

        for epoch in range(self.max_epochs):
            for batch_idx, batch in enumerate(dataloader):
                # === 1. 检查碳预算 ===
                if carbon_budget_g and self.accumulated_carbon_g >= carbon_budget_g:
                    self._pause_and_wait(optimal_window_hours=6)
                    continue

                # === 2. 执行训练步 ===
                self.optimizer.zero_grad()
                loss = self.model(batch)
                loss.backward()
                self.optimizer.step()

                # === 3. 实时累计碳排放 ===
                power_w = self._get_current_power_draw()  # RAPL/NVML
                ci = self.carbon_provider.get_current_ci(self.region)
                step_energy_kwh = power_w * self.step_duration_s / 3600.0 / 1000.0
                step_carbon_g = step_energy_kwh * ci
                self.accumulated_carbon_g += step_carbon_g

                # === 4. 高碳时段：降低并行度 ===
                if ci > self.carbon_provider.get_threshold(self.region, "high"):
                    self._reduce_parallelism()
                    # 增加 checkpoint 频率以避免进度损失_loss
                    if batch_idx % 50 == 0:
                        self._save_checkpoint(epoch, batch_idx)

                # === 5. 低碳时段：加速 ===
                elif ci < self.carbon_provider.get_threshold(self.region, "low"):
                    self._increase_parallelism()

                # === 6. 检查时间预算 ===
                elapsed = (datetime.now(pytz.UTC) - start_time).total_seconds() / 3600.0
                if elapsed >= max_hours:
                    self._save_checkpoint(epoch, batch_idx)
                    return

            self.scheduler.step()

    def _save_checkpoint(self, epoch: int, step: int):
        """DCP 分布式 Checkpoint 保存"""
        metadata = dcp.save(
            state_dict={
                "model": self.model.state_dict(),
                "optimizer": self.optimizer.state_dict(),
                "scheduler": self.scheduler.state_dict(),
            },
            checkpoint_id=f"{self.checkpoint_dir}/step_{epoch}_{step}",
        )

    def _load_latest_checkpoint(self):
        """恢复最新 Checkpoint"""
        # 实现：扫描 checkpoint_dir 找最新 DCP 索引文件
        checkpoint_id = self._find_latest_checkpoint()
        if checkpoint_id:
            dcp.load(
                state_dict={
                    "model": self.model.state_dict(),
                    "optimizer": self.optimizer.state_dict(),
                    "scheduler": self.scheduler.state_dict(),
                },
                checkpoint_id=checkpoint_id,
            )

    def _pause_and_wait(self, optimal_window_hours: float = 6):
        """高碳时段暂停，等待下一个清洁窗口"""
        print(f"[Carbon-Aware] 碳强度 {self.carbon_provider.get_current_ci(self.region):.0f} gCO₂/kWh，"
              f"超出阈值。累计排放 {self.accumulated_carbon_g:.1f}g，等待清洁窗口...")
        self._save_checkpoint(-1, -1)  # 紧急 checkpoint

        # 轮询等待碳强度下降
        while True:
            ci = self.carbon_provider.get_current_ci(self.region)
            if ci < 200:  # 清洁阈值
                print(f"[Carbon-Aware] 检测到低碳窗口 ({ci:.0f} gCO₂/kWh)，恢复训练")
                self._load_latest_checkpoint()
                return
            time.sleep(300)  # 每 5 分钟检查一次

4.2 碳配额交易市场模型

大型云厂商内部可以构建碳配额交易机制——不同团队/项目组拥有月度碳配额，超额排放需要从其他配额有结余处"购买"：

# carbon_quotas.py

@dataclass
class CarbonQuota:
    team: str
    monthly_budget_kg: float      # 月度碳配额 (kgCO₂)
    consumed_kg: float = 0.0       # 已消耗
    priority: str = "normal"       # critical/high/normal/low

    @property
    def remaining_kg(self) -> float:
        return max(0.0, self.monthly_budget_kg - self.consumed_kg)

    @property
    def utilization_rate(self) -> float:
        return self.consumed_kg / self.monthly_budget_kg if self.monthly_budget_kg > 0 else float('inf')

class CarbonPrincipalApproach:
    """碳预算感知 K8s 准入控制"""

    def __init__(self, quota_store: 'CarbonQuotaStore', carbon_exporter: 'CarbonExporter'):
        self.quota_store = quota_store
        self.carbon_exporter = carbon_exporter

    def admit_pod(self, pod: Pod) -> AdmissionResult:
        """Webhook 准入控制"""
        team = pod.labels.get("team", "default")
        quota = self.quota_store.get(team)

        # 估算 Pod 碳排放
        estimated_carbon = self.estimate_pod_carbon(pod)

        if quota.remaining_kg * 1000 < estimated_carbon:  # g vs kg
            return AdmissionResult(
                allowed=False,
                reason=f"碳配额不足: 剩余 {quota.remaining_kg:.2f}kg, "
                       f"任务预计排放 {estimated_carbon:.0f}g",
                suggested_action="申请增加配额或降低 carbon-tolerance"
            )

        # 如果是低优先级且有碳预算约束，打标延迟容忍
        if quota.utilization_rate > 0.80:
            pod.labels["carbon-tolerance"] = "weekly"
            pod.labels["carbon-budget-constrained"] = "true"

        return AdmissionResult(allowed=True)

    def estimate_pod_carbon(self, pod: Pod) -> float:
        """基于历史数据和资源请求估算碳排放"""
        cpu_cores = sum(c.resources.requests.get("cpu", 0)  for c in pod.spec.containers)
        gpu_count = sum(c.resources.requests.get("nvidia.com/gpu", 0) for c in pod.spec.containers)
        estimated_hours = float(pod.labels.get("estimated-duration-hours", "8"))

        # H100 GPU: ~700W, CPU per core ~12W
        power_w = gpu_count * 700 + cpu_cores * 12 + 200  # 基础功耗
        ci_avg = 350  # 假想区域平均碳强度
        emission_g = power_w * estimated_hours / 1000.0 * ci_avg

        return emission_g

五、生产级部署实践：Google 与 Microsoft 案例

5.1 Google Carbon-Intelligent Computing

Google 的 Carbon-Intelligent Computing 系统在 Google Cloud 上运行多年。核心设计原则：

1. 不移动计算，移动时间：将灵活的批处理作业服务器（如 YouTube 视频转码、BigQuery 非紧急查询）延迟到低碳时段
2. 地理转移：在低碳区域运行弹性负载（需网络带宽允许）
3. 机器学习预测：用 ML 预测未来 24 小时碳强度，提前规划调度

Google 在 2023 年发表的数据：通过碳智能计算，将灵活负载的清洁电力匹配率从 65% 提升到 86%。但实时工作负载（搜索、Gmail）的 CFE 仅从 65% 提升到 71%——越弹性的负载，碳感知效果越好。

5.2 Microsoft 的碳感知 Windows 更新

Microsoft 的 Carbon-Aware Windows Update 是消费级产品的典型：当电网碳强度高时推迟非紧急更新下载。工程要点：

• 使用 WattTime API 提供 5 分钟碳强度分辨率
• 区分"关键安全更新"和"功能更新"（只有后者可延迟）
• 默认 15 分钟轮询，可高至 1 小时延迟

六、碳感知调度的 SLA 与可靠性权衡

6.1 工作负载分类矩阵

不是所有工作负载都适合碳感知调度。以下是分类决策框架：

6.2 降级策略

碳感知调度的一个核心设计原则是 graceful degradation（优雅降级）：

class CarbonAwareFallback:
    """碳数据不可用时的降级策略"""

    STRATEGIES = {
        "ci_api_timeout": {
            "action": "use_cached_ci",
            "cache_ttl_seconds": 3600,
            "fallback": "power_only"
        },
        "ci_api_error": {
            "action": "use_temporal_heuristic",
            # 基于历史数据的经验规则：夜间碳强度高
            "heuristic": {
                "night_hours": {"multiplier": 1.5},   # 夜间权重降低
                "daytime_solar": {"multiplier": 0.6},  # 白天光伏权重提高
            }
        },
        "all_fallback_failed": {
            "action": "disable_carbon_aware",
            "alert": "slack:#carbon-alerts"
        }
    }

    def get_effective_ci(self, region: str) -> float:
        try:
            with timeout(seconds=5):
                return self.watttime_client.get_latest(region)
        except TimeoutError:
            cached = self.cache.get(f"ci:{region}")
            if cached and time.time() - cached["ts"] < 3600:
                return cached["value"]
            # 最终降级：使用时间启发式
            return self._apply_temporal_heuristic(region)

6.3 碳感知调度的财务价值

对于碳排成本内部化的企业，Carbon-Aware 调度的财务价值突出：

• 欧盟碳关税（CBAM）当前碳价约 €80/吨
• 1 张 H100 GPU 年耗电 ~6 MWh，在 CI = 400 的区域年排放 ~2.4 吨，碳成本约 €192/年
• 通过碳感知迁移到 CI = 150 的区域：排放 ~0.9 吨，碳成本节省 ~€120/GPU/年
• 千卡集群（8000 GPU）每年节省约 €960,000 碳成本

七、展望未来：24/7 CFE 的工程挑战

行业正在从 Carbon-Aware（碳感知）走向 24/7 CFE（24/7 碳免费能源匹配）——不仅是抵消排放，而是每小时都与清洁能源匹配。这面临三大工程挑战：

1. 储能系统集成：数据中心级储能（锂电池/氢能）使得"存储低碳电力用于高碳时段"成为可能，但充放电效率的建模极其复杂

2. 核聚变电力跟踪：随着小型模块化核反应堆（SMR）在数据中心部署，颗粒度到分钟的电力追踪需要新的计量基础设施

3. 国际标准互操作：24/7 CFE 需要全球统一的碳强度度量标准，避免"绿色洗白（Greenwashing）"

IBM 在 2024 年开源的 SEEM（Sustainable Energy Event Model） 正是为此设计——提供一个基于事件驱动的碳强度信号标准框架，已被 CNCF Green Computing WG 接纳为孵化项目。

结语

Carbon-Aware 调度不是一个单纯的"环保功能"，而是现代云基础设施的核心调度约束——与网络延迟、存储 I/O 权力、GPU 可用性并列。掌握它的工程细节，意味着你的 AI 训练成本可以再降低 15%—40%，同时在碳合规上抢占先机。

真正的绿色计算不在于壮志豪言，而在于代码中那一行 if ci > threshold: checkpoint_and_wait()。