月账单 35 万里 40% 在空转：FinOps 从标签归因到成本即代码的 6 个生产决策

# 必选标签（5 个维度）
Owner       = "张三"          # 责任人，出问题时找得到人
Team        = "payment"        # 团队，用于成本分摊
Project     = "order-service"  # 项目/产品线
Environment = "production"     # 环境：production/staging/development
CostCenter  = "CC-001"         # 成本中心，关联财务系统

#!/usr/bin/env python3
"""
标签合规检查脚本：扫描所有 ECS 实例，检查必选标签是否完整。
用于 CI/CD 流水线的资源创建门禁，或定时巡检。
"""
import json
import sys
from alibabacloud_ecs20140526.client import Client as EcsClient
from alibabacloud_tea_openapi.models import Config

REQUIRED_TAGS = ["Owner", "Team", "Project", "Environment", "CostCenter"]

def check_tag_compliance(access_key_id, access_key_secret, region_id="cn-hangzhou"):
    """扫描指定区域所有 ECS 实例的标签合规性"""
    config = Config(
        access_key_id=access_key_id,
        access_key_secret=access_key_secret,
        region_id=region_id,
        endpoint=f"ecs.{region_id}.aliyuncs.com"
    )
    client = EcsClient(config)

    # 获取所有实例
    instances = client.describe_instances(
        describe_instances_request={}
    ).body.instances.instance

    violations = []
    for inst in instances:
        instance_id = inst.instance_id
        instance_name = inst.instance_name
        existing_tags = {tag.tag_key: tag.tag_value for tag in (inst.tags.tag or [])}

        missing = [t for t in REQUIRED_TAGS if t not in existing_tags]
        if missing:
            violations.append({
                "instance_id": instance_id,
                "instance_name": instance_name,
                "missing_tags": missing,
                "existing_tags": existing_tags
            })

    return violations

if __name__ == "__main__":
    violations = check_tag_compliance(
        access_key_id=sys.argv[1],
        access_key_secret=sys.argv[2]
    )
    if violations:
        print(f"[FAIL] 发现 {len(violations)} 台标签不合规实例：")
        for v in violations:
            print(f"  {v['instance_id']} ({v['instance_name']}): 缺少 {v['missing_tags']}")
        sys.exit(1)
    else:
        print("[PASS] 所有实例标签合规")

# 批量修复标签大小写（阿里云 CLI 示例）
# 先列出所有标签，找出需要修复的实例
aliyun ecs DescribeTags --RegionId cn-hangzhou | \
  jq '.Tags.Tag[] | select(.TagKey | test("team"))' | \
  jq -r '.InstanceId'

# 对每个实例删除旧标签、添加标准标签
for instance_id in $(aliyun ecs DescribeInstances --RegionId cn-hangzhou | \
  jq -r '.Instances.Instance[] | select(.Tags.Tag[]? | .TagKey == "team") | .InstanceId'); do
    # 删除小写标签
    aliyun ecs RemoveTags --ResourceType instance --ResourceId "$instance_id" \
        --Tag.1.Key team --RegionId cn-hangzhou
    # 读取小写标签的值
    tag_value=$(aliyun ecs DescribeInstances --InstanceIds "[\"$instance_id\"]" | \
        jq -r '.Instances.Instance[0].Tags.Tag[] | select(.TagKey == "team") | .TagValue')
    # 添加大写标签
    aliyun ecs AddTags --ResourceType instance --ResourceId "$instance_id" \
        --Tag.1.Key Team --Tag.1.Value "$tag_value" --RegionId cn-hangzhou
done

┌─────────────────────────────────────────────────────┐
│                    OpenCost 架构                     │
├─────────────────────────────────────────────────────┤
│                                                     │
│  Prometheus  ──→  OpenCost  ──→  成本数据           │
│  (指标采集)       (成本计算引擎)    (API + UI)       │
│                                                     │
│  输入：                                             │
│  • node-exporter → 节点价格、规格                   │
│  • kube-state-metrics → Pod 资源请求               │
│  • cAdvisor → 实际资源使用                          │
│  • 云厂商定价 API → 节点单价                        │
│                                                     │
│  输出：                                             │
│  • 命名空间成本                                     │
│  • Pod 级成本                                       │
│  • 标签维度成本                                     │
│  • 资源效率（利用率 vs 请求值）                     │
│                                                     │
└─────────────────────────────────────────────────────┘

# 安装 OpenCost（需要现有 Prometheus）
kubectl apply -f https://raw.githubusercontent.com/opencost/opencost/develop/kubernetes/opencost.yaml -n opencost

# 验证部署
kubectl get pods -n opencost
# 期望输出：opencost 1/1 Running

# 端口转发访问 UI
kubectl port-forward -n opencost service/opencost 9003 9090

# 访问 http://localhost:9090 查看成本仪表盘

# OpenCost 自定义成本分配规则
# 通过 Pod 标签实现业务线维度成本归因
apiVersion: v1
kind: ConfigMap
metadata:
  name: opencost-config
  namespace: opencost
data:
  # 按标签分摊成本
  costLabels: "team,project,env"
  # 节点价格覆盖（自建 IDC 或非云环境需要手动配置）
  # 否则 OpenCost 会通过云厂商 API 自动获取节点价格
  pricing: |
    {
      "defaultCPUHourlyCost": 0.0,
      "defaultRAMHourlyCost": 0.0,
      "customPricesEnabled": true
    }    

# Pod 标签示例：成本归因标签
apiVersion: apps/v1
kind: Deployment
metadata:
  name: order-service
  namespace: production
spec:
  template:
    metadata:
      labels:
        app: order-service
        team: payment          # 团队
        project: order-system  # 项目
        env: production        # 环境

#!/usr/bin/env python3
"""
ECS 右尺寸巡检：基于 CloudMonitor 指标生成降配建议
依赖：alibabacloud_cloudmonitor20190801
"""
import json
from datetime import datetime, timedelta
from alibabacloud_cloudmonitor20190801.client import Client as CmsClient
from alibabacloud_tea_openapi.models import Config

def get_cpu_utilization(client, instance_id, days=7):
    """获取实例过去 N 天的 CPU 利用率 P95"""
    end_time = datetime.now()
    start_time = end_time - timedelta(days=days)
    period = 300  # 5 分钟一个数据点

    response = client.describe_metric_list(
        describe_metric_list_request={
            "MetricName": "CPUUtilization",
            "Dimensions": json.dumps({"instanceId": instance_id}),
            "StartTime": start_time.strftime("%Y-%m-%d %H:%M:%S"),
            "EndTime": end_time.strftime("%Y-%m-%d %H:%M:%S"),
            "Period": str(period)
        }
    )

    values = []
    for datapoint in response.body.datapoints:
        dp = json.loads(datapoint) if isinstance(datapoint, str) else datapoint
        if "Average" in dp:
            values.append(dp["Average"])

    if not values:
        return None

    # 计算 P95
    values.sort()
    p95_index = int(len(values) * 0.95)
    return values[p95_index]

def generate_rightsizing_report(client, instances):
    """生成右尺寸建议报告"""
    report = []
    for inst in instances:
        instance_id = inst["instance_id"]
        instance_type = inst["instance_type"]

        cpu_p95 = get_cpu_utilization(client, instance_id)
        if cpu_p95 is None:
            continue

        if cpu_p95 < 15:
            suggestion = "降配 50%"
            priority = "高"
        elif cpu_p95 < 30:
            suggestion = "降配 30%"
            priority = "中"
        elif cpu_p95 < 60:
            suggestion = "保持不变"
            priority = "低"
        else:
            suggestion = "升配或扩容"
            priority = "紧急"

        report.append({
            "instance_id": instance_id,
            "instance_type": instance_type,
            "cpu_p95": round(cpu_p95, 1),
            "suggestion": suggestion,
            "priority": priority
        })

    return report

if __name__ == "__main__":
    config = Config(
        access_key_id="YOUR_AK",
        access_key_secret="YOUR_SK",
        endpoint="metrics.cn-hangzhou.aliyuncs.com"
    )
    cms_client = CmsClient(config)

    # 实例列表（实际从 ECS API 获取）
    instances = [
        {"instance_id": "i-bp1xxx", "instance_type": "ecs.c5.2xlarge"},
        {"instance_id": "i-bp1yyy", "instance_type": "ecs.r5.large"},
    ]

    report = generate_rightsizing_report(cms_client, instances)
    print(f"{'实例ID':<20} {'规格':<20} {'CPU P95':<10} {'建议':<15} {'优先级'}")
    print("-" * 80)
    for r in report:
        print(f"{r['instance_id']:<20} {r['instance_type']:<20} {r['cpu_p95']:<10} {r['suggestion']:<15} {r['priority']}")

# K8s Spot 节点池 PodDisruptionBudget
apiVersion: policy/v1
kind: PodDisruptionBudget
metadata:
  name: spot-service-pdb
  namespace: production
spec:
  minAvailable: 2  # 至少保持 2 个 Pod 可用
  selector:
    matchLabels:
      node-role: spot
---
# Spot 节点上的 Deployment
apiVersion: apps/v1
kind: Deployment
metadata:
  name: batch-processor
spec:
  replicas: 6
  template:
    spec:
      nodeSelector:
        node-role: spot
      tolerations:
        - key: "spot-instance"
          operator: "Exists"
      containers:
        - name: processor
          image: registry.example.com/processor:v1.0
          lifecycle:
            preStop:
              exec:
                command:
                  - /bin/sh
                  - -c
                  - |
                    # 优雅退出：从负载均衡摘除，等待在途请求处理完
                    curl -s http://lb-controller/deregister
                    sleep 30                    
          readinessProbe:
            httpGet:
              path: /health
              port: 8080
            initialDelaySeconds: 5
            periodSeconds: 5

# GitLab CI 集成 Infracost 成本预估
# .gitlab-ci.yml
stages:
  - validate
  - cost-check
  - plan
  - apply

infracost-cost-check:
  stage: cost-check
  image: infracost/infracost:latest
  variables:
    INFRACOST_API_KEY: $INFRACOST_API_KEY
  script:
    # 生成成本预估报告
    - infracost breakdown --path . --format json --out-file infracost.json
    # 生成对比报告（与 main 分支的基线对比）
    - infracost diff --path . --compare-to infracost-base.json --format json --out-file infracost-diff.json
    # 提取月度成本变化
    - |
      MONTHLY_COST=$(jq -r '.projects[0].breakdown.totalMonthlyCost // 0' infracost.json)
      COST_DIFF=$(jq -r '.projects[0].diff.totalMonthlyCost // 0' infracost-diff.json)
      echo "本次变更月度成本：${MONTHLY_COST} USD"
      echo "与基线对比变化：${COST_DIFF} USD"
      # 预算门禁：月增超过 5000 USD 需要审批
      THRESHOLD=5000
      if (( $(echo "${COST_DIFF} > ${THRESHOLD}" | bc -l) )); then
        echo "⚠️ 成本增幅超阈值（${THRESHOLD} USD/月），需架构师审批"
        exit 1
      fi      
  artifacts:
    paths:
      - infracost.json
      - infracost-diff.json
  only:
    - merge_requests

#!/usr/bin/env python3
"""
日消费告警：检查当日云消费是否超过日均预算
通过阿里云 BSS（费用中心）API 获取实时消费数据
"""
import json
from alibabacloud_bssopenapi20171214.client import Client as BssClient
from alibabacloud_tea_openapi.models import Config

def check_daily_cost(access_key_id, access_key_secret, monthly_budget=280000):
    """
    检查当日消费是否异常
    monthly_budget: 月度预算（元），日均预算 = monthly_budget / 30
    """
    config = Config(
        access_key_id=access_key_id,
        access_key_secret=access_key_secret,
        endpoint="business.aliyuncs.com"
    )
    client = BssClient(config)

    # 查询当月账单
    from alibabacloud_bssopenapi20171214.models import QueryBillRequest
    import datetime
    now = datetime.datetime.now()
    response = client.query_bill(
        query_bill_request=QueryBillRequest(
            billing_cycle=f"{now.year}-{now.month:02d}"
        )
    )

    # 统计当日消费
    daily_cost = 0
    for item in response.body.data.account_bill_list.items:
        if item.billing_date == now.strftime("%Y-%m-%d"):
            daily_cost += float(item.pretax_amount)

    daily_budget = monthly_budget / 30
    ratio = daily_cost / daily_budget

    if ratio > 1.5:
        level = "CRITICAL"
    elif ratio > 1.2:
        level = "WARNING"
    elif ratio > 1.0:
        level = "NOTICE"
    else:
        level = "OK"

    return {
        "daily_cost": round(daily_cost, 2),
        "daily_budget": round(daily_budget, 2),
        "ratio": round(ratio, 2),
        "level": level
    }

if __name__ == "__main__":
    result = check_daily_cost("YOUR_AK", "YOUR_SK", monthly_budget=280000)
    print(json.dumps(result, indent=2, ensure_ascii=False))
    # CRITICAL 时通过钉钉/飞书机器人发送告警

def detect_cost_anomaly(client, instance_ids, days=7):
    """
    环比异常检测：当日成本与过去 N 天日均比较
    如果当日成本超过过去 N 天日均的 200%，判定为异常
    """
    import datetime
    now = datetime.datetime.now()

    costs = []
    for i in range(days + 1):
        date = (now - datetime.timedelta(days=i)).strftime("%Y-%m-%d")
        daily_cost = get_daily_cost(client, date)
        costs.append({"date": date, "cost": daily_cost})

    today_cost = costs[0]["cost"]
    avg_cost = sum(c["cost"] for c in costs[1:]) / days

    if avg_cost > 0 and today_cost > avg_cost * 2:
        return {
            "anomaly": True,
            "today_cost": today_cost,
            "avg_cost": round(avg_cost, 2),
            "increase_ratio": round(today_cost / avg_cost, 2),
            "message": f"当日成本 {today_cost} 元，超过过去 {days} 天日均 {round(avg_cost, 2)} 元的 {round(today_cost / avg_cost, 2)} 倍"
        }
    return {"anomaly": False, "today_cost": today_cost, "avg_cost": round(avg_cost, 2)}

#!/usr/bin/env python3
"""
闲置资源巡检：扫描 CPU 低利用率 ECS、无后端 SLB、未挂载云盘
每日执行，结果发钉钉/飞书通知
"""
import json
import datetime

def scan_idle_ecs(client, threshold=5, days=7):
    """扫描 CPU 利用率持续低于阈值的 ECS 实例"""
    idle_instances = []
    # 获取所有运行中实例
    instances = client.describe_instances(
        describe_instances_request={"status": "Running"}
    ).body.instances.instance

    for inst in instances:
        cpu_avg = get_cpu_avg(client, inst.instance_id, days)
        if cpu_avg is not None and cpu_avg < threshold:
            idle_instances.append({
                "instance_id": inst.instance_id,
                "instance_name": inst.instance_name,
                "instance_type": inst.instance_type,
                "cpu_avg": round(cpu_avg, 1),
                "idle_days": days,
                "monthly_cost": estimate_monthly_cost(inst.instance_type),
                "suggestion": "建议释放或降配"
            })
    return idle_instances

def scan_idle_slb(client):
    """扫描无后端服务器的 SLB 实例"""
    idle_slbs = []
    slbs = client.describe_load_balancers(
        describe_load_balancers_request={}
    ).body.load_balancers.load_balancer

    for slb in slbs:
        # 获取后端服务器列表
        backends = client.describe_health_status(
            describe_health_status_request={
                "load_balancer_id": slb.load_balancer_id
            }
        ).body.backend_servers.backend_server

        if not backends:
            idle_slbs.append({
                "slb_id": slb.load_balancer_id,
                "slb_name": slb.load_balancer_name,
                "address": slb.address,
                "monthly_cost": 30,  # SLB 基础费用约 30 元/月
                "suggestion": "无后端服务器，建议释放"
            })
    return idle_slbs

def scan_unattached_disks(client):
    """扫描未挂载的云盘"""
    idle_disks = []
    disks = client.describe_disks(
        describe_disks_request={"status": "Available"}  # Available = 未挂载
    ).body.disks.disk

    for disk in disks:
        idle_disks.append({
            "disk_id": disk.disk_id,
            "disk_name": disk.disk_name,
            "size_gb": disk.size,
            "disk_type": disk.category,
            "create_time": disk.creation_time,
            "monthly_cost": estimate_disk_monthly(disk.size, disk.category),
            "suggestion": "未挂载云盘，建议释放或挂载"
        })
    return idle_disks

def generate_report(idle_ecs, idle_slbs, idle_disks):
    """生成巡检报告"""
    total_waste = sum(i["monthly_cost"] for i in idle_ecs) + \
                  sum(s["monthly_cost"] for s in idle_slbs) + \
                  sum(d["monthly_cost"] for d in idle_disks)

    report = f"""
闲置资源巡检报告（{datetime.date.today()}）

闲置 ECS：{len(idle_ecs)} 台，月度浪费约 {sum(i['monthly_cost'] for i in idle_ecs)} 元
闲置 SLB：{len(idle_slbs)} 个，月度浪费约 {sum(s['monthly_cost'] for s in idle_slbs)} 元
未挂载云盘：{len(idle_disks)} 块，月度浪费约 {sum(d['monthly_cost'] for d in idle_disks)} 元

合计月度浪费：{total_waste} 元
"""
    if total_waste > 10000:
        report += "\n⚠️ 月度浪费超 1 万元，建议立即处理！"
    return report