云上省钱，别再从"找闲置"开始

先算单位成本，再排优化动作。顺序反了，省下来的钱会以另一种方式花回去。

几乎每个季度的成本优化，都是从同一张表开始的：按资源类型排序，找出 CPU 5% 以下的机器、没人绑的 EIP、三个月没访问的存储桶。这条路没错，见效也快。

但它有个天花板。闲置资源是一次性的，清完就没了；而真正决定账单结构的，是"每个服务花多少钱、换来多少业务量"——也就是单位成本。

先说结论：资源层决定你能省多少，单位成本决定你该不该省。

下文代码里的账单和监控数据由脚本生成，用来说明判断逻辑；接真实数据时替换 generate() 即可。

一、成本优化分三层

大部分团队停在第一层，因为第一层的动作最明确、最容易汇报。但连续做几个季度就会发现：能清的都清完了，账单还是一条缓慢上升的曲线。

二、单位成本：一个除法，三个坑

公式很简单：

每千次请求成本 = 服务月成本 / (月请求数 / 1000)

坑一：分母选错。 网关适合用请求数，交易服务适合用订单数，内容服务适合用活跃用户。分母选错，结论会完全反过来。一个"每千次请求很贵"的服务，可能只是因为它每次请求干的活更重。

坑二：共享成本不摊。 网关、日志、监控、CI 这些成本如果不按规则摊到服务上，每个服务看起来都很便宜，加起来却对不上总账单。

坑三：只算资源不算人力。 一个单位成本很低、但每周要两个人维护的服务，真实成本可能更高。这一项很难量化，但决策时不能假装它不存在。

示例数据里有个很直观的对比：gateway 是 ¥0.32/kreq，recommend 是 ¥40/kreq，差 125 倍。后者不是"降一档规格"能解决的问题——降配最多省 35%，而真正该回答的是"这个服务还有没有人在用"。

三、三张表必须先对齐

要做归因，至少需要三份数据：

对齐的键只有一个：resource_id。所以标签规范（service / env / owner）不是形式主义，它决定了你能不能把成本算到具体的人和具体的服务上。

没有这三个标签，后面的所有分析都会退化成"按资源类型看总额"——也就是本文开头那条路。

四、完整实现

逻辑很直白：逐条资源做规则判断，产出"动作 + 节省额 + 风险 + 依据"；再按服务算单位成本，找出异常。

"""cost_review.py —— 云成本体检：先算单位成本，再排优化动作。

单文件、零依赖：
    py -3 cost_review.py

账单与监控数据由脚本生成，用于演示判断逻辑；接真实数据时替换 generate()。
"""
from __future__ import annotations

import statistics
from dataclasses import dataclass
from collections import defaultdict


@dataclass
class Resource:
    rid: str
    service: str
    env: str
    rtype: str          # ecs / rds / obs / eip
    spec: str
    cpu_p95: float
    mem_p95: float
    cost: float         # 月成本（元）
    requests: int = 0   # 月请求数
    storage_gb: int = 0
    idle_days: int = 0  # 连续无流量天数
    night_idle: bool = False


@dataclass
class Finding:
    action: str
    rid: str
    service: str
    reason: str
    saving: float
    risk: str           # 低 / 中 / 高
    confidence: float


def review(r: Resource) -> list:
    out = []
    if r.env != "prod" and r.cpu_p95 < 2 and r.requests == 0 and r.idle_days >= 14:
        out.append(Finding("释放闲置资源", r.rid, r.service,
                           f"{r.idle_days} 天无流量，CPU P95 {r.cpu_p95}%",
                           r.cost, "低", 0.95))
        return out

    if r.rtype in ("ecs", "rds") and r.cpu_p95 < 15 and r.mem_p95 < 35:
        out.append(Finding("降配超配实例", r.rid, r.service,
                           f"CPU P95 {r.cpu_p95}%、内存 P95 {r.mem_p95}%，长期低水位",
                           round(r.cost * 0.35, 0), "中", 0.8))

    if r.rtype == "ecs" and r.env != "prod" and r.night_idle:
        out.append(Finding("非生产夜间关机", r.rid, r.service,
                           "仅工作时间有流量，22:00-08:00 可停机",
                           round(r.cost * 0.3, 0), "低", 0.9))

    if r.rtype == "obs" and r.idle_days > 90:
        out.append(Finding("冷数据转归档", r.rid, r.service,
                           f"{r.idle_days} 天未访问，{r.storage_gb} GB",
                           round(r.cost * 0.6, 0), "低", 0.85))

    if r.rtype == "eip" and r.requests == 0:
        out.append(Finding("释放闲置 EIP", r.rid, r.service,
                           "未绑定任何实例", r.cost, "低", 0.95))

    return out


def unit_cost(rows):
    """按服务算每千次请求成本，只统计生产环境、有流量的服务。"""
    cost, req = defaultdict(float), defaultdict(int)
    for r in rows:
        if r.env != "prod":
            continue
        cost[r.service] += r.cost
        req[r.service] += r.requests
    return {
        s: cost[s] / (req[s] / 1000)
        for s in cost if req[s] > 0
    }


def report(rows):
    findings = [f for r in rows for f in review(r)]
    total = sum(r.cost for r in rows)
    saving = sum(f.saving for f in findings)

    print(f"月度总支出        ¥ {total:,.0f}")
    print(f"可优化金额        ¥ {saving:,.0f}（{saving / total:.1%}）")
    low = sum(f.saving for f in findings if f.risk == "低")
    print(f"低风险可立即执行  ¥ {low:,.0f}（占可优化金额 {low / saving:.0%}）")
    print()

    uc = unit_cost(rows)
    median = statistics.median(uc.values())
    print(f"单位成本（每千次请求），中位数 ¥ {median:.2f}")
    for s, v in sorted(uc.items(), key=lambda kv: -kv[1]):
        flag = "  <-- 偏高" if v > median * 3 else ""
        print(f"  {s:<12}¥ {v:>7.2f} / kreq{flag}")
    print()

    print(f"{'动作':<12}{'项数':>4}{'月节省':>12}{'风险':>6}")
    print("-" * 40)
    by_action = defaultdict(list)
    for f in findings:
        by_action[f.action].append(f)
    for action, items in sorted(by_action.items(), key=lambda kv: -sum(f.saving for f in kv[1])):
        risks = {f.risk for f in items}
        risk = "低" if risks == {"低"} else "中"
        print(f"{action:<12}{len(items):>4}{sum(f.saving for f in items):>11,.0f}{risk:>5}")
    print()

    print("优先执行（按节省额 / 风险排序）：")
    for f in sorted(findings, key=lambda x: (x.risk != "低", -x.saving))[:6]:
        print(f"  [{f.risk}风险] {f.action}  {f.rid}（{f.service}）  ¥ {f.saving:,.0f}/月")
        print(f"            依据：{f.reason}")


def generate():
    rows = []

    def add(prefix, service, env, rtype, spec, cpu, mem, cost, req=0, storage=0,
            idle=0, count=1, night_idle=False):
        for i in range(count):
            rows.append(Resource(f"{prefix}-{i + 1:02d}", service, env, rtype, spec,
                                 cpu, mem, cost, req // count, storage // count,
                                 idle, night_idle))

    add("ecs-order", "order", "prod", "ecs", "8c16g", 52, 68, 1600, 3_200_000, count=4)
    add("ecs-cart", "cart", "prod", "ecs", "4c8g", 38, 55, 850, 1_600_000, count=4)
    add("ecs-search", "search", "prod", "ecs", "8c32g", 61, 74, 2100, 900_000, count=3)
    add("ecs-gw", "gateway", "prod", "ecs", "4c8g", 44, 40, 900, 6_000_000, count=3)
    add("ecs-reco", "recommend", "prod", "ecs", "8c16g", 9, 31, 1600, 120_000, count=3)
    add("ecs-report", "report", "prod", "ecs", "4c8g", 12, 28, 850, 20_000, count=2)
    add("ecs-ml", "ml-train", "prod", "ecs", "16c64g", 78, 85, 4200, count=2)

    add("ecs-stg", "order", "staging", "ecs", "4c8g", 1, 3, 850, idle=21, count=4)
    add("ecs-dev", "cart", "dev", "ecs", "2c4g", 0.5, 2, 320, idle=30, count=6)
    add("ecs-test", "search", "test", "ecs", "4c8g", 18, 22, 850, count=2, night_idle=True)

    add("rds-order", "order", "prod", "rds", "8c32g", 55, 70, 2600, count=2)
    add("rds-stg", "order", "staging", "rds", "4c16g", 1.5, 8, 980, idle=18, count=1)
    add("rds-ro", "report", "prod", "rds", "4c16g", 6, 25, 1400, count=2)

    add("obs-log", "log", "prod", "obs", "-", 0, 0, 900, storage=12_000, idle=180, count=3)
    add("obs-asset", "asset", "prod", "obs", "-", 0, 0, 500, storage=3_000, idle=5, count=2)

    add("eip-idle", "gateway", "dev", "eip", "-", 0, 0, 385, count=2)
    add("eip-gw", "gateway", "prod", "eip", "-", 0, 0, 385, req=5_000_000, count=2)

    return rows


if __name__ == "__main__":
    report(generate())

跑出来的结果：

月度总支出        ¥ 54,940
可优化金额        ¥ 12,456（22.7%）
低风险可立即执行  ¥ 9,200（占可优化金额 74%）

单位成本（每千次请求），中位数 ¥ 5.31
  report      ¥  225.00 / kreq  <-- 偏高
  recommend   ¥   40.00 / kreq  <-- 偏高
  search      ¥    7.00 / kreq
  order       ¥    3.62 / kreq
  cart        ¥    2.12 / kreq
  gateway     ¥    0.32 / kreq

动作            项数         月节省    风险
----------------------------------------
释放闲置资源        11      6,300    低
降配超配实例         7      3,256    中
冷数据转归档         3      1,620    低
释放闲置 EIP       2        770    低
非生产夜间关机        2        510    低

三个观察，比总数更重要：

1. 一半的节省额来自一次性动作。 11 项闲置资源贡献 6300，占可优化金额的 50.6%。但清完之后，下个季度不会再有了。把它当成"季度奖金"，别当成"持续收益"。

2. 降配是持续性的，但风险等级不同。 7 项超配合计 3256/月，每个月都能省；代价是要承担性能风险，必须灰度，不能批量执行。

3. 单位成本异常的两个服务，节省额加起来不如一次降配，但它们指向的问题更大。 report 是 ¥225/kreq，recommend 是 ¥40/kreq，而中位数只有 ¥5.31。这两个数字不该直接变成"降配工单"，而应该变成一个提问：这两个服务的业务价值，配得上这个成本吗？

五、执行顺序：按风险，不按金额

排序很简单，但很多人会按节省额从大到小做，结果第一批就踩了中风险动作。

低风险，直接做。 释放闲置、冷数据归档、闲置 EIP。前提是两步确认：释放前打快照，或确认没有任何绑定关系。这两步不能省。

中风险，先灰度。 降配一次只动一台，观察 7 天的 P95、错误率和下游依赖，再推全量。特别是有状态服务，降内存比降 CPU 危险得多。

高风险，先出方案。 降副本、换计费方式、下线服务——这些不该在季度末冲指标的时候做。指标压力会让你跳过验证。

六、三个反模式

1. 看均值降配。 平均值会掩盖尖峰：一台 8 核机器日均 10%，但每天 10 点冲到 90%，按均值降到 2 核，第二天就出事。所有降配判断都应该看 P95 或 P99。

2. 砍日志、备份、快照。 这类成本在账单上很显眼，省起来也快，但它们是事故时的唯一证据链。要优化就做生命周期分层，不要直接关掉。

3. 先买包年。 归因还没做清楚就锁定一到三年，等于把未来的优化空间提前卖掉。包年适合"用量稳定且已确认长期存在"的负载，不适合"还没想清楚要不要继续做"的服务。

七、落到华为云上

• 分账：成本中心按企业项目或标签出账，先把 service / env / owner 三个标签补齐，再谈归因。
• 指标：CES 取 P95 和连续低水位天数；应用侧指标用 AOM 补，避免只看基础设施。
• 存储：OBS 生命周期规则承接冷数据，标准 → 低频 → 归档逐级下沉，别手工搬。
• 快照：EVS 快照策略定期清理，快照费用悄悄超过盘本身的情况并不少见。
• 弹性：CCE 节点池配合弹性伸缩承接架构层优化，比单纯降配更可持续。
• 网络：闲置 EIP 直接释放，只解绑仍然会计费。

八、上线前 checklist

• [ ] 每个资源都能归到 service / env / owner
• [ ] 单位成本的分母按业务类型选对
• [ ] 共享成本有明确、可复查的分摊规则
• [ ] 降配依据是 P95/P99，不是均值
• [ ] 所有释放动作有快照，或确认无绑定
• [ ] 中风险动作先灰度 7 天再推全量
• [ ] 优化前后记录的是单位成本，不只是总额
• [ ] 每季度复盘一次"省下来的钱有没有变成事故"

写在最后

云成本优化不是一次清理活动，而是一条持续的度量线。

资源层帮你把明显的浪费扫掉，这件事该做，也有明确的终点。单位成本帮你判断哪些投入值得继续、哪些服务该合并或下线，这件事没有终点，但它决定了账单的长期形状。

先算那个除法，再动资源。

代码在文中，直接复制就能跑。如果你们团队也在做成本归因，欢迎聊聊共享成本是怎么摊的——这块比找闲置难得多。