AWS FinOps 指南：产品市场契合后 Series A 初创公司的 8 种成本模式

# 获取上个月按服务划分的 AWS 成本明细
# 这将成为你的基准值——保存到某个地方
aws ce get-cost-and-usage \
  --time-period Start=$(date -d 'last month' +%Y-%m-01),End=$(date +%Y-%m-01) \
  --granularity MONTHLY \
  --group-by Type=DIMENSION,Key=SERVICE \
  --metrics UnblendedCost \
  --query 'ResultsByTime[0].Groups[*].{Service:Keys[0],Cost:Metrics.UnblendedCost.Amount}' \
  --output table | sort -k3 -rn

Alice（功能/支付流程）的开发环境：
  EC2 m5.xlarge — 最后一次 CPU 活动：23 天前
  RDS db.t3.medium — 最后一次连接：19 天前
  EKS 命名空间 — 最后一次 Pod 调度：15 天前
  每月成本：$187
  状态：运行中

# 查找过去 14 天内平均 CPU 使用率低于 5% 的 EC2 实例
# 这些是空闲实例——可以考虑关闭或终止
aws cloudwatch get-metric-statistics \
  --namespace AWS/EC2 \
  --metric-name CPUUtilization \
  --period 1209600 \
  --statistics Average \
  --start-time $(date -d '14 days ago' --iso-8601=seconds) \
  --end-time $(date --iso-8601=seconds) \
  --dimensions Name=InstanceId,Value=YOUR_INSTANCE_ID \
  --query 'Datapoints[*].{Average:Average}' \
  --output table

# idle_environment_stopper.py
# 部署为由 EventBridge 定时触发的 Lambda 函数：cron(0 22 * * ? *)
# 在夜间和周末运行之前停止空闲的开发环境

import boto3
from datetime import datetime, timedelta, timezone

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

IDLE_CPU_THRESHOLD = 5.0      # 停止 CPU 平均使用率低于此阈值的实例
IDLE_DAYS = 7                  # 查看过去 7 天的 CloudWatch 数据
SNS_TOPIC_ARN = 'arn:aws:sns:us-east-1:YOUR_ACCOUNT:dev-environment-alerts'

def get_average_cpu(instance_id):
    """返回 EC2 实例的 7 天平均 CPU 利用率。"""
    response = cloudwatch.get_metric_statistics(
        Namespace='AWS/EC2',
        MetricName='CPUUtilization',
        Dimensions=[{'Name': 'InstanceId', 'Value': instance_id}],
        StartTime=datetime.now(timezone.utc) - timedelta(days=IDLE_DAYS),
        EndTime=datetime.now(timezone.utc),
        Period=604800,  # 一个 7 天周期
        Statistics=['Average']
    )
    datapoints = response.get('Datapoints', [])
    return datapoints[0]['Average'] if datapoints else 0.0

def lambda_handler(event, context):
    """停止空闲开发实例并通知其所有者。"""
    
    # 查找所有正在运行的开发实例
    response = ec2.describe_instances(
        Filters=[
            {'Name': 'instance-state-name', 'Values': ['running']},
            {'Name': 'tag:Environment', 'Values': ['dev', 'development']},
        ]
    )

    stopped = []
    skipped = []

    for reservation in response['Reservations']:
        for instance in reservation['Instances']:
            instance_id = instance['InstanceId']
            tags = {t['Key']: t['Value'] for t in instance.get('Tags', [])}

            # 跳过明确标记为保持运行的实例
            if tags.get('KeepAlive', '').lower() == 'true':
                skipped.append(instance_id)
                continue

            avg_cpu = get_average_cpu(instance_id)

            if avg_cpu < IDLE_CPU_THRESHOLD:
                # 停止前通知所有者
                owner = tags.get('Owner', 'unknown')
                sns.publish(
                    TopicArn=SNS_TOPIC_ARN,
                    Subject=f'开发环境已停止：{instance_id}',
                    Message=(
                        f'实例 {instance_id}（所有者：{owner}）在过去 {IDLE_DAYS} 天内的平均 CPU 利用率为 {avg_cpu:.1f}%，已被停止。\n\n'
                        f'为防止这种情况，请添加标签：KeepAlive=true\n'
                        f'要重新启动：aws ec2 start-instances --instance-ids {instance_id}'
                    )
                )
                ec2.stop_instances(InstanceIds=[instance_id])
                stopped.append({'id': instance_id, 'owner': owner, 'avg_cpu': avg_cpu})

    print(f"已停止 {len(stopped)} 个空闲实例。跳过了 {len(skipped)} 个 KeepAlive 实例。")
    return {'stopped': stopped, 'skipped': skipped}

staging-frontend:   $250/月   使用时间：周一至周五 09:00-18:00
staging-backend:    $250/月   使用时间：周一至周五 09:00-18:00
staging-ml:         $250/月   使用时间：周一至周五 10:00-17:00
staging-qa:         $250/月   使用时间：周一至周五 09:00-17:00
总计：           $1,000/月     运行时间：每天 24 小时，每周 7 天
实际使用率：       ~35%        你支付了 100%

# 查找标记为预发布的 EKS 节点组及其当前状态
aws eks list-nodegroups --cluster-name your-cluster-name --output table

# 检查标记为预发布的 EC2 实例及其启动时间
# 任何运行超过 30 天且没有周末停止计划的实例都是候选对象
aws ec2 describe-instances \
  --filters "Name=tag:Environment,Values=staging" "Name=instance-state-name,Values=running" \
  --query 'Reservations[*].Instances[*].{ID:InstanceId,Type:InstanceType,Launch:LaunchTime}' \
  --output table

# 方案 1：基于标签的调度（CloudFormation 解决方案）
# 在你的预发布 EC2 实例和 RDS 集群上添加以下标签：
# Schedule: office-hours
# 这将在工作日的 08:00 启动实例并在 20:00 停止它们
# 周末完全关闭

# 方案 2：快速 Lambda 解决方案 —— 工作日每天 20:00 停止所有预发布环境
aws events put-rule \
  --schedule-expression "cron(0 20 ? * MON-FRI *)" \
  --name stop-staging-environments \
  --state ENABLED

# 停止 Lambda —— 与模式 1 的模式相同，但针对预发布标签
# 在工作日的 07:30 添加相应的启动规则

# 一个带有命名空间隔离的共享预发布集群
# frontend-staging 和 backend-staging 共享节点，通过 Karpenter 实现
# 但通过命名空间级别的网络策略实现隔离
apiVersion: v1
kind: Namespace
metadata:
  name: staging-frontend
  labels:
    environment: staging
    team: frontend
---
apiVersion: v1
kind: Namespace
metadata:
  name: staging-backend
  labels:
    environment: staging
    team: backend

# 运行以下命令查看当前 NAT 网关的数据处理账单
aws ce get-cost-and-usage \
  --time-period Start=\((date -d 'last month' +%Y-%m-01),End=\)(date +%Y-%m-01) \
  --granularity MONTHLY \
  --filter '{
    "Dimensions": {
      "Key": "USAGE_TYPE",
      "Values": ["NatGateway-Bytes", "NatGateway-Hours"]
    }
  }' \
  --metrics UnblendedCost \
  --query 'ResultsByTime[0].Total.UnblendedCost.Amount' \
  --output text

# 首先获取你的 VPC ID 和路由表 ID
VPC_ID=$(aws ec2 describe-vpcs \
  --filters "Name=tag:Name,Values=your-vpc-name" \
  --query 'Vpcs[0].VpcId' --output text)

ROUTE_TABLE_ID=$(aws ec2 describe-route-tables \
  --filters "Name=vpc-id,Values=$VPC_ID" "Name=association.main,Values=true" \
  --query 'RouteTables[0].RouteTableId' --output text)

# S3 网关终端节点——免费创建，消除所有 S3 NAT 费用
aws ec2 create-vpc-endpoint \
  --vpc-id $VPC_ID \
  --service-name com.amazonaws.us-east-1.s3 \
  --route-table-ids $ROUTE_TABLE_ID

# DynamoDB 网关终端节点——同样免费
aws ec2 create-vpc-endpoint \
  --vpc-id $VPC_ID \
  --service-name com.amazonaws.us-east-1.dynamodb \
  --route-table-ids $ROUTE_TABLE_ID

# ECR API 终端节点——消除每次容器拉取的 NAT 费用
aws ec2 create-vpc-endpoint \
  --vpc-id $VPC_ID \
  --vpc-endpoint-type Interface \
  --service-name com.amazonaws.us-east-1.ecr.api \
  --subnet-ids $(aws ec2 describe-subnets \
    --filters "Name=vpc-id,Values=$VPC_ID" "Name=tag:Tier,Values=private" \
    --query 'Subnets[*].SubnetId' --output text)

# ECR Docker 终端节点——与 ECR API 一起用于镜像拉取
aws ec2 create-vpc-endpoint \
  --vpc-id $VPC_ID \
  --vpc-endpoint-type Interface \
  --service-name com.amazonaws.us-east-1.ecr.dkr \
  --subnet-ids $(aws ec2 describe-subnets \
    --filters "Name=vpc-id,Values=$VPC_ID" "Name=tag:Tier,Values=private" \
    --query 'Subnets[*].SubnetId' --output text)

Step 1: AWS 账单为每月 $100,000
Step 2: 购买每月 $70,000 的 Savings Plan 承诺
Step 3: 优化实例——实际支出降至 $60,000
Step 4: Savings Plan 覆盖 $70,000，但你只使用了 $60,000
Step 5: 你每月为未使用的计算资源支付 $28,000
         （Savings Plan 折扣应用于超额部分）
         
净结果：你在接下来的 12 个月内锁定了浪费

Step 1: 优化实例——支出从 $100,000 降至 $60,000
Step 2: 添加 Spot 实例用于测试环境——支出从 $60,000 降至 $45,000
Step 3: 将兼容的工作负载迁移到 Graviton 实例——支出降至 $36,000
Step 4: 现在购买覆盖每月 $25,000 的 Savings Plan（稳定状态的 70%）
Step 5: 有效月成本：$12,500（承诺部分）+ $11,000（按需部分）= $23,500

净结果：与原始账单相比，每月节省 $76,500

# 查看过去 30 天的 EC2 按需支出
# 这是你的优化基准——优化后的实际使用量
aws ce get-cost-and-usage \
  --time-period Start=\((date -d '30 days ago' +%Y-%m-%d),End=\)(date +%Y-%m-%d) \
  --granularity DAILY \
  --filter '{
    "And": [
      {"Dimensions": {"Key": "SERVICE", "Values": ["Amazon Elastic Compute Cloud - Compute"]}},
      {"Dimensions": {"Key": "PURCHASE_TYPE", "Values": ["On-Demand"]}}
    ]
  }' \
  --metrics UnblendedCost \
  --query 'ResultsByTime[*].{Date:TimePeriod.Start,Cost:Total.UnblendedCost.Amount}' \
  --output table

# 获取 AWS 自身基于你使用情况的 Savings Plan 建议
aws savingsplans get-savings-plans-purchase-recommendation \
  --savings-plans-type COMPUTE_SP \
  --term-in-years ONE_YEAR \
  --payment-option NO_UPFRONT \
  --lookback-period-in-days THIRTY_DAYS

# 检查当前跨可用区数据传输费用
aws ce get-cost-and-usage \
  --time-period Start=$(date -d 'last month' +%Y-%m-01),End=$(date +%Y-%m-01) \
  --granularity MONTHLY \
  --filter '{"Dimensions": {"Key": "USAGE_TYPE", "Values": ["DataTransfer-Regional-Bytes"]}}'  \
  --metrics UnblendedCost \
  --query 'ResultsByTime[0].Total.UnblendedCost.Amount' \
  --output text

# 检查你的数据库RDS实例所在的可用区
aws rds describe-db-instances \
  --query 'DBInstances[*].{ID:DBInstanceIdentifier,AZ:AvailabilityZone}' \
  --output table

# 检查你的应用程序Pod运行在哪些可用区
kubectl get pods -o wide -n production | awk '{print $7}' | sort | uniq -c

# topology-aware-routing.yaml
# 这告诉Kubernetes优先将Pod调度到与请求节点相同的可用区
# 以保持流量本地化

apiVersion: v1
kind: Service
metadata:
  name: payment-api
  namespace: production
  annotations:
    # 尽可能将流量路由到与调用者相同的可用区
    service.kubernetes.io/topology-mode: "Auto"
spec:
  selector:
    app: payment-api
  ports:
  - port: 8080
    targetPort: 8080

# 对于Pod本身——跨可用区分布但优先本地
# topologySpreadConstraints确保均匀分布
# 同时拓扑感知路由保持流量在可用区内

spec:
  topologySpreadConstraints:
  - maxSkew: 1
    topologyKey: topology.kubernetes.io/zone
    whenUnsatisfiable: DoNotSchedule
    labelSelector:
      matchLabels:
        app: payment-api

# 列出账户中所有的gp2卷及其大小和每月成本
aws ec2 describe-volumes \
  --filters Name=volume-type,Values=gp2 \
  --query 'Volumes[*].{
    ID:VolumeId,
    Size:Size,
    State:State,
    MonthlyCost_USD:Size
  }' \
  --output table

# 计算总数：卷的数量和总GB数
aws ec2 describe-volumes \
  --filters Name=volume-type,Values=gp2 \
  --query 'length(Volumes)' --output text

aws ec2 describe-volumes \
  --filters Name=volume-type,Values=gp2 \
  --query 'sum(Volumes[*].Size)' --output text

#!/bin/bash
# migrate_gp2_to_gp3.sh
# 将所有gp2卷迁移到gp3。在线操作——无需停机。
# 每次修改异步运行；卷在整个过程中保持可用。

echo "开始gp2到gp3的迁移..."

# 获取所有gp2卷的ID
VOLUMES=$(aws ec2 describe-volumes \
  --filters Name=volume-type,Values=gp2 \
  --query 'Volumes[*].VolumeId' \
  --output text)

COUNT=0
for VOL_ID in $VOLUMES; do
  echo "正在将$VOL_ID迁移到gp3..."
  aws ec2 modify-volume \
    --volume-id $VOL_ID \
    --volume-type gp3 \
    --no-cli-pager
  COUNT=$((COUNT + 1))
done

echo "已启动$COUNT个卷的迁移。"
echo "修改过程在线运行——无需停机。监控进度："
echo "aws ec2 describe-volumes-modifications --query 'VolumesModifications[*].{ID:VolumeId,State:ModificationState}'"

# 检查是否还有gp2卷
aws ec2 describe-volumes \
  --filters Name=volume-type,Values=gp2 \
  --query 'length(Volumes)' \
  --output text
# 预期结果：0

# 列出所有日志组及其保留设置
# 任何显示“retentionInDays: null”的组都是无限的——永远不会过期
aws logs describe-log-groups \
  --query 'logGroups[*].{Name:logGroupName,RetentionDays:retentionInDays,StoredBytes:storedBytes}' \
  --output table | grep -E "(None|null)"

# 统计有多少日志组未设置保留策略
aws logs describe-log-groups \
  --query 'length(logGroups[?retentionInDays==`null`])' \
  --output text

#!/bin/bash
# set_log_retention.sh
# 设置所有未设置策略的日志组的30天保留期
# 根据日志组类型调整保留期

echo "正在为没有过期策略的日志组设置保留策略..."

# 获取所有没有保留期的日志组
aws logs describe-log-groups \
  --query 'logGroups[?retentionInDays==`null`].logGroupName' \
  --output text | tr '\t' '\n' | while read LOG_GROUP; do

  # 跳过CloudTrail日志 — 这些日志需要更长的保留期以满足SOC2要求
  if echo "$LOG_GROUP" | grep -qi "cloudtrail"; then
    echo "跳过CloudTrail日志组: $LOG_GROUP"
    aws logs put-retention-policy \
      --log-group-name "$LOG_GROUP" \
      --retention-in-days 365
    continue
  fi

  # 对所有其他日志组设置30天保留期
  echo "正在为 $LOG_GROUP 设置30天保留期"
  aws logs put-retention-policy \
    --log-group-name "$LOG_GROUP" \
    --retention-in-days 30
done

echo "完成。超过保留期的日志将由CloudWatch自动删除。"

# 未附加的EBS卷 — 你正在为无内容的存储付费
aws ec2 describe-volumes \
  --filters Name=status,Values=available \
  --query 'Volumes[*].{
    ID:VolumeId,
    Size:Size,
    Created:CreateTime,
    MonthlyCost:Size
  }' \
  --output table

# 未关联的弹性IP — 当未附加到运行实例时，每个弹性IP每月收费3.60美元
aws ec2 describe-addresses \
  --query 'Addresses[?AssociationId==`null`].[PublicIp,AllocationId]' \
  --output table

# 旧快照 — 创建时间超过90天且不再需要
aws ec2 describe-snapshots \
  --owner-ids self \
  --query "Snapshots[?StartTime<='$(date -d '90 days ago' --iso-8601=seconds)'].[SnapshotId,StartTime,VolumeSize]" \
  --output table

# 空闲负载均衡器 — 活跃但路由零流量
aws elbv2 describe-load-balancers \
  --query 'LoadBalancers[*].{ARN:LoadBalancerArn,DNS:DNSName,State:State.Code}' \
  --output table

# orphan_resource_reporter.py
# 每周日20:00通过EventBridge运行
# 向Slack报告孤儿资源 — 不自动删除
# 删除需要人工决策。Lambda显示待删除资源候选。

import boto3
import json
import urllib.request
from datetime import datetime, timedelta, timezone

SLACK_WEBHOOK_URL = 'https://hooks.slack.com/services/YOUR/WEBHOOK/URL'

def get_orphaned_resources():
    """收集所有孤儿AWS资源及其估算的月度成本。"""
    ec2 = boto3.client('ec2')
    elbv2 = boto3.client('elbv2')
    report = {'total_monthly_waste': 0, 'resources': []}

    # 未附加的EBS卷 ($0.08/GB/月，gp3)
    volumes = ec2.describe_volumes(
        Filters=[{'Name': 'status', 'Values': ['available']}]
    )['Volumes']
    for vol in volumes:
        monthly_cost = round(vol['Size'] * 0.08, 2)
        report['resources'].append({
            'type': '未附加的EBS卷',
            'id': vol['VolumeId'],
            'detail': f"{vol['Size']}GB {vol['VolumeType']}",
            'monthly_cost': monthly_cost
        })
        report['total_monthly_waste'] += monthly_cost

    # 未关联的弹性IP ($3.60/月)
    addresses = ec2.describe_addresses()['Addresses']
    for addr in addresses:
        if 'AssociationId' not in addr:
            report['resources'].append({
                'type': '未关联的弹性IP',
                'id': addr['AllocationId'],
                'detail': addr['PublicIp'],
                'monthly_cost': 3.60
            })
            report['total_monthly_waste'] += 3.60

    # 创建时间超过90天的快照
    cutoff = (datetime.now(timezone.utc) - timedelta(days=90)).isoformat()
    snapshots = ec2.describe_snapshots(OwnerIds=['self'])['Snapshots']
    old_snapshots = [s for s in snapshots if s['StartTime'].isoformat() < cutoff]
    for snap in old_snapshots:
        monthly_cost = round(snap.get('VolumeSize', 0) * 0.05, 2)
        report['resources'].append({
            'type': '旧快照 (90+天)',
            'id': snap['SnapshotId'],
            'detail': f"创建于 {snap['StartTime'].strftime('%Y-%m-%d')}",
            'monthly_cost': monthly_cost
        })
        report['total_monthly_waste'] += monthly_cost

    return report

def post_to_slack(report):
    """将孤儿资源报告发送到Slack。"""
    resource_lines = '\n'.join([
        f"• {r['type']} `{r['id']}` — {r['detail']} — *${r['monthly_cost']}/月*"
        for r in report['resources']
    ])

message = {
    'text': (
        f":money_with_wings: *Weekly Orphaned Resource Report*\n\n"
        f"Found *{len(report['resources'])} orphaned resources* "
        f"costing *${report['total_monthly_waste']:.2f}/month*\n\n"
        f"{resource_lines}\n\n"
        f"Review and delete resources that are no longer needed."
    )
}

req = urllib.request.Request(
    SLACK_WEBHOOK_URL,
    data=json.dumps(message).encode(),
    headers={'Content-Type': 'application/json'}
)
urllib.request.urlopen(req)

def lambda_handler(event, context):
    report = get_orphaned_resources()
    post_to_slack(report)
    return {
        'resources_found': len(report['resources']),
        'monthly_waste': report['total_monthly_waste']
    }