AWS云成本监控、警报机制及优化方案：开发人员使用指南

在工程团队中，每个月都会出现这样一种常见的情景：有人会转发一张AWS账单的截图，显示出的费用数额比上个月要高。大家都表示认同，认为这笔费用确实应该更低一些，但实际上并不会采取任何具体的行动来改变这一状况，这样的循环就这样不断重复着。

本指南的目的就是打破这种循环，通过一种更为有效的方法来替代原有的处理方式——采用系统化、针对具体服务的分析方法，从而清楚地了解自己的支出情况、支出的原因，在问题真正形成账单之前就发现它们，并且在不需要猜测的情况下降低成本。

本书的结构设计便于读者随时查阅。每一部分都可以独立使用：如果你当前面临的是与RDS相关的问题，可以直接阅读相关章节；而如果你打算从零开始建立FinOps管理体系，那么可以按照指南的顺序从头到尾学习。书中的每一个命令都是可执行的，每一个脚本也都是可以实际应用的。

你将学到的内容

如何使用Athena设置成本和使用情况报告的查询功能，这是进行任何严肃的成本分析的基础
每个工程团队都需要的五款仪表盘，这些仪表盘都是利用目前就可以运行的查询语句构建而成的

一种三层警报机制，能够在不会导致用户产生疲劳感的情况下及时发现成本异常情况

针对EC2、S3、Lambda、RDS、DynamoDB以及数据传输等服务的具体优化方案

一个为期30天的实际操作计划，在第一个月内就能带来可量化的成本节约效果

让我们从头开始构建这个体系吧。

先决条件

在开始学习本指南之前，你需要具备以下条件：

知识要求：

对AWS旗下的服务有基本的了解和使用经验：EC2、S3、RDS、Lambda以及VPC
能够熟练阅读Python和SQL代码
对IAM策略和角色的工作原理有基本的认识

权限要求：

拥有具备计费功能的AWS账户。你使用的IAM用户或角色需要具备ce:GetCostAndUsage、ec2:Describe*、rds:Describe*以及这些权限
已经配置好了AWS CLI v2工具
拥有使用Athena的权限（用于第1部分中的CUR查询操作）

设置步骤：

如果尚未启用Cost Explorer，请立即进行启用。该工具是免费的，而且本指南中的大多数操作都需要使用它：

aws ce enable-cost-explorer --region us-east-1

需要配置“成本和使用报告”功能，并将其数据导出到S3存储桶中。如果尚未完成这些设置，可以参考AWS成本和使用报告设置指南进行操作。设置完成后需等待24小时，系统才会生成第一份数据文件。

第1部分：监控——了解每一分钱的去向

1.1 成本和使用报告——真实的成本数据来源

Cost Explorer能够显示各项服务的总使用费用，对于分析使用趋势来说非常有用，但在进行根本原因分析时还不够。当你需要了解到底是哪项具体资源导致了每月12,000美元的支出时，就需要通过Athena查询成本和使用报告了。

请根据你的成本和使用报告数据创建Athena表：

-- 在成本和使用报告开始生成数据后，在Athena中运行一次此命令
-- 将‘your-cur-bucket’和‘your-prefix’替换为实际的值

CREATE EXTERNAL TABLE IF NOT EXISTS cur_database.billing (
    billbilling_period_start_date  STRING,
    bill_payer_account_id           STRING,
    line_item_usage_start_date      STRING,
    line_item_resource_id           STRING,
    line_item_usage_type            STRING,
    line_item_usage_amount          DOUBLE,
    line_item_unblended_cost        DOUBLE,
    product_servicecode             STRING,
    product_instance_type           STRING,
    product_region                  STRING,
    resource_tags_user_environment  STRING,
    resource_tags_user_team         STRING,
    resource_tags_user_service      STRING,
    resource_tags_user_owner        STRING
)
PARTITIONED BY (year STRING, month STRING)
ROW FORMAT DELIMITED
FIELDS TERMINATED BY ','
LOCATION 's3://your-cur-bucket/your-prefix/'
TBLPROPERTIES ('skip_header.line.count'='1');

MSCK REPAIR TABLE cur_database.billing;

以下是你在第一天应该运行的三个查询语句：

-- 查询1：本月成本最高的20项资源
-- 先运行这个查询，以便确定需要重点关注的资源。
SELECT
    line_item_resource_id,
    product_servicecode,
    resource_tags_user_team     AS team,
    resource_tags_user_service  AS service,
    SUM(line_item_unblended_cost) AS total_cost_usd
FROM cur_database.billing
WHERE line_item_usage_start_date >= DATE_FORMAT(
    DATE_TRUNC('month', CURRENT_DATE), '%Y-%m-%d'
)
  AND line_item_unblended_cost > 0
GROUP BY 1, 2, 3, 4
ORDER BY total_cost_usd DESC
LIMIT 20;

-- 查询2：各服务的周环比成本增长情况
-- 通过这个查询可以找出增长最快的服务，这些服务需要进一步调查。
WITH weekly AS (
    SELECT
        DATE_TRUNC('week', CAST(line_item_usage_start_date AS DATE)) AS week,
        product_servicecode,
        SUM(line_item_unblended_cost) AS cost
    FROM cur_database.billing
    WHERE line_item_usage_start_date >=
          DATE_FORMAT(DATE_ADD('day', -42, CURRENT_DATE), '%Y-%m-%d')
    GROUP BY 1, 2
)
SELECT
    curr.product_servicecode AS service,
    ROUND(prev.cost, 2)      AS prev_week_cost,
    ROUND(curr.cost, 2)      AS curr_week_cost,
    ROUND(
        100.0 * (curr.cost - prev(cost) / NULLIF(prev.cost, 0),
        1
    ) AS pct_change
FROM weekly curr
JOIN weekly prev
  ON curr.product_servicecode = prev.product_servicecode
  AND curr.week = DATE_ADD('week', 1, prev_week)
WHERE curr.week = DATE_TRUNC('week', CURRENT_DATE)
  AND ABS((curr.cost - prev_cost) / NULLIF(prev(cost, 0)) > 0.20
ORDER BY pct_change DESC;

-- 查询3：那些在过去7天内产生了费用但使用量为零的资源——这些资源很可能是闲置或被遗忘的资源，因此应该被关闭
-- 查询结果包括资源ID、产品服务代码、资源所有者以及所属团队，同时还会显示过去7天的总成本
SELECT
    line_item_resource_id,
    product_servicecode,
    resource_tags_user_owner AS owner,
    resource_tags_user_team AS team,
    SUM(line_item_unblended_cost) AS cost_past_7_days
FROM cur_database.billing
WHERE line_item_usage_start_date >= DATE_FORMAT(DATE_ADD('day', -7, CURRENT_DATE), '%Y-%m-%d')
  AND line_item_usage_amount = 0
  AND line_item_unblended_cost > 5
GROUP BY 1, 2, 3, 4
ORDER BY cost_past_7_days DESC
LIMIT 30;

1.2 五大核心成本监控面板

这五个监控视图能够满足大多数工程团队的需求。您可以直接使用这些查询语句来生成相应的数据，无需任何第三方工具的帮助。

监控面板1：高管摘要（用于每周向领导层汇报)

# executive_summary.py
import boto3
from datetime import datetime, timedelta

ce = boto3.client('ce')

def weekly_summary():
    today = datetime.now()
    start_mtd = today.replace(day=1).strftime('%Y-%m-%d')
    today_str = today.strftime('%Y-%m-%d')

    # 本月至今的支出金额
    mtd = ce.get_cost_and_usage(
        TimePeriod {'Start': start_mtd, 'End': today_str},
        Granularity='MONTHLY',
        Metrics=['UnblendedCost']
    )
    mtd_spend = float(mtd['ResultsByTime'][0]['Total']['UnblendedCost']['Amount'])

    # 月末预测金额
    forecast = ce.get_cost_forecast(
        TimePeriod={
            'Start': today_str,
            'End': (today.replace(day=28) + timedelta(days=4)).replace(day=1).strftime('%Y-%m-%d'),
        },
        Metric='UNBLENDED_COST',
        Granularity='MONTHLY'
    )
    eom_forecast = float(forecast['Total']['Amount']) + mtd_spend

    # 前五名高成本服务
    by_service = ce.get_cost_and_usage(
        TimePeriod {'Start': start_mtd, 'End': today_str},
        Granularity='MONTHLY',
        Metrics=['UnblendedCost'],
        GroupBy=[{'Type': 'DIMENSION', 'Key': 'SERVICE'}]
    )
    services = sorted(
        [
            (g['Keys'][0], float(g['Metrics']['UnblendedCost']['Amount']))
            for g in by_service['ResultsByTime')[0]['Groups']
        ],
        key=lambda x: x[1],
        reverse=True
    )[:5]

    print(f"\n{'─'*48}")
    print(f"  AWS成本摘要 — {today.strftime('%B %Y')}")
    print(f"{'─'*48}")
    print(f"  本月至今的支出：${mtd_spend:>12,.2f}")
    print(f"  月末预测金额：${eom_forecast:>12,.2f}")
    print(f"\n  前五名高成本服务:")
    for name, cost in services:
        short = name.replace('Amazon ', '').replace('AWS ', '')
        print(f"    {short:<32} ${cost:>9,.2f}")
    print(f"{'─'*48}\n")

weekly_summary()

仪表盘2：团队成本明细（针对工程负责人）

# team_breakdown.py
import boto3
from datetime import datetime

ce = boto3.client('ce')

def team_breakdown():
    start = datetime.now().replace(day=1).strftime('%Y-%m-%d')
    end = datetime.now().strftime('%Y-%m-%d')

    response = ce.get_cost_and_usage(
        TimePeriod={'Start': start, 'End': end},
        Granularity='MONTHLY',
        Metrics=['UnblendedCost'],
        GroupBy=[
            {'Type': 'TAG', 'Key': 'Team'},
            {'Type': 'DIMENSION', 'Key': 'SERVICE'},
        ]
    )

    by_team = {}
    for group in response['ResultsByTime'][0].get('Groups', []):
        team_raw = group['Keys')[0]
        team = team_raw.replace('Team$', '') if team_raw else 'untagged'
        service = group['Keys][1]
        cost = float(group['Metrics']['UnblendedCost']['Amount'])

        if team not in by_team:
            by_team[team] = {'total': 0.0, 'by_service': {}}
        by_team[team]['total'] += cost
        by_team[team]['by_service'][service] = (
            by_team[team]['by_service'].get(service, 0.0) + cost
        )

    total_bill = sum(d['total'] for d in by_team.values())

    print(f"\n{'─'*58}")
    print(f"  团队成本明细 — 过去一个月 {datetime.now().strftime('%Y-%m-%d')}")
    print(f"  总金额：${total_bill:,.2f}")
    print(f"{'─'*58}")

    for team, data in sorted(by_team.items(), key=lambda x: x[1]['total'], reverse=True):
        pct = (data['total'] / total_bill * 100) if total_bill else 0
        print(f"\n  {team:<20}  ${data['total']:>10,.2f}  ({pct:.1f}%)")
        top3 = sorted(data['by_service'].items(), key=lambda x: x[1], reverse=True)[:3]
        for svc, cost in top3:
            short = svc.replace('Amazon ', '').replace('AWS ', '')
            print(f"    └─ {short:<30} ${cost:>8,.2f}")

    print()


team_breakdown()

仪表盘3：浪费检测（用于每周清理审查）：

# waste_detector.py
import boto3
from datetime import datetime, timezone, timedelta

ec2 = boto3.client('ec2')
elbv2 = boto3.client('elbv2')
cw = boto3.client('cloudwatch')

def detect_waste():
    report = {'items': [], 'total_monthly_waste': 0.0}

    # 未关联的EBS卷
    for vol in ec2.describe_volumes(
        Filters=[{'Name': 'status', 'Values': ['available']}]
    )['Volumes']:
        age = (datetime.now(timezone.utc) - vol['CreateTime']).days
        cost = round(vol['Size'] * 0.08, 2)
        tags = {t['Key']: t['Value'] for t in vol.get('Tags', [])}
        report['items'].append({
            'type': 'Unattached EBS Volume',
            'id': vol['VolumeId'],
            'detail': f"{vol['Size']}GB — 使用时长：{age}天",
            'owner': tags.get('Owner', '—'),
            'monthly_cost': cost,
        })
        report['total_monthly_waste'] += cost

    # 未关联的弹性IP
    for addr in ec2.describe_addresses()[‘Addresses’]:
        if 'AssociationId' not in addr:
            report['items'].append({
                'type': 'Unassociated Elastic IP',
                'id': addr.get('AllocationId', ""),
                'detail': addr['PublicIp'],
                'owner': '—',
                'monthly_cost': 3.60,
            })
            report['total_monthly_waste'] += 3.60

    # 闲置的负载均衡器（7天内接收到的请求少于100次）
    for lb in elbv2.describe_load_balancers()[‘LoadBalancers’]:
        metrics = cw.get_metric_statistics(
            Namespace='AWS/ApplicationELB',
            MetricName='RequestCount',
            Dimensions=[{'Name': 'LoadBalancer', 'Value': lb['LoadBalancerArn'].split(':loadbalancer/][-1]}],
            StartTime=datetime.now() - timedelta(days=7),
            EndTime(datetime.now(),
            Period=604800,
            Statistics=['Sum']
        )['Datapoints']
        total_requests = metrics[0]['Sum'] if metrics else 0

        if total_requests < 100:
            report['items'].append({
                'type': 'Idle Load Balancer',
                'id': lb['LoadBalancerName'],
                'detail': f"7天内接收到的请求次数：{int(total_requests)}次",
                'owner': '—',
                'monthly_cost': 22.0,
            })
            report['total_monthly_waste'] += 22.0

    print(f"\n  浪费检测报告 — {datetime.now().strftime('%Y-%m-%d')}")
    print(f"  预计的每月浪费金额：${report['total_monthly_waste']:.2f}\n")

    for item in sorted(report['items'], key=lambda x: x['monthly_cost'], reverse=True)[:20]:
        print(f"  [{item['type']}]")
        print(f"    ID：     {item['id']}")
        print(f"    详情：    {item['detail']}")
        print(f"    所有者：  {item['owner']}")
        print(f"    成本：   ${item['monthly_cost']:.2f}/月\n")

    return report


detect_waste()

1.3 标记策略——所有成本归属机制的基础

任何成本分配模型都依赖于标签。那些忽略标记工作的团队所构建的仪表盘只能显示总数，而无法提供任何解释。关键在于使标记工作具有结构化特征，而非仅仅遵循程序性流程；这种规范应该通过基础设施代码来强制执行，而不是依靠Confluence中的提醒功能。

必须使用的标签集合：

# terraform/variables.tf
variable "mandatory_tags" {
  description = "应用于该账户中所有资源的标签"
  type        = map(string)

  validation {
    condition = alltrue([
      contains(keys(var.mandatory_tags), "Environment"),
      contains(keys(var.mandatory_tags), "Team"),
      contains(keys(var.mandatory_tags), "Owner"),
      contains(keys(var.mandatory_tags), "Service"),
    ])
    error_message = "mandatory_tags必须包含Environment、Team、Owner和Service这些标签。"
  }
}

locals {
  common_tags = merge(var.mandatory_tags, {
    ManagedBy    = "terraform"
    LastModified = timestamp()
  })
}

resource "aws_instance" "api_server" {
  ami           = data.aws_ami.amazon_linux_2023.id
  instance_type = "t3.medium"
  tags          = merge(local.common_tags, {Name = "api-server-${var.environment}"})
}

每周都要查找并报告那些未被添加标签的资源：

#!/usr/bin/env bash
# find_untagged.sh

echo "未添加标签的EC2实例（缺少Team标签）:"
aws ec2 describe-instances \
  --filters "Name=instance-state-name,Values=running" \
  --query "Reservations[].Instances[?!not_null(Tags[?Key='Team'].Value|[0])].[InstanceId,InstanceType,LaunchTime]" \
  --output table

echo "未添加标签的RDS实例:"
aws rds describe-db-instances \
  --query "DBInstances[?!not_null(TagList[?Key='Team'].Value|[0])].DBInstanceIdentifier" \
  --output table

第2部分：警报机制——在费用激增成为正式账单之前及时发现并处理

如果没有主动的警报机制，典型的问题发现流程会是这样的：成本激增发生在5号，月度账单在20号寄出，有人在22号才发现这个问题，然后才开始调查，结果就是会有两周的时间白白浪费掉费用而无法挽回。而如果采用了主动警报机制，问题的发现时间可以在几小时内完成。

2.1 三层警报结构

过度接收警报与完全没有警报一样有害。三层警报结构通过将不同严重程度的警报发送到不同的渠道，并要求这些渠道采取不同的响应措施，从而确保警报信息能够得到有效处理。

# alert_router.py
import boto3
import json
import urllib.request
from enum import Enum

SLACK_INFO_WEBHOOK  = 'https://hooks.slack.com/services/INFO/WEBHOOK'
SLACK_ALERT/WebHOOK = 'https://hooks.slack.com/services/ALERT/WEBHOOK'
SNS_CRITICAL_TOPIC  = 'arn:aws:sns:us-east-1:YOUR_ACCOUNT:cost-critical'


class AlertTier(Enum):
    INFO     = 1
    WARNING  = 2
    CRITICAL = 3


def route_alert(tier: AlertTier, subject: str, message: str):
    """根据警报的严重程度，将其发送到相应的渠道。"""
    icons   = {AlertTier.INFO: ':information_source:',}
               AlertTier.WARNING: ':warning:', AlertTier.CRITICAL: ':rotating_light:'}
    payload = {'text': f"{icons[tier]} *{subject}*\n{message}"}

    if tier == AlertTier.INFO:
        _post_slack(SLACK_INFO_WEBHOOK, payload)
    elif tier == AlertTier.WARNING:
        _post_slack(SLACK_ALERT/WebHOOK, payload)
        _send_sns(SNS_critical_TOPIC, subject, f"WARNING: {message}")
    elif tier == AlertTier.CRITICAL:
        _post_slack(SLACK_alert\WebHOOK, payload)
        _send_sns(SNS_CRITICAL_topic, subject, f"CRITICAL: {message}")


def _post_slack(webhook: str, payload: dict):
    req = urllib.request.Request(
        webhook,
        data=json.dumps(payload).encode(),
        headers={'Content-Type': 'application/json'}
    )
    urllib.request.urlopen(req)


def _send_sns(topic_arn: str, subject: str, message: str):
    sns = boto3.client('sns')
    sns.publish(TopicArn=topic_arn, Subject=subject[:100], Message(message)

这三级警报系统及其对应的处理方式如下：第一级警报为信息提示，内容包括每日费用汇总、每周趋势分析以及标签合规性更新，用户无需采取任何行动。

第二级警报为警告通知，会通过Slack警报频道及电子邮件发送，适用于预算使用率超过75%、周环比增加25%或节能计划即将到期的情况，用户需在24小时内确认收到该警报。

第三级警报为紧急通知，会通过PagerDuty和短信发送，适用于预算使用率超过90%、24小时内费用增加100%、检测到加密货币挖矿行为，或预计支出金额超过预算计划的120%的情况，用户需在1小时内进行调查处理。

2.2 实时预算监控系统

默认情况下，AWS Budgets系统会每天发送一次警报。但由于每日只检查一次数据，因此如果当天8点之后出现费用激增的情况，那么直到第二天的警报才会被触发。而下面的Lambda函数会每小时运行一次，实时监测预算使用率以及每小时的费用变化情况。

# budget_monitor.py
# 这个Lambda函数每小时由EventBridge触发一次执行

import boto3
from datetime import datetime, timedelta
from alert_router import route_alert, AlertTier

ce      = boto3.client('ce')
budgets = boto3.client('budgets', region_name='us-east-1')
ACCOUNT_ID  = boto3.client('sts').get_caller_identity]['Account']
BUDGET_NAME = 'monthly-infrastructure'


def get_mtd_spend() -> float:
    start = datetime.now().replace(day=1).strftime('%Y-%m-%d')
    end   = datetime.now().strftime('%Y-%m-%d')
    r = ce.get_cost_and_usage(
        TimePeriod {'Start': start, 'End': end},
        Granularity='MONTHLY',
        Metrics=['UnblendedCost']
    )
    return float(r['ResultsByTime'][0]['Total']['UnblendedCost']['Amount')


def get_budget_limit() -> float:
    r = budgets.describe_budget(AccountId=ACCOUNT_ID, BudgetName=BUDGET_NAME)
    return float(r['Budget']['BudgetLimit']['Amount])


def get_hourly_costs(hours: int = 4) -> list:
    """返回过去N小时的每小时费用总额。"""
    end   = datetime.now()
    start = end - timedelta(hours=hours)
    r = ce.get_cost_and_usage(
        TimePeriod {'Start': start.strftime('%Y-%m-%d'), 'End': end.strftime('%Y-%m-%d')},
        Granularity='HOURLY',
        Metrics=['UnblendedCost']
    )
    return [
        float(period['Total']['UnblendedCost']['Amount'])
        for period in r['ResultsByTime']
    ]


def lambda_handler(event, context):
    mtd_spend    = get_mtd_spend()
    budget_limit = get_budget_limit()
    utilisation  = mtd_spend / budget_limit * 100

    days_elapsed  = datetime.now().day
    projected_eom = (mtd_spend / daysElapsed) * 30
    projected pct = projected_eom / budget_limit * 100

    if utilisation >= 90:
        route_alert(
            AlertTier.CRITICAL,
            f'预算使用率为{utilisation:.0f}%',
            f'本月的总支出为${mtd_spend:,.2f}，占预算总额的{utilisation:.0f}%。'
            f'预计耗尽预算的时间为：${projected_eom:,.2f}。'
        )
    elif utilisation >= 75:
        route_alert(
            AlertTierWARNING,
            f'预算使用率为{utilisation:.0f}%',
            f'本月的总支出为${mtd_spend:,.2f}，占预算总额的{utilisation:.0f}%。'
            f'预计耗尽预算的时间为：${projected_eom:,.2f}。'
        )

    # 检查每小时的费用变化情况
    hourly = get_hourly_costs(hours=4)
    if len(hourly) >= 2:
        last_hour = hourly[-1]
        prev_avg  = sum(hourly[:-1]) / len(hourly[:-1])
        if prev_avg > 0.10 and last_hour > prev_avg * 1.5:
            route_alert(
                AlertTierWARNING,
                '检测到每小时费用激增的情况，
                '最后一小时的支出为${last_hour:.2f}，而过去3小时的平均支出为${prev_avg:.2f}，
                '增幅为：{((last_hour/prev_avg - 1) * 100):.0f}%。'
            )

    return {
        'mtd_spend':       round(mtd_spend, 2),
        'utilisation_pct': round(utilisation, 1),
        'projected_eom':   round(projected_eom, 2),
    }

第3部分：按服务进行优化

3.1 EC2——按优先级顺序实施的七项优化措施

在大多数AWS账户中，EC2都是占比最大的一项服务，同时也提供了最多的优化选项。请按照以下顺序依次执行这些优化步骤，因为每完成一项优化，后续步骤所针对的基准值就会相应降低。

措施1：找出真正处于闲置状态的实例（CPU使用率连续14天低于1%）。

# ec2_idle_finder.py
import boto3
from datetime import datetime, timedelta

ec2 = boto3.client('ec2')
cw  = boto3.client('cloudwatch')

def findidle_instances(avg_cpu_threshold: float = 1.0, days: int = 14):
    instances = [
        inst
        for r in ec2.describeInstances(
            Filters=[{'Name': 'instance-state-name', 'Values': ['running']}]
        )['Reservations']
        for inst in r['Instances']
    ]

    idle = []
    for inst in instances:
        iid   = inst['InstanceId']
        stats = cw.get MetricStatistics(
            Namespace='AWS/EC2',
            MetricName='CPUUtilization',
            Dimensions=[{'Name': 'InstanceId', 'Value': iid}],
            StartTime datetime.utcnow() - timedelta(days=days),
            EndTime datetime.utcnow(),
            Period=days * 86400,
            Statistics=['Average']
        )['Datapoints']

        avg_cpu = stats[0]['Average'] if stats else 0.0
        if avg_cpu < avg_cpu_threshold:
            tags = {t['Key']: t['Value'] for t in inst.get('Tags', [])}
            idle.append({
                'instance_id':   iid,
                'instance_type': inst['InstanceType'],
                'avg_cpu':       round(avg_cpu, 2),
                'environment':   tags.get('Environment', '—'),
                'owner':         tags.get('Owner', '—'),
            })

    return sorted(idle, key=lambda x: x['avg_cpu'])

for inst in find_idle_instances():
    print(f"  {inst['instance_id']}  {inst['instance_type']}  "
          f"{inst['avg_cpu}% CPU  环境：{inst['environment']}  所有者：{inst['owner']}")

措施2：调整那些配置过度但实际使用率较低的实例的规模（确保CPU使用率持续低于20%）。

只需将脚本中的`avg_cpu_threshold`值改为`20.0`即可。这些实例是需要调整规模的对象，而不是需要关闭的对象。

措施3：对于那些被标记为`AutoShutdown=true`的实例，利用EventBridge规则来安排它们的关机操作。

措施4：只有在完成措施1至措施3之后，才考虑购买节省费用的计划。

措施5：迁移到Graviton上——其价格便宜20%，而且对于大多数工作负载来说，性能与现有方案相当。

措施6：对于那些需要具备容错能力的批处理任务或开发任务，可以使用Spot实例。

措施7：将基于容器的应用 workload迁移到EKS平台上，并利用Karpenter工具进行自动资源调度。

以下是计算Spot实例节省费用的具体代码：

# spot_price_analyser.py
import boto3

ec2 = boto3.client('ec2')

def spot_savings_estimate(instance_type: str) -> dict:
    spot_history = ec2.describe_spot_price_history(
        InstanceTypes=[instance_type],
        ProductDescriptions=['Linux/UNIX'],
        MaxResults=1
    )['SpotPriceHistory']
    spot_price = float(spot_history[0]['SpotPrice']) if spot_history else 0

    on_demand_approx = {
        't3.medium': 0.0416, 'm5.large': 0.096,
        'c5.xlarge': 0.17,   'r5.2xlarge': 0.504,
    }
    od_price = on_demand_approx.get(instance_type, 0)
    savings_pct = ((od_price - spot_price) / od_price * 100) if od_price else 0

    return {
        'instance_type': instance_type,
        'spot_price':    round(spot_price, 4),
        'ondemand':     od_price,
        'savings pct':   round(savings_pct, 1),
        'monthly_spot':  round(spot_price * 730, 2),
        'monthly_od':    round(od_price * 730, 2),
    }

for itype in ['t3.medium', 'm5.large', 'c5.xlarge']:
    r = spot_savings_estimate(itype)
    print(f"  {r['instance_type']:<15} Spot价格：{r['spot_price']}/小时  "
          f"常规价格：{r['on_demand']}/小时  节省费用百分比：{r['savings_pct}%")

3.2 S3——生命周期策略与存储类选择

S3优化主要包括两个方面：通过生命周期策略将不常被访问的数据转移到更便宜的存储类别中，以及消除诸如未完成的多部分上传这类资源浪费现象。

# s3_lifecycle_applier.py
import boto3

s3 = boto3.client('s3')

LOG_POLICY = {
    'Rules': [
        {
            'ID': 'standard-tiering',
            'Status': 'Enabled',
            'Filter': {'Prefix': ''},
            'Transitions': [
                {'Days': 30,  'StorageClass': 'STANDARD_IA'},
                {'Days': 90,  'StorageClass': 'GLACIER_IR'},
                {'Days': 365, 'StorageClass': 'DEEP_ARCHIVE'},
            ],
            'Expiration': {'Days': 2555},
            'AbortIncompleteMultipartUpload': {'DaysAfterInitiation': 7},
        },
    ],
}

TEMP_POLICY = {
    'Rules': [
        {
            'ID': 'temp-data-retention',
            'Status': 'Enabled',
            'Filter': {'Prefix': ''},
            'Expiration': {'Days': 30},
            'AbortIncompleteMultipartUpload': {'DaysAfterInitiation': 1},
        },
    ]
}

for bucket in s3.list_buckets]['Buckets']:
    name = bucket['Name']
    try:
        s3.get_bucket_lifecycle_configuration(Bucket=name)
        print(f"  {name} — 策略已存在，跳过此步骤")
    except s3.exceptions.ClientError:
        policy = TEMP_POLICY if any(k in name for k in ['temp', 'build', 'cache']) else LOG_policy
        s3.put_bucket_lifecycle_configuration(
            Bucket=name, LifecycleConfiguration=policy
        )
        print(f"  {name} — 已应用{'TEMP' if policy is TEMP-policy else 'LOG'}策略")

3.3 RDS——五种优化级别

级别	操作内容	通常能节省的成本百分比	风险等级
1	删除未使用的读取副本	可节省30–50%的副本成本	风险较低
2	将备份保留时间缩短至合规要求的最低限度	可节省20–30%的存储成本	风险较低
3	选择合适规模的实例类型（确保CPU使用率持续低于20%）	可节省20–40%的计算资源成本	风险中等
4	为生产环境购买预留实例	可节省30–60%的计算资源成本	风险较低
5	将负载可变的数据库迁移至Aurora Serverless v2	可节省40–70%的总成本	实施难度较高

如何识别那些配置过高的RDS实例：

# rds_rightsizer.py
import boto3
from datetime import datetime, timedelta

rds = boto3.client('rds')
cw  = boto3.client('cloudwatch')

def find_oversized_rds():
    instances  = rds.describe_db_instances]['DBInstances']
    candidates = []

    for inst in instances:
        iid    = inst['DBInstanceIdentifier']
        iclass = inst['DBInstanceClass']

        stats = cw.get_metric_statistics(
            Namespace='AWS/RDS',
            MetricName='CPUUtilization',
            Dimensions=[{'Name': 'DBInstanceIdentifier', 'Value': iid}],
            StartTime(datetime.utcnow() - timedelta(days=14),
            EndTime.datetime.utcnow(),
            Period=1209600,
            Statistics=['Average', 'Maximum']
        )['Datapoints']

        if not stats:
            continue

        avg_cpu = stats[0]['Average']
        max_cpu = stats[0]['Maximum']

        if avg_cpu < 20 and max_cpu < 50:
            candidates.append({
                'id':       iid,
                'class':    iclass,
                'avg_cpu':  round(avg_cpu, 1),
                'max_cpu':  round(max_cpu, 1),
                'engine':   inst['Engine'],
            })

    return candidates

for c in find_oversized_rds():
    print(f"  {c['id']}  {c['class']}  平均CPU使用率：{c['avg_cpu}%  最高CPU使用率：{c['max_cpu]%  使用的数据库引擎：{c['engine']}")

3.4 DynamoDB——按需使用模式与预配置模式的选择

按需使用模式虽然方便，但对于那些工作负载可预测的情况而言，其费用可能会是预配置模式的3到5倍。而采用自动扩展功能的预配置模式则能以显著更低的成本满足大多数需求。

# dynamodb_mode_advisor.py
import boto3
from datetime import datetime, timedelta

dynamodb = boto3.client('dynamodb')
cw = boto3.client('cloudwatch')

def analyse_tablebilling(table_name: str) -> dict:
    table = dynamodb.describe_table(TableName=table_name)['Table']
    mode = table.get('BillingModeSummary', {}).get('BillingMode', 'PROVISIONED')

    stats = {}
    for metric in ['ConsumedReadCapacityUnits', 'ConsumedWriteCapacityUnits']:
        data = cw.get_metric_statistics(
            Namespace='AWS/DynamoDB',
            MetricName=metric,
            Dimensions=[{'Name': 'TableName', 'Value': table_name}],
            StartTime=datetime.utcnow() - timedelta(days=30),
            EndTime(datetime.utcnow(),
            Period=86400,
            Statistics=['Average', 'Maximum']
        )['Datapoints']
        if data:
            avg = sum(d['Average'] for d in data) / len(data)
            peak = max(d['Maximum'] for d in data)
            stats[metric] = {'avg': round(avg, 1), 'peak': round(peak, 1)}

    if mode == 'PAY_PER_REQUEST':
        avg_rcu = stats.get('ConsumedReadCapacityUnits', {}).get('avg', 0)
        avg_wcu = stats.get('ConsumedWriteCapacityUnits',('').get('avg', 0)

        if avg_rcu > 2000 or avg_wcu > 500:
            return {
                'table': table_name,
                'current_mode': 'PAY_PER_REQUEST',
                'recommendation': '切换到带有自动扩展功能的预配置模式',
                'reason': f'平均每秒读取容量为{avg_rcu:.0f} RCU，平均每秒写入容量为{avg_wcu:.0f} WCU——这些数据符合可预测的工作负载模式'
            }

    return {'table': table_name, 'current_mode': mode, 'recommendation': '无需进行任何调整'}

for table in dynamodb.list_tables]['TableNames']:
    r = analyse_tablebilling(table)
    if r['recommendation'] != 'No change needed':
        print(f"{r['table']}: {r['recommendation']}")

3.5 数据传输——三种主要的资源浪费情况

在AWS的账单中，数据传输费用往往是最让人困惑的项目。以下是三种常见的资源浪费情况及其解决方法：

跨区域之间的数据传输：不同区域之间的服务进行数据传输时，每GB的数据传输费用为0.01美元。解决办法是在Kubernetes服务中使用能够识别网络拓扑结构的路由机制，或者确保应用程序层和数据库层位于同一区域。

通过NAT Gateway进行的内部数据传输会产生额外费用：对于通过NAT Gateway传输的S3、ECR、DynamoDB和SQS数据，每GB的费用为0.045美元。解决方法是使用VPC端点进行数据传输，从而完全避免这种费用。

不符合合规性要求却进行了跨区域数据复制：建议每季度检查一次S3的复制规则，确保它们符合实际的合规性要求。

# 查找具有活动复制功能的S3桶
for bucket in $(aws s3api list-buckets --query 'Buckets[*].Name' --output text); do
    result=$(aws s3api get-bucket-replication --bucket "$bucket" 2>&1)
    if ! echo "$result" | grep -q "ReplicationConfigurationNotFoundError"; then
        echo "  $bucket — 复制功能处于活动状态，需验证是否符合相关要求"
    fi
done

第4部分：为期30天的优化计划

这个优化计划在第一个月内就能带来可量化的成本节约效果。它适合由一名工程师每周投入两到四个小时来执行。

第1周 – 建立监控体系：启用Cost Explorer，设置Athena表格，运行为期三天的查询任务，将查询结果截图作为基准数据；为所有正在运行的EC2和RDS实例添加标签，并确定导致成本最高的三个因素，并针对每个因素制定详细的分析方案。

第2周 – 实施快速见效的措施：部署用于检测闲置资源的Lambda函数，为规模最大的三个S3桶应用生命周期策略，运行闲置实例检测工具，关闭那些平均CPU使用率低于1%且没有所有者提出异议的实例，并为S3、ECR和DynamoDB添加VPC终端节点。

第3周 – 进行资源优化：运行EC2资源优化分析工具，优先缩减那些置信度最高的闲置资源；运行RDS资源优化脚本，关闭那些流量极低的读复制实例。

第4周 – 建立警报机制与自动化流程：部署用于监控每小时预算消耗情况的Lambda函数，配置三级警报通知系统，设置每周一次的成本浪费报告机制；将Infracost GitHub Action添加到基础设施管理仓库中，并安排每月一次、持续30分钟的FinOps评审会议。

30天后的预期效果：AWS月度支出减少15%至25%，所有调整措施都有书面记录，同时建立起能够防止类似浪费再次发生的机制。

最佳实践总结

✅ 务必执行：在开展任何其他监控工作之前，先设置Cost Explorer和Athena。Cost Explorer只是一个起点，而Cost Explorer提供的数据才是最准确的依据。

✅ 务必执行：在Terraform或CloudFormation中强制要求资源添加标签。基于流程的标签管理可能会失效，但通过基础设施层面强制设置的标签则是永久有效的。

✅ 务必执行：每周运行闲置实例检测工具和成本浪费报告机制。成本浪费会持续累积，因此定期进行检查有助于将损失控制在较小范围内。

✅ 务必执行：采用三级警报机制。如果所有警报信息都集中在一个渠道中，很容易导致人们忽略其中的重要信息，因此应该分开设置不同的警报通道。

✅ 务必执行：按照合理的顺序依次实施EC2资源优化措施。在启用节省计划之前先进行资源优化，可以避免以后不得不以较低的价格继续承担不必要的成本开支。

✅ 务必执行：每季度检查DynamoDB的计费模式是否与实际使用情况相符。

❌ 切勿执行：未经调查就删除未添加标签的资源。未添加标签并不意味着这些资源没有被使用，而只是意味着它们尚未被认领而已。

❌ 请勿：在未先审核数据访问模式的情况下，强行应用过于严格的S3生命周期策略。如果数据的访问频率高于预期，使用Glacier进行数据检索所产生的费用可能会超过标准存储服务的成本。

❌ 请不要： 在首次部署时，启用自动删除功能来运行“ Waste Reporter Lambda”脚本。应先以仅生成报告的模式运行该脚本两周，以便验证其输出结果，之后再添加删除逻辑。

资源

AWS成本与使用报告数据字典: 本指南中所有涉及CUR Athena查询的列名参考
AWS Cost Explorer API参考手册: 关于使用Python的boto3库进行成本查询的详细说明
AWS计算优化器: 基于机器学习的资源配置建议工具，可用来验证手动分析脚本的结果
Amazon DynamoDB定价信息: 第3.4节中关于按配置方式与按需使用方式计算成本的相关说明
AWS实例调度器: AWS官方提供的基于标签的EC2及RDS任务调度解决方案
FinOps基础框架: 本指南所遵循的“信息收集、优化配置、运营执行”工作流程框架
配套资源仓库: 本指南中包含的所有脚本、Lambda函数及Terraform模块